Slack_タスク管理bot(Python)

ゆーしゃん
2022.05.07
456 views

動作概要

１、特定のリアクション(絵文字)をした投稿先リンクをbotにて通知させる

２、リアクション(絵文字)の設定、解除が反映可能
※：「ラーメン食べたい」投稿にてリアクション(絵文字)解除
※：「Pythonを好これ」投稿にてリアクション(絵文字)追加

３、URLをリンク設定する事で、「どんな投稿先か？」が分かるように設定可能

４、上記をDB登録済みユーザーが実行可能
※：別アカウント(ゆーしゃんbot)にて「ラーメン食べたい」投稿にてリアクション(絵文字)追加

プログラムを作成した理由

下記ツイートの通り、
Slackチャンネル内にて沢山の業務効率化の依頼を頂く為、
簡単に「現在進行中タスクの依頼文」を確認出来るようにしたかった為。

1つのチャンネル内に同時進行中のタスクがあると、依頼分まで遡るのに時間かかるんですよね💦
(毎回リンク先メモするのも面倒です)

なので、特定スタンプを押しておけば、
いつでも「貴方のタスクはこれです！」と辿れるようにしたい
— ゆーしゃん.py (@Mr_mura_ura) April 25, 2022

Slackbot用アプリ_導入手順

１、下記ツイートの「Slack ソケットモードの最も簡単な始め方」記事の通りにSlackアプリを設定する

あと、Event Subscriptions のデフォルトの設定は公開された URL が必要になるのですが、ソケットモードというものを有効にすると RTM と同様に接続できます。こちらの記事もぜひ見てみてください。 https://t.co/J4UXLAVbjJ
— Kazuhiro Sera (瀬良) (@seratch_ja) April 29, 2022

※助言頂き、本当にありがとうございました！！！

２、記事の通り「app.py」を作成し、
「こんにちは」とSlackに投稿したら「こんにちは〇〇さん！」とbotから返信が来る事を確認

「Slack_bold」のイベントとは？

「Slack ソケットモードの最も簡単な始め方」記事のサンプルコードとして、
「@app.message(“こんにちは”)」とありましたが、

こちらはExcelVBAで例えると「Workbook_Open」のように
「〇〇の状況になったら発動するトリガー」のようなものです。

Slackのイベント詳細は Slack_Bolt_for_Python 記事を参考にして頂ければと思いますが、
「いきなり公式記事見てもよく分からない…」という方は下記も併せてご参考下さい。

・「@app.message」…特定のメッセージが投稿された場合のイベント

@app.message("こんにちは")
def handle_messge_evnts(message, say):
    say(f"こんにちは <@{message['user']}> さん！")

@app.message("こんにちは")

def handle_messge_evnts(message, say):

say(f"こんにちは <@{message['user']}> さん！")

↑画像のように、
【「こんにちは」を含む投稿がされた】事をトリガーに、
「こんにちは〇〇さん！」と新規投稿されております。

このイベントでは、
別の人に「Aさんこんにちは！」と投稿したつもりでも、
botから常に「こんにちは〇〇さん！」と新規投稿がされてしまう為、
bot作成時は下記イベントを使う事になると思います。

・「@app.message」…メンション付き投稿された場合のイベント

@app.event("app_mention")
def handle_mention(context,body,say,logger):
    
    channel_id = context['channel_id']  #チャンネルID
    thread_ts = body["event"]['ts']     #投稿タイムスタンプ
    contents = body["event"]['text']    #投稿文章
    user_id = context['user_id']        #投稿ユーザー
    
    # botが行動出来るようにする
    client = context["client"]
    
    if "こんにちは" in contents:
                    
        # 返信
        client.chat_postMessage(
            channel = channel_id,
            text = "こんにちは <@" + user_id + "> さん！",
            thread_ts = thread_ts
        )

@app.event("app_mention")

def handle_mention(context,body,say,logger):

channel_id = context['channel_id'] #チャンネルID

thread_ts = body["event"]['ts'] #投稿タイムスタンプ

contents = body["event"]['text'] #投稿文章

user_id = context['user_id'] #投稿ユーザー

# botが行動出来るようにする

client = context["client"]

if "こんにちは" in contents:

# 返信

client.chat_postMessage(

channel = channel_id,

text = "こんにちは <@" + user_id + "> さん！",

thread_ts = thread_ts

)

↑画像のように、
【「タスク管理bot」がメンションされた】事をトリガーに、
「chat_postMessage」というslack_apiを用いて投稿に返信しております。

「slack_api」とは？

先程紹介した、「chat_postMessage」(返信)のように、
Slackから情報取得、情報送信するには必ずapiを使う事となります。

つまり、「slack_apiで何が出来るのか？」を知らないと
望み通りの処理は組めないという事です。

