어떤 이유로 slack로부터 약 1년간의 메시지 이력을 취득했으므로 Python에서의 구현 방법을 써 둡니다.
취득한 메세지를 분석하기 쉽도록 리포맷하거나 했습니다만, 여러가지 좋지 않으므로 전 공개는 할 수 없습니다. 공개할 수 있을 것 같은 부분이 있으면 또 기사를 쓰고 싶습니다.

Python에는 slackapi/python-slackclient이 있지만 이번에는 사용하지 않습니다. python-slackclient를 사용한 구현 방법을 알고 싶다면이 기사 이외의 내용을 읽는 것이 좋습니다.

환경

언어

Python3.8

주요 사용 라이브러리

requests==2.23.0

개발 보조 라이브러리

Pipenv

mypy

black

flake8

구현

클라이언트

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 방법은 conversations.history입니다. 요청 매개 변수에 대한 자세한 내용은 참조를 읽으십시오.
아래와 같이 파라미터를 코드에 떨어뜨리는 것으로 method로 리퀘스트 가능한 파라미터를 파악하기 쉽게 했습니다. 적절한 코멘트를 추가하면 코드가 훌륭한 참조가 될 수 있습니다.
일단, 1년간의 이력을 취득하는데 있어서 중요한 파라미터를 설명해 둡니다. 그들은 cursor와 oldest입니다. 커서는 재귀적으로 히스토리를 얻는 다음 토큰입니다. 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