Zenn
🤖

【Python】Slackにメッセージとファイルを同時に送信

に公開
Python
自動化
Slackbot
Slack API
自動送信
tech

【Python】Slackにメッセージとファイルを同時に送信

はじめに

Slackは現代のビジネスコミュニケーションに欠かせないツールとなっています。チームでの情報共有やプロジェクト管理において、テキストメッセージだけでなくファイルの共有も頻繁に行われます。この記事では、PythonからSlack APIを使用して、メッセージとファイルを同時に送信する方法を解説します。

自動化スクリプトやバッチ処理の結果をSlackに通知したい、定期的なレポートファイルを特定のチャンネルに自動投稿したいなど、様々なユースケースで活用できること間違いなしです！

前提条件

この記事のコードを実行するためには、以下の準備が必要です：

  1. Python 3.6以上の環境
  2. 必要なパッケージのインストール：requests
  3. Slack APIトークンの取得
  4. 投稿先のSlackチャンネルID
# requestsのインストール
pip install requests

追加の機能を使用する場合は、以下のパッケージも必要になります：

# 環境変数管理用
pip install python-dotenv

# データ分析例用
pip install pandas matplotlib

Slack APIトークンの取得方法

Slack APIを利用するには、APIトークンが必要です。以下の手順で取得できます：

  1. Slack API公式サイトにアクセス
  2. 「Create New App」をクリック
  3. アプリ名とワークスペースを選択
  4. 「OAuth & Permissions」セクションで以下の権限（スコープ）を追加：
    • files:write
    • chat:write
  5. アプリをワークスペースにインストール
  6. 「Bot User OAuth Token」（xoxb-で始まるもの）をコピー

詳細はこちらの記事を参考👇️
https://qiita.com/odm_knpr0122/items/04c342ec8d9fe85e0fe9

コードの全体像

まずは完全なコードを見てみましょう。このコードは、メッセージとファイルを同時にSlackに送信する機能を提供します。

import requests
import json
import os

class Slack:
    def __init__(self, token):
        self.token = token

    def upload_file_to_slack(self, channel_id, message=None, file_name=None, file_path=None):
        """
        Slackにファイルまたはメッセージを投稿する関数
        :param channel_id: 投稿先のチャンネルID
        :param message: メッセージ（任意）
        :param file_name: ファイル名（任意）
        :param file_path: ファイルパス（任意）
        :return: API応答のJSONデータ
        :raises FileNotFoundError: ファイルが見つからない場合
        :raises ValueError: メッセージもファイルも指定されていない場合
        :raises Exception: API呼び出しが失敗した場合
        """
        if file_path:
            # ファイルがある場合
            try:
                with open(file_path, 'rb') as f:
                    file_blob = f.read()
            except Exception as e:
                raise FileNotFoundError(f"ファイルが見つかりません: {file_path}") from e
            
            file_size = len(file_blob)

            # アップロードURL取得
            params = {
                'filename': file_name,
                'length': file_size
            }
            headers = {
                'Authorization': f'Bearer {self.token}'
            }
            upload_url_response = requests.get('https://slack.com/api/files.getUploadURLExternal', params=params, headers=headers)
            upload_url_json = upload_url_response.json()

            if not upload_url_json.get('ok', False):
                raise Exception(f"アップロードURLの取得に失敗しました: {upload_url_json.get('error')}")

            upload_url = upload_url_json['upload_url']
            file_id = upload_url_json['file_id']

            # ファイルをアップロード
            upload_headers = {
                'Content-Type': 'application/octet-stream'
            }
            upload_response = requests.post(upload_url, headers=upload_headers, data=file_blob)

            if upload_response.status_code != 200:
                raise Exception(f"ファイルのアップロードに失敗しました: {upload_response.text}")

            # アップロード完了通知
            complete_upload_payload = {
                'channel_id': channel_id,
                'files': [{'id': file_id, 'title': file_name}]
            }
            if message:
                complete_upload_payload['initial_comment'] = message  # Slack APIは initial_comment というキー

            complete_upload_headers = {
                'Authorization': f'Bearer {self.token}',
                'Content-Type': 'application/json'
            }
            complete_upload_response = requests.post(
                'https://slack.com/api/files.completeUploadExternal',
                headers=complete_upload_headers,
                json=complete_upload_payload
            )
            complete_upload_json = complete_upload_response.json()

            if not complete_upload_json.get('ok', False):
                raise Exception(f"アップロード完了通知に失敗しました: {complete_upload_json.get('error')}")

            print("[アップロード] ファイルのアップロードが完了しました。")
            return complete_upload_json

        else:
            # ファイルがない場合は単純なメッセージ送信
            if message is None:
                raise ValueError("file_path も message も両方Noneです。何も投稿するものがありません。")

            chat_payload = {
                'channel': channel_id,
                'text': message
            }
            chat_headers = {
                'Authorization': f'Bearer {self.token}',
                'Content-Type': 'application/json'
            }
            chat_response = requests.post(
                'https://slack.com/api/chat.postMessage',
                headers=chat_headers,
                json=chat_payload
            )
            chat_json = chat_response.json()

            if not chat_json.get('ok', False):
                raise Exception(f"メッセージの送信に失敗しました: {chat_json.get('error')}")

            print("[投稿] メッセージの送信が完了しました。")
            return chat_json

