仕様

Slack API + Python + Linux(cron)で定時作業を忘れていた場合にリマインドメッセージが飛ぶようにします。

具体的には、

・Slackbot で毎日9:00 に「ファイルアップロード！」というリマインドメッセージを飛ばす

・作業を忘れずに実施したときは上記のメッセージにスタンプをつける。(手動)

・リマインドメッセージを API によって取得し、メッセージの内容とスタンプの有無をチェックする

・スタンプがついていない(=作業を忘れている)場合、「ファイルのアップロードを忘れてますよ！」というメッセージを飛ばす

Slack アプリ作成

参考

https://zenn.dev/kou_pg_0131/articles/slack-api-post-message

1. ブラウザで Slack(https://slack.com/intl/ja-jp/) にログイン

2. Slack api の Your Apps ページ(https://api.slack.com/apps)にて「Create New App」を選択

3. 「From scratch」を選択し、App Name と対象ワークスペースを指定して「Create App」

4. 作成したアプリの詳細ページにて、「OAuth&Permissions」から「Scopes」を探す

5. Bot Token Scopes に以下の 3つの Scope を追加する。

 　　 - channels:history → パブリックチャンネルの情報取得用
　　  - chat:write → アプリによるメッセージ投稿用

6. 同ページ内の上の方にある「OAuth Tokens for Your Workspace」から「Install to Workspace」でアプリをインストール

7. 「Bot User OAuth Token」にトークンが発行されているので控えておく

 

チャンネルにアプリを登録

1. Workspace 内の App に作成したアプリが追加されていることを確認する

  追加されない場合はブラウザで一度 Slack を開いてみると良いかも

2. 追加したいチャンネルを右クリックし、「チャンネル詳細」>「インテグレーション」タブから

「アプリを追加する」を選択

3. 作成したアプリを追加

Slack API + Python 処理

まず公式SDKをインストールします。

pip install slack_sdk

細かい説明は後回しで先にプログラムを載せます。

import datetime, time
from slack_sdk import WebClient

# Slack API
SLACK_TOKEN = '1-7で確認したトークン'

#チャンネルをブラウザで開いたときの https://app.slack.com/client/hoge/fuga ← fugaの部分がID'
CHANNEL_ID = 'fugafuga' 

client = WebClient(token=SLACK_TOKEN)

#Slackbotのチェック対象メッセージのみを取得
tgt_msg = 'リマインダー : ファイルアップロード！'

try:
    response = client.conversations_history(channel=CHANNEL_ID,limit=10)
    current_ts = time.time()

    bot_msg = ""
    for msg in response['messages']:
        if msg.get('user') == 'USLACKBOT' and msg.get('text') == tgt_msg and (current_ts - float(msg.get('ts'))) < 300:
            bot_msg = msg

    #print(bot_msg)

    send_msg = '<!here> ファイルのアップロードを忘れてますよ！'


    if not bot_msg:
        pass
    elif 'reactions' in bot_msg.keys():
        pass
    else:
        client.chat_postMessage(channel=CHANNEL_ID,text=send_msg)

except SlackApiError as e:
    # Slack API呼び出しがエラーとなった場合の処理
    print(f"Error: {e.response['error']}")

では紹介。

このあたりが必須の部分です。トークンとチャンネルIDからWebClientをつくります。

from slack_sdk import WebClient

# Slack API
SLACK_TOKEN = '1-7で確認したトークン'

#チャンネルをブラウザで開いたときの https://app.slack.com/client/hoge/fuga ← fugaの部分がID'
CHANNEL_ID = 'fuga' 

client = WebClient(token=SLACK_TOKEN)

slackbot によるリマインダーメッセージには リマインダー :  がつくので、取得対象のメッセージは下記にする。

tgt_msg = 'リマインダー : ファイルアップロード！'

メッセージを取得するには client.conversations_history を使う。

取得するメッセージは最新から10通のみに絞る(デフォルトは100)。

response = client.conversations_history(channel=CHANNEL_ID,limit=10)

中身を見てみる(ID等はマスクして改行も加えてます。)

>>> print(response)
{'ok': True, 'messages': [
	{'client_msg_id': 'XXXX', 'type': 'message', 'text': 'test', 'user': 'XXXX', 'ts': '1705842699.105359', 'blocks': [{'type': 'rich_text', 'block_id': 'XXXX', 'elements':[{'type': 'rich_text_section', 'elements': [{'type': 'text', 'text': 'test'}]}]}], 'reactions': [{'name': 'eyes', 'users': ['XXXX'], 'count': 1}]}, 
	{'bot_id': 'B01', 'type': 'message', 'text': 'リマインダー : ファイルアップロード！', 'user': 'USLACKBOT', 'ts': '1705842141.709709', 'blocks': [{'type': 'rich_text', 'block_id': 'XXXX', 'elements': [{'type': 'rich_text_section', 'elements': [{'type': 'text', 'text': 'リマインダー : ファイルアップロード！'}]}]}], 'team': 'XXXX'}, 
	{'type': 'message', 'subtype': 'reminder_add', 'text': '  今日 22:02 , 日本標準時 にこのチャンネルでこれをリマインドするよう設定しました : “ファイルアップロー ド！” ', 'user': 'XXXX', 'ts': '1705842081.644479'}, 
	{'type': 'message', 'subtype': 'channel_join', 'ts': '1705841261.346399', 'user': 'XXXX', 'text': '<@XXXX>さんがチャンネルに参加しました', 'inviter': 'XXXX'}, 
	{'type': 'message', 'subtype': 'channel_join', 'ts': '1705839286.939959', 'user': 'XXXX', 'text': '<@XXXX>さんがチャンネルに参加しました'}], 
'has_more': False, 'pin_count': 0, 'channel_actions_ts': None, 'channel_actions_count': 0}
>>>

各要素はリストに辞書型で格納されているので、for 文で取得したい要素を指定すれば任意のフィールドで条件を組めます。

ただし、例えばスタンプがついていないと reactions は出てきません。

 

user が slackbot, text が取得対象メッセージ, かつ 5分以内のメッセージに一致するものを取得します。

for msg in response['messages']:
        if msg.get('user') == 'USLACKBOT' and msg.get('text') == tgt_msg and (current_ts - float(msg.get('ts'))) < 300:
            bot_msg = msg

メッセージの送信は client.chat_postMessage で行います。

bot_msg が空の場合と、対象メッセージにスタンプがついている場合は処理をスキップ。

send_msg = '<!here> スタンプに反応がありません！'
if not bot_msg:
    pass
elif 'reactions' in bot_msg.keys():
    pass
else:
    client.chat_postMessage(channel=CHANNEL_ID,text=send_msg)

メンションをつける場合は以下のように書くらしい。

・<@ユーザーID>

・<#チャンネルID>

・<!here>

 

最後のexcept 文は例外処理なので省略。

 

Linux(cron)設定

定時実行させるために Linux の cron を用います。

Cloud Scheduler なり GAS の定期実行(GASだと定時にはならないが…)でもできそう。

[test@test ~]$ crontab -l
3 9 * * * /usr/bin/python3.9 /home/hoge/work/check_work.py
[test@test ~]$

これで9:00のリマインダーに対して3分以内にスタンプの反応がなかったら、注意メッセージが飛んできます。

取得条件さえいろいろ変えれば汎用的に役に立ちそう。