LoginSignup
17
11

More than 1 year has passed since last update.

[コピペでOK]どのスタンプを誰がどんなチャンネルで使っているんだろう?(Slack API)

Last updated at Posted at 2021-12-08

Introduction

きっかけ

対面コミュニケーションにおいて、発せられる言葉から受け取る情報は7%で、そのほかの93%は言葉以外の部分から伝わるそうです。
非対面のテキストを通じたコミュニケーションによって抜け落ちてしまった情報は、スタンプ等を用いて感情表現を補完した経験が、皆さんにもあるかもしれません。

現職のSlackには、前職と比較して多くのスタンプが使われています。いわゆるカスタム絵文字というものですが、非常にユニークなものも多く毎日驚きと発見の連続です。それぞれのチャンネルを眺めているのも楽しいですが、Slackには各種APIが取り揃えられています。

そこで今回は、SlackのAPIを用いて、スタンプ使用率をさまざまな方法で調査したので、その手法を公開いたします。皆様も自分の社内Slackのスタンプ利用具合を調査してみてください!
なお個人が特定されるような公開できない情報が含まれる場合は、加工いたします。あらかじめご了承ください。

今回は、前回の記事の続編となります。
ソースコードの解説など基礎部分が知りたい方は、前回の記事も合わせてご参照ください。

過去に、似たようなことをやっている方が居たみたいです。Slack好きの集まる社風ですね。

結論

・ユーザーでランキングしたもの。

slack_user.png

・スタンプでランキングしたもの。

slack_sticker_ohayou.png

・チャンネルで分析したもの。

全てのチャンネルにてリアクションしたユーザーの上位10名は、下記のようになっています。
1位は、user1です。回数は23139回です。
2位は、user2です。回数は16770回です。
3位は、user3です。回数は12600回です。
4位は、user4です。回数は11820回です。
5位は、user5です。回数は10830回です。
6位は、user6です。回数は10207回です。
7位は、user7です。回数は9772回です。
8位は、user8です。回数は9178回です。
9位は、user9です。回数は6932回です。
10位は、user10です。回数は6600回です。

それぞれ上位10位までの結果は上記の通りです。
前回の記事のような基礎的な内容のみですと、使いづらい部分もあると思いましたので、実際のユースケースとして幾つかのパターンをまとめてみました。

下記の条件を調査対象としています。
・全ての公開チャンネル:858件。(アーカイブされたものも含む。2021年12月7日時点)
・それぞれのチャンネルの直近1000件の投稿。(Slack APIの上限が1000件のため)1
・集計対象は、リアクションに用いられたスタンプのみ。投稿内に使用されたスタンプは対象外。

本日のお品書き

・ SlackのWorkspaceにアプリをインストールする。(前回の内容)
・ Slack APIでごっそりとチャット履歴を取得する。(前回の内容)
・ Slack APIでユーザーごとの使用スタンプランキングを調べる。
・ Slack APIでスタンプごとの使用ユーザーランキングを調べる。
・ Slack APIでチャンネルごとのユーザーランキングを調べる。

対象者

・データ分析初心者の方
・1を聞いて10を理解できるエンジニア
・Slack APIとご無沙汰しておりますの方
Macユーザーの人
・Slack権限が一般の方2

非対象者

・データ分析ガチ勢
・説明下手な筆者を攻撃しようとするエンジニア
・Slack APIを熟知している方
Windowsユーザーの人
・Slackの管理者権限を持っている方

自己紹介

smile.jpg

自己紹介ページ

環境情報

macOS Big Sur ver:11.6
Python 3.8.5
slack-sdk==3.12.0

Cf.) Python2系とかPython3系の意味が分からない人へ!
Pythonには2系と3系があって、最近始めた人ならほとんどPython3系だと思います。一応バージョンを確認する方法を記載しておきます。

MacOSに入っているターミナル(Terminal)でPythonのインタラクティブモードを起動すればバージョン情報が表示されます。

$ Python
>Python 3.8.5 (default, Sep  4 2020, 02:22:02) 
[Clang 10.0.0 ] :: Anaconda, Inc. on darwin
Type "help", "copyright", "credits" or "license" for more information.>>>

インタラクティブモードを辞めるにはexit()を入力しましょう!

前提条件

