More than 3 years have passed since last update.

「進捗ありますか？」Slack Botを用いた研究室進捗確認ツール

Last updated at 2019-11-26Posted at 2019-06-24

概要

私の所属する研究室(情報分野)では、研究対象に応じて7、8人からなるグループを作成し、
グループごとに、Slackを用いて研究に関する情報や質問をすることができます。

今年からの取り組みとして、研究室での作業の取り組みを簡単にSlackで報告する方針に決めました。
決して、パノプティコンのような監視研究室を目指しているのではなく、ツールの導入や、エラーなどにより作業が進められないといったトラブルを次の進捗報告まで抱えるのではなく、Slackに気軽に報告し、詳しい人からアドバイスをもらおう！という__優しい世界__を目指しています。

そこで、指定日から今日まで(一週間程度を想定)Slackのチャンネルで発言していないユーザに対し、
タイトル詐欺にはなりますが、「進捗どうですか？」より2015倍捗る「困ってますか？」の文句を参考に、「進捗ありますか？」ではなく、「何か困っていませんか？」
と生存確認も込めてメッセージを送ります。

環境 & 使用Slack API

python3.5

使用するSlack API

API	用途	GET/POST
conversations.members	チャンネル内メンバー取得	GET
users.info	メンバーの詳細取得 (ユーザかbotの判定、名前の取得)	GET
conversations.history	チャンネルの履歴取得	GET
chat.postMessage	チャンネルにメッセージ投稿	POST

事前準備

Slack appの作成

作成したツールは、チャンネルに導入されたSlack Botを利用し、Botを通じてメッセージの投稿を行う方針です

Slack Botの導入
ここのSlack appを作成するを参考にアプリを作成してください。

Token & チャンネルIDの取得

Slack API使用時に必要なTokenと、対象のチャンネルのIDを事前に用意します。

Tokenの取得
今回は、OAuth Access TokenとBot User OAuth Access Tokenの
両方使用します。
(※初めは、Bot User OAuth Access Tokenで使用するAPIを扱う予定でしたが、API:conversation.historyで使用できなかったため、OAuth Access Tokenも使用します。)
Slack APIより、先ほど作成したアプリのOAuth & Permissionsにある2つのTokenをメモしてください。
- Scopeの設定
  conversation.historyでOAuth Access Tokenを用いるために、アプリに権限を与えます。
  対象アプリのメニュー
  Features > OAuth & PermissionsのScopes項目に以下の画像のように4つの権限(channels:history, groups:history, im:history, mpim:history)を付与します。
チャンネルIDの取得
Slack — APIに使う「チャンネルID」を取得する方法を参考に、チャンネルIDを取得してください。
e.g. https://hoge.slack.com/messages/GKV~~ -> GK〜〜がチャンネルID

Botをチャンネルに招く

Botアプリの再インストール
Scopeの設定などを反映させるため、アプリを再インストールします。
Slack APIの管理画面よりOAuth & PermissionsのReinstall Appボタンを押して再インストールしてください。
(１枚目の画像 Token情報下のReinstall Appボタン)
Botアプリを対象チャンネルに招待してください。
(手元で確認したところ、privateチャンネルにはBotを招待必要があるが、そうでないチャンネルはBotを招待しなくても問題なかった)

プログラムの作成

以下の手順で作成します.

チャンネル内のユーザ一覧を取得
指定日から今日までのチャンネル内の発話履歴から、発話ユーザ一覧を取得する。
発話していないユーザの＠メンション付きメッセージをチャンネルに投稿する。

チャンネル内のユーザ一覧を取得

ユーザの一覧取得は、はじめにAPI:conversations.membersを使用し、ユーザIDを取得する。
ユーザIDだと、ユーザを識別しにくいため、API:users.infoを用いてユーザ情報（name）と結びつける。

import requests
import json
import time
import datetime
import pprint

# Token & Channel ID
CHANNEL_ID = "{チャンネルID}"
OAuth_Token = "{OAuthトークンを入力}"
Bot_User_OAuth_Token = "{Bot User OAuthトークンを入力}"


Conversations_Members_API_URL = "https://slack.com/api/conversations.members"
def get_conversation_members():
    """ チャンネルにいるユーザidを取得 """
    params = {'token':Bot_User_OAuth_Token,
            'channel':CHANNEL_ID}
    r = requests.get(Conversations_Members_API_URL, params=params)
    json_data = r.json()
    if json_data.get('ok',False) == False:
        print('error channels memberの取得に失敗')
        return -1
    
    members = json_data['members']
    return members

