背景
slackの無料プランで過去3か月までの履歴しか見れなくなってしまったので、3か月以前のメッセージを残すためにコードを書きました。
コードの中身まで見るのが面倒くさい方は、SlackAPIの作り方とワークスペースに追加する方法まで見て、githubのコードを動かしてください。
完成品
https://github.com/shalXXXX/slack_message_log
以下のようなslackのメッセージをcsvに出力できます。
環境
- ubuntu 22.04
- python 3.7.13
- conda 4.12.0
必要なライブラリ
- requests 2.27.1
- pandas 1.3.5
- tqdm 4.64.0
slack API
Slack APIを作って、権限を追加し、ワークスペースに追加するまでの手順を記載する。
APIの作成
-
Slack APIサイトにアクセスし、create an appをクリック
- From scratchをクリック
- App Nameに適当な名前を入れて、Botを入れるワークスペースを選択し、Create Appを押す
権限の追加
左メニュー中のOAuth&Permissionsを選択し、下の方に行くとScopesという項目があるので、以下のスコープを追加する。
- channels:history
- channels:read
- groups:read
- im:history
- mpim:history
- users:read
スコープの追加が終わったら、上のOauth Tokens for Your Workspaceという項目の中にInstall to Workspaceというボタンがあるので、クリックしてBot User OAuth Tokenをコピーする。
このトークンを使ってAPIの認証を行う。
チャンネルにBotを追加する
slackのチャンネル名を右クリックして、チャンネル詳細を表示するをクリックする。
インテグレーションという項目があるので、クリックすると以下のような画面になるので、botを追加する。
※各チャンネル1つ1つにBotを追加しないと正しく動作しません。
APIの挙動を理解する
以下が今回使用するAPIです。
URL | 動作 |
---|---|
https://slack.com/api/users.list | ユーザ一覧を表示する |
https://slack.com/api/conversations.list | チャンネル一覧を取得する。 |
https://slack.com/api/conversations.history | メッセージ履歴を取得する。取得件数はデフォルトで200件で、上限は1000件。 |
https://slack.com/api/conversations.replies | リプライを表示する。 |
チャンネル一覧の取得
pythonのrequestsモジュールを使って、APIにGETリクエストを送る。
試しにconversations.list
にGETリクエストを送ってチャンネル一覧を取得してみる。
url = "https://slack.com/api/conversations.list"
token = 'XXXXXXXXXXXXXX'
headers = {
"Authorization": "Bearer {}".format(token)
}
res = requests.get(url, headers=headers)
print(res.json())
上記を実行すると、以下のようなレスポンスが返ってくる。
{'ok': True,
'channels': [{'id': 'C045RG56154',
'name': 'random',
'is_channel': True,
'is_group': False,
'is_im': False,
'is_mpim': False,
'is_private': False,
'created': 1665464104,
'is_archived': False,
'is_general': False,
'unlinked': 0,
'name_normalized': 'random',
'is_shared': False,
'is_org_shared': False,
'is_pending_ext_shared': False,
'pending_shared': [],
'context_team_id': 'T046MRBL9JL',
'parent_conversation': None,
'creator': 'U046APL83EV',
'is_ext_shared': False,
'shared_team_ids': ['T046MRBL9JL'],
'pending_connected_team_ids': [],
'is_member': False,
'topic': {'value': '', 'creator': '', 'last_set': 0},
...
'creator': 'U046APL83EV',
'last_set': 1665464142},
'previous_names': [],
'num_members': 3}],
'response_metadata': {'next_cursor': ''}}
チャンネルのメッセージ履歴の取得
チャンネルのメッセージ履歴を取得したい場合、上記レスポンスのid
をconversations.history
に投げる。
url = "https://slack.com/api/conversations.list"
token = 'XXXXXXXXXXXXXX'
headers = {
"Authorization": "Bearer {}".format(token)
}
payload = {
"channel": "C046APMSU81",
"limit": 1000
}
res = requests.get(url, headers=headers, params=payload)
このAPIを使うと、user情報やメンションの際の名前がIDとなってしまうので、users.list
のAPIと組み合わせて使用する必要がある。
{'ok': True,
'messages': [{'type': 'message',
'subtype': 'channel_join',
'ts': '1666148052.637859',
'user': 'U0473U7E9PV',
'text': '<@U0473U7E9PV>さんがチャンネルに参加しました',
'inviter': 'U046APL83EV'},
{'client_msg_id': 'b01b6a4a-aea0-4f51-a4ec-cf7f41b5cdba',
'type': 'message',
'text': 'テストメッセージ',
'user': 'U046APL83EV',
'ts': '1665543059.274109',
'blocks': [{'type': 'rich_text',
'block_id': 'FJnIx',
'elements': [{'type': 'rich_text_section',
'elements': [{'type': 'text', 'text': 'テストメッセージ'}]}]}],
'team': 'T046MRBL9JL'},
{'client_msg_id': '97da5b9b-31be-4742-98e2-91f04e0b59b6',
'type': 'message',
'text': 'aa',
'user': 'U046APL83EV',
'ts': '1665479545.353989',
'blocks': [{'type': 'rich_text',
'block_id': '/yu',
'elements': [{'type': 'rich_text_section',
...
コードの説明
今回のコードで使っている2つの関数の説明を記載します。
関数名 | 動作 |
---|---|
get_response() | APIにGETリクエストを投げてレスポンスを返す関数 |
get_messages() | 受け取ったレスポンスを整形してcsvに出力する関数 |
get_response()
GETリクエストを送る度に似たようなコードを書くのは面倒なので、関数化しました。
引数にはapiの名前とpayloadを受け取り、payloadがある場合とない場合でrequests.get()
の引数が変わるだけの簡単な関数です。
def get_reponse(api_url, payload=None):
BASE_URL = "https://slack.com/api/"
token = 'XXXXXXXXXXXXXX'
headers = {
"Authorization": "Bearer {}".format(token)
}
url = BASE_URL + api_url
if payload:
res = requests.get(url, headers=headers, params=payload)
return res.json()
else:
res = requests.get(url, headers=headers)
return res.json()
get_messages()
受け取ったレスポンスを整形してcsvに出力する関数です。
引数には以下を受け取ります。
- channel_id : conversations.listで受け取ったチャンネルのid
- channel_name : conversations.listで受け取ったチャンネルの名前
- users_dict : ユーザIDとユーザ名が格納された辞書型の変数。users.listで取得。
動作としては、
conversations.history
で取得したメッセージから
- ts:スレッドID
- user:ユーザID
- text:メッセージ
- reply_count:リプライの数
を取り出して、reply_count
が1以上である場合conversations.replies
からそのスレッドIDのリプライを取得し、0である場合はconversations.history
から必要な情報をデータフレームに追加していき、userIDをちゃんとした名前に置き換える。
だけの簡単な動作です。
def get_messages(channel_id, channel_name, users_dict):
api_url = "conversations.history"
payload = {
"channel": channel_id,
"limit": 1000
}
message_logs = get_reponse(api_url, payload=payload)
message_df = pd.DataFrame(columns=["thread", "user", "message", "reply_count", "type"])
reply_api_url = "conversations.replies"
for item in tqdm(message_logs["messages"]):
msg = item["text"]
sender = item["user"]
for user in users_dict.keys():
if user in msg:
msg = msg.replace(user, users_dict[user])
sender = sender.replace(sender, users_dict[sender])
if "reply_count" in item:
payload = {
"channel": channel_id,
"ts": item["ts"]
}
replies = get_reponse(reply_api_url, payload=payload)
reply_count = item["reply_count"]
reply_df = pd.DataFrame(columns=["thread", "user", "message", "reply_count", "type"])
for reply in replies["messages"]:
rep = reply["text"]
user_name = reply["user"]
for user in users_dict.keys():
if user in rep:
rep = rep.replace(user, users_dict[user])
user_name = user_name.replace(user_name, users_dict[user_name])
reply_df = reply_df.append({"thread": reply["thread_ts"], "user": user_name, "message": rep,"reply_count": 0, "type": "reply"}, ignore_index=True)
reply_df = reply_df.drop(reply_df.index[0])
reply_df = reply_df.iloc[::-1]
message_df = pd.concat([message_df, reply_df])
else:
reply_count = 0
message_df = message_df.append({"thread": item["ts"], "user": sender, "message": msg,"reply_count": reply_count, "type": "message"}, ignore_index=True)
message_df_sorted = message_df[::-1]
new_df = message_df_sorted[["user", "message", "reply_count", "type", "thread"]]
output_path = "output/" + channel_name + "_message_history.csv"
new_df.to_csv(output_path, encoding="utf-8", index=False)
main()
メイン関数では、userID、username、channel_listを取得して取得したいチャンネルのidをget_messages()
に投げるだけす。
if __name__ == "__main__":
user_url = "users.list"
users = get_reponse(user_url)
users_dict = {}
for user in users["members"]:
users_dict[user["id"]] = user["name"]
channels_url = "conversations.list"
channels = get_reponse(channels_url)
channel_list = ["取得したいチャンネル"]
for channel in channels["channels"]:
if not channel["name"] in channel_list:
continue
print(channel["id"], channel["name"])
get_messages(channel["id"], channel["name"], users_dict)
最後に
データフレーム操作周りがもう少しスムーズにできるんじゃないかなと思っています。
slack無料プランで3か月以前のメッセージを残したい方はぜひ使ってみてください。