ロボペイSlackリアクションランキング集計してみた

こんにちは、SRE課の塚本です。
今回はSlackAPIを使って、ずっとやろうと思ってやってなかった「Slackリアクションランキング」を集計していきます。

集計する対象についてはこんな感じで集計していきます

slackに投稿されたメッセージに対するリアクションの総数を集計
集計期間は2023年12月の１ヶ月間
パブリックチャンネルのみ
1~20位までを集計

取得する流れとしてはこうです

channel_idの一覧を取得
channel_idの一覧をもとにchannel_idに紐づくメッセージを取得
メッセージからリアクションを取り出し、ランキングを集計

引っかかったところとしては、今回使用したメソッドに関して、 Paginationが導入されており、取得できるデータがページごとに分割されるところです。cursorが空になるまでAPIを叩く必要がありました。

Paginationについて api.slack.com

また、APIのレート制限に引っかかったら処理を一定期間空けるといった対応も必要です。（今回はできてないのでガバガバ集計となってしまいました。。。）
ちなみに、今回スレッドに対するリアクションが拾えていないので、そこも改良の余地がありそうです。

気になる集計結果がこちら！

おはようが圧倒的に多いですね！（笑）あとは最近体調崩されている方が多いので、おだいじにが上位にランクインしてしまいました。。。みなさんお大事に！

ありがとう系のスタンプがたくさんあるのは嬉しいですね！これからも周りの方に感謝しながら仕事していければなと思います！

使用したコードがこちら↓

SLACK_API_TOKEN = "BotのUserTokenを貼る"

def main():
    channel_list = get_channel_list(SLACK_API_TOKEN)
    messages = []
    for channel in channel_list:
        print(channel)
        try:
            channel_messages = get_one_day_slack_message(SLACK_API_TOKEN, channel)
            messages.extend(channel_messages)
        except Exception as e:
            print(f"Error occurred while fetching messages for channel {channel}: {str(e)}")
            channel_messages = get_one_day_slack_message(SLACK_API_TOKEN, channel)
            continue
        
    result = lank_reaction(messages)
    return result

def get_one_day_slack_message(bot_token, channel):
    url = "https://slack.com/api/conversations.history"
    headers = {
        "Authorization": "Bearer " + bot_token,
        "Content-Type": "application/json; charset=utf-8"
    }
    
    messages = []
    cursor = None
    while True:
        payload = {
            "channel": channel,
            "limit": 100,
            "oldest": time.time() - 60 * 60 * 24 * 30,  # 30 days ago
        }
        if cursor:
            payload["cursor"] = cursor
        
        response = requests.post(url, headers=headers, json=payload)
        json_data = response.json()
        
        if not json_data["ok"]:
            print("Error:", json_data["error"])
            break
        
        messages.extend(json_data["messages"])
        
        if not json_data.get("response_metadata") or not json_data["response_metadata"].get("next_cursor"):
            break
        
        cursor = json_data["response_metadata"]["next_cursor"]
    
    return messages

def lank_reaction(messages):
    reactions_list = []
    for message in messages:
        if 'reactions' in message:
            for reaction in message['reactions']:
                reaction_info = {
                    'name': reaction['name'],
                    'count': reaction['count']
                }
                # 既に同じ名前のリアクションがリストに存在するか確認し、存在する場合はカウントを追加する
                found = False
                for existing_reaction in reactions_list:
                    if existing_reaction['name'] == reaction_info['name']:
                        existing_reaction['count'] += reaction_info['count']
                        found = True
                        break
                
                # もしリストに存在しない場合は新たに追加する
                if not found:
                    reactions_list.append(reaction_info)
    # countの値が大きい順にreactions_listを並べ替える
    sorted_reactions_list = sorted(reactions_list, key=lambda x: x['count'], reverse=True)
    # 上位10個の要素を取得する
    top_10_reactions = sorted_reactions_list[:20]
    return top_10_reactions

    
    
def get_channel_list(bot_token):
    url = "https://slack.com/api/conversations.list"
    headers = {
        "Authorization": "Bearer " + bot_token
    }
    channel_list = []
    cursor = None
    while True:
        payload = {
            "limit": 100
        }
        if cursor:
            payload["cursor"] = cursor
        
        response = requests.post(url, headers=headers)
        json_data = response.json()
        
        if not json_data["ok"]:
            print("Error:", json_data["error"])
            break
        
        channel_list.extend(channel['id'] for channel in json_data['channels'])
        
        
        if not json_data.get("response_metadata") or not json_data["response_metadata"].get("next_cursor"):
            break
        
        cursor = json_data["response_metadata"]["next_cursor"]
    return channel_list