scripts/runpod_scene_processor.py

#!/usr/bin/env python3
"""RunPod scene processor — autonomous batch VLM pipeline via vllm endpoint.

Designed to run ON a RunPod pod (or any machine running vllm). Replaces the
local transformers model with HTTP calls to a vllm OpenAI-compatible endpoint.
Processes videos in configurable batch sizes, checkpoints progress so it can
resume after a restart, and POSTs a webhook when each batch completes.

Prerequisites on the RunPod pod:
    pip install vllm          # start with: vllm serve Qwen/Qwen2.5-VL-72B-Instruct
    pip install openai        # for the API client
    pip install requests      # for webhook

Typical RunPod workflow:
    # 1. On local machine — sync data to pod:
    rsync -avz backend/scene-local-work/ runpod:/workspace/scene-local-work/
    rsync -avz backend/ runpod:/workspace/backend/  --exclude=videos --exclude=subtitles

    # 2. On pod — start vllm server (separate tmux):
    vllm serve Qwen/Qwen2.5-VL-72B-Instruct --tensor-parallel-size 1

    # 3. On pod — run this script:
    cd /workspace
    source backend/venv/bin/activate
    export SEARCH_UI_DATA_ROOT=/workspace/backend
    python scripts/runpod_scene_processor.py process \\
        --next 50 \\
        --vllm-url http://localhost:8000/v1 \\
        --vllm-model Qwen/Qwen2.5-VL-72B-Instruct \\
        --stop-on-error \\
        --webhook-url https://hooks.example.com/batch-done

    # 4. On local machine — sync results back:
    rsync -avz runpod:/workspace/backend/scene_index.db backend/scene_index.db
    rsync -avz runpod:/workspace/backend/scene-local-work/ backend/scene-local-work/
"""

from __future__ import annotations

import argparse
import base64
import json
import logging
import os
import re
import subprocess
import sys
import time
from typing import Any

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.dirname(SCRIPT_DIR)
BACKEND_DIR = os.path.join(REPO_ROOT, "backend")
if BACKEND_DIR not in sys.path:
    sys.path.insert(0, BACKEND_DIR)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Prompts (identical to vlm_scene_processor.py — keep in sync)
# ---------------------------------------------------------------------------

WINDOW_SYSTEM = (
    "You are a precise video description assistant. "
    "You write concise, factual descriptions of short video clips for a searchable index. "
    "You follow instructions exactly and return only valid JSON."
)

WINDOW_USER_TEMPLATE = """\
You are analyzing a 30-second window from a JW.org educational or documentary video.

Subtitle text spoken in this window:
{subtitle_block}

Task: Describe what is happening in these frames.

Strict rules:
1. Write EXACTLY 60-100 words. Count carefully before finalising.
2. Lead with the ACTION or RELATIONSHIP visible — not the setting. Do NOT start with \
"In this scene", "The video shows", "This window", or similar filler.
3. Combine what is CLEARLY VISIBLE in the frames with what is LITERALLY STATED in \
the subtitle text. Do NOT invent details.
4. Conservative: if you cannot clearly see something, do not describe it.
5. End-card detection: if this window is a standard JW.org end card — the jw.org logo \
or Watchtower logo on a plain black or dark background, with copyright text and NO human \
action — set skip=true and provide a skip_reason. Do NOT skip windows with meaningful content.

Also list any place names (cities, countries, regions) visible as ON-SCREEN TEXT only \
(chyrons, lower-thirds, signs, title cards, text overlays). Do NOT include places \
mentioned only in spoken dialogue.

Return ONLY this JSON — no markdown fences, no extra text:
{{
  "description": "<60-100 word description, or empty string if skip=true>",
  "onscreen_text_places": ["<place name>", ...],
  "skip": false,
  "skip_reason": ""
}}"""

SUMMARY_SYSTEM = (
    "You write precise video-level summaries for a searchable index. "
    "You follow word-count and formatting instructions exactly and return only valid JSON."
)

