PythonでのSlackbot開発

はじめに

本稿はSLP KBIT Advent Calendar 2019の24日目の記事です．
お久しぶりです，がっきーです．
今回は，Slackbotを作成したのでその記録を書いていこうと思います．

開発の経緯

私の所属しているサークルでは，主な連絡ツールとして 無料プラン でSlackを利用しています．
そういった中で，多くのメンバがワークスペースにしているため，
ときに関係のない話題でメンションが行ってしまうことがあるため，その軽減をできないかと思い，
ユーザグループ管理を行うことができるbotを作成することにしました．
(最もすべてのメッセージに対する通知を有効にされていては意味がありませんが…)

開発環境及び使用技術

• Ubuntu 18.04 LTS
• Python3.6
  • slackbot
  • slack-client

ワークスペースへのbotの導入

こちらのサイトを参考にさせていただきました．

開発の流れ

ここから本題に入っていきます．以下の内容に分けて書いていきます．

• 権限設定
• フォルダ構成
• コーディング

権限設定

今回作成したユーザグループの管理を行っていくためには，
botユーザとしての権限だけでは不可能だったため，以下のように必要な権限を設定しました．

フォルダ構成

フォルダ構成は，以下のような形になっています．

my_mention.pyとsubMethod.pyの方にbotユーザの機能を超えた処理を記述し，それをmainのrun.pyでインポートしています．

コーディング

今回は，ユーザグループを管理するためのコマンド群を実装しました．
また，それに付随してアンケート関係のコマンドも作成したので，それに関しても少し書いていきます．
主な機能は以下のとおりです．

• ユーザグループの作成
• ユーザグループの削除
• ユーザグループメンバーの編集
• ユーザグループメンション

まず前提として，今回作成したユーザグループの考え方から説明します．
先にも述べたとおり，botを導入したワークスペースは無料プランなので，
ユーザグループに関係するAPIのメソッドを利用することはできません．
そのため，keyをユーザグループ名，valueをグループメンバーの配列とした辞書を作り，
静的なファイルに保存することで，擬似的なユーザグループとすることにしました．
そのため，以降のプログラム内で辞書に関連する内容がでてきたらそういうことなのだと了解しておいてください．

ユーザグループの作成

プログラムは以下のとおりです．

@respond_to('create\s([a-zA-Z0-9]*)\s([\w\s,]+)')
def create_usergroup(message, usergroup_name, member):
    usergroup = subMethod.get_usergroup_list()
    member_list = subMethod.get_member()['members']
    for usergroup_dict in usergroup:
        if usergroup_dict['usergroup_name'] == usergroup_name:
            message.send("`" + usergroup_name+' is already exist.`\n> please choose another name.')
            return
    data = {}
    member_id = []
    data['usergroup_name'] = usergroup_name
    try:
        member_name = [x.strip() for x in member.split(',')]
    except AttributeError:
        member_name = []
        member_id = member
    ml_id = [ml['id'] for ml in member_list]
    ml_name = [ml['name'] for ml in member_list]
    ml_rname = [ml['real_name'] if 'real_name' in ml else 'no_name' for ml in member_list]
    ml_dname = [ml['profile']['display_name'] for ml in member_list]
    for mn in member_name:
        if mn in ml_name:
            member_id.append(ml_id[ml_name.index(mn)])
        elif mn in ml_rname:
            member_id.append(ml_id[ml_rname.index(mn)])
        elif mn in ml_dname:
            member_id.append(ml_id[ml_dname.index(mn)])
        else:
            message.send("`" + mn + " is not in this channel`")
    data['member'] = member_id
    usergroup.append(data)
    subMethod.set_usergroup_list(usergroup)
    message.send('Created a usergroup')

