背景

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を追加する。

※各チャンネル１つ１つに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か月以前のメッセージを残したい方はぜひ使ってみてください。