Zenn
MediiエンジニアブログPublicationへの投稿
😽

Slackでバリュースタンプの数を計測する仕組み

Tatsuya Watanabe
2024/12/27に公開
Python
Slack
Google Cloud
BigQuery
tech

はじめに

こんちには、株式会社Mediiの渡辺です
Mediiの特徴として、バリューを大切にしており、それが社員にちゃんと浸透していることが挙げられます
バリュー浸透のため、社内で色々な取り組みをしていますが、今回はその中の一つ、Slackでのバリュースタンプの数を計測する仕組みを作った話しをします

Mediiのバリューや取り組みについては、こちらの記事も是非ご覧ください！

ちょっとだけ変遷

今の仕組みになるまでに、何度か改良を重ねています
当初は楽をしていたのですが、キャパシティやコスト面で問題が出てきたため、徐々に改良を重ねてきました

  1. Zapier & スプレッドシート期
    • スプレッドシートのキャパを超えてしまった
  2. Zapier & BigQuery期
    • Zapierの使用料が1万タスク/月を超え、年間で30万円程度かかる状況に…
  3. Cloud Run functions & BigQuery期(今ここ)
    • 月額数百円程度で済むようになった

バリュースタンプの数を計測する仕組み

概要

【データを保存する】(本記事のスコープ)

  • Slack App
    • Event Subscriptionsで、reaction_addedイベントをサブスクライブする
    • イベントを検知したら、Cloud Run functionsを呼び出す
  • Cloud Run functions
    • バリュースタンプであるかの判定
    • ユーザーや実際のメッセージなどの付帯情報をSlack APIで取得
    • BigQueryにデータを書き込む

【データを利用する】(本記事のスコープ外、参考)

  • GAS
    • BigQueryからデータを取得
    • 新しくバリューを押されたメッセージを通知
    • バリュースタンプの数を集計して、定期的に通知
  • LookerStudio
    • BigQueryのデータを可視化

詳細

※本記事では、基本的な仕組みに焦点を当てているため、セキュリティ対応など一部省略しています

1. Cloud Run functionsの受け口をつくる

まず、Slack Appから呼び出すためのCloud Run functionsを作ります
この時点では、接続確認ができる最低限の実装で良いです

  1. Cloud Run functionsの関数を新規作成する

    • 環境をCloud Run functionにする
    • トリガーのタイプをHTTPSにする
    • URLをコピーしておく(後でコピーすることは可能です)
    • 認証は未認証の呼び出しを許可でOK
    • それ以外については自由に設定して良いです

      Cloud Run functionsの設定例

  2. Slack Appの接続テストを通すためのコードを書く
    本記事ではPython 3.12を使用しています

requirements.txt
# versionは適宜変更してください
functions-framework==3.*
requests==2.32.3
main.py
import functions_framework
import requests

@functions_framework.http
def webhook_slack_value(request):
    request_json = request.get_json(silent=True)
    # slackからの接続テストを成功させるため、challengeをそのまま返す必要
    if request_json and "type" in request_json:
        if request_json["type"] == "url_verification":
            return request_json["challenge"]

2. Slack Appを作成する

  1. Slack Appのページを開く
  2. Create New Appを選択
  3. From scratchを選択
  4. App Nameを入力し、slackのワークスペースを選択し、Create Appする
  5. 作成されたAppページのEvent Subscriptionsを選択
  6. Enable EventsをONにする
  7. Request URLに先ほど作成したCloud Run functionsのURLを入力
    正しく接続でき、challengeの値が返ってくると、Verifiedが表示される
  8. Subscribe to events on behalf of usersを開き、reaction_addedを追加
    Subscribe to bot eventsでも良いですが、botが入っているチャンネルでしかイベントを検知できないため、ここではSubscribe to events on behalf of usersを使います
  9. Save Changesします
    これで、Slack Appの設定は完了です
    slack上でスタンプが押されるたびにCloud Run functionsが呼び出されるようになります
    なお、reaction_addedで取得できる情報はこちらを参照してください

3. バリュースタンプの判定と保存(Cloud Run functions)

今回はBigQueryに保存していますが、他のデータベースでも構いません

