フェズでよく使われるSlackリアクションを数えてみた

はじめに

テクノロジービジネス推進部で部長をしている福田です。
小売様のDMP構築やデータ利活用支援の開発と、フェズのデータ基盤の構築・運用の二つの役割を担っています。

今回は「Slack」を使った社内コミュニケーションで、フェズのメンバーがどんなリアクションをしているのかカウントしてみたので、コードなどを含めご紹介します。

きっかけ、fezwitterとは？

フェズにはfezwitterと言う業務以外も気軽に書き込めるSlackチャンネルがあります。
某呟きサイトをもじった命名ですね。

今回ある書き込みからSlackのリアクションを数えてみよう！と言う動きがありました。
これをきっかけにAPI叩いて数えてみよ〜と動き出します。

（脱線ですが）
fezwitterは「飲みに行く人〜」とか「台風気をつけて」みたいな投稿から、BeerBashと言う定期イベント（今回は夏祭りがテーマ）で、盛り上がった様子なども投稿されてます。

Slackのリアクション（絵文字）カウント

実装の流れやちょっと詰まったポイントなども触れつつ、簡単にコード例などもご紹介

実装の流れ

ざっと調べたところAPIを使って

• チャンネル一覧を取得
• チャンネルごとに繰り返し投稿を取得
• 投稿のリアクションをカウント

でやりたいことはやれそうと目星を。

その後公式などもみつつトークンを払い出します。
※当たり前ですがトークンは公開しないように注意

コード例

いざチャンネル一覧から投稿を取得しようとすると、
ほげほげAPIでは、チャンネルに参加していないと履歴が取れなかったためまずはオープンチャンネルに対して参加しまくるところから始めます。

def join_channels(channels):
    for channel in channels:
        if channel["is_member"]:
            continue
        try:
            client.conversations_join(channel=channel["id"])
            logging.info('join: ' + channel["name"])
            time.sleep(0.1)
        except SlackApiError as e:
            print(f"Error joining channel {channel['id']}: {e}")

その後各チャンネルで、指定した期間（今回は1ヶ月固定）の履歴を取得してリアクションをカウントしていきます。

import os
import sys
import logging
import time
import datetime
from collections import defaultdict
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

logging.basicConfig(level=logging.INFO)
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]


def get_all_public_channels():
    channels = []
    next_cursor = None
    result = client.conversations_list(types="public_channel", exclude_archived=True)
    while True:
        result = client.conversations_list(
            types="public_channel",
            exclude_archived=True,
            cursor=next_cursor
        )
        for channel in result["channels"]:
            logging.debug({channel["id"]: channel["name"]})
            channels.append({
                "id": channel["id"],
                "name": channel["name"],
                "is_member": channel["is_member"]
                })
        next_cursor = result.get("response_metadata", {}).get("next_cursor")
        if not next_cursor:
            break
    return channels


def get_channel_history(channel, cursor, oldest):
    try:
        response = client.conversations_history(
            channel=channel["id"],
            oldest=str(oldest),
            cursor=cursor
        )
    except SlackApiError as e:
        print(f"Error channel history : {e}")
    return response


def join_channels(channels):
    for channel in channels:
        if channel["is_member"]:
            continue
        try:
            client.conversations_join(channel=channel["id"])
            logging.info('join: ' + channel["name"])
            time.sleep(0.1)
        except SlackApiError as e:
            print(f"Error joining channel {channel['id']}: {e}")

if __name__ == "__main__":
    client = WebClient(token=SLACK_BOT_TOKEN)
    emoji_reaction_count = defaultdict(int)
    now = time.time()
    count_history_from = now - (1 * 10 * 24 * 60 * 60)  # 1ヶ月前のタイムスタンプ
    try:
        channels = get_all_public_channels()
    except SlackApiError as e:
        print("get_all_public_channels")
        print(f"Error fetching conversations: {e}")

    try:
        join_channels(channels)
    except SlackApiError as e:
        print("join_channels")
        print(f"Error fetching conversations: {e}")

    try:
        all_channel_count = len(channels)
        channel_count = 1
        for channel in channels:
            logging.info(f"({channel_count}/{all_channel_count}) : {channel['name']}")

            has_more = True
            cursor = None

            while has_more:
                response = get_channel_history(channel, cursor, count_history_from)
                messages = response["messages"]
                for message in messages:
                    if "reactions" in message:
                        reactions = message["reactions"]
                        for reaction in reactions:
                            emoji = f":{reaction['name']}:"
                            count = int(reaction["count"])
                            emoji_reaction_count[emoji] += count
                cursor = response.get("response_metadata", {}).get("next_cursor")
                has_more = cursor is not None
                time.sleep(1)
            channel_count += 1
            print(emoji_reaction_count)
            if channel_count % 100 == 0:
                print('--------------------')
                print(len(emoji_reaction_count))
                for emoji, count in emoji_reaction_count.items():
                    print(f"{emoji}: {count}")
    except SlackApiError as e:
        print(f"Error fetching conversations: {e}")

    print('--------------------')
    print(len(emoji_reaction_count))
    for emoji, count in emoji_reaction_count.items():
        print(f"{emoji}: {count}")

出力はテキストですが、Slackに貼り付けるとこんな感じ

チャンネル数が多いとAPIコールがそれなりになりたまにエラーになるので、単純にwaitかけたらしてますがもっと上手くエラーハンドリング書きたいところです。

課題や今後

とりあえず出力することを優先したため雑なコードですが社内リポジトリに載せておきつつ、今後は結果の並び替えとかこのチャンネルでよく使われる絵文字をちょっとみてみよ〜みたいに使えるようにしていきたいですね。

まとめ

フェズにはfezwitterと言う業務以外も気軽に書き込めるSlackチャンネルがあり、何気ない書き込みからちょっとしたコードを書いてみたり社内交流が生まれたらする会社です。

絵文字をカウントした結果がまとまればどんな結果だったのか、ご紹介したコードがどうなったか？などの続報もお出しできればと思います。