今回の記事は、vscodeで実装しています。ただ、どちらのエディタでも問題ないと思います。知らんけど。

なお、ここからは前回の記事と同様の内容が続きます。
今回のメインテーマは、 Let's startからご覧ください。

SlackのWorkspaceにアプリをインストールする。

Slackの権限が、管理者と一般によって実装難易度が大きく異なります。
管理者の場合は、Slack APIを用いずともCSVファイル等を用いて分析することができるようです。

管理者権限で、CSVファイルを手に入れられる方は、下記の記事が参考になると思います。

自分は一般の権限しか持っていないため、下記の記事の手順ですすめました。下記の記事はRubyで書かれていますが、Slack APIの権限周りは開発言語に依存していないため、問題はない3です。

Slackのアプリを作成します。

Slackの公式でアプリを作成して、そのアプリを通じてSlackの各種APIを操作します。
そのためまずはアプリを作成してください。

Create an app をクリックして、アプリを作成してください。

slack_toppage.png

権限周りは下記の通りに設定しております。参考記事の通りです。 ユーザー別で分析をしない場合は、users:read は不要かもしれません。

slack_appscope.png

トークンを発行します。

トークン(token)が発行されたら、完了です。
自分はWorkspaceにインストールしたら、 xxxx-****-your-token みたいな長いトークンが発行されました。アプリを通じてSlack APIを操作するにはこちらのトークンが必須のようです。

Slack APIを使用する。

参考記事を読み進めていくと、この辺りでつまづきました4
色々調査を進めるうちに、APIの名前が統合されたり、名前が変わっている部分も多かったです。さらにSlack APIはSDKを通じて操作する方が簡単そうと判断しました5

結論としては、Slack SDKを用いてSlack APIを操作することに決めました。

SlackのSDKをインストールする。

下記の公式SDKの記事がわかりやすかったです。結論から述べると、Slack SDKはpip3でインストールできるようです。

pip3 install slack-sdk

インストールしたパッケージが使えているか確認するときは、下記のコードで実行できます。

api_test.py
from slack_sdk.web import WebClient

client = WebClient()
response = client.api_test()
print(response)
python api_test.py 

実行すると、{'ok': True, 'args': {}}との結果が返ってきます。
OKみたいですね!とりあえず安心です。

Let's Start

Slack APIでユーザーごとの使用スタンプランキングを調べる。

特定ユーザー&全てのチャンネル&スタンプランキング

自分が全ての公開チャンネルの中で、どんなスタンプを使っているのかをランキングにしたものです。

出力結果

slack_user.png

ソースコード

test_user.py
from collections import Counter
import time

from slack_sdk.web import WebClient


# トークン
SLACK_API_TOKEN = "xxxx-****-your-token"
# 集計結果を送信するチャンネル
POST_ID = "your_post_id"
# 集計対象者
SUBJECT_ID = "subject_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_public_list(CHANNELS_LIST:list):
    _public_list = []
    for channel in CHANNELS_LIST:
        channel_dict = {
            'name' : channel['name'],
            'id' : channel['id']
        }
        _public_list.append(channel_dict)
    return _public_list

def _get_sticker_list(CHANNELS_ID:str):
    # 特定の会話を、APIの上限まで取得する。
    response = client.conversations_history(channel=CHANNELS_ID, limit=1000)['messages']

    # 会話履歴から、一つの投稿ごとに処理する。
    for post in response:
        # リアクションのあったものだけを取り出す。
        if 'reactions' in post.keys():
            all_reaction_list.append(post['reactions'])

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

# 公開チャンネルを格納するリスト
public_list = _get_public_list(CHANNELS_LIST)

# それぞれの投稿についたリアクションをリストに貯めておく。
all_reaction_list = []
for index, row in enumerate(public_list):
    print(index, row)
    if index % 7 == 0:
        time.sleep(1)
    try:
        _get_sticker_list(row['id'])
    except Exception as e:
        print(e, len(all_reaction_list))
        time.sleep(3)
        pass

user_list = []
# 抽出したリアクションから、スタンプの数をカウントする。
for reaction_list in all_reaction_list:
    for row in reaction_list:
        max_count = row['count']
        count = 1
        while count <= max_count:
            try:
                user_list.append({
                    "name" : row['name'],
                    "user": row['users'][count - 1]
                })
            except IndexError as e:
                print(e)
                print(row['name'], count, max_count)
                pass
            count += 1