if __name__ == "__main__":
    # 環境変数からトークンを読み込む例
    # token = os.environ.get("SLACK_API_TOKEN")
    
    # テスト用（実際の使用時は上記の環境変数を使用することを推奨）
    token = "YOUR_SLACK_API_TOKEN"  # ここに自分のトークンを設定
    slack = Slack(token)
    slack.upload_file_to_slack(
        channel_id='CHANNEL_ID',  # ここにチャンネルIDを設定
        message="ファイルをアップロードします",
        file_name="sample.png",
        file_path="path/to/sample.png"  # 実際のファイルパスに変更
    )

⚠️ セキュリティ上の重要な注意: 実際のコードではAPIトークンをハードコーディングせず、環境変数などから安全に読み込むようにしましょう。トークンが漏洩すると、第三者があなたのSlackワークスペースにアクセスできてしまう可能性があります。

コードの詳細解説

1. Slackクラスの初期化

class Slack:
    def __init__(self, token):
        self.token = token

Slackクラスは初期化時にAPIトークンを受け取ります。このトークンは後続のAPI呼び出しで認証に使用されます。

2. ファイルアップロードのフロー

Slack APIを使ったファイルアップロードは、3つの主要なステップで構成されています：

  1. アップロードURLの取得
  2. ファイルの実際のアップロード
  3. アップロード完了の通知

これらのステップについて詳しく見ていきましょう。

2.1 アップロードURLの取得

# アップロードURL取得
params = {
    'filename': file_name,
    'length': file_size
}
headers = {
    'Authorization': f'Bearer {self.token}'
}
upload_url_response = requests.get('https://slack.com/api/files.getUploadURLExternal', params=params, headers=headers)
upload_url_json = upload_url_response.json()

if not upload_url_json.get('ok', False):
    raise Exception(f"アップロードURLの取得に失敗しました: {upload_url_json.get('error')}")

upload_url = upload_url_json['upload_url']
file_id = upload_url_json['file_id']

ここでは、files.getUploadURLExternalエンドポイントを呼び出して、一時的なアップロードURLとファイルIDを取得しています。このURLは次のステップでファイルをアップロードするために使用されます。

2.2 ファイルのアップロード

# ファイルをアップロード
upload_headers = {
    'Content-Type': 'application/octet-stream'
}
upload_response = requests.post(upload_url, headers=upload_headers, data=file_blob)

if upload_response.status_code != 200:
    raise Exception(f"ファイルのアップロードに失敗しました: {upload_response.text}")

取得したURLに対して、ファイルの内容（バイナリデータ）をPOSTリクエストで送信します。ヘッダーにはContent-Type: application/octet-streamを指定しています。

2.3 アップロード完了通知

# アップロード完了通知
complete_upload_payload = {
    'channel_id': channel_id,
    'files': [{'id': file_id, 'title': file_name}]
}
if message:
    complete_upload_payload['initial_comment'] = message  # Slack APIは initial_comment というキー

complete_upload_headers = {
    'Authorization': f'Bearer {self.token}',
    'Content-Type': 'application/json'
}
complete_upload_response = requests.post(
    'https://slack.com/api/files.completeUploadExternal',
    headers=complete_upload_headers,
    json=complete_upload_payload
)

最後に、files.completeUploadExternalエンドポイントを呼び出して、ファイルのアップロードを完了し、指定されたチャンネルに投稿します。このとき、オプションのメッセージ（initial_comment）を付けることができます。これにより、ファイルとメッセージが同時に送信されます。

