0
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

Slack無料プランで3か月前までの会話しか見えなくなるのをAPIを使って何とかする

Posted at

背景

slackの無料プランで過去3か月までの履歴しか見れなくなってしまったので、3か月以前のメッセージを残すためにコードを書きました。
コードの中身まで見るのが面倒くさい方は、SlackAPIの作り方とワークスペースに追加する方法まで見て、githubのコードを動かしてください。

完成品

https://github.com/shalXXXX/slack_message_log
以下のようなslackのメッセージをcsvに出力できます。
image.png

image.png

環境

  • 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の作成

  1. Slack APIサイトにアクセスし、create an appをクリック
    image.png
  2. From scratchをクリック
    image.png
  3. App Nameに適当な名前を入れて、Botを入れるワークスペースを選択し、Create Appを押す
    image.png

権限の追加

左メニュー中の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を追加する。
image.png

※各チャンネル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': ''}}

チャンネルのメッセージ履歴の取得

チャンネルのメッセージ履歴を取得したい場合、上記レスポンスのidconversations.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か月以前のメッセージを残したい方はぜひ使ってみてください。

0
2
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
0
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?