上記の関数より、チャンネル内にいるユーザIDのリストが得られる。

ユーザIDにはBotも含まれるため、わかりやすくするためにユーザIDと名前のタプルリストを作成する。

def get_user_list_info(user_id_lst=[]):
    user_infos = [] # [(user_id,display_name),...]
    for u_id in user_id_lst:
        u_info = get_user_info(u_id)
        if u_info.get('ok',False) == False:
            print('error user_id [{0}]'.format(u_id))
            continue
        # user info取得成功
        if u_info['user']['is_bot'] == True:
            bot_app_name = 'Bot-' + u_info['user']['real_name']
            user_infos.append(tuple([u_id,bot_app_name]))
            continue
        else:
            # 名前取得 優先 display_name -> name 
            u_display_name = u_info['user']['profile']['display_name']
            if u_display_name != '':
                user_infos.append(tuple([u_id,'(display)'+u_display_name]))
            else:
                u_name = '(name)'+u_info['user']['name']
                user_infos.append(tuple([u_id,u_name]))

    return user_infos

User_Info_API_URL = "https://slack.com/api/users.info"
def get_user_info(user_id):
    params = {'token':Bot_User_OAuth_Token,
            'user':user_id}
    r = requests.get(User_Info_API_URL, params=params)
    json_data = r.json()
    
    return json_data

上記のコードを用いて、ユーザIDと名前のリストを取得する。

members_in_channel = get_conversation_members()
user_list_info = get_user_list_info(members_in_channel)
print('===チャンネルにいるメンバー')
pprint.pprint(user_list_info)

実行結果

===チャンネルにいるメンバー
[('ID1', '(display)表示名A'), ('ID2', 'Bot-channelbot')]

名前はBotであれば先頭にBot-が付与される。
ユーザの場合、表示名(display_name)を優先的に取得し(dispaly)、表示名が存在しない場合には、name情報を取得することにする(name)。

Slack チャンネルの履歴取得

API:conversations.historyを用いてチャンネルのメッセージ履歴を取得する。
引数limitで取得する件数を指定。
(よくわかっていないが、100件以上取得するときはPaginationを使うそう。conversations.history)

Conversations_History_API_URL = "https://slack.com/api/conversations.history"
def get_history(limit=100):
    params = {'token':OAuth_Token,
            'channel':CHANNEL_ID,
            'limit':limit}
    r = requests.get(Conversations_History_API_URL, params=params)
    json_data = r.json()

    return json_data['messages']

得られたメッセージ履歴から指定日以後のユーザによる投稿を取得する。

メッセージ履歴には投稿時のtimestamp(ts)が付与されているのでその情報を活用する。
なお、timestampは日本時間を想定している。Slackのタイムゾーンが日本時間でない場合は変更すること
タイムゾーンを変更する
指定日のtimestamp取得
以下の関数(get_timestamp())により、文字列：日/月/年の形式を受け取り、floatのtimestampを取得する。

def get_timestamp(s='24/06/2019'):
    JST = datetime.timezone(datetime.timedelta(hours=+9), 'JST')
    return time.mktime(datetime.datetime.strptime(s, "%d/%m/%Y").replace(tzinfo=JST).timetuple())

past_timestamp = get_timestamp('24/06/2019')
print(past_timestamp)　　# -> 1561302000.0

指定日以後のメッセージから、発話者と発言回数を取得
以下の関数(get_reported_user())でメッセージ履歴と指定日時を受け取り、対象範囲内の発話者ID:発言回数の辞書を取得する。
なお、メッセージにはBotによるメッセージも含まれるため、ユーザの投稿をkey:subtypeを持たないものとして判断した。

def get_reported_user(message_history, past_timestamp: float):
    """ ユーザの投稿をカウントする．投稿はmessageのkeyに'subtype'を持たないもので判断 """
    reported_user = {}  # user_id:発言回数
    for m in message_history:
        m_ts = m['ts']
        m_type = m['type']
        if not (float(m_ts) >= past_timestamp and 'subtype' not in m.keys()):
            continue
        
        user_id = m['user']
        if user_id in reported_user.keys():
            reported_user[user_id] +=1
        else:
            reported_user[user_id] = 1
    
    return reported_user