sticker_list = []
for sticker in user_list:
    if SUBJECT_ID in sticker['user']:
        sticker_list.append(sticker['name'])

# IDだと視認性が低下するため、Slackで用いるユーザー名に変換する。
SUBJECT_NAME = _get_user_name(SUBJECT_ID)

# スタンプごとの使用ユーザーのランキングを作成する。
sticker_rank = Counter(sticker_list).most_common(RANK_NUMBER)

# スタンプのリストを取得する。
POST_CONTENTS = SUBJECT_NAME + "が、使った絵文字の上位" + str(RANK_NUMBER) + "個は、下記のようになっています。" + "\n"
for rank, sticker_row in enumerate(sticker_rank):
    # 文章を生成する。
    rank_string = str(rank+1) + "位は、:" + sticker_row[0] + ":です。回数は" + str(sticker_row[1]) + "回です。" + '\n'
    POST_CONTENTS += rank_string

# 自分のDMへ送信する。
client.chat_postMessage(channel=POST_ID, text=POST_CONTENTS)

大まかな流れ

  1. 全ての公開チャンネルのリストを取得する。
  2. チャンネルの会話履歴から、リアクション情報を取得する。
  3. 抽出したリアクションから、スタンプの数を取得する。
  4. 全てのスタンプリストから、特定ユーザーのスタンプの数のみ取得する。
  5. 特定ユーザーのスタンプの数をランキングにする。
  6. ランキング上位10項目をSlackへ送信するための文章に整形する。
  7. Slackへ送信する。

ソースコードの中で、わからない部分があればコメントください。解説いたします。

特定ユーザー&特定チャンネル&スタンプランキング

自分が#times_sendaチャンネルの中で、どんなスタンプを使っているのかをランキングにしたものです。

出力結果

slack_user_channel.png

ソースコード

test_user_channel.py
from collections import Counter
import time

from slack_sdk.web import WebClient


# トークン
SLACK_API_TOKEN = "xxxx-****-your-token"
# 調査対象のチャンネル
CHANNELS_ID = "your_channels_id"
# 集計結果を送信するチャンネル
POST_ID = "your_post_id"
# 集計対象者
SUBJECT_ID = "subject_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_sticker_list(CHANNELS_ID:str):
    # 特定の会話を、APIの上限まで取得する。
    response = client.conversations_history(channel=CHANNELS_ID, limit=1000)['messages']

    # 会話履歴から、一つの投稿ごとに処理する。
    for post in response:
        # リアクションのあったものだけを取り出す。
        if 'reactions' in post.keys():
            all_reaction_list.append(post['reactions'])

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

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

# それぞれの投稿についたリアクションをリストに貯めておく。
all_reaction_list = []
_get_sticker_list(CHANNELS_ID)

user_list = []
# 抽出したリアクションから、スタンプの数をカウントする。
for reaction_list in all_reaction_list:
    for row in reaction_list:
        max_count = row['count']
        count = 1
        while count <= max_count:
            try:
                user_list.append({
                    "name" : row['name'],
                    "user": row['users'][count - 1]
                })
            except IndexError as e:
                print(e)
                print(row['name'], count, max_count)
                pass
            count += 1

sticker_list = []
for sticker in user_list:
    if SUBJECT_ID in sticker['user']:
        sticker_list.append(sticker['name'])

# IDだと視認性が低下するため、Slackで用いるユーザー名に変換する。
SUBJECT_NAME = _get_user_name(SUBJECT_ID)

# スタンプごとの使用ユーザーのランキングを作成する。
sticker_rank = Counter(sticker_list).most_common(RANK_NUMBER)
CHANNELS_NAME = _get_channels_name(CHANNELS_ID)

# スタンプのリストを取得する。
POST_CONTENTS = "#" + CHANNELS_NAME + " チャンネルにて" + SUBJECT_NAME + "が、使った絵文字の上位" + str(RANK_NUMBER) + "個は、下記のようになっています。" + "\n"
for rank, sticker_row in enumerate(sticker_rank):
    # 文章を生成する。
    rank_string = str(rank+1) + "位は、:" + sticker_row[0] + ":です。回数は" + str(sticker_row[1]) + "回です。" + '\n'
    POST_CONTENTS += rank_string

