Slack通知設計：AIに「何を通知させるか」を教える

1
10:00 INFO: ジョブ正常完了
2
10:01 INFO: データ取得完了（120件）
3
10:02 WARNING: レスポンス 1.2秒（閾値: 2秒）
4
10:03 INFO: ジョブ正常完了
5
...（以下、同様の通知が続く）

1
CRITICAL（#alerts）: 即座に対応が必要
2
   → 本番エラー、認証失敗、CircuitBreaker発動
3

4
WARNING（#monitoring）: 数時間以内に確認
5
   → リトライ成功、閾値接近、ジョブ遅延
6

7
INFO（ログのみ）: アクション不要
8
   → 正常完了、定期レポート

1
# AIが書きがちなコード
2
def process_data(data):
3
    send_slack_message(channel="#alerts", text="処理開始")  # 不要
4
    for item in data:
5
        result = transform(item)
6
        send_slack_message(channel="#alerts", text=f"処理完了: {item.id}")  # 不要
7
    send_slack_message(channel="#alerts", text="全処理完了")  # INFOレベル

1
# フィードバック後にAIが書くコード
2
def process_data(data):
3
    try:
4
        for item in data:
5
            transform(item)
6
    except Exception as e:
7
        logger.error(f"Error: {e}")  # ログだけで通知なし

1
from myModules.callSlack import send_slack_message
2
from myModules.commonLogger import setupLogger
3

4
logger = setupLogger(__name__)
5

6

7
def notify_error(message: str, severity: str = "warning") -> None:
8
    """3層通知の振り分け関数。
9

10
    Args:
11
        message: 通知メッセージ
12
        severity: "critical", "warning", "info" のいずれか
13
    """
14
    if severity == "critical":
15
        send_slack_message(channel="#alerts", text=f"CRITICAL: {message}")
7 collapsed lines
16
        logger.error(f"[CRITICAL] {message}")
17
    elif severity == "warning":
18
        send_slack_message(channel="#monitoring", text=f"WARNING: {message}")
19
        logger.warning(f"[WARNING] {message}")
20
    else:
21
        # INFO level: log only, no Slack
22
        logger.info(f"[INFO] {message}")

1
def fetch_external_data(source: str) -> pd.DataFrame:
2
    try:
3
        df = api_client.get_data(source)
4
        notify_error(f"データ取得完了: {source} ({len(df)}件)", severity="info")
5
        return df
6
    except AuthenticationError as e:
7
        # 認証エラー → CRITICAL（即座に対応が必要）
8
        notify_error(f"認証失敗: {source} - {e}", severity="critical")
9
        raise
10
    except ConnectionError as e:
11
        # 接続エラー → WARNING（リトライで回復の可能性あり）
12
        notify_error(f"接続エラー: {source} - {e}", severity="warning")
13
        raise

1
from datetime import datetime
2

3

4
def send_daily_report(results: dict) -> None:
5
    """日次パフォーマンスレポートを固定時刻に送信する。
6

7
    22:00 JSTに送信。「来ない＝異常」の検知に使う。
8
    """
9
    report_time = datetime.now().strftime("%Y-%m-%d %H:%M")
10

11
    summary = (
12
        f"--- Daily Report ({report_time}) ---\n"
13
        f"処理件数: {results['total_count']}\n"
14
        f"成功率: {results['success_rate']:.1%}\n"
15
        f"平均処理時間: {results['avg_duration']:.2f}秒\n"
7 collapsed lines
16
        f"エラー件数: {results['error_count']}"
17
    )
18

19
    # 日次レポートはWARNINGチャンネルに送信
20
    # （毎日来ることが前提。来なければ異常）
21
    send_slack_message(channel="#monitoring", text=summary)
22
    logger.info(f"[DAILY_REPORT] {summary}")

1
22:00 レポート到着 → 正常
2
22:05 レポート未着 → ジョブが実行されていない可能性
3
22:15 レポート未着 → CRITICAL通知を手動または監視ジョブで発火

1
## Slack通知ルール
2

3
1. 通知レベルの3層分類を必ず守ること
4
   - CRITICAL（#alerts）: 認証失敗、DB接続不可、CircuitBreaker発動
5
   - WARNING（#monitoring）: リトライ成功、閾値接近、ジョブ遅延
6
   - INFO: ログのみ（Slack通知しない）
7

8
2. 正常完了の通知は禁止
9
   - 正常系はログのみ。Slackには送信しない
10
   - 日次レポートは例外として #monitoring に送信
11

12
3. notify_error() 関数を使うこと
13
   - 直接 send_slack_message() を呼ばない
14
   - severity パラメータで振り分けを統一する