SUMMARY_USER_TEMPLATE = """\
You have described all windows of a JW.org video titled: "{title}"

Window descriptions (chronological):
{window_block}

Subtitle context (all spoken text):
{subtitle_block}

Write a video-level summary. Strict rules:
1. tldr: 80-140 words, specific and factual — name who, what, where, when if present. \
No vague generalities.
2. themes: 5-8 SPECIFIC themes (e.g. "delegates arriving by plane at Yankee Stadium", \
not "travel"). Each theme is a concrete observable activity or subject in the video.
3. acts: 3-5 acts covering the video chronologically. The first act MUST have \
start_seconds=0. Each act description should be 1-2 sentences.
4. locations: countries, cities, or regions the video is SET IN or SUBSTANTIALLY ABOUT \
(drawn from narration/dialogue, not from onscreen text). Only include if the video is \
genuinely located there. Generic references like "many countries" do NOT count.

Return ONLY this JSON — no markdown fences, no extra text:
{{
  "tldr": "<80-140 words>",
  "themes": ["<specific theme>", ...],
  "acts": [
    {{"start_seconds": 0, "description": "<act 1>"}},
    ...
  ],
  "locations": ["<city or country>", ...]
}}"""


# ---------------------------------------------------------------------------
# JSON extraction (identical to vlm_scene_processor.py — keep in sync)
# ---------------------------------------------------------------------------

def _repair_truncated_json(fragment: str) -> str:
    """Best-effort repair of JSON truncated mid-output (hit max_tokens).

    Closes an unterminated string, then appends the closing brackets/braces
    needed to balance the structure. Recovers all fields that completed plus
    the (possibly slightly clipped) field that was being written. Returns the
    repaired string; the caller still json.loads() it and may still fail.
    """
    s = fragment.rstrip().rstrip(",")  # trailing comma would break the parse

    # Count unescaped double-quotes to decide if we're inside an open string.
    in_string = False
    escaped = False
    stack: list[str] = []
    for ch in s:
        if escaped:
            escaped = False
            continue
        if ch == "\\":
            escaped = True
            continue
        if ch == '"':
            in_string = not in_string
            continue
        if in_string:
            continue
        if ch in "{[":
            stack.append("}" if ch == "{" else "]")
        elif ch in "}]" and stack:
            stack.pop()

    if in_string:
        s += '"'        # close the dangling string value
    while stack:
        s += stack.pop()  # close open arrays/objects, innermost first
    return s


def extract_json(raw: str, context: str = "") -> dict:
    text = re.sub(r"```(?:json)?\s*", "", raw).strip()
    start = text.find("{")
    if start == -1:
        raise ValueError(
            f"No JSON object found in model output{(' (' + context + ')') if context else ''}.\n"
            f"Raw: {raw[:500]!r}"
        )
    depth = 0
    end = -1
    for i, ch in enumerate(text[start:], start):
        if ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                end = i + 1
                break
    if end == -1:
        # Output was truncated (hit max_tokens) before the JSON closed.
        # Attempt a best-effort repair rather than losing the whole video.
        try:
            repaired = _repair_truncated_json(text[start:])
            result = json.loads(repaired)
            log.warning(
                "Recovered truncated JSON via repair (%s) — consider raising max_tokens.",
                context,
            )
            return result
        except json.JSONDecodeError:
            raise ValueError(
                f"Unmatched braces in output ({context}); repair failed. "
                f"Raw: {raw[:500]!r}"
            )
    try:
        return json.loads(text[start:end])
    except json.JSONDecodeError as exc:
        raise ValueError(
            f"JSON parse error ({context}): {exc}\n"
            f"Extracted: {text[start:end][:500]!r}"
        ) from exc


# ---------------------------------------------------------------------------
# vllm client
# ---------------------------------------------------------------------------