# 自分のDMへ送信する。
client.chat_postMessage(channel=POST_ID, link_names=True, text=POST_CONTENTS)

大まかな流れ

  1. 特定のチャンネルの会話履歴から、リアクション情報を取得する。
  2. 抽出したリアクションから、スタンプの数を取得する。
  3. 全てのスタンプリストから、特定ユーザーのスタンプの数のみ取得する。
  4. 特定ユーザーのスタンプの数をランキングにする。
  5. ランキング上位10項目をSlackへ送信するための文章に整形する。
  6. Slackへ送信する。

ソースコードの中で、わからない部分があればコメントください。解説いたします。

Slack APIでスタンプごとの使用ユーザーランキングを調べる。

特定スタンプ&全てのチャンネル&ユーザーランキング

特定のスタンプが全ての公開チャンネルの中で、どんなユーザーに使われているのかをランキングにしたものです。
今回はよく使われる おはよう というスタンプの利用ランキングを調べてみました。

slack_greeting.png

ほとんど個人の名前になってしまうため、見せられませんが、出力としてはおおよその予測は付くかと思います。

出力結果

slack_sticker_ohayou.png

ソースコード

test_sticker.py
from collections import Counter
import time

from slack_sdk.web import WebClient


# トークン
SLACK_API_TOKEN = "xxxx-****-your-token"
# 集計結果を送信するチャンネル
POST_ID = "your_post_id"
# 集計対象者
SUBJECT_ID = "subject_id"
RANK_NUMBER = 10 # 上位いくつを集計するか。
STICKER_NAME = "ohayou"

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


def _get_public_list(CHANNELS_LIST:list):
    _public_list = []
    for channel in CHANNELS_LIST:
        channel_dict = {
            'name' : channel['name'],
            'id' : channel['id']
        }
        _public_list.append(channel_dict)
    return _public_list

def _get_sticker_list(CHANNELS_ID:str):
    # 特定の会話を、APIの上限まで取得する。
    response = client.conversations_history(channel=CHANNELS_ID, limit=1000)['messages']

    # 会話履歴から、一つの投稿ごとに処理する。
    for post in response:
        # リアクションのあったものだけを取り出す。
        if 'reactions' in post.keys():
            all_reaction_list.append(post['reactions'])

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

# 公開チャンネルを格納するリスト
public_list = _get_public_list(CHANNELS_LIST)

# それぞれの投稿についたリアクションをリストに貯めておく。
all_reaction_list = []
for index, row in enumerate(public_list):
    print(index, row)
    if index % 7 == 0:
        time.sleep(1)
    try:
        _get_sticker_list(row['id'])
    except Exception as e:
        print(e, len(all_reaction_list))
        time.sleep(3)
        pass

user_list = []
# 抽出したリアクションから、スタンプの数をカウントする。
for reaction_list in all_reaction_list:
    for row in reaction_list:
        max_count = row['count']
        count = 1
        while count <= max_count:
            try:
                user_list.append({
                    "name" : row['name'],
                    "user": row['users'][count - 1]
                })
            except IndexError as e:
                print(e)
                print(row['name'], count, max_count)
                pass
            count += 1

sticker_list = []
for sticker in user_list:
    if STICKER_NAME in sticker['name']:
        sticker_list.append(sticker['user'])

# スタンプごとの使用ユーザーのランキングを作成する。
sticker_rank = Counter(sticker_list).most_common(RANK_NUMBER)

# スタンプのリストを取得する。
POST_CONTENTS = ":" + STICKER_NAME + ":を使っているユーザーの上位" + str(RANK_NUMBER) + "名は、下記のようになっています。" + "\n"
for rank, sticker_row in enumerate(sticker_rank):

    # IDだと視認性が低下するため、Slackで用いるユーザー名に変換する。
    SUBJECT_NAME = _get_user_name(sticker_row[0])
    # 文章を生成する。
    rank_string = str(rank+1) + "位は、" + SUBJECT_NAME + "です。回数は" + str(sticker_row[1]) + "回です。" + '\n'
    POST_CONTENTS += rank_string
print(POST_CONTENTS)

# 自分のDMへ送信する。
client.chat_postMessage(channel=POST_ID, text=POST_CONTENTS)

