長い動画を Gemini に辿らせるとき、見つける問いと「無い」を確かめる問いは別々に投げています

import json
import time
from pathlib import Path
 
from google import genai
 
client = genai.Client()
LOG = Path("video_probe_log.jsonl")
 
 
def ask_agentic(video_uri: str, question: str, model: str = "gemini-3.7-flash") -> str:
    started = time.time()
    interaction = client.interactions.create(
        model=model,
        input=[
            {"type": "video", "uri": video_uri, "processing": "agentic"},
            {"type": "text", "text": question},
        ],
    )
    # SDK の版で属性名が揺れるため、両方を見てから諦める
    usage = getattr(interaction, "usage", None) or getattr(interaction, "usage_metadata", None)
    record = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "model": model,
        "mode": "agentic",
        "video": video_uri,
        "question": question,
        "elapsed_sec": round(time.time() - started, 2),
        "usage": usage.to_json_dict() if hasattr(usage, "to_json_dict") else str(usage),
        "answer": interaction.output_text,
    }
    with LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")
    return interaction.output_text

def plan_windows(duration_sec: float, window_sec: float = 300.0,
                 overlap_sec: float = 2.0) -> list[tuple[float, float]]:
    """[0, duration] を覆う (start, end) の並びを返す。
    overlap は境界で切れた出来事を落とさないための重なり。"""
    if duration_sec <= 0:
        raise ValueError("duration_sec must be positive")
    if overlap_sec >= window_sec:
        raise ValueError("overlap_sec must be smaller than window_sec")
 
    step = window_sec - overlap_sec
    windows: list[tuple[float, float]] = []
    start = 0.0
    while start < duration_sec:
        end = min(start + window_sec, duration_sec)
        windows.append((round(start, 3), round(end, 3)))
        if end >= duration_sec:
            break
        start += step
    return windows
 
 
def assert_covers(windows, duration_sec: float, tolerance: float = 1e-6) -> None:
    """隙間なく [0, duration] を覆っているかを確かめる。覆えていなければ例外。"""
    if not windows:
        raise AssertionError("no windows")
    ordered = sorted(windows)
    if ordered[0][0] > tolerance:
        raise AssertionError(f"head gap: 0.0 .. {ordered[0][0]}")
 
    reached = ordered[0][1]
    for start, end in ordered[1:]:
        if start > reached + tolerance:
            raise AssertionError(f"gap: {reached} .. {start}")
        reached = max(reached, end)
 
    if reached < duration_sec - tolerance:
        raise AssertionError(f"tail gap: {reached} .. {duration_sec}")
 
 
# 一時間三分の記録映像を、5分窓・2秒重なりで覆う
windows = plan_windows(3_780.0, window_sec=300.0, overlap_sec=2.0)
assert_covers(windows, 3_780.0)
print(len(windows), windows[0], windows[-1])
# → 13 (0.0, 300.0) (3576.0, 3780.0)

import subprocess
 
 
def cut_windows(src: str, windows, out_dir: str = "windows") -> list[str]:
    Path(out_dir).mkdir(exist_ok=True)
    paths = []
    for i, (start, end) in enumerate(windows):
        out = f"{out_dir}/win_{i:03d}.mp4"
        subprocess.run(
            [
                "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
                "-ss", f"{start}", "-to", f"{end}", "-i", src,
                "-c", "copy", out,
            ],
            check=True,
        )
        paths.append(out)
    return paths

def sweep_absence(window_files, question, done_path="sweep_done.jsonl"):
    """全窓を静的処理で走査する。1窓でも欠けたら『未確定』を返す。"""
    p = Path(done_path)
    done: dict[str, str] = {}
    if p.exists():
        for line in p.read_text(encoding="utf-8").splitlines():
            rec = json.loads(line)
            done[rec["window"]] = rec["answer"]
 
    for path in window_files:
        if path in done:
            continue
        interaction = client.interactions.create(
            model="gemini-3.7-flash",
            input=[
                {"type": "video", "uri": upload_file(path)},  # processing 未指定＝静的
                {"type": "text", "text": question},
            ],
        )
        done[path] = interaction.output_text
        with p.open("a", encoding="utf-8") as f:
            f.write(json.dumps({"window": path, "answer": done[path]},
                               ensure_ascii=False) + "\n")
 
    missing = [w for w in window_files if w not in done]
    hits = [(w, a) for w, a in done.items() if not a.startswith("該当なし")]
 
    if missing:
        return {"verdict": "未確定", "covered": len(done),
                "total": len(window_files), "missing": missing, "hits": hits}
    return {"verdict": "該当あり" if hits else "該当なし",
            "covered": len(done), "total": len(window_files), "hits": hits}