[python] [slack] files.upload API 廃止に伴う対応 requestsモジュール編

こんにちは。
マイクロアドのタカギです。

slackのfiles.upload API が廃止になるとのことで、対応が必要になるため、サンプルコードを作ってみました。
公式SDKを使用する場合はfiles.upload V2を使用すれば問題ないようですが、今回もrequestsモジュールを使用してSlack通知してみようと思います。

https://qiita.com/seratch/items/971980506da92ea6c8db

# やること
前回と同じことをやります。
- slackにメッセージを投稿する
- 投稿したメッセージのスレッドにファイルを投稿する

※前回

https://qiita.com/mtakagi555/items/02c24ff83988827330c0

# 準備
まずslackのトークンを環境変数にセットします。
これでうっかりトークンがインターネットに公開されてしまうリスクを減らします。

```bash
export SLACK_BOT_TOKEN="xxxxx"
```

# サンプルコード
省略していますが、mypyとかも使っています。

```.python-version
3.11.7
```

```pyproject.toml
[tool.poetry]
package-mode = false

[tool.poetry.dependencies]
python = "3.11.7"
requests = "2.31.0"
```

```app.py
import json
import logging
import os
import uuid

import requests  # type: ignore

LOGGER = logging.getLogger('app')
LOGGER.setLevel('INFO')
logging.basicConfig(level='INFO', format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S')

SLACK_BOT_TOKEN = os.environ.get('SLACK_BOT_TOKEN', '')
channel_id = 'xxx'


def main() -> None:
    """
    slack通知のサンプルコード
    """
    slack_msg = 'slackメッセージテスト'

    file_title = 'ファイルタイトルテスト'

    file_content = '''
    ファイル内容
    改行も含みます
    日本語も含みます
    '''

    # slackにメッセージを投稿する
    response_json = send_message(slack_msg, channel_id, SLACK_BOT_TOKEN)

    # メッセージ通知したスレッドにファイル形式で通知
    thread_ts = response_json['ts']
    send_file_content_on_message_thread(file_title, file_content, channel_id, thread_ts, SLACK_BOT_TOKEN)


def send_message(msg: str, channel_id: str, token: str) -> dict:
    """
    slackにメッセージを投稿する

    Args:
        msg: 通知メッセージ
        channel_id: slackのチャンネルID
        token: slackトークン
    Returns:
        APIのレスポンス情報
    """
    LOGGER.info('send message to slack.')
    url = 'https://slack.com/api/chat.postMessage'
    header = {'Authorization': f'Bearer {token}'}
    data = {'channel': channel_id, 'text': msg}

    response = requests.post(url, headers=header, data=data)
    response.raise_for_status()
    if not response.ok:
        raise Exception('failed to post messeage to slack.')

    response_json = response.json()
    LOGGER.info(response_json)
    return response_json


def send_file_content_on_message_thread(
    file_title: str, file_content: str, channel_id: str, thread_ts: str, token: str
) -> None:
    """
    slackの指定スレッドにファイルを投稿する

    Args:
        file_title: ファイルタイトル
        file_content: ファイル内容
        channel_id: slackのチャンネルID
        thread_ts: スレッドのタイムスタンプ
        token: slackトークン
    """
    # ファイル名はランダムな文字列
    file_name = f'{str(uuid.uuid4())}.txt'
    # 投稿するファイル内容のサイズを取得する
    length = len(file_content.encode(encoding='utf-8'))
    # ファイルアップロードURLとファイルIDを取得する
    upload_url, file_id = _get_upload_url_and_file_id(file_name, length, token)
    # ファイルをアップロードする
    _upload_file_content(upload_url, file_content)
    # ファイルアップロードを完了する
    _complete_upload(file_id, file_title, channel_id, thread_ts, token)


def _get_upload_url_and_file_id(filename: str, length: int, token: str) -> tuple[str, str]:
    """
    ファイルアップロードURLとファイルIDを取得する

    Args:
        filename: アップロードファイル名
        length: アップロードファイルサイズ
        token: slackトークン
    Returns:
        APIのレスポンス情報
        ファイルアップロードURLとファイルIDのタプル
    """
    LOGGER.info('get upload url from slack.')
    url = 'https://slack.com/api/files.getUploadURLExternal'
    header = {'Authorization': f'Bearer {token}'}
    params = {'filename': filename, 'length': length}

    response = requests.get(url, headers=header, params=params)
    response.raise_for_status()
    if not response.ok:
        raise Exception('failed to get upload url from slack')

    response_json = response.json()
    LOGGER.info(response_json)
    return response_json['upload_url'], response_json['file_id']


def _upload_file_content(upload_url: str, file_content: str) -> None:
    """
    ファイルをアップロードする

    Args:
        upload_url: ファイルアップロードURL
        file_content: ファイル内容
    """
    LOGGER.info('upload file to slack.')
    data = file_content.encode(encoding='utf-8')

    response = requests.post(upload_url, data=data)
    response.raise_for_status()
    if not response.ok:
        raise Exception('failed to upload file to slack.')


def _complete_upload(file_id: str, title: str, channel_id: str, thread_ts: str, token: str) -> None:
    """
    ファイルアップロードを完了する

    Args:
        file_id: ファイルID
        title: ファイルのタイトル
        channel_id: slackのチャンネルID
        thread_ts: スレッドのタイムスタンプ
        token: slackトークン
    """
    LOGGER.info("complete upload.")
    url = 'https://slack.com/api/files.completeUploadExternal'
    header = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json; charset=utf-8'}
    data = {'files': [{'id': file_id, 'title': title}], 'channel_id': channel_id, 'thread_ts': thread_ts}

    response = requests.post(url, headers=header, data=json.dumps(data))
    response.raise_for_status()
    if not response.ok:
        raise Exception('failed to complete upload.')


if __name__ == '__main__':
    main()

```

# 環境構築
poetryを使うとスムーズですね。
```bash
poetry install
poetry shell
```

# 実行
いざ！
```bash
python app.py
```

# 実行結果
メッセージが通知されて、
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/870283/421c1f6a-1f64-9672-e803-9541d4706755.png)

スレッドにも通知されました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/870283/3149f150-7aa1-7210-9f57-053b4e9b73b9.png)

# まとめ
requestsモジュールを使用して、files.upload API を使用せずに、slackにメッセージ投稿とファイル投稿ができました。
お疲れ様でした。

# 参考
- https://zenn.dev/y_matsubara/articles/slack-post-file-202405