大まかな流れ

  1. 全ての公開チャンネルのリストを取得する。
  2. チャンネルの会話履歴から、リアクション情報を取得する。
  3. 抽出したリアクションから、スタンプの数を取得する。
  4. 全てのスタンプリストから、特定スタンプの数のみ取得する。
  5. 特定スタンプのユーザー数をランキングにする。
  6. ユーザーのIDをSlackに登録されている名前に変換する。
  7. ランキング上位10項目をSlackへ送信するための文章に整形する。
  8. Slackへ送信する。

ソースコードの中で、わからない部分があればコメントください。解説いたします。

特定スタンプ&特定チャンネル&ユーザーランキング

eyeが #general チャンネルの中で、どんなユーザーに使われているのかをランキングにしたものです。

出力結果

slack_sticker_eye.png

ソースコード

test_sticker_channel.py
from collections import Counter
import time

from slack_sdk.web import WebClient


# トークン
SLACK_API_TOKEN = "xxxx-****-your-token"
# 調査対象のチャンネル
CHANNELS_ID = "your_channels_id"
# 集計結果を送信するチャンネル
POST_ID = "your_post_id"
# 集計対象者
SUBJECT_ID = "subject_id"
RANK_NUMBER = 10 # 上位いくつを集計するか。
STICKER_NAME = "eye"

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


def _get_sticker_list(CHANNELS_ID:str):
    # 特定の会話を、APIの上限まで取得する。
    response = client.conversations_history(channel=CHANNELS_ID, limit=1000)['messages']

    # 会話履歴から、一つの投稿ごとに処理する。
    for post in response:
        # リアクションのあったものだけを取り出す。
        if 'reactions' in post.keys():
            all_reaction_list.append(post['reactions'])

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

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

# それぞれの投稿についたリアクションをリストに貯めておく。
all_reaction_list = []
_get_sticker_list(CHANNELS_ID)

user_list = []
# 抽出したリアクションから、スタンプの数をカウントする。
for reaction_list in all_reaction_list:
    for row in reaction_list:
        max_count = row['count']
        count = 1
        while count <= max_count:
            try:
                user_list.append({
                    "name" : row['name'],
                    "user": row['users'][count - 1]
                })
            except IndexError as e:
                print(e)
                print(row['name'], count, max_count)
                pass
            count += 1

sticker_list = []
for sticker in user_list:
    if STICKER_NAME in sticker['name']:
        sticker_list.append(sticker['user'])

# スタンプごとの使用ユーザーのランキングを作成する。
sticker_rank = Counter(sticker_list).most_common(RANK_NUMBER)
CHANNELS_NAME = _get_channels_name(CHANNELS_ID)

# スタンプのリストを取得する。
POST_CONTENTS = "#" + CHANNELS_NAME + " チャンネルにて:" + STICKER_NAME + ":を使ったユーザーの上位" + str(RANK_NUMBER) + "名は、下記のようになっています。" + "\n"
for rank, sticker_row in enumerate(sticker_rank):

    # IDだと視認性が低下するため、Slackで用いるユーザー名に変換する。
    SUBJECT_NAME = _get_user_name(sticker_row[0])
    # 文章を生成する。
    rank_string = str(rank+1) + "位は、" + SUBJECT_NAME + "です。回数は" + str(sticker_row[1]) + "回です。" + '\n'
    POST_CONTENTS += rank_string
print(POST_CONTENTS)


# 自分のDMへ送信する。
client.chat_postMessage(channel=POST_ID, link_names=True, text=POST_CONTENTS)

大まかな流れ

  1. 特定チャンネルの会話履歴から、リアクション情報を取得する。
  2. 抽出したリアクションから、スタンプの数を取得する。
  3. 全てのスタンプリストから、特定スタンプの数のみ取得する。
  4. 特定スタンプのユーザー数をランキングにする。
  5. ユーザーのIDをSlackに登録されている名前に変換する。
  6. ランキング上位10項目をSlackへ送信するための文章に整形する。
  7. Slackへ送信する。

ソースコードの中で、わからない部分があればコメントください。解説いたします。

Slack APIでチャンネルごとのユーザーランキングを調べる。

全てのスタンプ&全てのチャンネル&ユーザーランキング