create {ユーザグループ名} {メンバー名,…} という形式のメッセージを受け取ると実行されます．
内部処理としては，まず指定されたユーザグループがすでに作成済みでないかを確認します．
確認の上，まだ存在しないものであれば，作成してメンバーを追加していきます．
このとき，指定されたメンバーがワークスペース内に存在するかを同時に確認します．
ワークスペースに存在するユーザは，user_listメソッドを利用することで一覧を得ることができます．
ただここで問題となるのがその形式です．
slackではユーザがuserID，fullname，displaynameの3つの形式の値を持つことができます．
メッセージを送るときに，IDを送ることは難しいので，fullnameかdisplaynameを指定することになると思うため，
そこの相互変換をすることができるようにしておきました．
この処理を経て，存在するユーザであればそのIDを追加，存在しなければエラーメッセージを送信するようになっています．
なぜIDを追加するのか，これに関しては以下グループメンションに関しての項目で説明します．
実行すると下のようになります．

ユーザグループの削除

プログラムは以下のとおりです．

@respond_to('delete_usergroup\s([a-zA-Z0-9]*)')
def delete_usergroup(message, usergroup_name):
    usergroup = subMethod.get_usergroup_list()
    usergroup_name_list = [x['usergroup_name'] for x in usergroup]
    if usergroup_name not in usergroup_name_list:
        message.send("`" + usergroup_name + ' is not exist.`\n> type `@secretary list` and check usergroup_name')
        return
    new_usergroup = []
    for usergroup_dict in usergroup:
        if usergroup_dict['usergroup_name'] == usergroup_name:
            continue
        new_usergroup.append(usergroup_dict)
    subMethod.set_usergroup_list(new_usergroup)
    message.send('Deleted a usergroup')

delete_usergroup {ユーザグループ名} という形式のメッセージを受けて実行します．
これに関しては，ほぼ辞書の操作だけになります．
指定されたユーザグループがあれば，それを辞書の中から削除するだけです．
なかった場合にはエラーメッセージを送信します．
使ってみるとこんな感じになります．

ユーザグループメンバーの編集

プログラムは以下のとおりです．

@respond_to('add\s([a-zA-Z0-9]*)\s([\w\s,]+)')
def add_member(message, usergroup_name, member):
    usergroup = subMethod.get_usergroup_list()
    usergroup_name_list = [usergroup_dict['usergroup_name'] for usergroup_dict in usergroup]
    if usergroup_name not in usergroup_name_list:
        message.send("`" + usergroup_name + " is not exist`\n> please type `@secretary list` and check usergroup_name.")
        return
    member_list = subMethod.get_member()['members']
    usergroup_member = subMethod.get_usergroup_member(usergroup_name)

    member_id = []
    try:
        member_name = [x.strip() for x in member.split(',')]
    except AttributeError:
        member_name = []
        member_id = member
    add_member_name = []
    for mn in member_name:
        if mn not in usergroup_member:
            add_member_name.append(mn)
        else:
            message.send("`" + mn + ' already belongs`')
    ml_id = [ml['id'] for ml in member_list]
    ml_name = [ml['name'] for ml in member_list]
    ml_rname = [ml['real_name'] if 'real_name' in ml else 'no_name' for ml in member_list]
    ml_dname = [ml['profile']['display_name'] for ml in member_list]
    for mn in add_member_name:
        if mn in ml_name:
            member_id.append(ml_id[ml_name.index(mn)])
        elif mn in ml_rname:
            member_id.append(ml_id[ml_rname.index(mn)])
        elif mn in ml_dname:
            member_id.append(ml_id[ml_dname.index(mn)])
        else:
            message.send("`" + mn + " is not in this channel`")
    if len(member_id) == 0:
        message.send("`No one will add`")
        return
    for usergroup_dict in usergroup:
        if usergroup_dict['usergroup_name'] == usergroup_name:
            usergroup_dict['member'].extend(member_id)
            usergroup_dict['member'] = list(set(usergroup_dict['member']))
            break
    subMethod.set_usergroup_list(usergroup)
    message.send('Added some member')

