とある理由でslackから約1年間のメッセージ履歴を取得したのでPythonでの実装方法を書いておきます。
取得したメッセージを分析しやすいようにリフォーマットしたりしたのですが、いろいろまずいので全公開はできません。公開できそうな部分があればまた記事を書きたいと思います。

Pythonにはslackapi/python-slackclientがありますが、今回は使用していません。python-slackclientを使った実装方法を知りたい方は本記事以外を読むことをオススメします。

環境

言語

Python3.8

主な使用ライブラリ

requests==2.23.0

開発補助用ライブラリ

Pipenv
mypy
black
flake8

実装

Client

slackのtokenは環境変数から取得、またはmainスクリプトに直接記述できるようにインスタンス変数になっています。pipenvを使っていれば自動的に.envを読み取ってくれるので環境変数にセットしたものをデフォルト値にしています。自分の開発環境に依存した実装ですが、pipenvを使っていない(環境変数にセットしたくない)ケースにも対応させました。
request関数の引数にmethod: BaseSlackMethodとしてありますが、これはslackの場合各APIエンドポイントをmethodと呼称しているのでそうしました。BaseSlackMethodの実装はあとで説明しますが、BaseSlackMethodを基底クラスにしてmethod用のクラスを増やせるようにしました。そうすることでリクエストのパラメータをコードで管理可能にしました。リファレンスをいちいち見にいく手間が省けます。やったね！

src/slack/client.py

import os
from dataclasses import dataclass
from typing import Any, ClassVar, Dict

import requests

from src.log import get_logger
from src.slack.exceptions import SlackRequestError
from src.slack.types import Headers
from src.slack.methods.base import BaseSlackMethod


SLACK_API_TOKEN = os.getenv("SLACK_API_TOKEN", "")

logger = get_logger(__name__)


@dataclass
class SlackClient:
    api_url: ClassVar[str] = "https://slack.com/api"
    token: str = SLACK_API_TOKEN

    def _get_headers(self, headers: Headers) -> Headers:
        """Get headers

        Args:
            headers (Headers)

        Returns:
            Headers
        """

        final_headers = {
            "Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
        }

        if self.token:
            final_headers["Authorization"] = f"Bearer {self.token}"

        final_headers.update(headers)

        return final_headers

    def request(
        self, method: BaseSlackMethod, headers: Dict[str, Any] = None,
    ) -> Dict[Any, Any]:
        """SlackへAPIリクエスト

        Args:
            method (BaseSlackMethod)
            headers (Dict[str, Any], optional): Defaults to None.

        Raises:
            SlackRequestError
            err

        Returns:
            Dict[Any, Any]: response body
        """

        if not isinstance(headers, dict):
            headers = {}
        headers = self._get_headers(headers)

        url = f"{self.api_url}/{method.endpoint}"

        try:
            res = requests.get(url, headers=headers, params=method.params)

            if res.ok is False:
                raise SlackRequestError(res.text)
        except Exception as err:
            logger.error(err)
            logger.error("slackからデータ取得失敗")
            raise err
        else:
            logger.info("slackからデータ取得完了")
            return res.json()

メッセージ履歴method

メッセージ履歴を取得するAPI methodはconversations.historyです。リクエストのパラメータの詳細はリファレンスを読んでください。
以下のようにパラメータをコードに落とし込むことでmethodでリクエスト可能なパラメータを把握しやすくしました。適宜コメントを加えることでコードが立派なリファレンスにもなり得ます。
一応、1年間の履歴を取得する上で重要なパラメータを説明しておきます。それらはcursorとoldestです。cursorは再帰的に履歴を取得するためのネクストトークンです。oldestは一般的な意味の通りい履歴の開始日時をを指定します。注意する点はoldestで指定できるのがUnix Timestampというぐらいですかね。

src/slack/methods/conversation.py

import os
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import ClassVar, Optional

from src.slack.types import SlackParams


SLACK_CHANNEL_ID = os.getenv("SLACK_CHANNEL_ID", "")


@dataclass
class ConversationsHistory:
    endpoint: ClassVar[str] = "conversations.history"

    channel: str = SLACK_CHANNEL_ID
    cursor: Optional[str] = None
    inclusive: bool = False
    limit: int = 100
    latest: float = datetime.now().timestamp()
    oldest: float = 0

    @property
    def params(self) -> SlackParams:
        self_dict = asdict(self)

        if self.cursor is None:
            del self_dict["cursor"]

        return self_dict

が基底クラスです。

src/slack/methods/base.py

from dataclasses import dataclass, asdict
from typing import ClassVar

from src.slack.types import SlackParams


@dataclass
class BaseSlackMethod:
    endpoint: ClassVar[str] = ""

    @property
    def params(self) -> SlackParams:
        return asdict(self)

mainスクリプト

1年間の履歴を取得したいので、datetime.now() - timedelta(days=365)の計算式を使用して1年前の日時を算出します。マイナスをプラスに変えれば1年後の日時も算出できるのでtimedeltaは便利です。ありがたや~~
あと1年間の履歴となると再帰的に取得しないといけないので今回は単純なwhileループを採用しました。ぶっちゃけ使い捨てのスクリプトなのでnext_cursorがあるか丁寧にif文を実装しなくても良かったんですが、KeyErrorで終わるのが気持ち悪かったのでそうしました。

src/slack/__main__.py

from datetime import datetime, timedelta

from src.utils import save_to_file
from src.slack.client import SlackClient
from src.slack.methods.conversation import ConversationsHistory


def main() -> None:
    tmp_oldest = datetime.now() - timedelta(days=365)
    oldest = tmp_oldest.timestamp()
    method = ConversationsHistory(inclusive=True, oldest=oldest)
    client = SlackClient()

    count = 1

    while True:
        res = client.request(method)
        save_to_file(res, f"outputs/tests/sample{count}.json")

        if (
            "response_metadata" in res
            and "next_cursor" in res["response_metadata"]
        ):
            method.cursor = res["response_metadata"]["next_cursor"]
            count += 1
        else:
            break


if __name__ == "__main__":
    main()

終わりに

1チャンネルの1年間の履歴を取得してみたら、1ファイル2000行以上が200ファイルほど作成されました。おそるべし

Reference

methods/conversations.history

【Python】Slackから1年間のメッセージ履歴を取得する

環境

言語