Pythonを使用してSlackのダイレクトメッセージやスレッドのやりとりをExcelにエクスポートする方法

この記事では、Pythonを使用してSlackのダイレクトメッセージやスレッドのやりとりをExcelにエクスポートする方法を説明します。（個人情報については注意ください。）

参考Github
https://github.com/thama125/slack-export


必要なライブラリのインポート
```    
import json
import logging
from pathlib import Path
from typing import Any, Literal
import pandas as pd
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
```
ロガーの設定
```
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
```
Slack APIクライアントの実装
```
class Client:
    def __init__(self, token) -> None:
        self._token = token
        self._headers = {"Authorization": "Bearer {}".format(token)}
        self._session = requests.Session()
        self._session.mount(
            "https://slack.com/",
            HTTPAdapter(max_retries=Retry(total=5, backoff_factor=3)),
        )
```
APIコールを行うヘルパーメソッド
```
 　　def _call(self, url, params=None) -> dict[str, Any]:
        if not params:
            params = {}

        response = self._session.get(url, headers=self._headers, params=params, timeout=3)
        response.raise_for_status()
        return response.json()

```
ユーザー情報の取得
```
    def fetch_users(self) -> list[dict[str, Any]]:
        response = self._call("https://slack.com/api/users.list")
        return response["members"]
```
ダイレクトメッセージチャンネルの取得
```
    def fetch_channels(self) -> list[dict[str, Any]]:
        response = self._call(
            "https://slack.com/api/conversations.list",
            params={"types": "im", "exclude_archived": True},
        )
        return [channel for channel in response["channels"] if channel["id"] == "ここにチャンネルID"]
```
メッセージの取得
```
    def fetch_messages(self, channel_id: str) -> list[dict[str, Any]]:
        messages = []
        next_cursor = None
        while True:
            params = {"channel": channel_id, "limit": 200}
            if next_cursor:
                params["cursor"] = next_cursor
            response = self._call("https://slack.com/api/conversations.history", params=params)
            messages += response["messages"]
            if response["has_more"]:
                next_cursor = response["response_metadata"]["next_cursor"]
            else:
                break
        return messages
```
スレッド内のメッセージの取得
```
      replies = []
        next_cursor = None
        while True:
            payload = {"channel": channel_id, "limit": 200, "ts": thread_ts}
            if next_cursor:
                payload["cursor"] = next_cursor

            response = self._call("https://slack.com/api/conversations.replies", params=payload)

            done = False
            for message in response["messages"]:
                if message["ts"] == thread_ts and len(replies) > 0:
                    done = True
                    break

                replies.append(message)

            if done:
                break
            elif response["has_more"]:
                next_cursor = response["response_metadata"]["next_cursor"]

        return replies
```
メイン関数
```
def main(
    token: str,
    output_dir: Path,
    output_format: Literal["json", "jsonl", "excel"] = "jsonl",
) -> None:
    output_dir.mkdir(parents=True, exist_ok=True)

    client = Client(token)

    logger.info("Fetching users")
    users = {user["id"]: user for user in client.fetch_users()}
    logger.info(f"{len(users)} users fetched")

    logger.info("Fetching channels")
    channels = client.fetch_channels()
    logger.info(f"{len(channels)} channels fetched")

    for channel in channels:
        channel_id = channel["id"]
        channel_name = channel.get("name") or users.get(channel.get("user"), {}).get("name", "unknown_channel")

        logger.info(f"Fetching messages: {channel_name=}")
        messages = client.fetch_messages(channel_id)

        messages_and_replies = []
        for message in reversed(messages):
            thread_ts = message.get("thread_ts")
            if not thread_ts:
                messages_and_replies.append(message)
                continue

            replies = client.fetch_replies(channel_id, thread_ts)
            messages_and_replies += replies

        logger.info(f"{len(messages_and_replies)} messages/replies fetched")

        if output_format == "excel":
            output_path = f"{output_dir / channel_name}.xlsx"
            df = pd.DataFrame(messages_and_replies)
            df.to_excel(output_path, index=False)
        else:
            output_path = f"{output_dir / channel_name}.{output_format}"
            with open(output_path, "w", encoding="utf-8") as f:
                if output_format == "json":
                    json.dump(
                        messages_and_replies,
                        f,
                        indent=4,
                        ensure_ascii=False,
                        sort_keys=True,
                    )
                elif output_format == "jsonl":
                    for message in messages_and_replies:
                        json.dump(message, f, ensure_ascii=False, sort_keys=True)
                        f.write("\n")
```
スクリプトの実行
```
main('', Path('output'), 'excel')
```