def _make_client(vllm_url: str, api_key: str) -> Any:
    try:
        from openai import OpenAI
    except ImportError as exc:
        raise RuntimeError(
            "openai package is required for RunPod mode. "
            "Run: pip install openai"
        ) from exc
    # max_retries=0: the SDK's own internal retries (default 2) would compound
    # with _chat_with_retry's backoff, inflating worst-case per-call stall to
    # ~8-12 min during a sustained 429 storm. Keep all retry/backoff logic in
    # one place (_chat_with_retry) so worst-case wait is the predictable ~335s.
    return OpenAI(base_url=vllm_url, api_key=api_key or "placeholder", max_retries=0)


def _image_content(path: str) -> dict:
    """Encode a local JPEG as a base64 data URI for the vllm API."""
    with open(path, "rb") as fh:
        b64 = base64.b64encode(fh.read()).decode()
    return {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}


def pick_frames(frame_paths: list[str], n: int = 3) -> list[str]:
    if not frame_paths:
        return []
    if len(frame_paths) <= n:
        return frame_paths
    indices = [round(i * (len(frame_paths) - 1) / (n - 1)) for i in range(n)]
    return [frame_paths[i] for i in indices]


def _chat_with_retry(client: Any, *, max_retries: int = 8, **kwargs: Any) -> Any:
    """Call chat.completions.create with exponential backoff on transient errors.

    OpenRouter's shared upstreams (e.g. Alibaba for qwen3-vl) intermittently
    return 429 "temporarily rate-limited". Without backoff, every window of a
    video fails and the whole video is skipped, churning the queue uselessly.
    Here we wait and retry on 429 / 5xx / timeout / connection errors so a
    rate-limit window just pauses the run instead of burning through videos.
    Non-transient errors (e.g. 400 bad request) raise immediately.
    """
    delay = 5.0
    cap = 90.0
    last_exc: Exception | None = None
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(**kwargs)
        except Exception as exc:  # noqa: BLE001 — classify below, re-raise if fatal
            last_exc = exc
            status = getattr(exc, "status_code", None)
            msg = str(exc).lower()
            is_rate = status == 429 or "429" in msg or "rate-limit" in msg or "rate limit" in msg
            is_5xx = isinstance(status, int) and 500 <= status < 600
            is_conn = "timeout" in msg or "connection" in msg or "temporarily" in msg
            if not (is_rate or is_5xx or is_conn) or attempt == max_retries - 1:
                raise
            wait = min(delay * (2 ** attempt), cap)
            log.warning(
                "Transient API error (status=%s); retry %d/%d in %.0fs",
                status if status is not None else "?", attempt + 1, max_retries, wait,
            )
            time.sleep(wait)
    assert last_exc is not None
    raise last_exc


def run_window_inference(
    client: Any,
    model_name: str,
    window: dict,
    n_frames: int = 3,
    max_tokens: int = 350,
    timeout: float = 120.0,
) -> dict:
    chosen_paths = pick_frames(window["frame_paths"], n_frames)
    if not chosen_paths:
        raise ValueError(f"Window {window['window_index']} has no frame_paths.")
    for p in chosen_paths:
        if not os.path.exists(p):
            raise FileNotFoundError(f"Frame file not found: {p}. Run prepare first.")

    sub_text = (window.get("subtitle_text") or "").strip()
    subtitle_block = f'"{sub_text}"' if sub_text else "(no spoken dialogue in this window)"
    user_text = WINDOW_USER_TEMPLATE.format(subtitle_block=subtitle_block)

    content: list[dict] = [_image_content(p) for p in chosen_paths]
    content.append({"type": "text", "text": user_text})

    response = _chat_with_retry(
        client,
        model=model_name,
        messages=[
            {"role": "system", "content": WINDOW_SYSTEM},
            {"role": "user", "content": content},
        ],
        max_tokens=max_tokens,
        temperature=0.0,
        timeout=timeout,
    )
    raw = response.choices[0].message.content or ""
    context = f"window {window['window_index']}"
    result = extract_json(raw, context)

    skip = bool(result.get("skip", False))
    description = str(result.get("description", "")).strip()

    if skip:
        if not result.get("skip_reason", "").strip():
            result["skip_reason"] = "boilerplate end card (auto-detected)"
        result["description"] = ""
        log.info("  window %d: SKIP — %s", window["window_index"], result["skip_reason"])
    else:
        if not description:
            raise ValueError(
                f"Window {window['window_index']}: empty description without skip=true. "
                f"Raw: {raw[:300]!r}"
            )
        word_count = len(description.split())
        # Hard-fail only if suspiciously short (<8 words = structural failure).
        # Shorter-than-target descriptions are quality warnings, not hard errors.
        if word_count < 8:
            raise ValueError(
                f"Window {window['window_index']}: description suspiciously short "
                f"({word_count} words). Raw: {description!r}"
            )
        if word_count < 60 or word_count > 130:
            log.warning("  window %d: %d words (target 60-100)", window["window_index"], word_count)
        else:
            log.info("  window %d: %d words OK", window["window_index"], word_count)
        result["description"] = description

    result.setdefault("onscreen_text_places", [])
    result.setdefault("skip", skip)
    return result