api詳細は slack_api_一覧 を参考にして頂ければと思いますが、
「いきなり公式記事見てもよく分からない…」という方は下記も併せてご参考下さい。

・「chat_postMessage」…チャンネルにメッセージ送信

# botが行動出来るようにする
client = context[“client”]
client.chat_postMessage(
)

必須：「channel」…下記コードの「channel_id」部分。
※黄色の「channel」は完全一致で記載する事。

channel_id = context[‘channel_id’] #チャンネルID
client.chat_postMessage(
channel = channel_id,
)

user_id = context[‘user_id’] #投稿ユーザー
client.chat_postMessage(
text = “こんにちは <@” + user_id + “> さん！”
)

任意：返信したい場合は、「ts」(下記赤文字部分)を指定
※黄色の「thread_ts」は完全一致で記載する事。

thread_ts = body[“event”][‘ts’] #投稿タイムスタンプ
client.chat_postMessage(
thread_ts = thread_ts
)

上記「chat_postMessage」で使用した下記項目が理解出来ると、
slack_apiの使い方が何となく見えてくると思います。

token (client = context[“client”]) #botが行動出来るようにする
channel (channel_id = context[‘channel_id’]) #チャンネルID
user_id (user_id = context[‘user_id’]) #投稿ユーザー
thread_ts (thread_ts = body[“event”][‘ts’]) #投稿タイムスタンプ

次に、本記事の「Slack_タスク管理_bot」で使用しているslack_apiを一つ紹介致します。

・「reactions.list」…ユーザーによる反応一覧を取得
※どのユーザーがどのリアクション(絵文字)をしたかを取得

「token」は変数「client」に格納しているものを使用
「オプションの引数」の「user」を見ると「このユーザーの反応を表示します」とあるので、ユーザーIDを指定

と下記コードを書くだけで「指定ユーザーのリアクション(絵文字)一覧」を取得する事が出来るようになります。

    user_id = context['user_id'] #投稿ユーザー 
    
    # リアクションリスト取得 
    reactions_list = client.reactions_list(
        user = user_id 
        )

user_id = context['user_id'] #投稿ユーザー

# リアクションリスト取得

reactions_list = client.reactions_list(

user = user_id

)

Slackタスク管理bot_ソースコード

「slack_bold」のイベント
「slack_api」の使い方

上記2点を解説したところで、
「Slackタスク管理bot」のソースコードを紹介致します。

from slack_sdk import WebClient
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

import logging
import os
import re

# 自作プロシージャ
import slack_task_bot_db
import get_sql_escape_contents

# 定数
# DBの列位置
DB_TASK_NAME_COLUMN_INDEX = 3
DB_EMOJI_NAME_COLUMN_INDEX = 3
DB_USER_ID_COLUMN_INDEX = 1

# トークン
SLACK_BOT_TOKEN = "xoxb-…"
SLACK_APP_TOKEN = "xapp-…"

# その他
SEARCH_LIMIT = 1000                  #絵文字リアクション取得上限数
DEFAULT_TASK_NAME_LEN_LIMIT = 100    #〇文字まで文字切取り

logging.basicConfig(level=logging.ERROR)

app = App(token=SLACK_BOT_TOKEN)

# 絵文字dbをbot起動時に初期化(手動でcsv変更したものを反映する為)
all_emoji_list = slack_task_bot_db.get_emoji_list("",True)

# 集合を用いて重複無しのユーザーリスト作成
all_user_list = set()
for user_list in all_emoji_list:
    all_user_list.add(user_list[DB_USER_ID_COLUMN_INDEX])


