【やってみた】pythonでslackのメッセージを取得して、csvに書き出す

どーも!

たかぽんです!

今回はあるチャンネルのslackのメッセージを取得して、その取得したメッセージをcsvに書き出してみる...というのをやってみようと思います!

やりたいこと

普段よく使っているslackで、あるチャンネルの投稿内容を分析したいなぁ...ということがありました。

投稿する方は以前webhook urlを用いて試したことがあるのですが、逆はまだないなというわけで、今回ちょっと遊んでみようと思います!

より具体的にしておくと、あるチャンネルにて、ある日時から現在までの投稿を取得します。

加えてその投稿に対するスレッドの有無やスタンプのリアクション有無も取得して、その投稿内容をcsvとして出力する...といったことをやってみたいと思います!

実際にやっていく上で、SlackAPIも必要になるので、改めてそこから解説していきます!

ではやっていきましょう!

事前準備

まずは事前準備をやっていきます。

以下のページにて、ログイン後、新しいSlack APIを新規作成します。

上記画面でログインをすると、以下のような"Your Apps"の画面が出てくると思います。

右上に"Create New App"というボタンがあるのでそこから新しいAppを作成します。

ボタンを押すと、以下のようなモーダルが出てくると思います。

今回は一から作っていくので"From Scratch"で作成しました。

App Nameは今回のアプリの名称を適当に入れます。

そして"Pick a workspace to develop your app in"は使いたいワークスペースを選択します。

今回は筆者の個人のワークスペースを選んでいます。

create Appをするとアプリが作成されます。

アプリが作成されたら、下記のような画面がでてきます。

まずは画面左の"Features"がら、"OAuth & Permissions"を選択します。

下にスクロールすると、"Scopes"という項目があると思うので、その中の"Bot Token Scopes"から、"Add an Oauth Scope"にて、"channels:history"と、"users:read"を追加します。

ここでやっていることは、このアプリのBot tokenを発行するのですが、そのトークンで行える権限範囲を定義しています。

チャンネルのヒストリーで投稿文を、また、投稿者名を取得するためにusers_infoを取得しています。

(一応historyでも投稿者のidは取得できますが、実際の名前を取得するために使っています。)

ここまでできたら、上の方に以下のような"OAuth Tokens for Your Workspace"という箇所があるため、そこから"Install to Workspace"を選択します。

ボタンを押すと以下のように確認画面が出てくるので、"Allow"をクリックします。

すると、元の画面に戻り、今度は先ほどの箇所に"Bot User OAuth Token"が生成されているかと思います。

このトークン情報は大切で、これが第三者にバレると、チャンネルの情報を第三者が閲覧できたり...等されるため、管理には気をつけてください。

トークンは上記にある通り、現時点では"xoxb-"といった具合で始まっているようです。

ここまで用意したら一旦SlackAPIはOKです!

さて、そしたら今度はslackに移ります。

まず簡単な検証用のチャンネルを作成し、適当に投稿をします。

今回は上記のような感じにしてみました。

そして、slackの左下にあるAppsから、先ほどインストールしたchat-analyzerを追加します。

もしAppsにない場合は、"Add apps"から追加してください。

追加されていることを確認できたら、対象のチャンネルの"Apps"にも追加していきます。

チャット画面の左上の"# chat-analyze-test"をクリックすると、以下のような画面が出てくるので、"Integrations"から"Add an App"で追加します。

押すと、以下のようにAppを選択できる画面になるので先ほど作ったものを探して"Add"を押します。

すると...

下記のように追加されるはずです。

これで事前準備はOKです!

あとは最後に、チャンネルのIDとワークスペースのドメインを調べておきます。

チャンネルIDは先ほどテスト用のチャンネルにAppを追加するときに開いた画面にあります。

一番下に"ChannelID"とありますね。これです。

この情報はコピペしておきます。

また、Slackの左上にあるワークスペース名をクリックすると、slackのドメインも確認することができます。

ここの"{ここをメモ}.slack.com"の箇所をメモしておきます。

次はやっとこさプログラムを書いていきましょう!

Pythonで情報を取得して、csvへの書き出しを行う

さて、先にプログラムをぺたっとしておきます。

事前に設定しておくべきは...

client = WebClient(token='xoxb-...あなたのBot tokenで置き換え...')
channel_id = 'C062TNB786B'
workspace_name = 'takapontest'

上記の箇所です。