@respond_to('delete\s([a-zA-Z0-9]*)\s([\w\s,]+)')
def delete_member(message, usergroup_name, member):
    usergroup = subMethod.get_usergroup_list()
    usergroup_name_list = [usergroup_dict['usergroup_name'] for usergroup_dict in usergroup]
    if usergroup_name not in usergroup_name_list:
        message.send("`" + usergroup_name + " is not exist`\n> type `@secretary list` and check usergroup_name")
        return
    member_list = subMethod.get_member()['members']
    member_id = []
    try:
        member_name = [x.strip() for x in member.split(',')]
    except AttributeError:
        member_name = []
        member_id = member
    ml_id = [ml['id'] for ml in member_list]
    ml_name = [ml['name'] for ml in member_list]
    ml_rname = [ml['real_name'] if 'real_name' in ml else 'no_name' for ml in member_list]
    ml_dname = [ml['profile']['display_name'] for ml in member_list]
    for mn in member_name:
        if mn in ml_name:
            member_id.append(ml_id[ml_name.index(mn)])
        elif mn in ml_rname:
            member_id.append(ml_id[ml_rname.index(mn)])
        elif mn in ml_dname:
            member_id.append(ml_id[ml_dname.index(mn)])
        else:
            message.send("`" + mn + " is not in this channel`")
    if len(member_id) == 0:
        message.send("`No one will delete`")
        return
    for usergroup_dict in usergroup:
        if usergroup_dict['usergroup_name'] == usergroup_name:
            for mi in member_id:
                if mi not in usergroup_dict['member']:
                    message.send("`" + ml_name[ml_id.index(mi)] + " doesn't belong to this`")
                else:
                    usergroup_dict['member'].remove(mi)
            break
    subMethod.set_usergroup_list(usergroup)
    message.send('Deleted some member')

add {ユーザグループ名} {メンバー名,…} というメッセージに対してユーザグループへのメンバーの追加
delete {ユーザグループ名} {メンバー名,…} というメッセージに対してユーザグループからのメンバーの削除を行います．
追加の処理は，指定されたユーザグループが存在すれば，あとは作成の処理と変わりません．
削除の処理は，指定されたユーザグループが存在したときに，そのユーザグループに所属するメンバーを一名ずつ削除していきます．
指定されたユーザグループ及びメンバーが存在しない場合はエラーです．

ユーザグループメンション

プログラムは以下のとおりです．

@listen_to('@[a-zA-Z0-9]+\s([\s\S]*)')
def reply_to_thread(message, text):
    usergroup = subMethod.get_usergroup_list()
    message.body['text'].replace('\n', ' ')
    mention = message.body['text'].split()[0].strip('@')
    mention_dict = []
    for dictionary in usergroup:
        if dictionary['usergroup_name'] == mention:
            mention_dict = dictionary
            break
    if len(mention_dict) == 0:
        message.send('`' + mention + ' is not exist`')
        return
    sentence = ""
    for member in mention_dict['member']:
        sentence = sentence + "<@" + member + "> "
    sentence = sentence + "\n"
    message.send(sentence, 
            thread_ts=message.thread_ts)

このメソッドは先までに書いた内容と系統が違います．
先程までのメソッドはbotに対してのメンションと合わせてでなければ実行されないものですが，
このメソッドはそれがなくても @{ユーザグループ名} {メッセージ} の形式のメッセージに反応します．
respondとして書いたメソッドはメンションが必要，listenとして書いたメソッドは不要ということです．
内部処理ですが，メッセージ冒頭に含まれる@{ユーザグループ名}を抜き出して，
指定されたユーザグループ名をkeyとした要素のvalueを取り出します．
valueは先に説明したとおり，配列の形式なので，for文で一つずつ要素を取り出して，結合してメッセージを作成します．
これがメンションのためのメッセージとなります．
ここで，メンションを行うときslackではIDを利用するようになっています．そのためIDを記憶しておく必要があったわけです．
作成したメッセージはTLに流してしまうと，ユーザグループのメンバーが非常に多かった場合邪魔にしかなりません．
そのため，今回はメッセージは元メッセージに対してスレッドの形式で送信するようにしました．
スレッドに送る場合のメソッドはmessage.send(sentence, thread_ts=message.thread_ts)です．
実際の画面は下のようになります．

ユーザグループに関連するメソッドとしては，おおよそ以上のようになります．
先に書いたもの以外に，グループ及びメンバーの一覧表示，
グループ名の変更，ユーザグループの結合といったものもありますが，これまでの組み合わせでできるので割愛します．