@app.event("app_mention")
def handle_mention(context,body,say,logger):
    
    channel_id = context['channel_id']
    thread_ts = body["event"]['ts']
    contents = body["event"]['text']
    user_id = context['user_id'] 
    
    # 「<@…\n」の全メンション除外
    contents = re.sub(r"<@[^\s]*\s", "" ,contents)
    
    # botが行動出来るようにする
    client = context["client"]
    
    # 返事先メンション
    MENTION = "<@" + user_id + ">\n"
    
    # 返信内容初期化
    rep_text = ""
    
    if "全員のタスク教えて" in contents:   
        
        # 全ユーザーループ
        for user_list_id in all_user_list:
            
            # リアクション(絵文字)が付いている一覧を取得
            rep_text = rep_text + get_task_contents(client, user_list_id)                                 
        
    # 指定文字が含まれる場合
    elif "タスク教えて" in contents:   
        
        # リアクション(絵文字)が付いている一覧を取得
        rep_text = get_task_contents(client, user_id)                                 
    
    # 指定文字が含まれる場合
    elif "タスク登録" in contents:
        
        # 文言内のURL一覧取得
        #例) ['<URL>', '<URL|タスク名>']
        new_task_lists = re.findall(r"<http[^>]*>", contents)
        
        # 「|」が含まれていないURLだけのものは除外
        #例) ['<URL|タスク名>']
        new_task_lists = [l for l in new_task_lists if "|" in l]
        
        for new_task_list in new_task_lists:
            
            # URLとリンク名で分割
            task_url, task_name = new_task_list.split("|")

            task_url = task_url.lstrip("<")     # 先頭1文字の「<」除外
            task_name = task_name.rstrip(">")   # 末尾1文字の「>」除外
            
            # DBにタスク登録　※返却値のタスクリストは受け取らない
            slack_task_bot_db.get_task_list(user_id, task_url, task_name)
            
            # デバック用
            # task_list = slack_task_bot_db.get_task_list(user_id, task_url, task_name)            
            # print(task_list)
            
            # タスク登録後の、リアクション(絵文字)が付いている一覧を取得
            rep_text = get_task_contents(client, user_id)
        
        # リンクが指定されている場合
        if len(new_task_lists) > 0:
            
            rep_text = "承知致しました！\n" + \
                        "登録後のタスクは下記となります！\n" + \
                        "==============================\n" + \
                        rep_text
                        
        # リンクが指定されなかった場合
        else:
            rep_text = "リンクを指定してから再度お呼び下さい！"
    
    else:
        # 想定外ワードの場合、オウム返ししておく
        rep_text = MENTION + contents
            
    # 返信用文言が設定されている場合のみ、返信
    if rep_text != "":
        
        # 返信
        client.chat_postMessage(
            channel = channel_id,
            text = rep_text,
            thread_ts = thread_ts
        )
    
    # 返信後のタスクdb表示
    print("タスク管理bot返信後のタスクdb\n")
    slack_task_bot_db.get_task_list("","","", "",True)

# 概要：リアクション(絵文字)が付いている一覧を取得
def get_task_contents(client, user_id):
    
    # 返事先メンション
    MENTION = "<@" + user_id + ">"
    
    # リアクションリスト取得
    reactions_list = client.reactions_list(
        count = SEARCH_LIMIT,
        limit = SEARCH_LIMIT,
        user = user_id
    )
    
    # 初期化
    # permalink = ""
    rep_complete_text = ""
    
    # ユーザー毎の絵文字リスト取得
    emoji_list = slack_task_bot_db.get_emoji_list(user_id)
    
    # 絵文字数ループ
    for emoji in emoji_list:
        
        # 初期化
        rep_text = ""
        
        # 絵文字取得
        search_emoji = emoji[DB_EMOJI_NAME_COLUMN_INDEX]
        
        # リアクション(絵文字)した投稿全てのループ
        for contents in reactions_list["items"]:
            
            # メッセージ取得
            contents = contents["message"]

            # 付いている絵文字の数だけループ
            for reactions in contents["reactions"]:
                
                # 指定絵文字の場合
                if reactions["name"] == search_emoji:
                    
                    # 前回と同じURLではない場合
                    # if permalink != contents["permalink"]:
                    
                    # リンク取得
                    permalink = contents["permalink"]
                    
                    # エスケープ文字調整
                    permalink = get_sql_escape_contents.get_sql_escape_contents(permalink)
                    
                    # タスク本文に記載されていない新しいURLの場合
                    if not permalink in rep_text:
                                                                    
                        # DB登録済みのタスクリスト取得
                        task_list = slack_task_bot_db.get_task_list(user_id, permalink)
                        
                        # DB登録済みのタスク名取得
                        task_name = ""
                        for task in task_list:
                            task_name = task[DB_TASK_NAME_COLUMN_INDEX]
                            break
                        
                        # bot返信用文章作成                        
                        if task_name == "":
                            
                            # 左から上限文字数まで抽出(上限より文字数が少ない場合は何も変わらない)
                            after_contents = contents["text"][:DEFAULT_TASK_NAME_LEN_LIMIT]
                            
                            # 文字数調整された場合は、末尾に「…」結合
                            if contents["text"] != after_contents:
                                after_contents = after_contents + "…"
                                
                            # DB未登録のタスクの場合は、そのままURLを載せる
                            # rep_text = rep_text + "・" + contents["text"] + "\n" + contents["permalink"] + "\n\n"
                            rep_text = rep_text + "・" + after_contents + "\n" + contents["permalink"] + "\n\n"
                            
                        else:
                            
                            # DB登録済みのタスクの場合は、リンクを生成する                            
                            rep_text = rep_text + "・<" + contents["permalink"] + "|" + task_name+ ">\n\n"

        # 「<@…\n」の全メンション除外
        rep_text = re.sub(r"<@[^\s]*\s", "" ,rep_text)
    
        # 指定絵文字でリアクションが無かった場合
        if rep_text == "":
            rep_complete_text = rep_complete_text + MENTION + "が:" + search_emoji + ":で反応したものはありませんでした。\n"
        
        # 指定絵文字でリアクションがあった場合
        else:
            rep_complete_text = rep_complete_text + MENTION + "が:" + search_emoji + ":で反応したものです\n\n" + rep_text + "\n"
    
    # タスク本文を返却
    return rep_complete_text

