PythonでChatworkに自動通知する仕組みを作ってみた

Chatworkへの自動通知が必要になった理由

前職でChatworkを使っているクライアントとやりとりする機会が多く、「バッチ処理の完了通知」「エラーアラート」「日次レポートの送信」を手動でやっていました。Slackと違ってWebhookが使いにくいChatworkでしたが、REST APIを使えば問題なく自動化できます。

今回はChatwork APIとPythonで通知を自動送信する仕組みの作り方を解説します。

Chatwork APIトークンの取得

Chatwork APIトークン管理ページにアクセスし、APIトークンを発行します。トークンは.envファイルに保存してください。

CHATWORK_API_TOKEN=your_api_token_here
CHATWORK_ROOM_ID=123456789

基本実装：メッセージを送信する

import os
import requests
from dotenv import load_dotenv

load_dotenv()

API_TOKEN = os.environ["CHATWORK_API_TOKEN"]
ROOM_ID = os.environ["CHATWORK_ROOM_ID"]
BASE_URL = "https://api.chatwork.com/v2"


def send_message(room_id: str, message: str) -> dict:
    """Chatworkのルームにメッセージを送信する"""
    url = f"{BASE_URL}/rooms/{room_id}/messages"
    headers = {"X-ChatWorkToken": API_TOKEN}
    payload = {"body": message, "self_unread": 0}

    response = requests.post(url, headers=headers, data=payload, timeout=10)
    response.raise_for_status()
    return response.json()


# テスト送信
result = send_message(ROOM_ID, "テスト送信です")
print(f"送信完了: message_id={result['message_id']}")

Chatworkのメッセージ記法

Chatworkにはアカウントへのメンションやタスク追加などの独自記法があります。

def format_mention(account_id: int, name: str) -> str:
    """メンション記法を生成する"""
    return f"[To:{account_id}]{name}さん"


def format_info(title: str, content: str) -> str:
    """情報ボックス記法を生成する"""
    return f"[info][title]{title}[/title]{content}[/info]"


def format_code(code: str) -> str:
    """コード記法を生成する"""
    return f"[code]{code}[/code]"

実践：日次レポートの自動送信

バッチ処理の結果を毎日定時にChatworkに送信するスクリプトです。

from datetime import date
from typing import Optional


class ChatworkNotifier:
    def __init__(self, api_token: str, room_id: str):
        self.api_token = api_token
        self.room_id = room_id
        self.base_url = "https://api.chatwork.com/v2"

    def _post(self, endpoint: str, payload: dict) -> dict:
        url = f"{self.base_url}{endpoint}"
        headers = {"X-ChatWorkToken": self.api_token}
        response = requests.post(url, headers=headers, data=payload, timeout=10)
        response.raise_for_status()
        return response.json()

    def send(self, message: str) -> str:
        result = self._post(
            f"/rooms/{self.room_id}/messages",
            {"body": message, "self_unread": 0}
        )
        return result["message_id"]

    def send_daily_report(
        self,
        processed_count: int,
        error_count: int,
        elapsed_seconds: float,
        detail_lines: Optional[list[str]] = None,
    ) -> None:
        today = date.today().strftime("%Y/%m/%d")
        status = "SUCCESS" if error_count == 0 else "WARNING"
        icon = "" if error_count == 0 else "[warning]"

        lines = [
            f"{icon}【日次バッチ完了】{today}",
            "",
            f"ステータス: {status}",
            f"処理件数: {processed_count:,}件",
            f"エラー件数: {error_count}件",
            f"処理時間: {elapsed_seconds:.1f}秒",
        ]

        if detail_lines:
            lines.append("")
            lines.append("[info][title]詳細[/title]")
            lines.extend(detail_lines)
            lines.append("[/info]")

        self.send("\n".join(lines))


# 使用例
notifier = ChatworkNotifier(API_TOKEN, ROOM_ID)
notifier.send_daily_report(
    processed_count=1523,
    error_count=0,
    elapsed_seconds=45.2,
    detail_lines=["商品A: 800件", "商品B: 723件"]
)

エラー通知デコレータ

重要なバッチ処理にデコレータとして付けておくと、失敗時に自動でChatworkに通知できます。

import traceback
import functools


def chatwork_on_error(notifier: ChatworkNotifier, mention_account_id: Optional[int] = None):
    """エラー時にChatwork通知するデコレータ"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                mention = ""
                if mention_account_id:
                    mention = f"[To:{mention_account_id}]担当者 "

                message = (
                    f"{mention}[警告]\n"
                    f"関数 `{func.__name__}` でエラーが発生しました\n\n"
                    f"エラー: {type(e).__name__}: {e}\n\n"
                    f"[code]{traceback.format_exc()}[/code]"
                )
                notifier.send(message)
                raise
        return wrapper
    return decorator


# 使用例
@chatwork_on_error(notifier, mention_account_id=12345)
def important_batch():
    # 失敗するとChatworkに通知される
    raise RuntimeError("データベース接続失敗")

cronで定期実行する

# 毎日18時に日次レポートを送信
0 18 * * * /path/to/venv/bin/python /path/to/send_report.py >> /tmp/chatwork_bot.log 2>&1

まとめ

Chatwork APIを使った自動通知のポイントをまとめます。

Slackほど情報が多くないChatworkのAPIですが、十分な機能が揃っています。手動で送っていた定期連絡があれば、ぜひ自動化してみてください。