背景
この度会社で忘年会があり、同僚から「クイズを作るのにSlackで使用された回数の多いスタンプを集計してくれ」と依頼されたので久しぶりにPythonを使って集計するためのコードを実装してみました。
実装手順
以下コマンドで仮想環境を構築し、必要なパッケージをインストールします。
1. 仮想環境を構築
python -m venv venv
source venv/bin/activate
2. 必要パッケージをインストール
pip install requests python-dotenv matplotlib pandas
3. .envファイルを作成し、Slack Tokenを設定
プロジェクトディレクトリ直下に.envファイルを作成し、以下を追記します。
SLACK_TOKEN=xoxb-<Your-Slack-Bot-Token>
※SLACK_TOKENについて
Slackトークンに必要な権限（スコープ）が付与されている必要があります。
スコープ設定手順
SlackのAPI管理ページにアクセスし、「Create New App」をクリック。
AFrom scratchを選び、アプリ名とワークスペースを入力して「Create App」をクリック。
続いて、アプリが必要とするスコープを設定します。
1. OAuth & Permissionsメニューに移動
作成したアプリの管理ページで左側メニューから「OAuth & Permissions」を選択。
2. スコープを追加
Bot Token Scopesセクションに移動します。
「Add an OAuth Scope」をクリックし、以下のスコープを追加します。
channels:read // ワークスペースのパブリックチャンネルの情報を取得する
channels:history // パブリックチャンネル内のメッセージを取得する
groups:history // プライベートチャンネル内のメッセージを取得する
reactions:read // メッセージに付けられたスタンプ情報を取得する
channels:join // 特定のチャンネルにBotを参加させる（全チャンネル対応の場合に必要）
OAuth & Permissionsページで「Install App to Workspace」をクリック
ボタンをクリックすると、スコープの許可を求められます。
必要なスコープを確認して「許可」をクリック。
アプリがインストールされると、「OAuth & Permissions」ページにBot User OAuth Tokenが表示されます。
このトークンをコピーして、プロジェクトの.envファイルに追加します。
例：SLACK_TOKEN=xoxb-xxxxxxxxxx-xxxxxxxxxx-xxxxxxxxxx
4. ranking.pyファイルを作成し、script実装
コード全体
import os
import requests
from collections import Counter
from dotenv import load_dotenv
import matplotlib.pyplot as plt
import pandas as pd
from datetime import datetime
# 環境変数の読み込み
load_dotenv()
SLACK_TOKEN = os.getenv("SLACK_TOKEN")
# Slack APIヘッダー
headers = {"Authorization": f"Bearer {SLACK_TOKEN}"}
# チャンネルリストを取得する関数
def fetch_channel_list():
"""全チャンネルのリストを取得"""
url = "https://slack.com/api/conversations.list"
params = {"limit": 1000, "types": "public_channel,private_channel"}
response = requests.get(url, headers=headers, params=params)
response_data = response.json()
if not response_data.get("ok"):
raise Exception(f"Slack API Error: {response_data.get('error')}")
return response_data.get("channels", [])
def fetch_channel_messages(channel_id, oldest, latest):
"""指定したチャンネルからメッセージを取得する関数"""
url = "https://slack.com/api/conversations.history"
params = {
"channel": channel_id,
"oldest": oldest,
"latest": latest,
"limit": 1000
}
response = requests.get(url, headers=headers, params=params)
response_data = response.json()
if not response_data.get("ok"):
raise Exception(f"Slack API Error: {response_data.get('error')}")
return response_data.get("messages", [])
def count_reactions(messages):
"""メッセージ内のスタンプを集計する関数"""
reaction_counter = Counter()
for message in messages:
reactions = message.get("reactions", [])
for reaction in reactions:
reaction_counter[reaction["name"]] += reaction["count"]
return reaction_counter
def visualize_reactions(reaction_counter):
"""スタンプの使用回数を可視化する関数"""
reactions, counts = zip(*reaction_counter.most_common(10))
plt.barh(reactions, counts)
plt.xlabel("使用回数")
plt.ylabel("スタンプ")
plt.title("Slackスタンプ使用ランキング")
plt.gca().invert_yaxis() # ランキングを上位から下位に並べる
plt.show()
def save_to_csv(reaction_counter, filename="reaction_ranking.csv"):
"""ランキングをCSVに保存する関数"""
df = pd.DataFrame(reaction_counter.most_common(), columns=["スタンプ", "使用回数"])
df.to_csv(filename, index=False, encoding="utf-8-sig")
def to_unix_timestamp(date_string):
"""日付文字列をUnixタイムスタンプに変換する関数"""
dt = datetime.strptime(date_string, "%Y-%m-%d %H:%M:%S")
return int(dt.timestamp())
if __name__ == "__main__":
print("Slack API スタンプ集計ツール")
# 全チャンネルを取得
print("\n利用可能なチャンネルを取得中...")
channels = fetch_channel_list()
for i, channel in enumerate(channels):
print(f"{i + 1}. {channel['name']} (ID: {channel['id']})")
# ユーザー入力: チャンネルIDと期間
channel_index = int(input("\n対象のチャンネル番号を入力してください: ")) - 1
channel_id = channels[channel_index]["id"]
oldest_date = input("開始日時 (YYYY-MM-DD HH:MM:SS) を入力してください: ")
latest_date = input("終了日時 (YYYY-MM-DD HH:MM:SS) を入力してください: ")
# 日付をUnixタイムスタンプに変換
oldest = to_unix_timestamp(oldest_date)
latest = to_unix_timestamp(latest_date)
# メッセージを取得して集計
print("メッセージを取得中...")
messages = fetch_channel_messages(channel_id, oldest, latest)
if not messages:
print("指定された期間内にメッセージがありません。")
exit()
print("スタンプを集計中...")
reaction_counter = count_reactions(messages)
# 結果表示
print("\nスタンプ使用回数ランキング")
for rank, (reaction, count) in enumerate(reaction_counter.most_common(), start=1):
print(f"{rank}. {reaction}: {count}回")
# グラフ表示
visualize_reactions(reaction_counter)
# CSV保存
save_to_csv(reaction_counter)
print("ランキングをreaction_ranking.csvに保存しました。")
詳細解説
def to_unix_timestamp(date_string)関数
集計期間をスクリプト実行時に入力する際、YYYY-MM-DD HH:MM:SSの形式で入力できるようにして内部でUnixタイムスタンプに変換する処理を追加しました。
集計期間を細かく設定できます。
def to_unix_timestamp(date_string):
dt = datetime.strptime(date_string, "%Y-%m-%d %H:%M:%S")
return int(dt.timestamp())
def fetch_channel_list():関数
ワークスペース内のすべてのチャンネルを取得して選択できるようにしました。
これで、特定のチャンネルIDを取得し、そのチャンネルにおけるスタンプ利用回数の集計ができます。
https://slack.com/api/conversations.listはチャンネルIDを取得するためのエンドポイントです。
def fetch_channel_list():
url = "https://slack.com/api/conversations.list"
params = {"limit": 1000, "types": "public_channel,private_channel"}
response = requests.get(url, headers=headers, params=params)
response_data = response.json()
if not response_data.get("ok"):
raise Exception(f"Slack API Error: {response_data.get('error')}")
return response_data.get("channels", [])
def fetch_and_process_all_channels():関数
Slackワークスペース内のすべてのチャンネルについて、メッセージを取得し、スタンプの使用回数を集計するための関数です。複数のチャンネルを自動的に処理し、集計結果を個別に保存できます。
def fetch_and_process_all_channels():
print("全チャンネルのメッセージを取得し集計します...")
channels = fetch_channel_list()
for channel in channels:
try:
print(f"チャンネル {channel['name']} を処理中...")
messages = fetch_channel_messages(channel["id"], oldest, latest)
reaction_counter = count_reactions(messages)
save_to_csv(reaction_counter, filename=f"{channel['name']}_reaction_ranking.csv")
except Exception as e:
print(f"チャンネル {channel['name']} の処理でエラー: {e}")
エラーが発生した場合
not_in_channel エラー
対象のチャンネルにBotが参加していません。
conversations.join を使用してBotをチャンネルに参加させる必要があります。
missing_scope エラー
必要なスコープが不足しています。
アプリ管理ページでスコープを追加して再インストールしてください。
最後に
Slack APIではこのほかにもチャンネルやユーザー情報の会話履歴、自動化などさまざまな情報を提供しているので次回はお遊びではなく業務効率化に繋げる実装もしてみたいと思ってます！