Slackのスラッシュコマンドを作成する

本記事は以前投稿したSlackで自分の投稿にリアクションされたら通知するにスラッシュコマンドを追加する内容となります。
前述の記事はSlackアプリを作成する内容となっています。

コマンドの概要

リアクションが追加された時の通知対象を変更出来るように選択式のフォームを表示させ、
設定内容はDBに保管するというものです。

DBの作成

DBはAzure Cosmos DBを使用してみました。
格納するデータはチャンネルID、ユーザID、通知させる投稿対象の設定値のみです。

受信許可するIPアドレスをApp Serviceと自分の端末（グローバルIP）のみに設定します。

App Serviceの送信IPアドレスはApp Serviceのプロパティで見れます。

コマンドの作成

アプリ設定のSlash Commands画面でCreate New Commandボタンを押下してコマンドを作成します。

名前や説明等記入します。
Request URLはChallengeパラメータを返すURLと同じです。

保存するとcommandsスコープが追加されます。

スコープの追加がある場合はアプリの再インストールを行って反映します。

Interactivity & Shortcuts画面でInteractivityを有効化します。

App Serviceの設定を追加

Pythonアプリから読み取るようにするためDB名、エンドポイント、コンテナー名、およびアクセスキーをアプリケーション設定に追加します。

ソースコード修正

スラッシュコマンド実行時の画面とボタン選択時の処理を追加します。
画面のコードはBlock Kit Builderを使うと簡単に作成できます。
修正後はApp Serviceにデプロイします。

app.py

from slack_bolt import App, Ack, BoltResponse, BoltContext, Respond, Say
from slack_sdk import WebClient
import azure.cosmos.cosmos_client as cosmos_client
from enum import IntEnum, auto
import copy

# 通知モード種別
class Notification(IntEnum):
    NONE = 0
    SPOT = auto()
    ALL = auto()


USER_ID = "user_id"
NOTIFICATION_MODE = "notification_mode"
template_doc_main = {"id": "", "channel_id": "", "user": []}

# DBクライアント
cosmosclient = cosmos_client.CosmosClient(
    os.environ["DB_ENDPOINT"],
    os.environ["DB_KEY"],
)

db_id = os.environ["DB_ID"]
container_id = os.environ["DB_CONTAINER_ID"]
dbclient = cosmosclient.get_database_client(db_id)
containerclient = dbclient.get_container_client(container_id)
query_user = "SELECT u.user_id, u.notification_mode FROM config c JOIN u IN c.user WHERE c.channel_id = @channel_id AND u.user_id = @user_id"


@app.command("/rewatch")
def notification(ack: Ack, body: dict, respond: Respond):
    ack(
        blocks=[
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"通知タイプを選択してください\n\n*投稿毎* ：:{target_reaction}:のリアクションがある場合のみ通知します\n*全て* ：すべての投稿のリアクションを通知します\n*なし* ：通知しません",
                },
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "action_id": "notification_spot",
                        "text": {"type": "plain_text", "text": "投稿毎", "emoji": True},
                        "value": f"{Notification.SPOT}",
                    },
                    {
                        "type": "button",
                        "action_id": "notification_all",
                        "text": {"type": "plain_text", "text": "全て", "emoji": True},
                        "value": f"{Notification.ALL}",
                    },
                    {
                        "type": "button",
                        "action_id": "notification_none",
                        "text": {"type": "plain_text", "text": "なし", "emoji": True},
                        "value": f"{Notification.NONE}",
                    },
                ],
            },
        ],
    )


@app.action("notification_spot")
def start(ack: Ack, body: dict, logger, respond):
    ack()
    logger.info(body)
    upsert_notification(body, logger, respond)


@app.action("notification_all")
def start(ack: Ack, body: dict, logger, respond):
    ack()
    logger.info(body)
    upsert_notification(body, logger, respond)


@app.action("notification_none")
def start(ack: Ack, body: dict, logger, respond):
    ack()
    logger.info(body)
    upsert_notification(body, logger, respond)


def upsert_notification(body: dict, logger, respond):
    exist_channel = list(
        containerclient.query_items(
            query="SELECT c.id FROM config c WHERE c.id=@id",
            parameters=[{"name": "@id", "value": body["channel"]["id"]}],
            enable_cross_partition_query=True,
        )
    )

    channel = body["channel"]["id"]
    user = body["user"]["id"]
    notification_mode = int(body["actions"][0]["value"])

    # チャンネル追加
    upsert_item = copy.deepcopy(template_doc_main)
    logger.info(upsert_item)
    upsert_item["id"] = channel
    upsert_item["channel_id"] = channel
    upsert_item["user"] += [{USER_ID: user, NOTIFICATION_MODE: notification_mode}]
    logger.info(upsert_item)

    res_message = f"通知モードを「" + body["actions"][0]["text"]["text"] + f"」に設定しました。"

    if len(exist_channel) == 0:
        containerclient.upsert_item(body=upsert_item)
        respond(res_message)
        return

    # 通知モードの更新
    channel_items = containerclient.read_item(item=channel, partition_key=channel)
    for channel_item in channel_items["user"]:
        if channel_item["user_id"] == user:
            channel_item["notification_mode"] = notification_mode
            containerclient.upsert_item(body=channel_items)
            respond(res_message)
            return

    # ユーザ追加
    channel_items["user"] += [{USER_ID: user, NOTIFICATION_MODE: notification_mode}]
    containerclient.upsert_item(body=channel_items)
    respond(res_message)
    return

動作確認

アプリをインストールしたチャンネルで設定したコマンド/rewatchを実行します。

画面が表示され3つの選択肢が出ました。
全てを選択すると…

設定完了のメッセージが表示されます。

コマンドの動作はこのようになります。

これでスラッシュコマンドの作成完了です。
後はリアクション追加の通知処理の中でDBの値を見るように変更すればユーザ毎に通知タイプを変更できます。