Slackで参加している全チャンネルのログをcsvで保存する

背景

slackの規約改定に伴い，2022年9月より90日間を超えた記録はフリープランだとアクセス出来なくなる．
ここに記録を残しておくためのプログラムを書いたので記録として残しておく．

目的

新しい規約が適用される前にslackでのデータやり取りをslack apiで出力し，保存する．

手順

Slack APIで新規Appを作成する
Scopeにchannels:history channels:read groups:history groups:read users:readを加える．
作成したAppのTokenをコピーしておく

実行ファイル

import numpy as np
import pandas as pd
from pandas.io.json import json_normalize
import json
import datetime
import requests

url_channel_list = "https://slack.com/api/conversations.list"
url = "https://slack.com/api/conversations.history"
url_user = "https://slack.com/api/users.info?user="
url_reply = "https://slack.com/api/conversations.replies"

token="xoxp-...." #your logger app token
header={"Authorization": "Bearer {}".format(token)}
payload_list  = {"types" : "public_channel,private_channel",}

# チャンネル一覧の取得 > df_list
res_list = requests.get(url_channel_list, headers=header,params=payload_list)
df_list = pd.json_normalize(res_list.json()["channels"])
df_list.to_csv("df_list.csv", encoding='UTF-8')

# チャンネルごとに会話を保存
for channel, cname in zip(df_list["id"], df_list["name"]):
    print(channel, cname)
    #channel設定
    payload  = {
    "channel" : channel,
    "ts" : "1622007981.001400",
    "limit": "1000",
    }
    payload_user  = {
        "channel" : channel,
        }
    
    # チャンネル情報読み込み
    res = requests.get(url, headers=header, params=payload)
    res_json = res.json()
    # 最初のページのデータフレーム化, しかしこのdfには返信の内容が情報が入っていない．
    df = pd.json_normalize(res_json["messages"])
    
    # pagingされているのであれば，それも全て取りに行きdfに加える
    # 次のページがあるか否かはres_json["has_more"]に情報が入っている．
    # next_cursorを取ってきてpayloadを更新しながらdfを行方向に結合し続ける
    while res_json["has_more"]:
        next_cursor = res_json['response_metadata']['next_cursor']
        payload["cursor"] = next_cursor
        res = requests.get(url, headers=header, params=payload)
        res_json = res.json()
        df_tmp = pd.json_normalize(res_json["messages"])
        df = pd.concat([df,df_tmp])
    
    if (df.shape[0] == 0):
        print("No messages were detected.") # dfが空の時はログを保存せずエラー回避
    else:
    # 投稿時間tsを参照して，apiで返信のツリーを取りに行く > df_reply
        df_reply = pd.DataFrame()
        for i in range(df.shape[0]):
            payload_tmp  = {
                "channel" : channel,
                "ts": df.iloc[i,:]["ts"]
                }
            res_reply = requests.get(url_reply, headers=header, params=payload_tmp)
            df_reply_tmp = pd.json_normalize(res_reply.json()["messages"])
            if i == 0:
                df_reply = df_reply_tmp
            else:
                df_reply = pd.concat([df_reply,df_reply_tmp])

        if (df_reply.shape[0] == 0):
            print("No messages were detected.") # df_replyが空にはなり得ないが念の為
        else:
            # useridをname(slack上の表示名)に置換する
            if "reply_users" in df_reply.columns:
                df_reply["reply_users"] = df_reply["reply_users"].apply(lambda x: ', '.join(sorted(x)) if type(x) == list else "")
            # 対応表の作成
            usernames=df_reply["user"].unique()
            realnames=[]
            for username in usernames:
                res_user = requests.get(url_user+username+"&pretty=1", headers=header, params=payload_user)
                realname = res_user.json()["user"]["real_name"]
                realnames.append(realname)
            # 置換
            for username, realname in zip(usernames, realnames):
                df_reply=df_reply.replace(username, realname, regex=True)

            # タイムスタンプをUNIX時間から人間が読みやすい時間に変更
            tscols = ["ts", "thread_ts", "latest_reply", "edited.ts"]
            for tscol in tscols:
                if tscol in df_reply.columns:
                    df_reply[tscol]=df_reply[tscol].astype(float).apply(lambda x: datetime.datetime.fromtimestamp(x) if ~np.isnan(x) else np.nan)

            # 保存
            df_reply.to_csv("slack_log_"+cname+".csv", encoding='UTF-8')

今回の方法で保存できるのはメッセージのみ．添付ファイル等は保存できない．
時間が出来た時に定期実行に対応出来るよう改変したい．