おまけ

slackでアンケートを取るときに，メッセージに対してリアクションをつけてもらうようにしていたのですが，
この形式だと一つずつリアクションを見て，誰がどこに投票したか確認する必要があり非常に手間でした．
ユーザグループを作成したときに，アンケートの解答対象者をユーザグループにまとめれば，
botの方でハンドリングをできるのではないかと思い，2つほど機能を作りました．(最近SimplePollなるものが出てるらしいですけどね…)

アンケート集計

プログラムは以下のとおりです．

@respond_to('count')
def count_up_reaction(message):
    response = subMethod.get_message(message.body['channel'], 
                                    message.thread_ts)
    if not response:
        message.direct_reply("Can't use count method in DM")
        return
    sentence = ''
    if 'reactions' in response['messages'][0]:
        data = response['messages'][0]['reactions']
        sorted_data = sorted(data, reverse=True, key=lambda x:x['count'])
        sentence = response['messages'][0]['text'] + '\n\n*Result*\n'
        for datum in sorted_data:
            sentence = sentence + ":" + datum['name'] + ":" + " "
            for user in datum['users']:
                sentence = sentence + "<@" + user + "> "
            sentence = sentence + "\n"
    else:
        sentence = 'No reactions'
    message.direct_reply(sentence)

リアクションの集計を取りたいメッセージのスレッドにcountと送ることで，実行されます．
処理としては，スレッドのトップのメッセージを取得すると，中にリアクションのデータをまとめた部分があるため，それを整理します．
リアクションのデータは下のような形で入っています．結局は辞書なのでデータを取得して成形してcountと送信してきたユーザにDMで送信します．
実行画面は下のようになります．

アンケート差分集計

プログラムは以下のとおりです．

@respond_to('diff')
def check_reactor(message):
    response = subMethod.get_message(message.body['channel'],
                                    message.thread_ts)
    if not response:
        message.direct_reply("Can't use count method in DM")
        return
    target_usergroup = response['messages'][0]['text'].replace('\n', ' ').split()[0].strip('@')
    all_target_audience = subMethod.get_usergroup_member_id(target_usergroup)
    if len(all_target_audience) == 0:
        sentence = 'No specified user group'
    elif 'reactions' in response['messages'][0]:
        data = response['messages'][0]['reactions']
        reacted_users = []
        reacted_users.extend([user for datum in data for user in datum['users']])
        target_audience = []
        target_audience.extend([user for user in all_target_audience if user not in reacted_users])
        sentence = "*Hasn't yet reacted*\n"
        for user in target_audience:
            sentence = sentence + "<@" + user + ">\n"
    else:
        sentence = "*Hasn't yet reacted*\n"
        for user in all_target_audience:
            sentence = sentence + "<@" + user + ">\n"
    message.direct_reply(sentence)

ユーザグループメンションと同様に，対象となるユーザグループをメッセージの冒頭で指定します．
その上で，対象のメッセージのスレッドにdiffと送ることで集計を行います．
まず対象のユーザグループをメッセージから抜き出して，メンバーの配列を取得します．
次に，リアクションをつけているユーザの一覧を何らかの形で取得します．
この2つを取得できたら，2つの配列のどちらか一方にしか存在しないユーザを抽出し，
それらを整えて，DMで送信します．
実行画面は下のようになります．

おわりに

slackbotの作成は導入自体はGUIを利用して直感的にできる上に，
機能実装もPythonに関しては，いいライブラリがあるためわりと簡単に取り組むことができるという印象でした．
現在はまだ機能も少ないため，今後も継続してコーディングしていこうと思います．
また毎度のことですが，同じ処理を色んな所で書いたり，
ファイル名もやばいことになっていたりとひどい状態なので，
リファクタリングをしっかりとやっておきたいと考えています．
最後に今回のコードはGitHub上にアップしているので，よければご覧になってください．
Dockerのあれこれも書いているので，Dockerを使える環境のある方は，トークンの設定さえしてもらえればすぐに利用してもらえると思います．