all_reported_users = get_reported_user(history,past_timestamp)
print('===発言user_id & 発言回数 {0}'.format(all_reported_users))
# -> {'ID1': 5,'ID2': 3, ...}

発言していないユーザへの＠メンションメッセージの投稿

発言していないユーザの取得
チャンネルのメンバーリスト(Botを除外したidリスト:user_id_in_channel)と、指定日以後に発言したユーザ:回数辞書を用いて発言していないユーザIDリストを得る。

# 発言していないユーザIDリスト
user_in_channel = [u for u in user_list_info if not u[1][:4] == 'Bot-'] # Botでないユーザリスト
user_id_in_channel = [u[0] for u in user_in_channel]    # user idのリスト

# 未発言ユーザidリスト
not_reported_users = [u_id for u_id in user_id_in_channel if u_id not in all_reported_users.keys()]
print('===未発言user_id {0}'.format(not_reported_users))

ユーザへのメンション付きメッセージの投稿

チャンネルへのメッセージ投稿はAPI:chat.postMessageを用います。
なお、このAPIはPOST通信を利用しており、以下のコード(post_message関数)のようにheaderの'Authorization':'Bearer {Bot User OAuthトークン}'としてトークンを付与します。
SlackのWebAPIを直接呼ぶときに困ったこと

def post_message(msg=''):
    headers = {'Content-Type': 'application/json; charset=utf-8',
                'Authorization':"Bearer {0}".format(Bot_User_OAuth_Token)}
    data = {'channel':CHANNEL_ID, 'text':msg, 'username':'進捗母ット'}
    r = requests.post(BOT_POST_API_URL,data=json.dumps(data),headers=headers)
    print(r.json())

メンション付きの投稿
APIを用いたメッセージの内容はAPIの引数textで与えます。
ここで@メンションは、
<@ユーザID>hogeの形式で記述します。
ここでは、未発言ユーザIDリストを受け取り、@メンション付きのメッセージをgenerate_msg関数を作成しました。

def generate_msg(not_reported_users:list):
    if len(not_reported_users) == 0:
        print('未発言ユーザはいません')
        return -1
    msg = ''
    for u_id in not_reported_users:
        msg += '<@' + u_id + '>さん'
    msg += '何か困っていませんか？'
    return msg

以下のコードでチャンネルにメッセージを投稿します。

msg = generate_msg(not_reported_users)
print(msg)
if not msg == -1: 
    post_message(msg)

実行結果

Tips & 改良

Slack API
Slack APIはかなり数が多く、似たような機能を持つAPIが複数あるように感じました。細かい仕様については、APIの公式ドキュメントを読む必要があると思います。
また、各APIはブラウザからテスト可能ですので、プログラム作成前や、APIの挙動をまずはテストするといいです。
アプリの改良について
メッセージ履歴を取得したのち、何か処理し、Botを用いて応答する流れを示すことができたと思います。
Slackでユーザメッセージを読み取った上で、分析、あるいはBot等で応答するアプリを作る一助になれば幸いです。

課題としては、このツールは手元でコードを動かす必要があります。したがって、完全な常駐Botにするという改良はあると思います。(サーバなど必要なんでしょうか？詳しいことはわかりません。)
注意点
Botによるメッセージ__「何か困っていませんか？/進捗ありますか？」__は、このコードでは、__チャンネルにいるユーザ全員が対象__になるので、チャンネルに先生や上司などがいる場合には、(適切に？)送信対象から外さないと問題が起こるかもしれません（笑）。(ツールによる問題について、一切の責任は負い兼ねます)

ソースコード

# -*- coding: utf-8 -*-
import requests
import json
import time
import datetime
import pprint
### Token & Channel ID
CHANNEL_ID = "{チャンネルID}"
OAuth_Token = "{OAuthトークンを入力}"
Bot_User_OAuth_Token = "{BotUserOAuthトークンを入力}"
######################

Conversations_Members_API_URL = "https://slack.com/api/conversations.members"
def get_conversation_members():
    """ チャンネルにいるユーザidを取得 """
    params = {'token':Bot_User_OAuth_Token,
            'channel':CHANNEL_ID}
    r = requests.get(Conversations_Members_API_URL, params=params)
    json_data = r.json()
    if json_data.get('ok',False) == False:
        print('error channels memberの取得に失敗')
        return -1
    
    members = json_data['members']
    return members