ここだけ各自の情報に置き換えてください。


import os
import csv
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import time
from datetime import datetime, timezone, timedelta
import sys

client = WebClient(token='xoxb-...あなたのBot tokenで置き換え...')
channel_id = 'C062TNB786B'
workspace_name = 'takapontest'

def fetch_messages(channel_id, from_date):
    try:
        # 日付をUNIXタイムスタンプに変換
        oldest = from_date.timestamp()

        # @see: https://api.slack.com/methods/conversations.history
        result = client.conversations_history(channel=channel_id, oldest=oldest, limit=1000)
        print(result)
        messages = result['messages']

        data = []
        for message in messages:
            user_id = message.get('user')
            timestamp = float(message['ts'])
            text = message['text']
            thread_ts = message.get('thread_ts')
            reactions = message.get('reactions', [])
            
            # 投稿者名を取得 @see: https://api.slack.com/methods/users.info
            user_info = client.users_info(user=user_id)
            user_name = user_info['user']['real_name']

            # 投稿日時を変換
            dt = datetime.fromtimestamp(timestamp, timezone.utc).astimezone()
            formatted_date = dt.strftime('%Y-%m-%d %H:%M:%S')

            # メッセージへのリンクを生成
            link = f"https://{workspace_name}.slack.com/archives/{channel_id}/p{str(timestamp).replace('.', '')}"

            data.append([
                user_name,
                formatted_date,
                text,
                bool(thread_ts),
                bool(reactions),
                link
            ])
        
        return data

    except SlackApiError as e:
        print(f"Error fetching conversations: {e.response['error']}")
        return []

def write_csv(data, filename):
    with open(filename, mode='w', newline='', encoding='utf-8') as file:
        writer = csv.writer(file)
        writer.writerow(["投稿者名", "投稿日時", "本文", "スレッド有無", "スタンプ有無", "元チャットへのリンク"])
        for row in data:
            writer.writerow(row)

# 引数は"2023-10-01"等、"YYYY-MM-DD"のみ受け付ける
if __name__ == "__main__":
    if len(sys.argv) == 2:
        from_date_str = sys.argv[1]
        from_date = datetime.strptime(from_date_str, '%Y-%m-%d').replace(tzinfo=timezone.utc)
    else:
        # 日付指定がない場合は30日前から最新のものまで取得
        from_date = datetime.now(timezone.utc) - timedelta(days=30)

    messages = fetch_messages(channel_id, from_date)
    write_csv(messages, "slack_messages.csv")
    print('slack_messages.csv was successfully created.')

もしslackのsdkが入っていない場合は以下でインストールも必要になります。

pip install slack-sdk

使い方としては以下の通りです。

taka@Taka python % python3 slack_csv.py
slack_messages.csv was successfully created.
taka@Taka python % cat slack_messages.csv
投稿者名,投稿日時,本文,スレッド有無,スタンプ有無,元チャットへのリンク
chat-analyzer,2023-10-29 01:01:24,<@U063B13CTLL> has joined the channel,False,False,https://takapontest.slack.com/archives/C062TNB786B/p1698508884371239
takapon,2023-10-28 20:25:51,スタンプありチャットその一,False,True,https://takapontest.slack.com/archives/C062TNB786B/p1698492351845559
takapon,2023-10-28 20:25:28,すれっどありパターンのテストチャット,True,False,https://takapontest.slack.com/archives/C062TNB786B/p1698492328132519
takapon,2023-10-28 20:25:14,分析用テストちゃっとその1,False,False,https://takapontest.slack.com/archives/C062TNB786B/p1698492314741979
takapon,2023-10-28 20:24:33,<@U01FG389ZAQ> has joined the channel,False,False,https://takapontest.slack.com/archives/C062TNB786B/p1698492273052189
taka@Taka python %

"slack_csv.py"として先ほどのコードを保存して実行をすると、同じディレクトリに"slack_messages.csv"として書き出されます。

みていただければわかりますが、csvでカンマ区切りで書き出されているので、このファイルをspreadsheetやnumbersなどでひらけば...

このようになります。

また、今回は引数として"2023-10-01"といった形式で入力をされると、その期間の情報のみに絞ることができるように設定しています。

もし引数がない場合はい30日前から現在までです。

上記に関しては日付経過が必要なので一旦検証は省いていますが...おそらく動くはずです...!

今度数日後また確認しようと思います。