全ての公開チャンネルの中で、どんなユーザーが合計で何回リアクション6をしたのかをランキングにしたものです。

出力結果

全てのチャンネルにてリアクションしたユーザーの上位10名は、下記のようになっています。
1位は、user1です。回数は23139回です。
2位は、user2です。回数は16770回です。
3位は、user3です。回数は12600回です。
4位は、user4です。回数は11820回です。
5位は、user5です。回数は10830回です。
6位は、user6です。回数は10207回です。
7位は、user7です。回数は9772回です。
8位は、user8です。回数は9178回です。
9位は、user9です。回数は6932回です。
10位は、user10です。回数は6600回です。

ソースコード

test_channels.py
from collections import Counter
import time

from slack_sdk.web import WebClient


# トークン
SLACK_API_TOKEN = "xxxx-****-your-token"
# 集計結果を送信するチャンネル
POST_ID = "your_post_id"
# 集計対象者
SUBJECT_ID = "subject_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_public_list(CHANNELS_LIST:list):
    _public_list = []
    for channel in CHANNELS_LIST:
        # print(j)
        channel_dict = {
            'name' : channel['name'],
            'id' : channel['id']
        }
        _public_list.append(channel_dict)
    return _public_list

def _get_sticker_list(CHANNELS_ID:str):
    # 特定の会話を、APIの上限まで取得する。
    response = client.conversations_history(channel=CHANNELS_ID, limit=1000)['messages']

    # 会話履歴から、一つの投稿ごとに処理する。
    for post in response:
        # リアクションのあったものだけを取り出す。
        if 'reactions' in post.keys():
            all_reaction_list.append(post['reactions'])

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

# 公開チャンネルを格納するリスト
public_list = _get_public_list(CHANNELS_LIST)

# それぞれの投稿についたリアクションをリストに貯めておく。
all_reaction_list = []
for index, row in enumerate(public_list):
    print(index, row)
    if index % 7 == 0:
        time.sleep(1)
    try:
        _get_sticker_list(row['id'])
    except Exception as e:
        print(e, len(all_reaction_list))
        time.sleep(3)
        pass

user_list = []
# 抽出したリアクションから、スタンプの数をカウントする。
for reaction_list in all_reaction_list:
    for row in reaction_list:
        max_count = row['count']
        count = 1
        while count <= max_count:
            try:
                user_list.append(row['users'][count - 1])
            except IndexError as e:
                print(e)
                print(row['name'], count, max_count)
                pass
            count += 1

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

# スタンプのリストを取得する。
POST_CONTENTS = "全てのチャンネルにてリアクションしたユーザーの上位" + str(RANK_NUMBER) + "名は、下記のようになっています。" + "\n"
for rank, user_row in enumerate(user_rank):
    # IDだと視認性が低下するため、Slackで用いるユーザー名に変換する。
    SUBJECT_NAME = _get_user_name(user_row[0])
    # 文章を生成する。
    rank_string = str(rank+1) + "位は、" + SUBJECT_NAME + "です。回数は" + str(user_row[1]) + "回です。" + '\n'
    POST_CONTENTS += rank_string
print(POST_CONTENTS)

# 自分のDMへ送信する。
client.chat_postMessage(channel=POST_ID, text=POST_CONTENTS)

大まかな流れ

  1. 全ての公開チャンネルのリストを取得する。
  2. チャンネルの会話履歴から、リアクション情報を取得する。
  3. 抽出したリアクションから、ユーザーごとのスタンプの合計数を取得する。
  4. ユーザーごとのリアクション数をランキングにする。
  5. ユーザーのIDをSlackに登録されている名前に変換する。
  6. ランキング上位10項目をSlackへ送信するための文章に整形する。
  7. Slackへ送信する。

ソースコードの中で、わからない部分があればコメントください。解説いたします。

全てのスタンプ&特定チャンネル&ユーザーランキング

#times_senda チャンネルの中で、どんなユーザーが合計で何回リアクション6をしたのかをランキングにしたものです。

出力結果

#times_senda チャンネルにてリアクションしたユーザーの上位10名は、下記のようになっています。
1位は、user1です。回数は86回です。
2位は、user2です。回数は43回です。
3位は、user3です。回数は42回です。
4位は、user4です。回数は41回です。
5位は、user5です。回数は33回です。
6位は、user6です。回数は17回です。
7位は、user7です。回数は12回です。
8位は、user8です。回数は10回です。
9位は、user9です。回数は9回です。
10位は、user10です。回数は8回です。