if __name__ == "__main__":
    handler = SocketModeHandler(app, SLACK_APP_TOKEN)
    handler.start()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

from slack_sdk import WebClient

from slack_bolt import App

from slack_bolt.adapter.socket_mode import SocketModeHandler

import logging

import os

import re

# 自作プロシージャ

import slack_task_bot_db

import get_sql_escape_contents

# 定数

# DBの列位置

DB_TASK_NAME_COLUMN_INDEX = 3

DB_EMOJI_NAME_COLUMN_INDEX = 3

DB_USER_ID_COLUMN_INDEX = 1

# トークン

SLACK_BOT_TOKEN = "xoxb-…"

SLACK_APP_TOKEN = "xapp-…"

# その他

SEARCH_LIMIT = 1000 #絵文字リアクション取得上限数

DEFAULT_TASK_NAME_LEN_LIMIT = 100 #〇文字まで文字切取り

logging.basicConfig(level=logging.ERROR)

app = App(token=SLACK_BOT_TOKEN)

# 絵文字dbをbot起動時に初期化(手動でcsv変更したものを反映する為)

all_emoji_list = slack_task_bot_db.get_emoji_list("",True)

# 集合を用いて重複無しのユーザーリスト作成

all_user_list = set()

for user_list in all_emoji_list:

all_user_list.add(user_list[DB_USER_ID_COLUMN_INDEX])

@app.event("app_mention")

def handle_mention(context,body,say,logger):

channel_id = context['channel_id']

thread_ts = body["event"]['ts']

contents = body["event"]['text']

user_id = context['user_id']

# 「<@…\n」の全メンション除外

contents = re.sub(r"<@[^\s]*\s", "" ,contents)

# botが行動出来るようにする

client = context["client"]

# 返事先メンション

MENTION = "<@" + user_id + ">\n"

# 返信内容初期化

rep_text = ""

if "全員のタスク教えて" in contents:

# 全ユーザーループ

for user_list_id in all_user_list:

# リアクション(絵文字)が付いている一覧を取得

rep_text = rep_text + get_task_contents(client, user_list_id)

# 指定文字が含まれる場合

elif "タスク教えて" in contents:

# リアクション(絵文字)が付いている一覧を取得

rep_text = get_task_contents(client, user_id)

# 指定文字が含まれる場合

elif "タスク登録" in contents:

# 文言内のURL一覧取得

#例) ['<URL>', '<URL|タスク名>']

new_task_lists = re.findall(r"<http[^>]*>", contents)

# 「|」が含まれていないURLだけのものは除外

#例) ['<URL|タスク名>']

new_task_lists = [l for l in new_task_lists if "|" in l]

for new_task_list in new_task_lists:

# URLとリンク名で分割

task_url, task_name = new_task_list.split("|")

task_url = task_url.lstrip("<") # 先頭1文字の「<」除外

task_name = task_name.rstrip(">") # 末尾1文字の「>」除外

# DBにタスク登録　※返却値のタスクリストは受け取らない

slack_task_bot_db.get_task_list(user_id, task_url, task_name)

# デバック用

# task_list = slack_task_bot_db.get_task_list(user_id, task_url, task_name)

# print(task_list)

# タスク登録後の、リアクション(絵文字)が付いている一覧を取得

rep_text = get_task_contents(client, user_id)

# リンクが指定されている場合

if len(new_task_lists) > 0:

rep_text = "承知致しました！\n" + \

"登録後のタスクは下記となります！\n" + \

"==============================\n" + \

rep_text

# リンクが指定されなかった場合

else:

rep_text = "リンクを指定してから再度お呼び下さい！"

else:

# 想定外ワードの場合、オウム返ししておく

rep_text = MENTION + contents

# 返信用文言が設定されている場合のみ、返信

if rep_text != "":

# 返信

client.chat_postMessage(

channel = channel_id,

text = rep_text,

thread_ts = thread_ts

)

# 返信後のタスクdb表示

print("タスク管理bot返信後のタスクdb\n")

slack_task_bot_db.get_task_list("","","", "",True)

# 概要：リアクション(絵文字)が付いている一覧を取得

def get_task_contents(client, user_id):

# 返事先メンション

MENTION = "<@" + user_id + ">"

# リアクションリスト取得

reactions_list = client.reactions_list(

count = SEARCH_LIMIT,

limit = SEARCH_LIMIT,

user = user_id

)

# 初期化

# permalink = ""

rep_complete_text = ""

# ユーザー毎の絵文字リスト取得

emoji_list = slack_task_bot_db.get_emoji_list(user_id)

# 絵文字数ループ

for emoji in emoji_list:

# 初期化

rep_text = ""

# 絵文字取得

