夜間ログを毎朝3分で読み切る — Gemini API で Daily Brief 風の運用ダイジェストを自作した記録

# collect_logs.py — 直近24時間の夜間ログを1つのJSONペイロードに集約する
import json
from datetime import datetime, timedelta
from pathlib import Path
 
LOG_SOURCES = {
    "article_publish": Path("logs/publish"),   # 記事自動公開のログ
    "app_reviews": Path("logs/reviews"),       # レビュー返信下書き生成のログ
    "seo_snapshot": Path("logs/search"),       # 検索データ日次取得のログ
    "backup": Path("logs/backup"),             # バックアップのログ
}
 
MAX_CHARS_PER_SOURCE = 8000  # 1ソースあたりの上限。古い行から捨てる
 
def collect_yesterday() -> dict:
    since = datetime.now() - timedelta(hours=24)
    payload = {}
    for name, root in LOG_SOURCES.items():
        if not root.exists():
            payload[name] = "(no logs)"
            continue
        chunks = []
        for f in sorted(root.glob("*.log")):
            if datetime.fromtimestamp(f.stat().st_mtime) < since:
                continue
            chunks.append(f.read_text(encoding="utf-8", errors="replace"))
        text = "\n".join(chunks).strip()
        # 上限超過時は「新しい行を残す」— 末尾側に直近の結果が書かれているため
        payload[name] = text[-MAX_CHARS_PER_SOURCE:] if text else "(no logs)"
    return payload
 
if __name__ == "__main__":
    data = collect_yesterday()
    print(json.dumps({k: len(v) for k, v in data.items()}, ensure_ascii=False))
    # 期待する出力例: {"article_publish": 4213, "app_reviews": 1877, "seo_snapshot": 902, "backup": 311}

# summarize.py — Gemini 3.5 Flash に構造化ダイジェストを生成させる
import json
from google import genai
from google.genai import types
 
client = genai.Client()  # 環境変数 GEMINI_API_KEY を読み込む
 
DIGEST_SCHEMA = types.Schema(
    type=types.Type.OBJECT,
    properties={
        "headline": types.Schema(type=types.Type.STRING),
        "incidents": types.Schema(
            type=types.Type.ARRAY,
            items=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "source": types.Schema(type=types.Type.STRING),
                    "severity": types.Schema(
                        type=types.Type.STRING,
                        enum=["info", "warn", "critical"],
                    ),
                    "summary": types.Schema(type=types.Type.STRING),
                    "action": types.Schema(type=types.Type.STRING),
                },
                required=["source", "severity", "summary"],
            ),
        ),
    },
    required=["headline", "incidents"],
)
 
SYSTEM_INSTRUCTION = (
    "あなたは個人開発者の運用アシスタントです。"
    "渡された夜間ログを読み、朝の判断に必要な情報だけを抽出してください。"
    "正常に完了した処理は incidents に含めず、headline で1行に集約します。"
    "エラー・リトライ・想定外の出力は incidents に入れ、"
    "severity と、開発者が今朝とるべき action を日本語で付けてください。"
)
 
def summarize(payload: dict) -> dict:
    res = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=json.dumps(payload, ensure_ascii=False),
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_INSTRUCTION,
            response_mime_type="application/json",
            response_schema=DIGEST_SCHEMA,
            temperature=0.2,
        ),
    )
    return json.loads(res.text)

# deliver.py — ダイジェストをプレーンテキストのメールに整形して送る
import smtplib
from email.mime.text import MIMEText
 
def render(digest: dict) -> str:
    lines = [f"■ {digest['headline']}", ""]
    if not digest["incidents"]:
        lines.append("incidents なし。すべてのパイプラインが正常に完了しています。")
    for it in digest["incidents"]:
        mark = {"info": "・", "warn": "[warn]", "critical": "[CRIT]"}.get(it["severity"], "・")
        lines.append(f"{mark} {it['source']}: {it['summary']}")
        if it.get("action"):
            lines.append(f"    → {it['action']}")
    return "\n".join(lines)
 
def send(body: str) -> None:
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = "Morning Ops Digest"
    msg["From"] = "ops@example.com"
    msg["To"] = "me@example.com"
    with smtplib.SMTP("localhost") as s:
        s.send_message(msg)

# crontab — 毎朝 6:40 に実行（リトライ込みでも 7:00 前に配信が完了する）
40 6 * * * cd /home/me/ops-digest && /usr/bin/python3 run_digest.py >> cron.log 2>&1

# run_digest.py — 要約に失敗しても配信は止めないエントリポイント
import time
from collect_logs import collect_yesterday
from summarize import summarize
from deliver import render, send
 
RETRIES = 3
 
def degraded_digest(payload: dict) -> str:
    # Gemini が利用できない朝のための「生ログ縮約版」
    lines = ["■ [劣化モード] 要約APIが利用できないため各ソースの末尾を表示します", ""]
    for name, text in payload.items():
        tail = text.strip().splitlines()[-5:]
        lines.append(f"--- {name} ---")
        lines.extend(tail)
        lines.append("")
    return "\n".join(lines)
 
def main() -> None:
    payload = collect_yesterday()
    for attempt in range(1, RETRIES + 1):
        try:
            send(render(summarize(payload)))
            return
        except Exception:
            if attempt == RETRIES:
                send(degraded_digest(payload))
                return
            time.sleep(30 * attempt)  # 30秒 → 60秒と間隔を広げて再試行
 
if __name__ == "__main__":
    main()