ソースコード

test_channels_channel.py
from collections import Counter
import time

from slack_sdk.web import WebClient


# トークン
SLACK_API_TOKEN = "xxxx-****-your-token"
# 調査対象のチャンネル
CHANNELS_ID = "your_channels_id"
# 集計結果を送信するチャンネル
POST_ID = "your_post_id"
# 集計対象者
SUBJECT_ID = "subject_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_sticker_list(CHANNELS_ID:str):
    # 特定の会話を、APIの上限まで取得する。
    response = client.conversations_history(channel=CHANNELS_ID, limit=1000)['messages']

    # 会話履歴から、一つの投稿ごとに処理する。
    for post in response:
        # リアクションのあったものだけを取り出す。
        if 'reactions' in post.keys():
            all_reaction_list.append(post['reactions'])

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

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

# それぞれの投稿についたリアクションをリストに貯めておく。
all_reaction_list = []
_get_sticker_list(CHANNELS_ID)

user_list = []
# 抽出したリアクションから、スタンプの数をカウントする。
for reaction_list in all_reaction_list:
    for row in reaction_list:
        max_count = row['count']
        count = 1
        while count <= max_count:
            try:
                user_list.append(row['users'][count - 1])
            except IndexError as e:
                print(e)
                print(row['name'], count, max_count)
                pass
            count += 1

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

CHANNELS_NAME = _get_channels_name(CHANNELS_ID)

# スタンプのリストを取得する。
POST_CONTENTS = "#" + CHANNELS_NAME + " チャンネルにてリアクションしたユーザーの上位" + str(RANK_NUMBER) + "名は、下記のようになっています。" + "\n"
for rank, user_row in enumerate(user_rank):

    # IDだと視認性が低下するため、Slackで用いるユーザー名に変換する。
    SUBJECT_NAME = _get_user_name(user_row[0])

    # 文章を生成する。
    rank_string = str(rank+1) + "位は、" + SUBJECT_NAME + "です。回数は" + str(user_row[1]) + "回です。" + '\n'
    POST_CONTENTS += rank_string
print(POST_CONTENTS)

# 自分のDMへ送信する。
client.chat_postMessage(channel=POST_ID, link_names=True, text=POST_CONTENTS)

大まかな流れ

  1. 特定のチャンネルの会話履歴から、リアクション情報を取得する。
  2. 抽出したリアクションから、ユーザーごとのスタンプの合計数を取得する。
  3. ユーザーごとのリアクション数をランキングにする。
  4. ユーザーのIDをSlackに登録されている名前に変換する。
  5. ランキング上位10項目をSlackへ送信するための文章に整形する。
  6. Slackへ送信する。

ソースコードの中で、わからない部分があればコメントください。解説いたします。

まとめ

今回の内容は前回の記事を踏まえて、実際にどんな出力ができるのかを試してみたものとなります。皆さんのSlack生活がより豊かになることを願っております。マイナースタンプランキングを作ってみても面白いと思います。ここまでみていただきありがとうございました。

To be Continued

終わりに

今後もTwitterでこのようなデータの可視化を発信しています。興味があったらご覧ください。 大抵はくだらないことです。

Sempleのツイッター

参考記事


  1. 時間の指定や他の方法を駆使すれば、直近1000件以前を遡ることもできます。参考情報はこちら 

  2. 管理者権限でない一般の権限という意味です。上手い言い方が思い浮かびませんでした。アイデア募集中です。 

  3. 嘘です。強がりました。こちらの記事をもとに実装していましたが、Rubyがどうしても理解できず、途中から自己流で実装しています。大まかな流れは、上記の記事をもとにしています。 

  4. もちろんRubyが読めないからです。本文では責任転嫁していますが、単純にRuby記事を読み進めていった自分に問題があると思います。皆さんは自己を過信しすぎないでください。 

  5. SDKとかAPIという用語は、きちんと理解せずに使っています。認識違いがありましたら、ご指摘お願いいたします。 

  6. スタンプによる全てのリアクションを総合したものです。 

17
11
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
17
11