3. メッセージのみの送信

# ファイルがない場合は単純なメッセージ送信
if message is None:
    raise ValueError("file_path も message も両方Noneです。何も投稿するものがありません。")

chat_payload = {
    'channel': channel_id,
    'text': message
}
chat_headers = {
    'Authorization': f'Bearer {self.token}',
    'Content-Type': 'application/json'
}
chat_response = requests.post(
    'https://slack.com/api/chat.postMessage',
    headers=chat_headers,
    json=chat_payload
)

ファイルパスが指定されていない場合は、chat.postMessageエンドポイントを使用して、テキストメッセージのみを送信します。

使用例

モジュールとして使用するための準備

まずは、上記のコードをslack_uploader.pyとして保存しましょう。これにより、他のPythonスクリプトからインポートして使用できるようになります。

基本的な使用方法

import os
from slack_uploader import Slack

# 環境変数からトークンを取得
token = os.environ.get("SLACK_API_TOKEN")
# トークンが取得できない場合のフォールバック（実際の運用では使用しないでください）
if not token:
    token = "xoxb-your-token"  # 環境変数から取得するなど安全な方法で

# Slackインスタンスの作成
slack = Slack(token)

# メッセージとファイルを送信
slack.upload_file_to_slack(
    channel_id='C01234ABCDE',
    message="月次レポートを共有します",
    file_name="monthly_report.pdf",
    file_path="./reports/monthly_report.pdf"
)

# メッセージのみを送信
slack.upload_file_to_slack(
    channel_id='C01234ABCDE',
    message="明日の会議は15時からです"
)

実践的な使用例：データ分析結果の自動投稿

以下の例では、Pandasを使用して売上データを分析し、Matplotlibでグラフを作成した後、結果をSlackに投稿しています。

import os
import pandas as pd
import matplotlib.pyplot as plt
import time
from slack_uploader import Slack

# データ分析を実行
df = pd.read_csv('sales_data.csv')
monthly_sales = df.groupby('month')['amount'].sum()

# グラフ作成
plt.figure(figsize=(10, 6))
monthly_sales.plot(kind='bar')
plt.title('Monthly Sales')
plt.savefig('monthly_sales.png')
plt.close()  # リソースを解放

# 分析結果をSlackに投稿
token = os.environ.get('SLACK_API_TOKEN')
slack = Slack(token)

message = f"""
*月間売上レポート*
- 総売上: {monthly_sales.sum():,.0f}円
- 最高月: {monthly_sales.idxmax()} ({monthly_sales.max():,.0f}円)
- 前月比: {(monthly_sales.iloc[-1] / monthly_sales.iloc[-2] - 1) * 100:.1f}%
"""

try:
    slack.upload_file_to_slack(
        channel_id='C01234ABCDE',
        message=message,
        file_name="monthly_sales.png",
        file_path="./monthly_sales.png"
    )
    print("売上レポートの送信に成功しました")
except Exception as e:
    print(f"エラーが発生しました: {e}")

定期実行の例：毎日の自動レポート送信

Pythonスクリプトをcrontabなどで定期実行することで、自動的にレポートを送信できます。以下は、そのためのスクリプト例です。

import os
import datetime
from slack_uploader import Slack

def send_daily_report():
    today = datetime.datetime.now().strftime('%Y-%m-%d')
    report_path = f"./reports/daily_report_{today}.pdf"
    
    # レポートが存在するか確認
    if not os.path.exists(report_path):
        print(f"レポートファイルが見つかりません: {report_path}")
        return False
    
    # Slackに送信
    token = os.environ.get('SLACK_API_TOKEN')
    slack = Slack(token)
    
    try:
        slack.upload_file_to_slack(
            channel_id='C01234ABCDE',
            message=f"*{today}の日次レポート*\n今日のハイライト:\n- 売上目標達成率: 95%\n- 新規顧客獲得数: 12件",
            file_name=f"日次レポート_{today}.pdf",
            file_path=report_path
        )
        return True
    except Exception as e:
        print(f"レポート送信中にエラーが発生しました: {e}")
        return False

if __name__ == "__main__":
    success = send_daily_report()
    exit(0 if success else 1)  # 終了コードでスクリプトの成功/失敗を通知

注意点とベストプラクティス

APIトークンの安全な管理