def get_user_list_info(user_id_lst=[]):
    user_infos = [] # [(user_id,display_name),...]
    for u_id in user_id_lst:
        u_info = get_user_info(u_id)
        if u_info.get('ok',False) == False:
            print('error user_id [{0}]'.format(u_id))
            continue
        # user info取得成功
        if u_info['user']['is_bot'] == True:
            bot_app_name = 'Bot-' + u_info['user']['real_name']
            user_infos.append(tuple([u_id,bot_app_name]))
            continue
        else:
            # 名前取得 優先 display_name -> name 
            u_display_name = u_info['user']['profile']['display_name']
            if u_display_name != '':
                user_infos.append(tuple([u_id,'(display)'+u_display_name]))
            else:
                u_name = '(name)'+u_info['user']['name']
                user_infos.append(tuple([u_id,u_name]))

    return user_infos

User_Info_API_URL = "https://slack.com/api/users.info"
def get_user_info(user_id):
    params = {'token':Bot_User_OAuth_Token,
            'user':user_id}
    r = requests.get(User_Info_API_URL, params=params)
    json_data = r.json()

    return json_data

Conversations_History_API_URL = "https://slack.com/api/conversations.history"
def get_history(limit=100):
    params = {'token':OAuth_Token,
            'channel':CHANNEL_ID,
            'limit':limit}
    r = requests.get(Conversations_History_API_URL, params=params)
    json_data = r.json()

    return json_data['messages']

def get_timestamp(s='24/06/2019'):
    JST = datetime.timezone(datetime.timedelta(hours=+9), 'JST')
    return time.mktime(datetime.datetime.strptime(s, "%d/%m/%Y").replace(tzinfo=JST).timetuple())

def get_reported_user(message_history, past_timestamp: float):
    """ ユーザの投稿をカウントする．投稿はmessageのkeyに'subtype'を持たないもので判断 """
    reported_user = {}  # user_id:発言回数
    for m in message_history:
        m_ts = m['ts']
        m_type = m['type']
        if not (float(m_ts) >= past_timestamp and 'subtype' not in m.keys()):
            continue
        
        user_id = m['user']
        if user_id in reported_user.keys():
            reported_user[user_id] +=1
        else:
            reported_user[user_id] = 1
    
    return reported_user

BOT_POST_API_URL = "https://slack.com/api/chat.postMessage"
def post_message(msg=''):
    headers = {'Content-Type': 'application/json; charset=utf-8',
                'Authorization':"Bearer {0}".format(Bot_User_OAuth_Token)}
    data = {'channel':CHANNEL_ID, 'text':msg, 'username':'進捗母ット'}
    r = requests.post(BOT_POST_API_URL,data=json.dumps(data),headers=headers)
    print(r.json())

def generate_msg(not_reported_users:list):
    if len(not_reported_users) == 0:
        print('未発言ユーザはいません')
        return -1
    msg = ''
    for u_id in not_reported_users:
        msg += '<@' + u_id + '>さん'
    msg += '何か困っていませんか？'
    return msg

def main():
    # チャンネル内メンバー取得
    members_in_channel = get_conversation_members()
    user_list_info = get_user_list_info(members_in_channel)
    print('===チャンネルにいるメンバー')
    pprint.pprint(user_list_info)

    # メッセージ履歴取得
    history = get_history(limit=100)
    # print(history)

    # 検索対象の過去の日時の指定
    past_timestamp = get_timestamp('24/06/2019')        # 指定日時以降のメッセージを対象にする
    print(past_timestamp)

    # 対象日時以降のメッセージより，発話ユーザIDと回数を取得
    all_reported_users = get_reported_user(history,past_timestamp)
    print('===発言user_id & 発言回数 {0}'.format(all_reported_users))

    # チャンネル内ユーザIDリスト
    user_in_channel = [u for u in user_list_info if not u[1][:4] == 'Bot-'] # Botでないユーザリスト
    user_id_in_channel = [u[0] for u in user_in_channel]    # user idのリスト

    # 未発言ユーザidリスト
    not_reported_users = [u_id for u_id in user_id_in_channel if u_id not in all_reported_users.keys()]
    print('===未発言user_id {0}'.format(not_reported_users))

    # post message
    msg = generate_msg(not_reported_users)
    print(msg)
    if not msg == -1: 
        post_message(msg)
        print('送信完了')

if __name__ == "__main__":
    main()

参考

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up