search_emoji = emoji[DB_EMOJI_NAME_COLUMN_INDEX]

# リアクション(絵文字)した投稿全てのループ

for contents in reactions_list["items"]:

# メッセージ取得

contents = contents["message"]

# 付いている絵文字の数だけループ

for reactions in contents["reactions"]:

# 指定絵文字の場合

if reactions["name"] == search_emoji:

# 前回と同じURLではない場合

# if permalink != contents["permalink"]:

# リンク取得

permalink = contents["permalink"]

# エスケープ文字調整

permalink = get_sql_escape_contents.get_sql_escape_contents(permalink)

# タスク本文に記載されていない新しいURLの場合

if not permalink in rep_text:

# DB登録済みのタスクリスト取得

task_list = slack_task_bot_db.get_task_list(user_id, permalink)

# DB登録済みのタスク名取得

task_name = ""

for task in task_list:

task_name = task[DB_TASK_NAME_COLUMN_INDEX]

break

# bot返信用文章作成

if task_name == "":

# 左から上限文字数まで抽出(上限より文字数が少ない場合は何も変わらない)

after_contents = contents["text"][:DEFAULT_TASK_NAME_LEN_LIMIT]

# 文字数調整された場合は、末尾に「…」結合

if contents["text"] != after_contents:

after_contents = after_contents + "…"

# DB未登録のタスクの場合は、そのままURLを載せる

# rep_text = rep_text + "・" + contents["text"] + "\n" + contents["permalink"] + "\n\n"

rep_text = rep_text + "・" + after_contents + "\n" + contents["permalink"] + "\n\n"

else:

# DB登録済みのタスクの場合は、リンクを生成する

rep_text = rep_text + "・<" + contents["permalink"] + "|" + task_name+ ">\n\n"

# 「<@…\n」の全メンション除外

rep_text = re.sub(r"<@[^\s]*\s", "" ,rep_text)

# 指定絵文字でリアクションが無かった場合

if rep_text == "":

rep_complete_text = rep_complete_text + MENTION + "が:" + search_emoji + ":で反応したものはありませんでした。\n"

# 指定絵文字でリアクションがあった場合

else:

rep_complete_text = rep_complete_text + MENTION + "が:" + search_emoji + ":で反応したものです\n\n" + rep_text + "\n"

# タスク本文を返却

return rep_complete_text

if __name__ == "__main__":

handler = SocketModeHandler(app, SLACK_APP_TOKEN)

handler.start()

import sqlite3
import pandas as pd
import sys

# 自作クラス
import get_sql_escape_contents

# 共通定数
DB_FOLD_NAME = "07_db"
USER_ID_COLUMN_NAME = "user_id"
USER_NAME_COLUMN_NAME = "user_name"

# 指定のDBをCSV情報に初期化する
def db_reset(db, db_table_name, query, csv_file_name):
    
    # クエリー実行用
    db_query = db.cursor()
    
    # 初期化用のCSVをDBに反映するSQL作成    
    df = pd.read_csv(csv_file_name, encoding = "shift-jis", dtype=str)
    df.to_sql(db_table_name, db, if_exists = "replace")
    
    # db操作を確定させる
    db.commit()
    
    # クエリー実行
    db_query.execute(query)
    
    # 1行ずつ表示
    for row in db_query.fetchall():
        print(row)
    
    # print("処理を中止しました。")
    
    # # 処理強制終了
    # sys.exit()        

# 概要：ユーザー毎の絵文字リストを取得
def get_emoji_list(user_id = "", DB_RESET_FLG = False):    
    
    # 定数
    # EMOJI_NAME_COLUMN_NAME = "emoji_name"
    EMOJI_FILE_NAME = "slack_emoji_list"
    DB_EMOJI_TABLE_NAME = "T_emoji_list"
    QUERY_ALL_SELECT = "select * from " + DB_EMOJI_TABLE_NAME   # 全レコード表示するクエリー
    
    # フルパス
    db_name = DB_FOLD_NAME + "//" + EMOJI_FILE_NAME + ".db"
    csv_emoji_name = DB_FOLD_NAME + "//" + EMOJI_FILE_NAME + ".csv"
    
    # DB接続(存在しない場合は新たに作成)
    db = sqlite3.connect(db_name)

    # クエリー実行用
    db_query = db.cursor()
    
    # DBを初期化csv情報に初期化
    if DB_RESET_FLG:
        
        print("絵文字db_csvインポート開始")
        db_reset(db, DB_EMOJI_TABLE_NAME, QUERY_ALL_SELECT, csv_emoji_name)
        print("絵文字db_csvインポート終了")
        
    # ユーザーIDが指定されている場合、ユーザー毎のリストを返却
    if user_id != "":
        
        # SQLで指定し易いようにシングルクォーテーションで囲む
        user_id = "'" + user_id + "'"

        # ユーザー毎の絵文字リスト取得
        QUERY_USER_LIST = QUERY_ALL_SELECT + \
                " where " + USER_ID_COLUMN_NAME + " = " + user_id

        # クエリー実行
        db_query.execute(QUERY_USER_LIST)

    # ユーザーIDが指定されていない場合、全員のリストを返却
    else:
        
        # クエリー実行
        db_query.execute(QUERY_ALL_SELECT)        
    
    # クエリー実行後結果取得
    emoji_lists = []
    for row in db_query.fetchall():        
        emoji_lists.append(row)

    # dbを閉じる
    db.close
    
    # リスト返却
    return emoji_lists
        