APIトークンはソースコードにハードコーディングせず、環境変数や設定ファイルから安全に読み込むようにしましょう。特に公開リポジトリにコードをアップロードする場合は注意が必要です。

import os
from dotenv import load_dotenv

# .envファイルから環境変数を読み込み
load_dotenv()

# 環境変数からトークンを取得
token = os.environ.get('SLACK_API_TOKEN')
if not token:
    raise ValueError("SLACK_API_TOKENが設定されていません")

.envファイルの例:

SLACK_API_TOKEN=xoxb-your-token-here
SLACK_CHANNEL_ID=C01234ABCDE

このファイルは.gitignoreに追加して、バージョン管理システムでの追跡から除外してください。

エラーハンドリング

本番環境では、より詳細なエラーハンドリングを行うことをお勧めします。エラーの種類に応じて適切な対応を行いましょう。

import logging

# ロガーの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("slack_uploader.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("slack_uploader")

try:
    slack.upload_file_to_slack(
        channel_id='C01234ABCDE',
        message="テスト",
        file_name="test.txt",
        file_path="./test.txt"
    )
    logger.info("送信成功")
except FileNotFoundError as e:
    logger.error(f"ファイルが見つかりません: {e}")
except requests.exceptions.ConnectionError:
    logger.error("ネットワーク接続エラー: インターネット接続を確認してください")
except requests.exceptions.Timeout:
    logger.error("タイムアウトエラー: サーバーの応答が遅いか、接続が切断されました")
except Exception as e:
    logger.error(f"エラーが発生しました: {e}", exc_info=True)

API制限への対応

Slack APIには以下の制限があります：

  1. レート制限: Tier 3のアプリは50リクエスト/分の制限があります。
  2. ファイルサイズ制限: 1ファイルあたり最大50MBまで。
  3. チャネルあたりのファイル数: 保存期間によって制限が異なります。

特に大量のリクエストを送信する場合は、レート制限に達した場合の対応を実装しましょう。

import time

# レート制限に対応した実装例
def upload_with_rate_limit(slack, channel_id, files, max_retries=3):
    for file_info in files:
        retries = 0
        success = False
        
        while not success and retries < max_retries:
            try:
                slack.upload_file_to_slack(
                    channel_id=channel_id,
                    message=file_info.get('message'),
                    file_name=file_info.get('name'),
                    file_path=file_info.get('path')
                )
                success = True
                logging.info(f"ファイル '{file_info.get('name')}' のアップロードに成功しました")
            except Exception as e:
                if "rate_limited" in str(e).lower():
                    wait_time = 60  # デフォルト待機時間（秒）
                    # エラーメッセージからRetry-Afterヘッダーの値を抽出できる場合は使用
                    if 'retry_after' in str(e).lower():
                        import re
                        match = re.search(r'retry_after: (\d+)', str(e), re.IGNORECASE)
                        if match:
                            wait_time = int(match.group(1))
                    
                    logging.warning(f"レート制限に達しました。{wait_time}秒待機します...")
                    time.sleep(wait_time)
                    retries += 1
                else:
                    logging.error(f"アップロード中にエラーが発生しました: {e}")
                    break
            
            # 連続送信の間隔（成功時）
            if success:
                time.sleep(1)
        
        if not success:
            logging.error(f"ファイル '{file_info.get('name')}' のアップロードに失敗しました（最大再試行回数に達しました）")

ファイルサイズの確認

大きなファイルをアップロードする前に、サイズを確認することをお勧めします。

def check_file_size(file_path, max_size_mb=50):
    """ファイルサイズを確認し、制限を超えている場合はFalseを返す"""
    max_size_bytes = max_size_mb * 1024 * 1024  # MBからバイトに変換
    
    try:
        file_size = os.path.getsize(file_path)
        if file_size > max_size_bytes:
            logging.warning(
                f"ファイルサイズが制限を超えています: {file_size / (1024 * 1024):.2f}MB "
                f"(最大: {max_size_mb}MB)"
            )
            return False
        return True
    except OSError as e:
        logging.error(f"ファイルサイズの確認中にエラーが発生しました: {e}")
        return False

発展的な使い方

複数ファイルの送信

現在のAPIでは、1回のリクエストで1つのファイルしか送信できません。複数のファイルを送信したい場合は、それぞれのファイルに対して個別にリクエストを送信する必要があります。

import time