def run_summary_inference(
    client: Any,
    model_name: str,
    request: dict,
    window_results: list[dict],
    max_tokens: int = 1500,  # tldr + 5-8 themes + 3-5 acts + locations can be long
    timeout: float = 120.0,
) -> dict:
    title = request.get("title", "")
    window_lines = []
    sub_lines = []
    for req_w, res_w in zip(request["windows"], window_results):
        if res_w.get("skip"):
            continue
        start, end = req_w["start_seconds"], req_w["end_seconds"]
        desc = res_w.get("description", "").strip()
        window_lines.append(f"[{start:.0f}s–{end:.0f}s] {desc}")
        sub = (req_w.get("subtitle_text") or "").strip()
        if sub:
            sub_lines.append(f"[{start:.0f}s] {sub}")

    if not window_lines:
        raise ValueError(f"[{request['natural_key']}] All windows skipped; cannot summarise.")

    user_text = SUMMARY_USER_TEMPLATE.format(
        title=title,
        window_block="\n".join(window_lines),
        subtitle_block="\n".join(sub_lines) if sub_lines else "(no spoken dialogue)",
    )

    response = _chat_with_retry(
        client,
        model=model_name,
        messages=[
            {"role": "system", "content": SUMMARY_SYSTEM},
            {"role": "user", "content": user_text},
        ],
        max_tokens=max_tokens,
        temperature=0.0,
        timeout=timeout,
    )
    raw = response.choices[0].message.content or ""
    result = extract_json(raw, "video summary")

    tldr = str(result.get("tldr", "")).strip()
    if not tldr:
        raise ValueError(f"[{request['natural_key']}] Summary returned empty tldr. Raw: {raw[:500]!r}")
    word_count = len(tldr.split())
    if word_count < 8:
        raise ValueError(f"[{request['natural_key']}] tldr suspiciously short ({word_count} words).")
    if word_count < 60 or word_count > 160:
        log.warning("  video summary: tldr %d words (target 80-140)", word_count)
    else:
        log.info("  video summary: tldr %d words OK", word_count)

    result.setdefault("themes", [])
    result.setdefault("acts", [])
    result.setdefault("locations", [])
    return result


# ---------------------------------------------------------------------------
# Per-video pipeline
# ---------------------------------------------------------------------------