# 概要：ユーザー毎にタスク登録&取得
def get_task_list(user_id, task_url, task_name = "", DB_RESET_FLG = False, ALL_SHOW_FLG = False):
    
    # 定数
    TASK_LIST_FILE_NAME = "slack_task_list"
    DB_TASK_TABLE_NAME = "T_task_list"
    TASK_URL_COLUMN_NAME = "task_url"
    TASK_NAME_COLUMN_NAME = "task_name"
    QUERY_ALL_SELECT = "select * from " + DB_TASK_TABLE_NAME    # 全レコード表示するクエリー
    
    # フルパス
    db_name = DB_FOLD_NAME + "//" + TASK_LIST_FILE_NAME + ".db"
    csv_task_name = DB_FOLD_NAME + "//" + TASK_LIST_FILE_NAME + ".csv"
    
    # DB接続(存在しない場合は新たに作成)
    db = sqlite3.connect(db_name)

    # クエリー実行用
    db_query = db.cursor()
    
    # テスト用：DBを初期化csv情報に初期化し、処理強制終了
    if DB_RESET_FLG:
        
        print("タスクdb_csvインポート開始")
        db_reset(db, DB_TASK_TABLE_NAME, QUERY_ALL_SELECT, csv_task_name)
        print("タスクdb_csvインポート終了")
    
    # テスト用：全レコード表示
    if ALL_SHOW_FLG:
        
        # クエリー実行して全件表示
        print("タスクdb_全件表示_開始")
        
        db_query.execute(QUERY_ALL_SELECT)
        for row in db_query.fetchall():
            print(row)        
            
        print("タスクdb_全件表示_終了")
        
        return

    # SQLで指定し易いようにシングルクォーテーションで囲む
    user_id = "'" + user_id + "'"
    task_url = "'" + task_url + "'"
    
    # エスケープ文字調整
    task_url = get_sql_escape_contents.get_sql_escape_contents(task_url)

    # タスク名が指定された場合はDBに登録する
    # ※指定されていない場合は、リスト取得のみ
    if task_name != "":
            
        # SQLで指定し易いようにシングルクォーテーションで囲む        
        task_name = "'" + task_name + "'"
        
        ### 新規タスク追加 ############################################################
        
        # URLのみ登録済判定(ユーザー毎)
        QUERY_ALREADY_SET = QUERY_ALL_SELECT + \
                " where " + USER_ID_COLUMN_NAME + " = " + user_id + \
                " and " + TASK_URL_COLUMN_NAME + " = " + task_url 
                
        # クエリー実行
        db_query.execute(QUERY_ALREADY_SET)

        # DB追加フラグ初期化(True：新規追加する False：タスク名を更新する)
        INSERT_FLG = True

        # DB編集結果を表示
        for row in db_query.fetchall():
            INSERT_FLG = False
            break

        # URLが登録されていない場合、新規追加
        if INSERT_FLG:
            
            # タスク追加
            QUERY_CREATE = "insert into " + DB_TASK_TABLE_NAME + \
                        " ( " + USER_ID_COLUMN_NAME +  ", " + TASK_URL_COLUMN_NAME + ", " + TASK_NAME_COLUMN_NAME + ")" + \
                        "values (" + user_id + ", " + task_url + ", " + task_name + ")"
                        
            db_query.execute(QUERY_CREATE)
            
            # db操作を確定させる
            db.commit()
        
        ### タスク名更新 ############################################################
        # URL、タスク名共に完全一致判定(ユーザー毎)
        QUERY_ALREADY_SET = QUERY_ALL_SELECT + \
                " where " + USER_ID_COLUMN_NAME + " = " + user_id + \
                " and " + TASK_URL_COLUMN_NAME + " = " + task_url + \
                " and " + TASK_NAME_COLUMN_NAME + " = " + task_name
                
        # クエリー実行
        db_query.execute(QUERY_ALREADY_SET)

        # DB追加フラグ初期化(True：新規追加する False：タスク名を更新する)
        UPDATE_FLG = True

        # DB編集結果を表示
        for row in db_query.fetchall():
            UPDATE_FLG = False
            break
        
        # URLが登録されていない場合、新規追加
        if UPDATE_FLG:
            
            # タスク名更新
            QUERY_UPDATE = "update " + DB_TASK_TABLE_NAME + \
                        " set " + TASK_NAME_COLUMN_NAME + " = " + task_name + \
                        " where " + USER_ID_COLUMN_NAME + " = " + user_id + \
                        " and " + TASK_URL_COLUMN_NAME + " = " + task_url
            
            db_query.execute(QUERY_UPDATE)
            
            # db操作を確定させる
            db.commit()
            
    # ユーザー毎のタスク取得
    QUERY_USER_LIST = QUERY_ALL_SELECT + \
            " where " + USER_ID_COLUMN_NAME + " = " + user_id + \
            " and " + TASK_URL_COLUMN_NAME + " = " + task_url 

    # クエリー実行
    db_query.execute(QUERY_USER_LIST)

    # ユーザー×一致URLのリストを返却(1件しか存在しない想定だが)
    task_lists = []
    for row in db_query.fetchall():        
        task_lists.append(row)
        
    # dbを閉じる
    db.close
    
    # リスト返却
    return task_lists
    