files = [
    {"path": "./report.pdf", "name": "report.pdf", "message": "月次レポート"},
    {"path": "./graph.png", "name": "graph.png", "message": "売上グラフ"},
    {"path": "./data.csv", "name": "data.csv", "message": "生データ"}
]

# すべてのファイルが存在するか確認
all_files_exist = all(os.path.exists(file_info["path"]) for file_info in files)

if all_files_exist:
    for file_info in files:
        try:
            slack.upload_file_to_slack(
                channel_id='C01234ABCDE',
                message=file_info['message'],
                file_name=file_info['name'],
                file_path=file_info['path']
            )
            print(f"{file_info['name']} を送信しました")
            time.sleep(1)  # 連続送信の間隔
        except Exception as e:
            print(f"{file_info['name']} の送信中にエラーが発生しました: {e}")
else:
    print("一部のファイルが見つかりません。パスを確認してください。")

リッチメッセージの送信

Slackはマークダウン形式のテキストをサポートしています。リッチなメッセージを作成することで、より見やすい通知を送信できます。

# リッチメッセージの例
rich_message = """
*プロジェクト進捗レポート* :chart_with_upwards_trend:

>*概要*
>期限までの残り日数: 7日
>完了タスク: 25/30 (83%)

*主要マイルストーン*:
• :white_check_mark: 要件定義 - 完了
• :white_check_mark: 設計フェーズ - 完了
• :large_blue_circle: 開発フェーズ - 進行中 (85%)
• :black_circle: テストフェーズ - 未開始

詳細はプロジェクト管理ツールを確認してください: <https://example.com/project|プロジェクトダッシュボード>
"""

slack.upload_file_to_slack(
    channel_id='C01234ABCDE',
    message=rich_message,
    file_name="project_gantt.png",
    file_path="./project_gantt.png"
)

よくあるエラーとその対処法

  1. 認証エラー (not_authed, invalid_auth, token_revoked)

    • APIトークンが無効または失効している可能性があります。新しいトークンを取得してください。

  2. スコープ不足 (missing_scope)

    • トークンに必要な権限（スコープ）が設定されていません。アプリの設定でfiles:writeとchat:writeを追加してください。

  3. チャンネルが見つからない (channel_not_found)

    • 指定したチャンネルIDが無効か、ボットがそのチャンネルに招待されていない可能性があります。
    • チャンネルIDを確認し、必要に応じてボットをチャンネルに招待してください。

  4. レート制限 (rate_limited)

    • 短時間に多くのリクエストを送信しすぎています。
    • エラーメッセージに含まれるRetry-Afterの値（秒）だけ待機してから再試行してください。

  5. ファイルサイズ超過 (file_too_large)

    • ファイルサイズが上限（50MB）を超えています。
    • ファイルを分割するか、圧縮してサイズを小さくしてください。

（備考）Slackメッセージのフォーマット

Slackでは、メッセージに特殊なフォーマットを適用できます。これを活用すると、より見やすく情報を伝えることができます。

基本的なフォーマット

  • *太字* → 太字
  • _イタリック_ → イタリック
  • ~取り消し線~ → 取り消し線
  • `コード` → コード
  • コードブロック → コードブロック

リストとブロック引用

• 箇条書きリスト
> 引用テキスト

絵文字と特殊リンク

  • 絵文字: :smile: → 😊
  • リンク: <https://example.com|テキスト> → クリック可能なリンク
  • チャンネル: <#C01234ABCDE> → チャンネルへのリンク
  • ユーザー: <@U01234ABCDE> → ユーザーへのメンション

詳細なフォーマットガイドはSlack APIドキュメントを参照してください。

まとめ

この記事では、PythonからSlack APIを使ってメッセージとファイルを同時に送信する方法を紹介しました。以下のポイントを押さえることで、効果的なSlack通知システムを構築できます：

  1. Slack APIの認証とトークン管理
  2. ファイルアップロードの3ステップ（URL取得、アップロード、完了通知）
  3. メッセージとファイルの同時送信
  4. エラーハンドリングとレート制限への対応
  5. セキュリティ対策（トークン管理、ファイルサイズチェック）

このコードを活用して、チームのコミュニケーションを効率化し、様々な自動化を実現してください。

※筆者は、機械学習プログラムの結果を随時送信することで、正常に学習が進んでいるかを監視するために利用しています。

参考リソース

この記事が皆さんのプロジェクトやタスクの自動化に役立つことを願っています。
質問やフィードバックがありましたら、コメントをお待ちしています。

Discussion