LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@yucca1(ゆっか)

Slackから特定期間の投稿数・リアクション状況を取得する

Posted at

はじめに

slackの活性状況をチームの状況把握に役立てたいなと思い、slackの定量的な情報を取得できないかと思い、調べてみると、記事がたくさん見つかりました。私もせっかくなのでやってみたことを残そうと思い、記事に残します。

結論：できるようになったこと

対象期間、対象チャンネルを設定して以下のようにスクリプトを実行すると、Slackの情報を取得できるようになりました。

$ python run.py 
#テスト チャンネルの状況
2024-05-09 18:30:00 ~ 2024-05-10 23:59:59
総メッセージ数: 7
リアクション回数: 
  sato.​taro: 4回
  yamada.hanako: 2回

環境

macOS Big Sur
Python 3.6.8

取得情報

目的は、チームのコミュニケーション状況把握なので、具体的に取得したい情報としては、以下とします。

  • 各チャンネルの特定期間における投稿数
  • 各チャンネルの特定期間におけるリアクション数

Slackワークスペース作成

色々試しやすいように、以下から新規でワークスペースをフリープランで作成しました。

基本的にガイドに沿って、必要最低限の項目を入力していくと作成できました。
ただし、以下のような最後のプラン選択では「フリープラン」の方を間違えずに選択してください。

Slack Appを作成

Slackに対してCLIやAPIでの操作は、アプリを経由して実行します。

（１） アプリケーション作成開始

  • ブラウザ上で作成したSlackのワークスペースにログインした状態で、https://api.slack.com/apps にアクセス
  • 「Create an App」を押下
  • 「From scratch」を選択
  • アプリ名を適当なものを入力（ここでは「Test Team App」とした）
  • 組み込むワークスペースを選択（先程作成したワークスペースを選択）

（2）許可するスコープの選択
「OAuth & Permissions」メニュー＞「Scopes」＞「User Token Scopes」において、今回は目的に合わせて以下のスコープを許可しました。

  • channels:history: 公開チャンネルのメッセージ取得
  • channels:read: 公開チャンネルの情報取得
  • emoji:read： ユーザーが追加した絵文字情報の取得
  • reactions:read: 絵文字のリアクション情報の取得
  • users:read: ユーザー情報の取得

（３）IP制限

今回はアクセス元IPが限られていたので、「OAuth & Permissions」メニュー＞「Restrict API Token Usage」セクションにおいて、許可するIPアドレス範囲を設定
※編集後は「Save IP address ranges」を押下しないと反映されないので注意

（４）インストール

「Install App」メニューから「Install to Workspace」を押下

以下の画面に遷移するので、「許可」を選択

インストール完了後、以下の赤枠のところに発行されたトークンが表示されるので、これを解析スクリプト内で使用する

情報取得スクリプト作成

今回はPythonで解析します。
venvで仮想環境を作り、必要パッケージをインストールして環境を構築します。

$ mkdir <作業ディレクトリ>
$ cd <作業ディレクトリ>
$ python -m venv .
$ source bin/activate
$ pip install slack-sdk pytz

※Windowsの場合、source bin/activateではなくsource Scripts/activateになります。

主に以下のSDKのメソッドを用いて

スクリプトを書きます。

run.py
from collections import Counter
import time
import pytz
from datetime import datetime
from slack_sdk.web import WebClient


# 対象期間
START_JST = datetime(2024, 5, 9, 18, 30, 0)
END_JST = datetime(2024, 5, 10, 23, 59, 59)

# トークン
SLACK_API_TOKEN = "your-slack-api-token"
# 調査対象のチャンネル
CHANNELS_ID = "target-channel-id"
# 上位いくつまでを集計するか。
RANK_NUMBER = 10

client = WebClient(token=SLACK_API_TOKEN)
# Workspaceの全てのユーザー情報を取得する。
USERS_LIST = client.users_list()['members']
# Workspaceの全ての公開チャンネル情報を取得する。
CHANNELS_LIST = client.conversations_list(limit=1000)['channels']

def _get_conversations_info(CHANNELS_ID:str):
    all_posts_reactions = []
    message_count = 0
    # 特定の会話を、APIの上限まで取得する。
    start_utc = START_JST.astimezone(pytz.utc).timestamp()
    end_utc = END_JST.astimezone(pytz.utc).timestamp()
    posts = client.conversations_history(channel=CHANNELS_ID, limit=1000, oldest=start_utc, latest=end_utc)['messages']
    for post in posts:
        # print(post['text'])
        message_count = message_count + 1
        # リアクションのあったものだけを取り出す。
        if 'reactions' in post.keys():
            all_posts_reactions.append(post['reactions'])
        # 返信があるpostの場合、返信を取得（ただし、返信時にチェックを入れてチャンネルにも投稿されたpostの場合は除く）
        if 'thread_ts' in post.keys() and 'subtype' not in post.keys():
            # 簡単さと実際の利用状況の観点から、返信は100件までのカウントとする
            res = client.conversations_replies(channel=CHANNELS_ID,
                ts=post['thread_ts'], limit=100, oldest=start_utc, latest=end_utc)
            # 0番目のメッセージは最初のpostであり集計済みなので、それ以外の返信を確認
            for reply in res['messages'][1:]:
                # チャンネルにも投稿済みの返信は集計済みなのでスキップ
                if 'subtype' in reply.keys():
                    continue
                # print(reply['text'])
                message_count = message_count + 1
                if 'reactions' in reply.keys():
                    all_posts_reactions.append(reply['reactions'])
    return message_count, all_posts_reactions

def _get_user_name(USER_ID:str):
    # ユーザーのIDを名前へと変換する。
    for row in USERS_LIST:
        if USER_ID == row['id']:
            return row['real_name']

def _get_channels_name(CHANNELS_ID:str):
    # 公開チャンネルのIDを名前へと変換する。
    for channel in CHANNELS_LIST:
        if CHANNELS_ID == channel['id']:
            return channel['name']

# 総メッセージ数と全リアクション情報を取得
message_count, all_posts_reactions = _get_conversations_info(CHANNELS_ID)

user_list = []
# 抽出したリアクションから、スタンプの数をカウントする。
for post_reactions in all_posts_reactions:
    for reaction in post_reactions:
        user_list.extend(reaction['users'])

# ユーザーごとのランキングを作成する。
user_rank = Counter(user_list).most_common(RANK_NUMBER)

CHANNELS_NAME = _get_channels_name(CHANNELS_ID)

# 結果出力
result = "#" + CHANNELS_NAME + " チャンネルの状況" + "\n"
result += f"{START_JST} ~ {END_JST}\n"
result += f"総メッセージ数: {message_count}\n"
result += "リアクション回数: \n"
for user_row in user_rank:
    # IDからユーザー名に変換
    user_name = _get_user_name(user_row[0])
    # 文章を生成する。
    result += f"  {user_name}: {user_row[1]}回\n"
print(result)

以下を参考にさせていただきました。ありがとうございます！

上記スクリプトを実行すると、冒頭に示した「結論」の結果が得られます。

さいごに

Slackのワークスペースを作ること自体初めてやってみましたが、こんなチャットサービスが無料で使えるってほんとすごいなとあらためて思いました。また、スクリプトを作成する際は、たくさんの参考記事はあるもののやっぱり自分で実際に書いて、動かして、所々出力してデータ構造を見ていくことでしっかり理解できると感じました。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?