# タスクdb初期化用
# print(get_task_list("","","", "",True))

# 絵文字db初期化用
# print(get_emoji_list("",True))

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

import sqlite3

import pandas as pd

import sys

# 自作クラス

import get_sql_escape_contents

# 共通定数

DB_FOLD_NAME = "07_db"

USER_ID_COLUMN_NAME = "user_id"

USER_NAME_COLUMN_NAME = "user_name"

# 指定のDBをCSV情報に初期化する

def db_reset(db, db_table_name, query, csv_file_name):

# クエリー実行用

db_query = db.cursor()

# 初期化用のCSVをDBに反映するSQL作成

df = pd.read_csv(csv_file_name, encoding = "shift-jis", dtype=str)

df.to_sql(db_table_name, db, if_exists = "replace")

# db操作を確定させる

db.commit()

# クエリー実行

db_query.execute(query)

# 1行ずつ表示

for row in db_query.fetchall():

print(row)

# print("処理を中止しました。")

# # 処理強制終了

# sys.exit()

# 概要：ユーザー毎の絵文字リストを取得

def get_emoji_list(user_id = "", DB_RESET_FLG = False):

# 定数

# EMOJI_NAME_COLUMN_NAME = "emoji_name"

EMOJI_FILE_NAME = "slack_emoji_list"

DB_EMOJI_TABLE_NAME = "T_emoji_list"

QUERY_ALL_SELECT = "select * from " + DB_EMOJI_TABLE_NAME # 全レコード表示するクエリー

# フルパス

db_name = DB_FOLD_NAME + "//" + EMOJI_FILE_NAME + ".db"

csv_emoji_name = DB_FOLD_NAME + "//" + EMOJI_FILE_NAME + ".csv"

# DB接続(存在しない場合は新たに作成)

db = sqlite3.connect(db_name)

# クエリー実行用

db_query = db.cursor()

# DBを初期化csv情報に初期化

if DB_RESET_FLG:

print("絵文字db_csvインポート開始")

db_reset(db, DB_EMOJI_TABLE_NAME, QUERY_ALL_SELECT, csv_emoji_name)

print("絵文字db_csvインポート終了")

# ユーザーIDが指定されている場合、ユーザー毎のリストを返却

if user_id != "":

# SQLで指定し易いようにシングルクォーテーションで囲む

user_id = "'" + user_id + "'"

# ユーザー毎の絵文字リスト取得

QUERY_USER_LIST = QUERY_ALL_SELECT + \

" where " + USER_ID_COLUMN_NAME + " = " + user_id

# クエリー実行

db_query.execute(QUERY_USER_LIST)

# ユーザーIDが指定されていない場合、全員のリストを返却

else:

# クエリー実行

db_query.execute(QUERY_ALL_SELECT)

# クエリー実行後結果取得

emoji_lists = []

for row in db_query.fetchall():

emoji_lists.append(row)

# dbを閉じる

db.close

# リスト返却

return emoji_lists

# 概要：ユーザー毎にタスク登録&取得

def get_task_list(user_id, task_url, task_name = "", DB_RESET_FLG = False, ALL_SHOW_FLG = False):

# 定数

TASK_LIST_FILE_NAME = "slack_task_list"

DB_TASK_TABLE_NAME = "T_task_list"

TASK_URL_COLUMN_NAME = "task_url"

TASK_NAME_COLUMN_NAME = "task_name"

QUERY_ALL_SELECT = "select * from " + DB_TASK_TABLE_NAME # 全レコード表示するクエリー

# フルパス

db_name = DB_FOLD_NAME + "//" + TASK_LIST_FILE_NAME + ".db"

csv_task_name = DB_FOLD_NAME + "//" + TASK_LIST_FILE_NAME + ".csv"

# DB接続(存在しない場合は新たに作成)

