Pythonで業務改善Slack Botを作って単純作業を減らしたい！

この記事はFLINTERS Advent Calendar 2022の20日目です。

業務上のコミュニケーションはSlackで行なっている会社が多いですよね。私のチームでもSlackを使用していて、Slack Botを活用すれば便利になる部分がたくさんあると思ったので、まずは手始めとして簡単な数値計算Botを作ってみたというお話です。

Slack APIについて

基礎

Slackを操作するためのAPIが公開されてます。

APIのアーキテクチャはRPC（Remote Procedure Call）スタイルらしいです。RESTではない。
Boltという公式フレームワークが出ており、そちらを利用すると簡潔な記述でAPIを操作できるので積極的に使っていきたいですね。

JavaScript、Python、Javaで使用できます。

Socket Mode

2021年ごろにリリースされた便利機能。WebSocket通信規格を通してSlackアプリを構築できます。

通常であれば、Boltを使用してリクエストハンドラを作成、特定のリクエストに対する処理を記述してアプリを起動、そのアプリをどこかのインターネットに公開（Public URLを取得）、SlackのCallBack URLにそのPublic URLを指定...などの手順を踏む必要があると思います。

しかしこのSocketModeを使用すると、そのPublic URLが必要無くて、とりあえずLocalでもどこでもアプリを起動しておけば、Slackのリクエストに反応してくれるようになるという認識です。便利ですね。

Eventの受信

Slack Boltを使用してイベントリスナーを構築しておくと、以下のようなすべてのイベントを受信することができます。

例えばメッセージの投稿というイベントを受信すると、以下のようなオブジェクトを受け取れます。

{
	"type": "message",
	"channel": "C2147483705",
	"user": "U2147483697",
	"text": "Hello world",
	"ts": "1355517523.000005"
}

アプリ側でのイベントリスナーの記述例。以下のようにすると、helloが含まれるメッセージが送信された際に上記のオブジェクトを取得することができます。

@app.message("hello")
def say_hello(message, say):
    user = message['user']
    say(f"Hi there, <@{user}>!")

このイベントリスナーを記述する際に指定可能な引数がかなり存在していてややこしいのですが、例えばmessageとsayを指定すると以下のようなことができます。

message

messageを関数の引数に設定することで、イベントで発生したメッセージオブジェクトを変数名messageで参照できるようになります。

say

say() utility function, which calls chat.postMessage API with the associated channel ID

sayを関数の引数に指定すると、Sayクラスのオブジェクトを取得できます。Sayクラスオブジェクトのインスタンス化には通常channel IDとclientが必要ですが、すでにそれらが渡された状態でインスタンス化されたオブジェクトを受け取れるようです。

上記リンクのソースコードを見てみると、Sayクラスの__call__メソッドにEvents APIのchat.postMessageを使用してメッセージを送信する記述がありました。そのため、sayを引数に指定した関数中で下記のように記述することでイベントが発生したchannelにメッセージを送ることができるようです。

say("aiueo")

特定のスレッドに返信したい場合などは、thread_tsを指定します。

say("aiueo", thread_ts=12345.6789)

役に立ちそうな機能を考える

スレッド内数値計算

仮にこんな感じのスレッドで勤務時間を報告している会社があった場合を考えてみましょう。

テストケース: チーム合計の勤務時間内訳を出したい

この会社の管理職は、上記のスレッドから毎回チーム（部署内）の合計営業時間、合計事務作業時間などを電卓で計算して、毎日上司や業務改善課etc.に報告しています。
毎日この作業に10〜20分くらい取られているので、これを自動化しましょう。

成果物

githubにアップロードしました。

使い方

公式ガイドに沿ってアプリの作成、トークンとアプリのインストールを行います。

今回の操作に必要なScope

• Bot Token Scopes
  • channels:history
    • スレッド内のメッセージを取得するために必要です。
  • chat:write
    • Botが計算結果を書き込むために必要です。
• App-Level-Tokens
  • connections:write
    • ソケットモード通信で発生したSlackイベントを受け取るために必要です。

ガイドでは省略されていますが、Socketモードを有効にした後、Event SubscriptionsからSubscribe to bot eventsを選択、Bot User Eventにmessage.channelsを追加してください。これがないとpythonアプリを起動してSlack側でメッセージを送信しても、pythonアプリ側でイベントが受信されていないようでした。

次に、使用したいチャンネルにこのAppを追加します。

最後に、今回のPythonファイルを実行する環境の環境変数SLACK_BOT_TOKENとSLACK_APP_TOKENに上記で作成したAppのトークンを設定して実行すればOKです。
うまくいけばチャンネルに投稿されたSlackのメッセージを受信できるようになっているはずです。

使用例

以下のようなスレッドで試してみます。

計算というワードに反応して計算した結果を書き込みます。

数値合ってそうです、良かった！

コード解説

Dataclassの活用