requirements.txt
functions-framework==3.*
requests==2.32.3
google-cloud-bigquery==3.25.0
main.py
import functions_framework
import requests
from google.cloud import bigquery

@functions_framework.http
def webhook_slack_value(request):
    request_json = request.get_json(silent=True)
    # slackからの接続テストを成功させるため、challengeをそのまま返す必要
    if request_json and "type" in request_json:
        if request_json["type"] == "url_verification":
            return request_json["challenge"]
    
    events = request_json["event"]
    if not is_value_stamp(events):
        # バリュースタンプじゃなかった場合は処理を終了する
        return "Not value stamp"
    
    # バリュースタンプだった場合は保存処理などを行う
    insert_bigquery(events)

    return "success"

# バリュースタンプのリスト
VALUES = [
    "value1",
    "value2",
    "value3",
    "value4",
    "value5",
]

# バリュースタンプか判定
def is_value_stamp(events):
    reaction = events["reaction"]
    return reaction in VALUES

def insert_bigquery(events):
    # eventプロパティから必要な情報を取得
    # スタンプ押されたスタンプ名
    reaction = events["reaction"]
    # スタンプ押したユーザーID
    reaction_user = events["user"]
    # スタンプ押されてメッセージが投稿されたチャンネルID
    channel = events["item"]["channel"]
    # スタンプ押されたメッセージID
    message_id = events["item"]["ts"]
    # スタンプ押されたメッセージを送信したユーザーID
    message_user = events["item_user"]
    # イベントが実行されたタイムスタンプ
    event_ts = events["event_ts"]

    # 追加で情報を取りたい場合は、別途Slack APIを叩いて情報を取得します
    # メッセージ文だったり、ユーザーの表示名などはAPIから取得する必要があります
    # ここでは省略します

    # BigQueryに保存するデータ
    # データセット、テーブル、カラムは事前に作成しておく必要があります
    insert_data = {
        "reaction": reaction,
        "reaction_user": reaction_user,
        "channel": channel,
        "message_id": message_id,
        "message_user": message_user,
        "event_ts": event_ts,
    }

    # BigQueryにデータを保存する
    client = bigquery.Client()
    errors = client.insert_rows_json(
        "your-project.your-dataset.your-table",
        [insert_data],
    )

    if errors:
        # BQへの保存に失敗した場合の処理
        pass

(おまけ)最低限のセキュリティ対応

現状では、誰でもCloud Run functionsにアクセスできてしまいます
少なくても、Slackからのアクセスであるかのチェックを入れることをおすすめします

main.py
import time

# slackからのアクセスであるかのチェック
def is_access_from_slack_api(request):
    slack_signature = request.headers.get("X-Slack-Signature")
    slack_request_timestamp = request.headers.get("X-Slack-Request-Timestamp")

    if slack_signature is None or slack_request_timestamp is None:
        print("Slack以外からのアクセスです")
        return False

    # リクエストのタイムスタンプが古すぎないかチェック（5分以内のリクエストを許可）
    if abs(time.time() - float(slack_request_timestamp)) > 60 * 5:
        print("認証タイムアウトです")
        return False

    return True

その他特記事項

  • BigQueryに保存する場合は事前にデータセットとテーブルを作成しておく必要があります
    • また、Cloud Run functionsのサービスアカウントがBigQueryに書き込み権限を持っている必要があります
  • エラー処理やリトライ処理などは省略しています
  • SlackのEvent Subscriptionsでは3秒以内にレスポンスを返さないとタイムアウトします
    • そのため、処理時間がかかる場合は、Pub/Subなどで実処理の関数を非同期で呼び出すなどの工夫が必要です

おわりに

あくまでもバリュー浸透のための取り組みの一つであり、バリュースタンプの数だけで社員の価値を測るものではありません
しかし、バリュースタンプの数を計測・可視化することで、バリューに対する意識が高まり、バリューを実践する機会が増えることは期待できます

これからもMediiでは、バリューを大切にし、ミッションを達成するための取り組みを続けていきます！

Mediiエンジニアブログ
MediiエンジニアブログPublication

Discussion