静かに止まる自動処理を防ぐ — Gemini 基盤変更に備える事前検証ゲートの設計

# preflight.py — Gemini 自動処理のための事前検証ゲート
import os
import re
import sys
import pathlib
from google import genai
from google.genai import errors as genai_errors
 
REPO_ROOT = pathlib.Path(os.environ.get("REPO_ROOT", "."))
MODEL = os.environ.get("PREFLIGHT_MODEL", "gemini-2.5-flash-lite")
 
class Violation(Exception):
    """前提違反。メッセージがそのまま停止理由になる。"""
 
def check_key_is_usable_and_restricted() -> str:
    """キーが有効かつ制限拒否されないかを、最小リクエストで確認する。"""
    key = os.environ.get("GEMINI_API_KEY")
    if not key:
        raise Violation("GEMINI_API_KEY が未設定です")
    client = genai.Client(api_key=key)
    try:
        # 1トークンで済む最小の呼び出し。課金・レイテンシを最小化する
        client.models.generate_content(
            model=MODEL,
            contents="ok",
            config={"max_output_tokens": 1},
        )
    except genai_errors.ClientError as e:
        code = getattr(e, "status_code", None) or getattr(e, "code", None)
        # 403 系は「制限なしキー拒否」や「参照元不許可」を含む。前提違反として扱う
        if code in (401, 403):
            raise Violation(
                f"キーが拒否されました (HTTP {code})。"
                "アプリ制限/API制限の設定、またはキーの有効性を確認してください"
            )
        raise
    return "key ok"

LEGACY_CLI = re.compile(r"(?<![\w-])gemini\s+(generate|chat|config|mcp)\b")
SCAN_SUFFIXES = {".sh", ".bash", ".yml", ".yaml", ".mjs", ".py"}
 
def check_no_legacy_cli() -> str:
    """EOL 済みの旧 gemini CLI 呼び出しがコードに残っていないか走査する。"""
    hits = []
    for path in REPO_ROOT.rglob("*"):
        if path.suffix.lower() not in SCAN_SUFFIXES:
            continue
        if any(part in {".git", "node_modules", ".next"} for part in path.parts):
            continue
        try:
            text = path.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        for i, line in enumerate(text.splitlines(), 1):
            if line.lstrip().startswith("#"):
                continue  # コメント行は除外
            if LEGACY_CLI.search(line):
                hits.append(f"{path.relative_to(REPO_ROOT)}:{i}")
    if hits:
        raise Violation(
            "EOL 済みの旧 Gemini CLI 呼び出しを検出しました: " + ", ".join(hits[:8])
        )
    return "no legacy CLI"

LEGACY_CALL = re.compile(r"\.generate_content\(|\.generateContent\(")
 
def audit_migration_debt(max_allowed: int = 0) -> str:
    """旧 generateContent コールサイト数を数え、しきい値超過を WARN する。"""
    count = 0
    sites = []
    for path in REPO_ROOT.rglob("*"):
        if path.suffix.lower() not in {".py", ".ts", ".js", ".mjs"}:
            continue
        if any(p in {".git", "node_modules", ".next"} for p in path.parts):
            continue
        try:
            text = path.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        for i, line in enumerate(text.splitlines(), 1):
            if LEGACY_CALL.search(line):
                count += 1
                sites.append(f"{path.relative_to(REPO_ROOT)}:{i}")
    if count > max_allowed:
        print(f"::warning:: Interactions API 未移行のコールサイト {count} 件 "
              f"(上限 {max_allowed}): {', '.join(sites[:5])}")
    return f"migration debt: {count}"

def main() -> int:
    checks = [
        ("key", check_key_is_usable_and_restricted, True),   # True = FAIL 対象
        ("legacy-cli", check_no_legacy_cli, True),
        ("migration-debt", lambda: audit_migration_debt(0), False),  # WARN のみ
    ]
    failed = False
    for name, fn, is_fatal in checks:
        try:
            print(f"[ok] {name}: {fn()}")
        except Violation as v:
            level = "FAIL" if is_fatal else "WARN"
            print(f"[{level}] {name}: {v}", file=sys.stderr)
            if is_fatal:
                failed = True
    return 1 if failed else 0
 
if __name__ == "__main__":
    sys.exit(main())

- name: Preflight gate
  run: python preflight.py
  env:
    GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
    REPO_ROOT: ${{ github.workspace }}
 
- name: Generate articles
  run: node generate.mjs   # preflight が失敗すると、ここには到達しない

# preflight が 0 を返したときだけ本処理へ進む
python preflight.py && node generate.mjs || \
  echo "$(TZ=Asia/Tokyo date '+%F %T') preflight blocked run" >> preflight.log