db = sqlite3.connect(db_name)

# クエリー実行用

db_query = db.cursor()

# テスト用：DBを初期化csv情報に初期化し、処理強制終了

if DB_RESET_FLG:

print("タスクdb_csvインポート開始")

db_reset(db, DB_TASK_TABLE_NAME, QUERY_ALL_SELECT, csv_task_name)

print("タスクdb_csvインポート終了")

# テスト用：全レコード表示

if ALL_SHOW_FLG:

# クエリー実行して全件表示

print("タスクdb_全件表示_開始")

db_query.execute(QUERY_ALL_SELECT)

for row in db_query.fetchall():

print(row)

print("タスクdb_全件表示_終了")

return

# SQLで指定し易いようにシングルクォーテーションで囲む

user_id = "'" + user_id + "'"

task_url = "'" + task_url + "'"

# エスケープ文字調整

task_url = get_sql_escape_contents.get_sql_escape_contents(task_url)

# タスク名が指定された場合はDBに登録する

# ※指定されていない場合は、リスト取得のみ

if task_name != "":

# SQLで指定し易いようにシングルクォーテーションで囲む

task_name = "'" + task_name + "'"

### 新規タスク追加 ############################################################

# URLのみ登録済判定(ユーザー毎)

QUERY_ALREADY_SET = QUERY_ALL_SELECT + \

" where " + USER_ID_COLUMN_NAME + " = " + user_id + \

" and " + TASK_URL_COLUMN_NAME + " = " + task_url

# クエリー実行

db_query.execute(QUERY_ALREADY_SET)

# DB追加フラグ初期化(True：新規追加する False：タスク名を更新する)

INSERT_FLG = True

# DB編集結果を表示

for row in db_query.fetchall():

INSERT_FLG = False

break

# URLが登録されていない場合、新規追加

if INSERT_FLG:

# タスク追加

QUERY_CREATE = "insert into " + DB_TASK_TABLE_NAME + \

" ( " + USER_ID_COLUMN_NAME + ", " + TASK_URL_COLUMN_NAME + ", " + TASK_NAME_COLUMN_NAME + ")" + \

"values (" + user_id + ", " + task_url + ", " + task_name + ")"

db_query.execute(QUERY_CREATE)

# db操作を確定させる

db.commit()

### タスク名更新 ############################################################

# URL、タスク名共に完全一致判定(ユーザー毎)

QUERY_ALREADY_SET = QUERY_ALL_SELECT + \

" where " + USER_ID_COLUMN_NAME + " = " + user_id + \

" and " + TASK_URL_COLUMN_NAME + " = " + task_url + \

" and " + TASK_NAME_COLUMN_NAME + " = " + task_name

# クエリー実行

db_query.execute(QUERY_ALREADY_SET)

# DB追加フラグ初期化(True：新規追加する False：タスク名を更新する)

UPDATE_FLG = True

# DB編集結果を表示

for row in db_query.fetchall():

UPDATE_FLG = False

break

# URLが登録されていない場合、新規追加

if UPDATE_FLG:

# タスク名更新

QUERY_UPDATE = "update " + DB_TASK_TABLE_NAME + \

" set " + TASK_NAME_COLUMN_NAME + " = " + task_name + \

" where " + USER_ID_COLUMN_NAME + " = " + user_id + \

" and " + TASK_URL_COLUMN_NAME + " = " + task_url

db_query.execute(QUERY_UPDATE)

# db操作を確定させる

db.commit()

# ユーザー毎のタスク取得

QUERY_USER_LIST = QUERY_ALL_SELECT + \

" where " + USER_ID_COLUMN_NAME + " = " + user_id + \

" and " + TASK_URL_COLUMN_NAME + " = " + task_url

# クエリー実行

db_query.execute(QUERY_USER_LIST)

# ユーザー×一致URLのリストを返却(1件しか存在しない想定だが)

task_lists = []

for row in db_query.fetchall():

task_lists.append(row)

# dbを閉じる

db.close

# リスト返却

return task_lists

# タスクdb初期化用

# print(get_task_list("","","", "",True))

# 絵文字db初期化用

# print(get_emoji_list("",True))

# 概要：db登録時、参照時に問題になる文字を置換して返却
def get_sql_escape_contents(text):
       
    text = text.replace("&amp;amp;","&amp;")    
    
    return text

# 概要：db登録時、参照時に問題になる文字を置換して返却

def get_sql_escape_contents(text):

text = text.replace("&amp;","&")

return text

注意事項

導入時に必要な「絵文字、タスク管理dbを設定するcsvファイル」は未紹介の為、
本コードをコピペしただけでは動きませんのでご注意下さい。
(「どのユーザー、どの絵文字を検索対象にするか？」のdbを初期化するcsv)

※今後需要がありましたら、実行可能なサンプルコードでダウンロード出来るように致します。