@dataclass
class PersonWorkTime:
    sales: Dict[str, Any] = field(
        default_factory = lambda: {"pattern": "営業", "hour": 0}
    )
    office_work: Dict[str, Any] = field(
        default_factory = lambda: {"pattern": "事務作業", "hour": 0}
    )
    phone_support: Dict[str, Any] = field(
        default_factory = lambda: {"pattern": "電話対応", "hour": 0}
    )
    total: Dict[str, Any] = field(
        default_factory = lambda: {"pattern": "合計", "hour": 0}
    )

クラスオブジェクトが、1メッセージごとの営業X時間、事務作業Y時間などを保存する形にしています。このオブジェクトの状態で、全てのオブジェクトのhourを加算して合計時間を計算します。注意したこととしては、仮に計測対象の作業が増えた場合（例えば資料作成を追加など）、この定義部分の変更だけで対応をできるよう気を付けています。

クラスを使用する考え方としては、クラスを使用しないでAPIから受け取ったオブジェクト（おそらくは複雑な辞書形式）をそのまま時間の計算処理をしていくよりも、どこかで一度オブジェクトの形式を確定させてしまった方が計算処理で何をしているのかが読みやすくなるという点があります。読みやすさに加えてテストもその方が書きやすいことが多い印象です。

ただデータクラスのインスタンス変数をミュータブルな値で初期化する際はdefault_factoryを使用しなければいけないなど結構クセがあったので、その辺りでかなり詰まってしまいました。ミュータブルな値で初期化するなら普通のクラスでも良かったかもしれないです。（今回はデータクラスに慣れるために使いました）

メッセージオブジェクト

thread_msgs = client.conversations_replies(token=token, channel=channel_id, ts=thread_ts)["messages"]

でスレッド内のメッセージ情報を取得しています。引数などに何が必要かはSlack API Documentに書いてあります。

成功すれば上記リンクのCommon successful responseに書いてあるような形式で返ってくるため、messageキーの中身を取り出すと、メッセージオブジェクト（Dict形式）のリストが得られます。

メッセージオブジェクトは以下のようになっており、textキーにメッセージの文字列が入っています。

{
    "type": "message",
    "user": "U061F7AUR",
    "text": "island",
    "thread_ts": "1482960137.003543",
    "reply_count": 3,
    "subscribed": true,
    "last_read": "1484678597.521003",
    "unread_count": 0,
    "ts": "1482960137.003543"
},

このメッセージ文字列をもとに、作業時間の抽出などを行います。

メッセージから作業時間の抽出処理

def find_match_pattern_and_add_work_time(self, message_text: str) -> None:
    values = self.get_ins_var_values()
    for value in values:
        calculate_class = value["pattern"]
        pattern = rf"[\s\S]*{calculate_class}\s([0-9]+)時間[\s\S]*"
        if m := re.match(pattern, message_text):
            hour = int(m.group(1))
            value["hour"] += hour

時間抽出処理は、上記のようにPersonWorkTimeクラスのインスタンスメソッドとして記述しました。引数は先述したスレッドのメッセージ文字列です。
self._get_ins_var_values()はインスタンス変数の値を全て取ってきます。オブジェクトを作成してすぐ実行した場合は、以下のようなリストが返ってきます。

[
    {"pattern": "営業", "hour": 0},
    {"pattern": "事務作業", "hour": 0},
    {"pattern": "電話対応", "hour": 0},
    {"pattern": "合計", "hour": 0}
]

営業、事務作業、etc.ごとにループさせて、指定したパターンにマッチした場合にマッチした([0-9]+)の数値分をクラスオブジェクトの値に追加していきます。

pattern = rf".*{calculate_class}\s([0-9]+)時間.*"だと改行後に事務作業 X時間という記述があった場合にマッチしないということに気付かずちょっと詰まりました。正規表現の.は"改行以外"の全ての文字列でしたね。

作業時間合計処理

def __add__(self, other):
    values = self.get_ins_var_values()
    other_values = other.get_ins_var_values()
    for value, other_value in zip(values, other_values):
        value["hour"] += other_value["hour"]
    return self

全てのオブジェクトのhourを加算して合計時間を計算するために、オブジェクト同士の加算を上記のように規定しました。こうすると、合計処理が以下のようにだいぶ簡潔に書けます。ここはクラスに格納した恩恵を受けた気がしました。

def calculate_all_person_work_time(pwt_iter: Iterable[PersonWorkTime]) -> PersonWorkTime:
    all_pwt = PersonWorkTime()
    for pwt in pwt_iter:
        all_pwt += pwt
    return all_pwt

感想

公式ドキュメントも充実していて、比較的すんなりと実装できました。
QiitaでSlack Botと検索すると、数百件も記事がヒットして、今回のものより全然複雑な構成のものがたくさん紹介されています。こういった記事を参考に、もっと便利なものを作って活用していきたいですね。

明日（21日目）の記事は@rikuta_flさんがCRA→Vite移行について書いてくれるようです、お楽しみに。