Gemini の画像生成をワークフローに組み込む — プロンプト設計から動画→サムネイル自動生成まで

import os
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# Nano Banana 2 = gemini-3.1-flash-image（2026-06 GA）。
# 動画を文脈として渡せるのはこの系統です。
IMAGE_MODEL = "gemini-3.1-flash-image"
 
def thumbnail_from_video(video_path: str, brief: str) -> bytes:
    with open(video_path, "rb") as f:
        video_bytes = f.read()
 
    response = client.models.generate_content(
        model=IMAGE_MODEL,
        contents=[
            types.Part.from_bytes(data=video_bytes, mime_type="video/mp4"),
            brief,
        ],
        config=types.GenerateContentConfig(
            response_modalities=["TEXT", "IMAGE"],
        ),
    )
 
    for part in response.candidates[0].content.parts:
        if part.inline_data is not None:
            return part.inline_data.data
    raise RuntimeError("画像パートが返りませんでした（テキストのみ応答）")

def thumbnail_from_large_video(video_path: str, brief: str) -> bytes:
    uploaded = client.files.upload(file=video_path)
    response = client.models.generate_content(
        model=IMAGE_MODEL,
        contents=[uploaded, brief],
        config=types.GenerateContentConfig(
            response_modalities=["TEXT", "IMAGE"],
        ),
    )
    for part in response.candidates[0].content.parts:
        if part.inline_data is not None:
            return part.inline_data.data
    raise RuntimeError("画像が返りませんでした")

import os
import subprocess
import tempfile
 
def clip_window(video_path: str, start_sec: float, dur_sec: float = 12.0) -> str:
    """山場の前後だけを切り出し、渡す動画そのものを軽くする。"""
    out = os.path.join(tempfile.mkdtemp(), "window.mp4")
    subprocess.run(
        [
            "ffmpeg", "-y",
            "-ss", str(start_sec),   # -i より前に置くと高速シークになります
            "-t", str(dur_sec),
            "-i", video_path,
            "-an",                       # 音声は画像生成に使われないので落とす
            "-vf", "fps=2,scale=-2:720", # 秒あたりのフレームと解像度も下げる
            out,
        ],
        check=True,
        capture_output=True,
    )
    return out

from PIL import Image
from io import BytesIO
 
def passes_quality_gate(img_bytes: bytes, target_ratio: float = 16 / 9,
                        tol: float = 0.08) -> bool:
    img = Image.open(BytesIO(img_bytes)).convert("RGB")
    w, h = img.size
    ratio = w / h
    if abs(ratio - target_ratio) / target_ratio > tol:
        return False  # アスペクト比が目標から外れすぎ
 
    # ほぼ単色（のっぺり/破綻）の検出: 縮小して画素分散を見る
    small = img.resize((32, 18))
    px = list(small.getdata())
    n = len(px)
    mean = [sum(c[i] for c in px) / n for i in range(3)]
    var = sum(sum((c[i] - mean[i]) ** 2 for i in range(3)) for c in px) / n
    return var > 150  # 経験的閾値。下回るとのっぺり/破綻が多い

import hashlib
import json
import time
from pathlib import Path
 
LEDGER = Path("generated/ledger.jsonl")
 
def record(brief: str, model: str, out_path: Path, adopted: bool) -> str:
    """1回の生成を1行として追記する。key は同一条件の再現用。"""
    key = hashlib.sha1(f"{model}\n{brief}".encode("utf-8")).hexdigest()[:12]
    LEDGER.parent.mkdir(parents=True, exist_ok=True)
    with LEDGER.open("a", encoding="utf-8") as f:
        f.write(json.dumps({
            "key": key,
            "model": model,
            "brief": brief,
            "file": str(out_path),
            "adopted": adopted,
            "at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        }, ensure_ascii=False) + "\n")
    return key
 
def find_adopted(keyword: str) -> list[dict]:
    """採用済みの生成だけを、ブリーフ文の語で引き当てる。"""
    if not LEDGER.exists():
        return []
    rows = [json.loads(line) for line in LEDGER.read_text(encoding="utf-8").splitlines() if line]
    return [r for r in rows if r["adopted"] and keyword in r["brief"]]

def generate_in_series(reference_path: str, brief: str) -> bytes:
    """採用済みの1枚をアンカーにして、同じ空気の続きを作る。"""
    with open(reference_path, "rb") as f:
        reference = f.read()
 
    response = client.models.generate_content(
        model=IMAGE_MODEL,
        contents=[
            types.Part.from_bytes(data=reference, mime_type="image/png"),
            "この画像の配色・質感・余白の取り方を保ったまま、次の内容で1枚作ってください。\n" + brief,
        ],
        config=types.GenerateContentConfig(
            response_modalities=["TEXT", "IMAGE"],
        ),
    )
    for part in response.candidates[0].content.parts:
        if part.inline_data is not None:
            return part.inline_data.data
    raise RuntimeError("画像が返りませんでした")

# モデル ID は一箇所に集約する。preview 停止に1行で追従できるようにする。
IMAGE_MODELS = {
    "fast": "gemini-3.1-flash-image",   # Nano Banana 2（GA）
    "quality": "gemini-3-pro-image",    # Nano Banana Pro（GA）
}
 
def resolve_image_model(tier: str = "fast") -> str:
    try:
        return IMAGE_MODELS[tier]
    except KeyError as e:
        # 未知のティアを既定へ無言フォールバックさせず、明示的に落とす
        raise ValueError(f"未知の画像モデルティア: {tier}") from e