さて、次にプログラムについて少しだけ解説を残して終わりたいと思います!

プログラムの解説

では実行から〜の流れに沿って簡単に解説をしていきます。

まず実行をされると以下の箇所が実行されます。

# 引数は"2023-01-01"等、"YYYY-MM-DD"のみ受け付ける
if __name__ == "__main__":
    if len(sys.argv) == 2:
        from_date_str = sys.argv[1]
        from_date = datetime.strptime(from_date_str, '%Y-%m-%d').replace(tzinfo=timezone.utc)
    else:
        # 日付指定がない場合は30日前から最新のものまで取得
        from_date = datetime.now(timezone.utc) - timedelta(days=30)

    messages = fetch_messages(channel_id, from_date)
    write_csv(messages, "slack_messages.csv")
    print('slack_messages.csv was successfully created.')

もし引数が"実行ファイル名+引数)"で2であれば、入力された日付をdatetimeに変換してfrom_dateに格納しています。

もし引数がなければ、現在から30日分引いた日付を設定します。

そしてその値を用いてfetch_messagesメソッドの処理に移ります。

def fetch_messages(channel_id, from_date):
    try:
        # 日付をUNIXタイムスタンプに変換
        oldest = from_date.timestamp()

        # @see: https://api.slack.com/methods/conversations.history
        result = client.conversations_history(channel=channel_id, oldest=oldest, limit=1000)

        messages = result['messages']

        data = []
        for message in messages:
            user_id = message.get('user')
            timestamp = float(message['ts'])
            text = message['text']
            thread_ts = message.get('thread_ts')
            reactions = message.get('reactions', [])
            
            # 投稿者名を取得 @see: https://api.slack.com/methods/users.info
            user_info = client.users_info(user=user_id)
            user_name = user_info['user']['real_name']

            # 投稿日時を変換
            dt = datetime.fromtimestamp(timestamp, timezone.utc).astimezone()
            formatted_date = dt.strftime('%Y-%m-%d %H:%M:%S')

            # メッセージへのリンクを生成
            link = f"https://{workspace_name}.slack.com/archives/{channel_id}/p{str(timestamp).replace('.', '')}"

            data.append([
                user_name,
                formatted_date,
                text,
                bool(thread_ts),
                bool(reactions),
                link
            ])
        
        return data

    except SlackApiError as e:
        print(f"Error fetching conversations: {e.response['error']}")
        return []

こちらはちょっと長いのですが...かいつまんで説明していきます。

以下でAPIでメッセージを一括で取得します。

result = client.conversations_history(channel=channel_id, oldest=oldest, limit=1000)

そして、APIの結果から、messagesを取得し、forを回しています。

for message in messages:

あとは、messageから必要な情報(投稿者のIDや投稿日時、本文、スレッドのタイムスタンプやリアクション)を取得し、それぞれ整形や、idから投稿者名を取得してdataに累積させていき、全て終わったらリターンします。

try部で失敗したらSlackのAPIエラーを表示する例外処理もいれています。

この処理が終わったら今度は"write_csv"メソッドが呼ばれます。

def write_csv(data, filename):
    with open(filename, mode='w', newline='', encoding='utf-8') as file:
        writer = csv.writer(file)
        writer.writerow(["投稿者名", "投稿日時", "本文", "スレッド有無", "スタンプ有無", "元チャットへのリンク"])
        for row in data:
            writer.writerow(row)

こちらはあまり説明は不要かなと思いますが...

writerowでヘッダー部を追加、その後先ほど整形した書き込むデータも同じく書き込んでいます。

一通り書き込みが終わったら、完了...といった流れですね!

まとめ

さて、今回はpythonとslackAPIを用いて、Slackのチャンネルからデータを取得し、csvに書き出す...といったことをしてみました!

実はお仕事でそういうことちょっとしてみたいなぁと思うことがあってお試しに触ってみた感じなのですが、来週は実際に実務でも試してみようと思います。

Slackに投稿された内容を分析することはあまりないかもですが、Slackのユースケース分析とかもできると、コミュニケーションの課題とかもいい感じに探せそうな気がします。

本格的にガッツリ実行はあまり想定してないのでアレですが...(トークンベタ貼りとかイケテナイ。)

localでちょっと一回やってみたいなぁとか、その程度であれば全然有効に使えそうなので、引き続き色々試してみ用と思います。

それでわ!

おすすめの記事