def process_one_video(
    *,
    natural_key: str,
    language: str,
    label: str,
    client: Any,
    model_name: str,
    work_dir: str,
    n_frames: int,
    no_prepare: bool,
    no_persist: bool,
) -> None:
    video_work = os.path.join(work_dir, natural_key)
    request_path = os.path.join(video_work, "request.json")
    output_path = os.path.join(video_work, "response.json")

    if not no_prepare and not os.path.exists(request_path):
        log.info("[%s] Running prepare ...", natural_key)
        _run_prepare(natural_key, language, label, work_dir)
    if not os.path.exists(request_path):
        raise FileNotFoundError(
            f"[{natural_key}] request.json not found at {request_path}. "
            f"Run: python scripts/scene-index-local.py prepare --keys {natural_key}"
        )

    with open(request_path, "r", encoding="utf-8") as fh:
        request = json.load(fh)

    log.info(
        "[%s] %d windows (%s subs) → response.json",
        natural_key, len(request["windows"]), request.get("subtitle_source", "?"),
    )

    window_results: list[dict] = []
    t_vlm = time.time()
    for i, window in enumerate(request["windows"]):
        log.info(
            "  window %d/%d [%ds–%ds] ...",
            i + 1, len(request["windows"]),
            window["start_seconds"], window["end_seconds"],
        )
        result = run_window_inference(client, model_name, window, n_frames=n_frames)
        result["window_index"] = window["window_index"]
        window_results.append(result)

    log.info("  windows done in %.1fs", time.time() - t_vlm)
    log.info("  running video summary ...")
    summary = run_summary_inference(client, model_name, request, window_results)

    response = {
        "natural_key": natural_key,
        "windows": [
            {
                "window_index": r["window_index"],
                "description": r.get("description", ""),
                "onscreen_text_places": r.get("onscreen_text_places", []),
                **({"skip": True, "skip_reason": r.get("skip_reason", "")}
                   if r.get("skip") else {}),
            }
            for r in window_results
        ],
        "locations": summary.get("locations", []),
        "video_summary": {
            "tldr": summary["tldr"],
            "themes": summary.get("themes", []),
            "acts": summary.get("acts", []),
        },
        "_vlm_model": model_name,
        "_generated_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
    }
    with open(output_path, "w", encoding="utf-8") as fh:
        json.dump(response, fh, indent=2, ensure_ascii=False)
    log.info("  wrote %s", output_path)

    if not no_persist:
        log.info("[%s] Running persist ...", natural_key)
        _run_persist(natural_key, language, work_dir)


def _run_prepare(natural_key: str, language: str, label: str, work_dir: str) -> None:
    cmd = [
        sys.executable,
        os.path.join(SCRIPT_DIR, "scene-index-local.py"),
        "prepare",
        "--keys", natural_key,
        "--language", language,
        "--label", label,
        "--work-dir", work_dir,
    ]
    r = subprocess.run(cmd, capture_output=False)
    if r.returncode != 0:
        raise RuntimeError(f"prepare failed for {natural_key} (exit {r.returncode}).")


def _run_persist(natural_key: str, language: str, work_dir: str) -> None:
    cmd = [
        sys.executable,
        os.path.join(SCRIPT_DIR, "scene-index-local.py"),
        "persist",
        "--keys", natural_key,
        "--language", language,
        "--work-dir", work_dir,
    ]
    r = subprocess.run(cmd, capture_output=False)
    if r.returncode != 0:
        raise RuntimeError(
            f"persist failed for {natural_key} (exit {r.returncode}). "
            f"DB is unchanged. Fix the issue before continuing."
        )


# ---------------------------------------------------------------------------
# Key selection
# ---------------------------------------------------------------------------

def select_next_keys(language: str, n: int, db_path: str) -> list[str]:
    from catalog_priority import load_cached_catalog
    from scene_processing import scene_db
    from scene_processing.exclusions import load_exclusions, should_exclude
    from scene_processing.index_filter import priority_tier, should_index

    catalog = load_cached_catalog(language)
    exclusions = load_exclusions()
    with scene_db.open_db(db_path) as conn:
        done = scene_db.successful_run_keys(conn, language)
    kept: list[tuple[str, dict]] = []
    for key, item in catalog.items():
        if not should_index(item):
            continue
        excluded, _ = should_exclude(key, item, exclusions)
        if excluded or key in done:
            continue
        kept.append((key, item))
    kept.sort(key=lambda kv: (priority_tier(kv[1]), kv[1].get("duration") or 0, kv[0]))
    return [k for k, _ in kept[:n]]


# ---------------------------------------------------------------------------
# Checkpoint
# ---------------------------------------------------------------------------

