Gemini CLI 停止に備え、夜間バッチに google-genai SDK フォールバックを組み込む

import os
from google import genai
 
_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def run_via_sdk(prompt: str, model: str = "gemini-3.5-flash") -> str:
    """対話を必要としない処理は SDK 直叩きに寄せる。
    CLI の生死と無関係に動くフォールバック先。"""
    response = _client.models.generate_content(
        model=model,
        contents=prompt,
    )
    return response.text

def run_via_cli(prompt: str, timeout_sec: int = 120) -> str:
    result = subprocess.run(
        ["gemini", "-p", prompt],
        capture_output=True,
        text=True,
        timeout=timeout_sec,
    )
    if result.returncode != 0:
        raise RuntimeError(f"gemini CLI exited {result.returncode}: {result.stderr[:200]}")
    return result.stdout.strip()

import time
 
def generate(prompt: str, *, prefer_cli: bool = False, max_attempts: int = 3) -> str:
    """CLI を優先したい処理だけ prefer_cli=True にする。
    それ以外は最初から SDK を使い、CLI が落ちていれば自動で SDK に切り替える。"""
    last_error = None
 
    if prefer_cli and cli_available():
        for attempt in range(max_attempts):
            try:
                return run_via_cli(prompt)
            except (subprocess.TimeoutExpired, RuntimeError) as exc:
                last_error = exc
                time.sleep(2 ** attempt)  # 1, 2, 4 秒の指数バックオフ
 
    for attempt in range(max_attempts):
        try:
            return run_via_sdk(prompt)
        except Exception as exc:  # APIError 等
            last_error = exc
            time.sleep(2 ** attempt)
 
    raise RuntimeError(f"CLI と SDK の両方が失敗しました: {last_error}")

import hashlib
import pathlib
 
STATE_DIR = pathlib.Path.home() / ".cache" / "nightly-batch"
 
def already_done(site: str, slug: str, run_date: str) -> bool:
    """同じ日に同じ記事を二度処理しないための冪等キー。
    バッチが途中で落ちて再実行されても二重投稿を防げる。"""
    key = hashlib.sha1(f"{site}:{slug}:{run_date}".encode()).hexdigest()[:16]
    marker = STATE_DIR / key
    if marker.exists():
        return True
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    marker.write_text(run_date)
    return False

def nightly_job(site: str, slug: str, prompt: str, run_date: str) -> None:
    if already_done(site, slug, run_date):
        print(f"skip: {site}/{slug} は {run_date} に処理済みです")
        return
    body = generate(prompt, prefer_cli=False)  # 無人生成は SDK で十分
    publish(site, slug, body)                  # 既存の公開処理に渡す