def load_checkpoint(checkpoint_path: str) -> set[str]:
    """Load set of already-completed keys from checkpoint file."""
    if not os.path.exists(checkpoint_path):
        return set()
    with open(checkpoint_path, "r", encoding="utf-8") as fh:
        data = json.load(fh)
    return set(data.get("completed", []))


def save_checkpoint(checkpoint_path: str, completed: set[str]) -> None:
    with open(checkpoint_path, "w", encoding="utf-8") as fh:
        json.dump({"completed": sorted(completed), "updated_at": time.strftime("%Y-%m-%dT%H:%M:%S")},
                  fh, indent=2)


# ---------------------------------------------------------------------------
# Webhook
# ---------------------------------------------------------------------------

def send_webhook(url: str, payload: dict) -> None:
    if not url:
        return
    try:
        import requests
        r = requests.post(url, json=payload, timeout=10)
        log.info("Webhook sent: HTTP %d", r.status_code)
    except Exception as exc:
        log.warning("Webhook failed (non-fatal): %s", exc)


# ---------------------------------------------------------------------------
# CLI commands
# ---------------------------------------------------------------------------

def _data_root() -> str:
    from runtime_paths import get_data_root
    return get_data_root()


def cmd_process(args: argparse.Namespace) -> int:
    from runtime_paths import ensure_runtime_dirs
    ensure_runtime_dirs()

    work_dir = args.work_dir or os.path.join(_data_root(), "scene-local-work")
    db_path = args.db or os.path.join(_data_root(), "scene_index.db")
    os.makedirs(work_dir, exist_ok=True)

    checkpoint_path = args.checkpoint or os.path.join(work_dir, "runpod_checkpoint.json")
    completed = load_checkpoint(checkpoint_path)
    log.info("Checkpoint: %d already completed", len(completed))

    # Resolve key list
    if args.keys:
        all_keys = [k.strip() for k in args.keys.split(",") if k.strip()]
    else:
        log.info("Selecting next %d video(s) from priority queue ...", args.next)
        all_keys = select_next_keys(args.language, args.next, db_path)
        if not all_keys:
            print("Nothing to process — all priority videos are already indexed.")
            return 0

    # Subtract already-completed
    keys = [k for k in all_keys if k not in completed]
    if not keys:
        print(f"All {len(all_keys)} selected video(s) are already in the checkpoint. Done.")
        return 0

    log.info(
        "Processing %d video(s) (%d already done, %d remaining)",
        len(all_keys), len(all_keys) - len(keys), len(keys),
    )

    # Build vllm client
    client = _make_client(args.vllm_url, args.api_key)

    # Verify the API endpoint is reachable using a plain HTTP request —
    # the Python OpenAI SDK v2 has a response-parsing incompatibility with
    # some providers (together.ai returns a bare list, not a paged object).
    try:
        import urllib.request
        base = args.vllm_url.rstrip("/")
        req = urllib.request.Request(
            f"{base}/models",
            headers={"Authorization": f"Bearer {args.api_key}"},
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read())
        # Response may be a list or {"data": [...]}
        items = data if isinstance(data, list) else data.get("data", [])
        available = [m.get("id", "") for m in items if isinstance(m, dict)]
        log.info("API connected. Models found: %d", len(available))
        if args.vllm_model not in available:
            log.warning(
                "Model '%s' not in server list — proceeding anyway "
                "(provider may use a different alias).",
                args.vllm_model,
            )
        else:
            log.info("Model '%s' confirmed available.", args.vllm_model)
    except Exception as exc:
        # 403 from together.ai usually means read-only mode (needs deposit)
        # but the models list endpoint itself may still block Python user-agents.
        # Don't abort — let the first real inference call be the true test.
        log.warning(
            "Could not verify API connection (%s). "
            "Proceeding — first inference call will confirm if the key works.",
            exc,
        )

    batch_start = time.time()
    failed: list[str] = []

    for i, key in enumerate(keys, 1):
        print(f"\n[{i}/{len(keys)}] {key}")
        try:
            process_one_video(
                natural_key=key,
                language=args.language,
                label=args.label,
                client=client,
                model_name=args.vllm_model,
                work_dir=work_dir,
                n_frames=args.frames,
                no_prepare=args.no_prepare,
                no_persist=not args.persist,
            )
            completed.add(key)
            save_checkpoint(checkpoint_path, completed)
            print(f"  [{key}] DONE — checkpoint saved")

            # Webhook on batch boundary
            if args.webhook_url and i % args.batch_size == 0:
                elapsed = time.time() - batch_start
                send_webhook(args.webhook_url, {
                    "event": "batch_complete",
                    "completed": i,
                    "total": len(keys),
                    "failed": len(failed),
                    "elapsed_seconds": round(elapsed),
                    "latest_key": key,
                })

        except Exception as exc:
            log.error("[%s] FAILED: %s", key, exc)
            failed.append(key)
            if args.stop_on_error:
                print(f"\nStopping on first error (--stop-on-error). Failed: {key}")
                if args.webhook_url:
                    send_webhook(args.webhook_url, {
                        "event": "stopped_on_error",
                        "failed_key": key,
                        "error": str(exc),
                        "completed_before_stop": i - 1,
                    })
                return 1

    elapsed = time.time() - batch_start
    print(
        f"\n{'All' if not failed else str(len(keys) - len(failed)) + '/' + str(len(keys))} "
        f"video(s) processed in {elapsed:.0f}s. "
        f"{'Failed: ' + ', '.join(failed) if failed else 'No failures.'}"
    )

    if args.webhook_url:
        send_webhook(args.webhook_url, {
            "event": "run_complete",
            "processed": len(keys) - len(failed),
            "failed": len(failed),
            "failed_keys": failed,
            "elapsed_seconds": round(elapsed),
        })

    return 1 if failed else 0


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    sub = parser.add_subparsers(dest="command", required=True)

    p_proc = sub.add_parser(
        "process",
        help="Run vllm-backed VLM on videos and persist to scene_index.db",
    )
    p_proc.add_argument("--keys", default=None,
                        help="Comma-separated natural_keys. If omitted, uses --next.")
    p_proc.add_argument("--next", type=int, default=50,
                        help="Pick next N priority videos (default 50)")
    p_proc.add_argument("--language", default="E")
    p_proc.add_argument("--label", default="720p")
    p_proc.add_argument("--vllm-url", default="http://localhost:8000/v1",
                        help="vllm OpenAI-compatible base URL (default: http://localhost:8000/v1)")
    p_proc.add_argument("--vllm-model", default="Qwen/Qwen2.5-VL-72B-Instruct",
                        help="Model name as registered in vllm (default: Qwen/Qwen2.5-VL-72B-Instruct)")
    p_proc.add_argument("--api-key", default="",
                        help="API key for vllm (usually empty for local deployments)")
    p_proc.add_argument("--frames", type=int, default=3,
                        help="Frames to send per window (default 3)")
    p_proc.add_argument("--batch-size", type=int, default=50,
                        help="Send webhook every N videos (default 50)")
    p_proc.add_argument("--persist", action="store_true", default=True,
                        help="Run persist after each video (default: True)")
    p_proc.add_argument("--no-persist", dest="persist", action="store_false",
                        help="Skip persist step (write response.json only)")
    p_proc.add_argument("--no-prepare", action="store_true",
                        help="Skip prepare; fail if request.json is missing")
    p_proc.add_argument("--stop-on-error", action="store_true",
                        help="Halt on first video failure (default: continue)")
    p_proc.add_argument("--checkpoint", default=None,
                        help="Path to checkpoint JSON (default: <work-dir>/runpod_checkpoint.json)")
    p_proc.add_argument("--webhook-url", default=None,
                        help="POST progress updates here after each batch and at completion")
    p_proc.add_argument("--work-dir", default=None)
    p_proc.add_argument("--db", default=None)
    p_proc.set_defaults(func=cmd_process)

    args = parser.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())