MEV_code / qwen /generate_multi_event_annotations.py
Upload MEV code
7a5c845 verified 6 months ago
92.7 kB
#!/usr/bin/env python3
from __future__ import annotations

import argparse
import ast
import json
import logging
import re
import subprocess
import sys
import tempfile
import textwrap
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable


VIDEO_SUFFIXES = {".mp4", ".mov", ".m4v", ".avi", ".mkv", ".webm"}
SEGMENT_RE = re.compile(
    r"^(?P<uuid>.+)-(?P<start>\d{2}\.\d{2}\.\d{2}\.\d{3})-(?P<end>\d{2}\.\d{2}\.\d{2}\.\d{3})-seg(?P<segid>\d+)\.(?P<ext>[A-Za-z0-9]+)$"
)
EVENT_CAMERA_LABELS = [
    "dolly-in",
    "dolly-out",
    "pedestal-up",
    "pedestal-down",
    "truck-left",
    "truck-right",
    "pan-left",
    "pan-right",
    "tilt-up",
    "tilt-down",
    "roll-CW",
    "roll-CCW",
]
EVENT_CAMERA_LABEL_MAP = {label.lower(): label for label in EVENT_CAMERA_LABELS}
SUBJECT_CATEGORIES = {"human", "animal", "body_part", "object", "unknown"}
GLOBAL_CAMERA_PREFIXES = {
    "static",
    "moving-throughout",
    "intermittent",
    "motion-changes",
    "mixed",
    "near-static",
}
GLOBAL_CAMERA_PREFIX_MAP = {prefix.lower(): prefix for prefix in GLOBAL_CAMERA_PREFIXES}
DEFAULT_ROOT = Path(__file__).resolve().parent
DEFAULT_CAMERA_GUIDE = DEFAULT_ROOT / "camera_motion.md"


@dataclass(frozen=True)
class SegmentInfo:
    uuid: str
    path: Path
    start_time: float
    end_time: float
    duration: float
    seg_id: int


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Generate multi-event annotations with a local Qwen3-VL checkpoint."
    )
    parser.add_argument(
        "--input",
        required=True,
        help="Path to a directory of segmented videos or a single segmented video file.",
    )
    parser.add_argument(
        "--output",
        required=True,
        help="Output JSON path.",
    )
    parser.add_argument(
        "--model-path",
        default=str(DEFAULT_ROOT / "checkpoints"),
        help="Local Qwen3-VL checkpoint directory.",
    )
    parser.add_argument(
        "--camera-motion-guide",
        default=str(DEFAULT_CAMERA_GUIDE),
        help="Path to camera_motion.md used to prime the prompts.",
    )
    parser.add_argument(
        "--recursive",
        action="store_true",
        help="Recursively search for segment videos when the input is a directory.",
    )
    parser.add_argument(
        "--uuids",
        nargs="*",
        default=None,
        help="Optional UUID allow-list.",
    )
    parser.add_argument(
        "--uuids-file",
        default=None,
        help="Optional text file containing one UUID per line.",
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=None,
        help="Only process the first N UUIDs after sorting and filtering.",
    )
    parser.add_argument(
        "--num-shards",
        type=int,
        default=1,
        help="Deterministically split the UUID list into N shards and only process one shard in this process.",
    )
    parser.add_argument(
        "--shard-index",
        type=int,
        default=0,
        help="0-based shard index used together with --num-shards.",
    )
    parser.add_argument(
        "--print-uuids-json",
        action="store_true",
        help="Print the discovered UUID list as JSON and exit. Useful for external multi-GPU launchers.",
    )
    parser.add_argument(
        "--overwrite",
        action="store_true",
        help="Overwrite an existing output JSON instead of failing.",
    )
    parser.add_argument(
        "--resume",
        action="store_true",
        help="Merge into an existing output JSON and skip UUIDs that are already present.",
    )
    parser.add_argument(
        "--continue-on-error",
        action="store_true",
        help="Log UUID failures and continue instead of stopping immediately.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Only validate file grouping and print the discovered UUIDs.",
    )
    parser.add_argument(
        "--keep-temp",
        action="store_true",
        help="Keep temporary stitched videos for inspection.",
    )
    parser.add_argument(
        "--temp-dir",
        default=None,
        help="Optional temp directory root for stitched videos.",
    )
    parser.add_argument(
        "--video-suffixes",
        default=",".join(sorted(VIDEO_SUFFIXES)),
        help="Comma-separated video suffixes to include.",
    )
    parser.add_argument(
        "--fps",
        type=float,
        default=1.0,
        help="Sampling FPS passed to the Qwen video processor.",
    )
    parser.add_argument(
        "--max-pixels",
        type=int,
        default=360000,
        help="Optional max_pixels hint passed to the Qwen video processor.",
    )
    parser.add_argument(
        "--context-max-new-tokens",
        type=int,
        default=1024,
        help="Max new tokens for the global context pass.",
    )
    parser.add_argument(
        "--event-max-new-tokens",
        type=int,
        default=1024,
        help="Max new tokens for each event pass.",
    )
    parser.add_argument(
        "--global-max-new-tokens",
        type=int,
        default=4096,
        help="Max new tokens for the final global prompt pass.",
    )
    parser.add_argument(
        "--temperature",
        type=float,
        default=0.2,
        help="Sampling temperature.",
    )
    parser.add_argument(
        "--top-p",
        type=float,
        default=0.9,
        help="Sampling top-p.",
    )
    parser.add_argument(
        "--top-k",
        type=int,
        default=20,
        help="Sampling top-k.",
    )
    parser.add_argument(
        "--repetition-penalty",
        type=float,
        default=1.05,
        help="Generation repetition penalty.",
    )
    parser.add_argument(
        "--greedy",
        action="store_true",
        help="Disable sampling and use greedy decoding.",
    )
    parser.add_argument(
        "--device-map",
        default="auto",
        help="Transformers device_map argument.",
    )
    parser.add_argument(
        "--dtype",
        default="bfloat16",
        choices=["auto", "float16", "bfloat16", "float32"],
        help="Torch dtype for model loading.",
    )
    parser.add_argument(
        "--attn-implementation",
        default="flash_attention_2",
        help="Optional attn_implementation passed to from_pretrained. Use an empty string to disable.",
    )
    parser.add_argument(
        "--ffmpeg-bin",
        default="ffmpeg",
        help="ffmpeg executable used to stitch per-UUID segments.",
    )
    parser.add_argument(
        "--log-level",
        default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
        help="Python log level.",
    )
    parser.set_defaults(trust_remote_code=True)
    parser.add_argument(
        "--no-trust-remote-code",
        dest="trust_remote_code",
        action="store_false",
        help="Disable trust_remote_code when loading the model and processor.",
    )
    return parser.parse_args()


def setup_logging(level: str) -> None:
    logging.basicConfig(
        level=getattr(logging, level),
        format="%(asctime)s | %(levelname)s | %(message)s",
    )


def parse_video_suffixes(raw: str) -> set[str]:
    suffixes: set[str] = set()
    for item in raw.split(","):
        token = item.strip().lower()
        if not token:
            continue
        suffixes.add(token if token.startswith(".") else f".{token}")
    return suffixes or set(VIDEO_SUFFIXES)


def timecode_to_seconds(value: str) -> float:
    hours, minutes, seconds, millis = value.split(".")
    total = (
        int(hours) * 3600
        + int(minutes) * 60
        + int(seconds)
        + int(millis) / 1000.0
    )
    return round(total, 3)


def parse_segment_filename(path: Path) -> SegmentInfo:
    match = SEGMENT_RE.match(path.name)
    if not match:
        raise ValueError(
            f"Unsupported segment filename: {path.name}. Expected uuid-start-end-segN.ext."
        )
    start_time = timecode_to_seconds(match.group("start"))
    end_time = timecode_to_seconds(match.group("end"))
    duration = round(max(end_time - start_time, 0.0), 3)
    return SegmentInfo(
        uuid=match.group("uuid"),
        path=path.resolve(),
        start_time=start_time,
        end_time=end_time,
        duration=duration,
        seg_id=int(match.group("segid")),
    )


def discover_segment_files(
    input_path: Path,
    recursive: bool,
    suffixes: set[str],
) -> list[Path]:
    if input_path.is_file():
        return [input_path.resolve()]
    if not input_path.is_dir():
        raise FileNotFoundError(f"Input path does not exist: {input_path}")
    iterator = input_path.rglob("*") if recursive else input_path.iterdir()
    return sorted(
        path.resolve()
        for path in iterator
        if path.is_file() and path.suffix.lower() in suffixes
    )


def collect_segment_groups(
    input_path: Path,
    recursive: bool,
    suffixes: set[str],
    allow_uuids: set[str] | None,
) -> dict[str, list[SegmentInfo]]:
    groups: dict[str, list[SegmentInfo]] = {}
    skipped = 0
    for path in discover_segment_files(input_path, recursive, suffixes):
        try:
            segment = parse_segment_filename(path)
        except ValueError as exc:
            logging.warning("%s", exc)
            skipped += 1
            continue
        if allow_uuids and segment.uuid not in allow_uuids:
            continue
        groups.setdefault(segment.uuid, []).append(segment)
    for uuid, segments in groups.items():
        groups[uuid] = sorted(
            segments,
            key=lambda item: (item.start_time, item.seg_id, item.path.name),
        )
    if skipped:
        logging.info("Skipped %d non-matching files.", skipped)
    return dict(sorted(groups.items(), key=lambda item: item[0]))


def load_camera_motion_guide(path: Path) -> str:
    if path.exists():
        return path.read_text(encoding="utf-8").strip()
    fallback_lines = [
        "# Camera Motion Labels",
        "",
        "| Camera Motion Label | Explanation |",
        "|---|---|",
    ]
    explanations = {
        "dolly-in": "The camera moves forward relative to the ground plane and the initial frame.",
        "dolly-out": "The camera moves backward relative to the ground plane and the initial frame.",
        "pedestal-up": "Select this when the camera moves upward clearly and consistently relative to the ground or the orientation of the initial frame.",
        "pedestal-down": "Select this when the camera moves downward clearly and consistently relative to the ground or the orientation of the initial frame.",
        "truck-left": "The camera physically moves to the left, changing its position relative to the initial frame.",
        "truck-right": "The camera physically moves to the right, changing its position relative to the initial frame.",
        "pan-left": "The camera rotates its angle by pivoting left with respect to the initial frame.",
        "pan-right": "The camera rotates its angle by pivoting right with respect to the initial frame.",
        "tilt-up": "The camera rotates its angle upward vertically with respect to the initial frame.",
        "tilt-down": "The camera rotates its angle downward vertically with respect to the initial frame.",
        "roll-CW": "The camera performs a clear and consistent clockwise roll by rotating around its own optical center.",
        "roll-CCW": "The camera performs a clear and consistent counterclockwise roll by rotating around its own optical center.",
    }
    for label in EVENT_CAMERA_LABELS:
        fallback_lines.append(f"| `{label}` | {explanations[label]} |")
    return "\n".join(fallback_lines)


def apply_uuid_shard(uuids: list[str], num_shards: int, shard_index: int) -> list[str]:
    if num_shards < 1:
        raise ValueError(f"--num-shards must be >= 1, got {num_shards}")
    if shard_index < 0 or shard_index >= num_shards:
        raise ValueError(
            f"--shard-index must be in [0, {num_shards - 1}], got {shard_index}"
        )
    if num_shards == 1:
        return list(uuids)
    return [uuid for idx, uuid in enumerate(uuids) if idx % num_shards == shard_index]


def load_uuid_allow_list(raw_uuids: list[str] | None, uuids_file: str | None) -> set[str] | None:
    values: list[str] = []
    if raw_uuids:
        values.extend(raw_uuids)
    if uuids_file:
        path = Path(uuids_file).expanduser().resolve()
        for line in path.read_text(encoding="utf-8").splitlines():
            token = line.strip()
            if token:
                values.append(token)
    if not values:
        return None
    return set(values)


def normalize_whitespace(value: Any) -> str:
    if value is None:
        return ""
    return re.sub(r"\s+", " ", str(value)).strip()


def normalize_subject_category(value: Any) -> str:
    category = normalize_whitespace(value).lower().replace("-", "_").replace(" ", "_")
    if category in SUBJECT_CATEGORIES:
        return category
    return "unknown"


def coerce_bool(value: Any, default: bool = False) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)):
        return bool(value)
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in {"true", "yes", "y", "1"}:
            return True
        if lowered in {"false", "no", "n", "0"}:
            return False
    return default


def coerce_optional_bool(value: Any) -> bool | None:
    if value is None:
        return None
    if isinstance(value, str) and value.strip().lower() in {"none", "null", ""}:
        return None
    return coerce_bool(value)


def normalize_camera_motion_label(value: Any) -> str:
    label = normalize_whitespace(value)
    if not label:
        return ""
    lowered = label.lower()
    if lowered in {"static", "none", "no-motion", "no motion", "n/a", "na"}:
        return ""
    return EVENT_CAMERA_LABEL_MAP.get(lowered, "")


def normalize_global_camera_prefix(value: Any) -> str:
    prefix = normalize_whitespace(value)
    if not prefix:
        return ""
    lowered = prefix.lower()
    if lowered in GLOBAL_CAMERA_PREFIX_MAP:
        return GLOBAL_CAMERA_PREFIX_MAP[lowered]
    return ""


def word_count(text: str) -> int:
    if not text:
        return 0
    return len(re.findall(r"\b\S+\b", text))


def normalize_camera_motion_surface_text(value: Any) -> str:
    cleaned = normalize_whitespace(value)
    if not cleaned:
        return ""
    cleaned = re.sub(r"\bdollys\b", "dollies", cleaned, flags=re.IGNORECASE)
    cleaned = re.sub(r"\bdolly-ins\b", "dollies in", cleaned, flags=re.IGNORECASE)
    cleaned = re.sub(r"\bdolly-outs\b", "dollies out", cleaned, flags=re.IGNORECASE)
    cleaned = re.sub(
        r"\b(as|while|when)\s+(the camera|camera|it)\s+dolly[\s-]*(in|out)\b",
        lambda match: f"{match.group(1)} {match.group(2)} dollies {match.group(3).lower()}",
        cleaned,
        flags=re.IGNORECASE,
    )
    cleaned = re.sub(
        r"\b(the camera|camera|it)\s+dolly[\s-]*(in|out)\b",
        lambda match: f"{match.group(1)} dollies {match.group(2).lower()}",
        cleaned,
        flags=re.IGNORECASE,
    )
    return cleaned


def describe_camera_motion(label: str, scope: str = "event") -> str:
    fragments = {
        "dolly-in": "dollies in",
        "dolly-out": "dollies out",
        "pedestal-up": "pedestals up",
        "pedestal-down": "pedestals down",
        "truck-left": "trucks left",
        "truck-right": "trucks right",
        "pan-left": "pans left",
        "pan-right": "pans right",
        "tilt-up": "tilts up",
        "tilt-down": "tilts down",
        "roll-CW": "rolls clockwise",
        "roll-CCW": "rolls counterclockwise",
    }
    if label not in fragments:
        return ""
    if scope == "global":
        return f"The camera {fragments[label]} through most of the video."
    return f"The camera {fragments[label]}."


def is_following_camera_description(text: str) -> bool:
    lowered = normalize_whitespace(text).lower()
    if not lowered:
        return False
    keywords = ("follow", "follows", "following", "track", "tracking", "tracked", "keeps pace")
    return any(keyword in lowered for keyword in keywords)


def camera_motion_fragment(label: str) -> str:
    sentence = describe_camera_motion(label)
    if not sentence:
        return ""
    fragment = sentence.removeprefix("The camera ").strip()
    return fragment[:-1] if fragment.endswith(".") else fragment


def derive_global_camera_from_events(
    events: list[dict[str, Any]],
) -> tuple[str, str, str]:
    sequence_description = summarize_global_camera_sequence(events)
    moving_events = [event for event in events if coerce_bool(event.get("has_camera_motion"))]
    labels = [event.get("camera_motion_label", "") for event in moving_events if event.get("camera_motion_label")]
    unlabeled_descs = [
        normalize_whitespace(event.get("camera_motion_desc"))
        for event in moving_events
        if not event.get("camera_motion_label") and normalize_whitespace(event.get("camera_motion_desc"))
    ]
    if not moving_events:
        return (
            "static",
            "",
            sequence_description,
        )
    if len(moving_events) == len(events):
        unique_labels = {label for label in labels if label}
        if len(unique_labels) == 1:
            label = next(iter(unique_labels))
            return ("moving-throughout", label, sequence_description)
        if unlabeled_descs and all(is_following_camera_description(desc) for desc in unlabeled_descs):
            return (
                "moving-throughout",
                "",
                sequence_description,
            )
        return (
            "motion-changes",
            "",
            sequence_description,
        )
    unique_labels = {label for label in labels if label}
    if len(unique_labels) == 1:
        label = next(iter(unique_labels))
        return (
            "intermittent",
            label,
            sequence_description,
        )
    if unlabeled_descs and all(is_following_camera_description(desc) for desc in unlabeled_descs):
        return (
            "intermittent",
            "",
            sequence_description,
        )
    return (
        "motion-changes",
        "",
        sequence_description,
    )


def global_camera_conflicts_with_events(
    prefix: str,
    label: str,
    events: list[dict[str, Any]],
) -> bool:
    moving_events = [event for event in events if coerce_bool(event.get("has_camera_motion"))]
    total_events = len(events)
    labeled_events = [event for event in moving_events if event.get("camera_motion_label")]
    unique_labels = {event.get("camera_motion_label", "") for event in labeled_events if event.get("camera_motion_label")}

    if prefix in {"static", "near-static"}:
        return bool(moving_events)

    if prefix == "moving-throughout":
        if total_events == 0 or len(moving_events) != total_events:
            return True
        if label and unique_labels and unique_labels != {label}:
            return True
        return False

    if prefix == "intermittent":
        if total_events == 0 or len(moving_events) == 0 or len(moving_events) == total_events:
            return True
        if label and unique_labels and label not in unique_labels:
            return True
        return False

    if prefix == "motion-changes":
        return not moving_events

    if prefix in {"mixed"}:
        return not moving_events

    return False


def camera_motion_description_to_phrase(text: str) -> str:
    sentence = normalize_whitespace(text).rstrip(".")
    if not sentence:
        return ""
    lowered = sentence.lower()
    for prefix in ("the camera ", "camera "):
        if lowered.startswith(prefix):
            sentence = sentence[len(prefix):]
            break
    return sentence_to_clause(sentence)


def event_camera_phase(event: dict[str, Any]) -> tuple[str, str]:
    if not coerce_bool(event.get("has_camera_motion")):
        return ("static", "the camera holds a static framing")
    label = normalize_camera_motion_label(event.get("camera_motion_label"))
    if label:
        return (label, f"the camera {camera_motion_fragment(label)}")
    description = normalize_whitespace(event.get("camera_motion_desc"))
    if is_following_camera_description(description):
        return (
            "following",
            "the camera remains relatively stable in framing while following the subject",
        )
    if description:
        phrase = camera_motion_description_to_phrase(description)
        if phrase.startswith("the "):
            return (f"desc:{normalized_caption_key(description)}", phrase)
        return (f"desc:{normalized_caption_key(description)}", f"the camera {phrase}")
    return ("moving", "the camera moves relative to the scene")


def summarize_global_camera_sequence(events: list[dict[str, Any]]) -> str:
    if not events:
        return "The camera remains static or near-static throughout the video."

    phases: list[tuple[str, str]] = []
    for event in events:
        key, phrase = event_camera_phase(event)
        if phases and phases[-1][0] == key:
            continue
        phases.append((key, phrase))

    if not phases:
        return "The camera remains static or near-static throughout the video."

    if len(phases) == 1:
        key, phrase = phases[0]
        if key == "static":
            return "The camera remains static or near-static throughout the video."
        return f"{phrase[0].upper() + phrase[1:]} throughout the video."

    clauses = [phrase for _, phrase in phases[:4]]
    if len(phases) == 2:
        return f"{clauses[0][0].upper() + clauses[0][1:]} at first, then {clauses[1]}."
    if len(phases) == 3:
        return f"{clauses[0][0].upper() + clauses[0][1:]} at first, then {clauses[1]}, and later {clauses[2]}."
    return (
        f"{clauses[0][0].upper() + clauses[0][1:]} at first, then {clauses[1]}, "
        f"later {clauses[2]}, and eventually {clauses[3]}."
    )


def compose_global_camera_movement(prefix: str, label: str, description: str) -> str:
    cleaned = normalize_camera_motion_surface_text(description)
    if "|" in cleaned:
        parts = [normalize_whitespace(part) for part in cleaned.split("|") if normalize_whitespace(part)]
        filtered = [
            part
            for part in parts
            if not normalize_global_camera_prefix(part) and not normalize_camera_motion_label(part)
        ]
        cleaned = " ".join(filtered) if filtered else " ".join(parts)
    if cleaned:
        return cleaned
    if prefix in {"static", "near-static"}:
        return "The camera remains static or near-static throughout the video."
    if label:
        return describe_camera_motion(label, scope="global")
    if prefix == "moving-throughout":
        return "The camera keeps moving throughout the video."
    if prefix == "intermittent":
        return "The camera alternates between static framing and periods of movement."
    if prefix == "motion-changes":
        return "The camera motion changes across the video."
    return "The camera behavior remains visually consistent across the video."


def normalize_camera_movement_description_output(value: Any) -> str:
    cleaned = normalize_camera_motion_surface_text(value)
    if not cleaned:
        return ""
    if "|" in cleaned:
        parts = [normalize_whitespace(part) for part in cleaned.split("|") if normalize_whitespace(part)]
        non_meta = [
            part
            for part in parts
            if not normalize_global_camera_prefix(part) and not normalize_camera_motion_label(part)
        ]
        cleaned = " ".join(non_meta) if non_meta else ""
    if normalize_global_camera_prefix(cleaned) or normalize_camera_motion_label(cleaned):
        return ""
    return cleaned


def fallback_subject_profile(subject_registry: list[dict[str, str]]) -> str:
    if not subject_registry:
        return ""
    if len(subject_registry) == 1:
        subject = subject_registry[0]
        term = subject.get("term", "subject")
        profile = subject.get("profile", "")
        category = subject.get("category", "unknown")
        if category == "animal":
            prefix = f"A {term} is the main subject of the video, and no human is visibly established."
        elif category in {"body_part", "object"}:
            prefix = f"A {term} is the main visible subject of the video, and no full human subject is clearly established."
        else:
            prefix = f"A {term} is the main subject of the video."
        if profile:
            return f"{prefix} {profile}"
        return prefix
    chunks = []
    for subject in subject_registry:
        term = subject.get("term", "subject")
        profile = subject.get("profile", "")
        category = subject.get("category", "unknown")
        prefix = f"{term} ({category})" if category != "unknown" else term
        chunks.append(f"{prefix}: {profile}" if profile else prefix)
    return "Multiple recurring subjects appear across the video. " + "; ".join(chunks)


def normalize_focus_object(value: Any, subject_registry: list[dict[str, str]]) -> str:
    focus = normalize_whitespace(value)
    if focus:
        return focus
    if subject_registry:
        return subject_registry[0].get("term", "main subject")
    return "main subject"


def normalized_caption_key(text: str) -> str:
    lowered = normalize_whitespace(text).lower()
    return re.sub(r"[^a-z0-9]+", " ", lowered).strip()


def sentence_to_clause(text: str) -> str:
    sentence = normalize_whitespace(text).rstrip(".")
    if not sentence:
        return ""
    if sentence[0].isupper():
        return sentence[0].lower() + sentence[1:]
    return sentence


def join_clauses(clauses: list[str]) -> str:
    items = [item.strip() for item in clauses if item and item.strip()]
    if not items:
        return ""
    if len(items) == 1:
        return items[0]
    if len(items) == 2:
        return f"{items[0]} and {items[1]}"
    return f"{', '.join(items[:-1])}, and {items[-1]}"


def disambiguate_repeated_caption(
    current_caption: str,
    previous_caption: str,
    event: dict[str, Any],
) -> str:
    if normalized_caption_key(current_caption) != normalized_caption_key(previous_caption):
        return current_caption
    additions = []
    if event.get("has_appearance"):
        additions.append("with a new subject entering view")
    if event.get("has_disappearance"):
        additions.append("with a subject leaving view")
    if event.get("focus_object"):
        additions.append(f"with attention settling on the {event['focus_object']}")
    if event.get("camera_motion_desc"):
        additions.append(f"as {sentence_to_clause(event['camera_motion_desc'])}")
    if event.get("has_lighting_change"):
        additions.append("under a noticeable lighting shift")
    if event.get("has_environmental_change") and event.get("focus_object"):
        additions.append(f"with further visible change around the {event['focus_object']}")
    elif event.get("has_environmental_change"):
        additions.append("with further visible change in the scene detail")
    if not additions:
        additions.append("with a subtle visible shift from the previous segment")
    base = current_caption.rstrip().rstrip(".")
    return f"{base}, {join_clauses(additions)}."


def extract_json_text(raw_text: str) -> str:
    text = raw_text.strip()
    fenced_matches = re.findall(r"```(?:json)?\s*(.*?)```", text, flags=re.DOTALL | re.IGNORECASE)
    candidates = [match.strip() for match in fenced_matches if match.strip()]
    candidates.append(text)
    for candidate in candidates:
        if candidate.startswith("{") or candidate.startswith("["):
            return candidate
    for open_char, close_char in (("{", "}"), ("[", "]")):
        start = text.find(open_char)
        if start == -1:
            continue
        depth = 0
        in_string = False
        escape = False
        for index in range(start, len(text)):
            char = text[index]
            if in_string:
                if escape:
                    escape = False
                elif char == "\\":
                    escape = True
                elif char == '"':
                    in_string = False
                continue
            if char == '"':
                in_string = True
                continue
            if char == open_char:
                depth += 1
            elif char == close_char:
                depth -= 1
                if depth == 0:
                    return text[start : index + 1]
    raise ValueError("Failed to extract a JSON object from the model response.")


def parse_json_payload(raw_text: str) -> dict[str, Any]:
    json_text = extract_json_text(raw_text)
    try:
        payload = json.loads(json_text)
    except json.JSONDecodeError:
        try:
            payload = ast.literal_eval(json_text)
        except (SyntaxError, ValueError) as exc:
            snippet = raw_text.strip().replace("\n", " ")[:500]
            raise ValueError(f"Model output is not valid JSON: {snippet}") from exc
    if not isinstance(payload, dict):
        raise ValueError(f"Expected a JSON object, got {type(payload).__name__}.")
    return payload


def load_existing_output(
    output_path: Path,
    overwrite: bool,
    resume: bool,
) -> dict[str, dict[str, Any]]:
    if not output_path.exists():
        return {}
    if overwrite:
        return {}
    if not resume:
        raise FileExistsError(
            f"Output already exists: {output_path}. Use --overwrite or --resume."
        )
    payload = json.loads(output_path.read_text(encoding="utf-8"))
    videos = payload.get("videos", [])
    if not isinstance(videos, list):
        raise ValueError(f"Existing output has no valid videos list: {output_path}")
    return {
        item["uuid"]: item
        for item in videos
        if isinstance(item, dict) and "uuid" in item
    }


def write_output(output_path: Path, videos_by_uuid: dict[str, dict[str, Any]]) -> None:
    ordered_videos = [videos_by_uuid[uuid] for uuid in sorted(videos_by_uuid)]
    payload = {
        "total_videos": len(ordered_videos),
        "videos": ordered_videos,
    }
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def format_subject_registry(subject_registry: list[dict[str, str]]) -> str:
    if not subject_registry:
        return "[]"
    return json.dumps(subject_registry, ensure_ascii=False, indent=2)


def build_global_context_prompt(
    uuid: str,
    total_events: int,
    camera_motion_guide: str,
) -> str:
    return textwrap.dedent(
        f"""
        You are annotating a stitched multi-event video composed of {total_events} chronologically ordered event clips from the same original video.
        UUID: {uuid}
        Return strict JSON only. Do not use markdown fences. Use double-quoted JSON strings and JSON booleans/null.
        Goals:
        1. Identify whether the content is sports footage. Sports videos are the only exception to the "adjacent event captions should not repeat" rule.
        2. Build a stable subject registry that will be reused in every event caption and in the global prompt.
        3. Prefer specific human nouns such as man, woman, boy, girl, delivery man, worker, bride, groom, or chef instead of person when the subject is visually clear.
        4. If the primary subject is an animal and no human is clearly visible, use a specific animal noun such as dog, cat, horse, bird, or duck instead of person-like wording.
        5. If only a body part or non-human visible subject is central, such as a hand, paw, brush, or vehicle, record that faithfully instead of inventing a full person.
        6. The term field in subject_registry is the canonical noun for that recurring subject, and the category field must distinguish whether it is human, animal, body_part, object, or unknown.
        7. Example: if the chosen term is woman, that exact noun should be reused everywhere. Global captions may introduce the subject as a woman and later refer to the woman, but the noun itself must stay woman rather than lady, person, female, or individual.
        8. Avoid sensitive or racial labels in subject descriptions. If skin tone is visually reliable and helpful, use neutral phrasing like light-to-medium skin tone.
        9. Only state visually high-confidence facts.
        Camera motion guide:
        {camera_motion_guide}
        Return JSON in exactly this schema:
        {{
          "is_sports": false,
          "has_animal": false,
          "is_cooking": false,
          "subject_registry": [
            {{
              "subject_id": "subject_1",
              "category": "human",
              "term": "woman",
              "profile": "young adult woman with very long light brown hair, a rust-colored top, and blue jeans"
            }}
          ],
          "stable_scene_summary": "One or two concise sentences about globally stable environment and framing.",
          "stable_camera_summary": "One concise sentence about globally stable filming behavior."
        }}
        """
    ).strip()


def build_event_prompt(
    uuid: str,
    event_index: int,
    total_events: int,
    previous_caption: str | None,
    previous_focus_object: str | None,
    context: dict[str, Any],
    camera_motion_guide: str,
) -> str:
    subject_registry = format_subject_registry(context.get("subject_registry", []))
    prev_caption = previous_caption or "<none>"
    prev_focus = previous_focus_object or "<none>"
    is_first_event = event_index == 1
    return textwrap.dedent(
        f"""
        You are annotating event {event_index} of {total_events} for UUID {uuid}.
        Return strict JSON only. Do not use markdown fences. Use double-quoted JSON strings and JSON booleans/null.
        Stable context from the full stitched video:
        - is_sports: {json.dumps(context.get("is_sports", False))}
        - has_animal: {json.dumps(context.get("has_animal", False))}
        - is_cooking: {json.dumps(context.get("is_cooking", False))}
        - subject_registry: {subject_registry}
        - stable_scene_summary: {json.dumps(context.get("stable_scene_summary", ""))}
        - stable_camera_summary: {json.dumps(context.get("stable_camera_summary", ""))}
        Previous event caption: {json.dumps(prev_caption)}
        Previous focus object: {json.dumps(prev_focus)}
        Rules:
        - Caption only this event clip, not the whole stitched video.
        - The caption should focus on the event-specific action, motion, interaction, or state change that makes this segment distinct.
        - Do not spend words restating stable global information unless that information is needed to make the event action clear and complete.
        - There is no hard word limit for the event caption. Keep it as concise as possible, but long enough to express the action completely and unambiguously.
        - The rules below define what information must be clear, not a fixed sentence template.
        - Write in natural, varied English. Avoid repetitive stock openings or boilerplate phrasing across events when the content differs.
        - Distinguish subtle but visible differences from the previous event when they are present. Fine-grained changes are meaningful and should be described instead of collapsing them into the same generic action.
        - Examples of meaningful distinctions include closed-mouth smiling versus smiling with teeth visible, looking down versus looking toward the frame's left, a slight head lift versus a head turn, a hand resting versus a hand lifting, or maintaining contact with an object versus releasing it.
        - If two adjacent events share the same broad action, describe the most reliable fine-grained difference that makes this event specific.
        - If the repeated action visibly changes an object, material, or scene over time, distinguish the event through that evolving state. Examples include more ingredients being added, fewer petals remaining, a surface becoming more covered, a container becoming fuller or emptier, or another clear cumulative visual change.
        - If the subject keeps moving in a similar way across adjacent events, distinguish the event through what the subject passes, approaches, leaves behind, enters, or moves alongside. Examples include passing a large tree, moving beside a wall, approaching a doorway, or walking beyond parked cars.
        - Prefer these concrete scene-detail or object-state differences over repeating a generic caption such as the man walks or the hand adds ingredients.
        - Unless this is sports footage, the caption must not be textually identical to the previous event caption.
        - The subject_registry provides the canonical subject nouns for this UUID. Reuse those exact nouns in the caption instead of synonyms.
        - subject_registry may include humans, animals, body parts, or non-human objects. Reuse those established noun categories faithfully and do not hallucinate a human subject when the main visible subject is an animal or non-human entity.
        - Any recurring subject already established in subject_registry is globally known for this UUID, so event captions should use definite reference such as the woman or the man, even in event 1.
        - Example: if subject_registry uses woman, write the woman when referring to that established subject in event captions, including the first event. Do not switch to lady, person, female, or individual.
        - Likewise, if subject_registry uses bird, cat, dog, horse, hand, or brush, keep using the bird, the cat, the dog, the horse, the hand, or the brush rather than introducing a person who is not clearly visible.
        - Only use A/An for a truly new subject that is not already established in subject_registry.
        - Keep subject nouns strictly consistent with the subject_registry across all events whenever the same subject is being referenced.
        - If a new human, animal, object, hand, or body part enters the frame, explicitly indicate appearance with wording like appears, enters the frame, or comes into view.
        - If a human, animal, object, hand, or body part leaves the frame, explicitly indicate disappearance with wording like disappears, leaves the frame, or goes out of view.
        - The directional, turn, hand, and camera examples below are semantic guidance. Do not copy them mechanically when a more natural wording can express the same meaning with equal precision.
        - Use frame-aware direction wording when relevant: toward the frame's left, toward the frame's right, into the background, toward the camera.
        - For turns, describe the final orientation, such as turns to face the frame's left, turns away from the camera to face the background, or turns toward the camera to face the viewer.
        - If a subject uses a hand, specify his/her/their left hand or right hand.
        - If a subject is holding a camera inside the scene, call it the man's camera, the woman's camera, or an equivalent possessive phrase so it is not confused with the filming camera.
        - Camera motion must capture only the dominant major motion. Ignore tiny shifts and ordinary shake.
        - If the camera is truly static or near-static, set camera_motion_label to "" and camera_motion_desc to "".
        - If the camera is moving to follow a subject but the subject stays relatively stable in the frame, and you cannot assign a high-confidence major motion label, set camera_motion_label to "" but keep has_camera_motion as true and write camera_motion_desc to explicitly say that the framing remains relatively stable while the camera follows the subject.
        - Use standard English camera-motion verb forms when writing camera_motion_desc. Write dollies in/out, pedestals up/down, trucks left/right, pans left/right, tilts up/down, rolls clockwise/counterclockwise, and zooms in/out. Never write malformed forms such as dollys out or as it dolly-in.
        - has_focus_shift is relative to the previous event and only applies when the filming camera does not move. The first event must use null.
        - If has_camera_motion is true, has_environmental_change must also be true.
        - is_multi_person means multiple human subjects are visible in this event clip.
        - has_occlusion is true only when a newly introduced or newly moved subject/object meaningfully blocks a previously visible subject in a changed way, not for ordinary partial overlap.
        Camera motion guide:
        {camera_motion_guide}
        Return JSON in exactly this schema:
        {{
          "caption": "A complete event-specific caption with no fixed word limit. Prioritize the distinctive action in this event, including subtle differences, evolving object or scene state, or passed landmarks when relevant, and avoid unnecessary repetition of stable global context.",
          "has_appearance": false,
          "has_disappearance": false,
          "has_camera_motion": false,
          "has_focus_shift": {json.dumps(None) if is_first_event else "false"},
          "is_multi_person": false,
          "has_environmental_change": false,
          "has_occlusion": false,
          "has_lighting_change": false,
          "has_animal": false,
          "is_cooking": false,
          "focus_object": "brief noun phrase",
          "camera_motion_label": "",
          "camera_motion_desc": ""
        }}
        """
    ).strip()


def build_event_caption_refine_prompt(
    uuid: str,
    event_index: int,
    total_events: int,
    previous_caption: str,
    draft_caption: str,
    context: dict[str, Any],
    attempt: int = 1,
) -> str:
    subject_registry = format_subject_registry(context.get("subject_registry", []))
    extra_rules = ""
    if attempt >= 2:
        extra_rules = textwrap.dedent(
            """
            - This is a second rewrite attempt because the caption still sounds too close to the previous one.
            - Change the sentence structure as well as the distinguishing detail so the result does not feel like a lightly edited template.
            """
        ).strip()
    return textwrap.dedent(
        f"""
        You are revising the caption for event {event_index} of {total_events} for UUID {uuid}.
        Return strict JSON only. Do not use markdown fences. Use double-quoted JSON strings and JSON booleans/null.
        Stable context:
        - is_sports: {json.dumps(context.get("is_sports", False))}
        - subject_registry: {subject_registry}
        Previous event caption: {json.dumps(previous_caption)}
        Current draft caption: {json.dumps(draft_caption)}
        The draft caption is too similar to the previous event caption. Rewrite it so the current event is distinguished by the finest reliable visible difference.
        Rules:
        - Reuse the exact canonical subject nouns from subject_registry.
        - Keep the caption focused on this event only.
        - Write in natural, non-formulaic English. Do not merely append generic boilerplate to the draft.
        - If the broad action stays similar, describe the subtle but visible difference, such as mouth open versus mouth closed while smiling, teeth becoming visible, a shift in gaze direction, a head lift, a hand movement, a small pose change, a change in object contact, or another reliable fine-grained state change.
        - If the repeated action changes the scene or object over time, use that evolving detail to distinguish the event, such as more ingredients accumulating, fewer petals remaining, a surface becoming more covered, or another cumulative visible change.
        - If the subject keeps moving in a similar way, distinguish the event through what the subject passes, approaches, leaves behind, or enters, such as a tree, wall, doorway, vehicle, table, or other reliable landmark.
        - Do not invent details that are not visually reliable.
        - Do not restate stable global context unless needed for clarity.
        - The rewritten caption must not be textually identical to the previous event caption.
        - The earlier examples are semantic guidance only. Prefer natural phrasing over rigid reuse of example wording.
        {extra_rules}
        Return JSON in exactly this schema:
        {{
          "caption": "A refined event-specific caption that explicitly distinguishes this event from the previous one."
        }}
        """
    ).strip()


def build_event_camera_motion_repair_prompt(
    uuid: str,
    event_index: int,
    total_events: int,
    draft_payload: dict[str, Any],
    context: dict[str, Any],
    camera_motion_guide: str,
    attempt: int = 1,
) -> str:
    subject_registry = format_subject_registry(context.get("subject_registry", []))
    extra_rules = ""
    if attempt >= 2:
        extra_rules = textwrap.dedent(
            """
            - This is a second repair attempt because the camera fields are still incomplete.
            - Rewrite the description more concretely and naturally instead of falling back to vague stock wording.
            """
        ).strip()
    return textwrap.dedent(
        f"""
        You are repairing the camera motion fields for event {event_index} of {total_events} for UUID {uuid}.
        Return strict JSON only. Do not use markdown fences. Use double-quoted JSON strings and JSON booleans/null.
        Stable context:
        - subject_registry: {subject_registry}
        - current_caption: {json.dumps(normalize_whitespace(draft_payload.get("caption")))}
        - draft_has_camera_motion: {json.dumps(coerce_bool(draft_payload.get("has_camera_motion")))}
        - draft_camera_motion_label: {json.dumps(normalize_camera_motion_label(draft_payload.get("camera_motion_label")))}
        - draft_camera_motion_desc: {json.dumps(normalize_whitespace(draft_payload.get("camera_motion_desc")))}
        Rules:
        - Repair only the camera motion fields.
        - Use a major motion label only when it is high-confidence.
        - If the camera is truly static or near-static, set has_camera_motion to false and leave both camera_motion_label and camera_motion_desc empty.
        - If the camera is moving to follow a subject but the subject remains relatively stable in the frame and no major motion label is high-confidence, set has_camera_motion to true, leave camera_motion_label empty, and write a natural camera_motion_desc explaining that the framing stays relatively stable while the camera follows the subject.
        - Write natural prose. Do not use shorthand such as static | pan-right | ...
        - Use standard English camera-motion verb forms. Write dollies in/out, pedestals up/down, trucks left/right, pans left/right, tilts up/down, rolls clockwise/counterclockwise, and zooms in/out. Never write malformed forms such as dollys out or as it dolly-in.
        - Do not invent camera motion that is not visually reliable.
        {extra_rules}
        Camera motion guide:
        {camera_motion_guide}
        Return JSON in exactly this schema:
        {{
          "has_camera_motion": false,
          "camera_motion_label": "",
          "camera_motion_desc": ""
        }}
        """
    ).strip()


def build_global_camera_repair_prompt(
    uuid: str,
    context: dict[str, Any],
    events: list[dict[str, Any]],
    draft_payload: dict[str, Any],
    derived_prefix: str,
    derived_label: str,
    derived_description: str,
    attempt: int = 1,
) -> str:
    event_camera_summary = [
        {
            "event_id": event["event_id"],
            "has_camera_motion": event["has_camera_motion"],
            "camera_motion_label": event["camera_motion_label"],
            "camera_motion_desc": event["camera_motion_desc"],
        }
        for event in events
    ]
    extra_rules = ""
    if attempt >= 2:
        extra_rules = textwrap.dedent(
            """
            - This is a second repair attempt because the previous global camera description was still unusable or inconsistent.
            - Rewrite it in more specific natural language while staying fully consistent with the event-level camera evidence.
            """
        ).strip()
    return textwrap.dedent(
        f"""
        You are repairing the global camera motion fields for UUID {uuid}.
        Return strict JSON only. Do not use markdown fences. Use double-quoted JSON strings and JSON booleans/null.
        Stable context:
        - subject_registry: {format_subject_registry(context.get("subject_registry", []))}
        - stable_camera_summary: {json.dumps(context.get("stable_camera_summary", ""))}
        - event_camera_summary:
        {json.dumps(event_camera_summary, ensure_ascii=False, indent=2)}
        - current_draft_prefix: {json.dumps(normalize_global_camera_prefix(draft_payload.get("camera_movement_prefix")))}
        - current_draft_label: {json.dumps(normalize_camera_motion_label(draft_payload.get("camera_movement_label")))}
        - current_draft_description: {json.dumps(normalize_whitespace(draft_payload.get("camera_movement_description")))}
        - event_derived_prefix: {json.dumps(derived_prefix)}
        - event_derived_label: {json.dumps(derived_label)}
        - event_derived_description: {json.dumps(derived_description)}
        Rules:
        - The result must stay consistent with the event-level camera evidence.
        - camera_movement_description must be natural prose. Do not use shorthand such as static | pan-right | ...
        - Use standard English camera-motion verb forms. Write dollies in/out, pedestals up/down, trucks left/right, pans left/right, tilts up/down, rolls clockwise/counterclockwise, and zooms in/out. Never write malformed forms such as dollys out or as it dolly-in.
        - If the camera behavior changes across the video, describe the sequence directly in natural language.
        - If the camera is globally static, say so directly in a sentence.
        - If the camera often follows the subject while keeping the framing relatively stable, you may leave camera_movement_label empty and describe that behavior in natural language.
        - Prefer specific, UUID-appropriate wording over generic stock sentences when the evidence supports it.
        {extra_rules}
        Return JSON in exactly this schema:
        {{
          "camera_movement_prefix": "static",
          "camera_movement_label": "",
          "camera_movement_description": "A natural-language sentence or paragraph."
        }}
        """
    ).strip()


def build_subject_profile_repair_prompt(
    uuid: str,
    context: dict[str, Any],
    events: list[dict[str, Any]],
    draft_payload: dict[str, Any],
    attempt: int = 1,
) -> str:
    event_subject_summary = [
        {
            "event_id": event["event_id"],
            "caption": event["caption"],
            "is_multi_person": event["is_multi_person"],
            "has_animal": event["has_animal"],
            "focus_object": event["focus_object"],
        }
        for event in events
    ]
    extra_rules = ""
    if attempt >= 2:
        extra_rules = textwrap.dedent(
            """
            - This is a second repair attempt because subject_profile is still missing or unusable.
            - Rewrite it as a natural paragraph rather than a stock formula.
            """
        ).strip()
    return textwrap.dedent(
        f"""
        You are repairing the subject_profile field for UUID {uuid}.
        Return strict JSON only. Do not use markdown fences. Use double-quoted JSON strings and JSON booleans/null.
        Stable context:
        - subject_registry: {format_subject_registry(context.get("subject_registry", []))}
        - current_draft_subject_profile: {json.dumps(normalize_whitespace(draft_payload.get("subject_profile")))}
        - event_subject_summary:
        {json.dumps(event_subject_summary, ensure_ascii=False, indent=2)}
        Rules:
        - Reuse the canonical nouns from subject_registry.
        - Explicitly distinguish whether the recurring subjects are human-only, animal-only, mixed human-animal, or non-human-only.
        - If no human is clearly visible, do not imply one.
        - Write a natural paragraph, not a stock formula.
        - Avoid sensitive racial labels. If skin tone is visually reliable and useful, use neutral phrasing.
        - Do not narrate event order.
        {extra_rules}
        Return JSON in exactly this schema:
        {{
          "subject_profile": "A natural paragraph describing the recurring main subject or subjects."
        }}
        """
    ).strip()


def build_global_refine_prompt(
    uuid: str,
    context: dict[str, Any],
    events: list[dict[str, Any]],
) -> str:
    event_summary = [
        {
            "event_id": event["event_id"],
            "caption": event["caption"],
            "focus_object": event["focus_object"],
            "camera_motion_label": event["camera_motion_label"],
            "camera_motion_desc": event["camera_motion_desc"],
            "has_appearance": event["has_appearance"],
            "has_disappearance": event["has_disappearance"],
            "has_environmental_change": event["has_environmental_change"],
            "has_occlusion": event["has_occlusion"],
            "has_lighting_change": event["has_lighting_change"],
        }
        for event in events
    ]
    return textwrap.dedent(
        f"""
        You are writing the global annotation for UUID {uuid} using the full stitched video as primary evidence and the event summaries as supporting evidence.
        Return strict JSON only. Do not use markdown fences. Use double-quoted JSON strings and JSON booleans/null.
        Stable context:
        - is_sports: {json.dumps(context.get("is_sports", False))}
        - subject_registry: {format_subject_registry(context.get("subject_registry", []))}
        - stable_scene_summary: {json.dumps(context.get("stable_scene_summary", ""))}
        - stable_camera_summary: {json.dumps(context.get("stable_camera_summary", ""))}
        - event_summaries:
        {json.dumps(event_summary, ensure_ascii=False, indent=2)}
        Rules:
        - The global prompt must describe stable, high-confidence information from the stitched whole video.
        - Do not narrate event order. Do not use first, then, next, afterwards, before, after, event 1, or event 2.
        - The structured fields do not have fixed sentence-count or word-count limits. Make them as short or as detailed as needed, but only include visually high-confidence and globally stable information.
        - The field descriptions define coverage, not canned wording. Avoid making every field sound like the same stock template.
        - Write the structured fields in the style of compact descriptive paragraphs rather than short labels.
        - Use natural prose and vary sentence openings across fields when appropriate instead of repeatedly starting with the same formula such as "The video shows" or "The scene is set".
        - background should describe the stable physical setting in concrete visual detail: scene type, layout, major surfaces, furniture, props, vegetation, architecture, visible depth layers, and other recurring environmental elements.
        - theme should function like a theme description rather than an abstract keyword. It should explain the stable situation, the recurring human or animal roles, the likely activity domain, and the central visual focus without narrating event order.
        - style should describe the overall visual presentation and footage character, such as documentary, candid, cinematic, stock-footage-like, surveillance-like, tutorial-like, vlog-like, observational, staged, polished, raw, or naturalistic.
        - shot_type should describe the dominant framing and composition: shot scale, camera angle, viewpoint height, subject placement, depth of field, whether the framing isolates the subject or includes the environment, and any stable compositional pattern.
        - camera_movement_description should read like a natural standalone sentence or paragraph, summarizing the overall filming behavior, what motion is dominant, whether motion is sustained or intermittent, and whether the framing stays fixed or follows the subject.
        - Do not write camera_movement_description in shorthand such as static | pan-right | ... . The final text should be natural prose.
        - Use standard English camera-motion verb forms. Write dollies in/out, pedestals up/down, trucks left/right, pans left/right, tilts up/down, rolls clockwise/counterclockwise, and zooms in/out. Never write malformed forms such as dollys out or as it dolly-in.
        - If the camera behavior changes across the video, you may describe that sequence directly in natural language, such as static at first, then following the subject, and later panning right.
        - Avoid generic stock camera descriptions when a more specific, high-confidence description can be written for this UUID.
        - lighting should describe stable lighting source and quality in concrete terms, such as natural sunlight, indoor ambient light, overhead fixtures, warm or cool tones, contrast level, shadows, highlights, diffusion, and visibility.
        - atmosphere should describe the stable mood and experiential tone suggested by the visual evidence, such as calm, tense, playful, intimate, energetic, routine, joyful, solemn, tranquil, busy, or contemplative.
        - subject_profile should read like a compact profile of the recurring main subject or subjects, covering consistent identity terms, apparent age group, visible build or posture when reliable, hair, clothing, accessories, recurring carried objects, and stable appearance cues.
        - subject_profile must explicitly distinguish whether the stable primary subjects are human-only, animal-only, mixed human-animal, or non-human-only.
        - If no human is clearly visible, do not imply one. For animal-only videos, describe the animal as the primary subject using species or animal-type details, markings, fur, feathers, or body shape when reliable. For body-part-led or object-led videos, state that the visible hand, tool, object, or other non-human subject is the primary visible subject.
        - The short caption should be about 20 words.
        - The middle caption should be about 80 words.
        - The long caption should be about 200 words.
        - The detailed caption should be about 500-800 words.
        - The detailed caption may synthesize and reuse the structured fields plus stable recurring evidence from the event summaries, but it must not turn into a segment-by-segment narration.
        - The subject_registry defines the canonical subject nouns for the UUID. Reuse those same nouns across subject_profile and all global captions instead of switching to synonyms.
        - Example: if subject_registry uses woman, the global descriptions may introduce her as a woman and later refer to the woman, but should not switch to lady, person, female, or individual.
        - The subject_profile should avoid sensitive racial labels and should reuse the same subject nouns chosen in the subject_registry.
        - The global camera_movement should summarize only the overall stable filming pattern using:
          - static
          - moving-throughout
          - intermittent
          - motion-changes
          - mixed
          - near-static
        - For static or near-static global motion, camera_movement_label can be empty.
        Return JSON in exactly this schema:
        {{
          "background": "A paragraph-style description of the stable physical environment: scene type, layout, major surfaces, props, furniture, architecture, vegetation, and recurring depth layers. No fixed length. Use only high-confidence information.",
          "theme": "A paragraph-style theme description of the stable situation, recurring roles, and central visual focus. Do not narrate event order. No fixed length. Use only high-confidence information.",
          "style": "A paragraph-style description of the overall visual style or footage character, such as documentary, candid, cinematic, naturalistic, stock-footage-like, tutorial-like, or observational. No fixed length. Use only high-confidence information.",
          "shot_type": "A paragraph-style description of dominant shot scale, camera angle, viewpoint height, framing, composition, subject placement, and depth-of-field behavior. No fixed length. Use only high-confidence information.",
          "camera_movement_prefix": "static",
          "camera_movement_label": "",
          "camera_movement_description": "A natural-language sentence or paragraph describing the global filming behavior, without any pipe-delimited shorthand. If motion changes over time, describe the sequence directly in prose. No fixed length. Use only high-confidence information.",
          "lighting": "A paragraph-style description of stable lighting source, quality, color temperature, contrast, shadows, highlights, and overall visibility. No fixed length. Use only high-confidence information.",
          "atmosphere": "A paragraph-style description of the stable mood and experiential tone created by the scene, subjects, pacing, and visual presentation. No fixed length. Use only high-confidence information.",
          "subject_profile": "A paragraph-style profile of the recurring main subject or subjects, explicitly distinguishing human-only, animal-only, mixed human-animal, or non-human-only setups. Reuse the canonical nouns from subject_registry, and if no human is clearly visible, do not imply one. No fixed length. Use only high-confidence information.",
          "short_caption": "About 20 words.",
          "middle_caption": "About 80 words.",
          "long_caption": "About 200 words.",
          "detailed_caption": "About 500-800 words."
        }}
        """
    ).strip()


class Qwen3VLRunner:
    def __init__(
        self,
        model_path: Path,
        device_map: str,
        dtype: str,
        attn_implementation: str,
        trust_remote_code: bool,
        fps: float,
        max_pixels: int | None,
        temperature: float,
        top_p: float,
        top_k: int,
        repetition_penalty: float,
        greedy: bool,
    ) -> None:
        try:
            import torch
            from transformers import AutoProcessor, Qwen3VLForConditionalGeneration
        except ModuleNotFoundError as exc:
            raise RuntimeError(
                "Missing runtime dependencies. Install torch and transformers in the inference environment."
            ) from exc

        self.torch = torch
        self.temperature = temperature
        self.top_p = top_p
        self.top_k = top_k
        self.repetition_penalty = repetition_penalty
        self.greedy = greedy
        self.fps = fps if fps and fps > 0 else None
        self.max_pixels = max_pixels if max_pixels and max_pixels > 0 else None
        self.process_vision_info = None

        try:
            from qwen_vl_utils import process_vision_info  # type: ignore
        except ModuleNotFoundError:
            process_vision_info = None
        self.process_vision_info = process_vision_info

        dtype_map = {
            "auto": "auto",
            "float16": torch.float16,
            "bfloat16": torch.bfloat16,
            "float32": torch.float32,
        }
        load_kwargs: dict[str, Any] = {
            "device_map": device_map,
            "torch_dtype": dtype_map[dtype],
            "trust_remote_code": trust_remote_code,
        }
        if attn_implementation:
            load_kwargs["attn_implementation"] = attn_implementation

        self.model = Qwen3VLForConditionalGeneration.from_pretrained(
            str(model_path),
            **load_kwargs,
        )
        self.processor = AutoProcessor.from_pretrained(
            str(model_path),
            trust_remote_code=trust_remote_code,
        )

    def _build_messages(self, video_paths: list[Path], prompt: str) -> list[dict[str, Any]]:
        user_content: list[dict[str, Any]] = []
        for path in video_paths:
            video_item: dict[str, Any] = {
                "type": "video",
                "video": str(path.resolve()),
            }
            if self.fps is not None:
                video_item["fps"] = self.fps
            if self.max_pixels is not None:
                video_item["max_pixels"] = self.max_pixels
            user_content.append(video_item)
        user_content.append({"type": "text", "text": prompt})
        return [
            {
                "role": "system",
                "content": (
                    "You are a careful video annotation assistant. "
                    "Return strict JSON only and prefer high-confidence visual facts."
                ),
            },
            {
                "role": "user",
                "content": user_content,
            },
        ]

    def _prepare_inputs(self, messages: list[dict[str, Any]]):
        if self.process_vision_info is not None:
            text = self.processor.apply_chat_template(
                messages,
                tokenize=False,
                add_generation_prompt=True,
            )
            image_inputs, video_inputs = self.process_vision_info(messages)
            inputs = self.processor(
                text=[text],
                images=image_inputs,
                videos=video_inputs,
                padding=True,
                return_tensors="pt",
            )
        else:
            try:
                inputs = self.processor.apply_chat_template(
                    messages,
                    tokenize=True,
                    add_generation_prompt=True,
                    return_dict=True,
                    return_tensors="pt",
                )
            except Exception as exc:
                raise RuntimeError(
                    "Failed to prepare Qwen video inputs. Install qwen-vl-utils or use a transformers build that supports direct multimodal chat templating."
                ) from exc
        return inputs.to(self.model.device)

    def generate_json(
        self,
        video_paths: list[Path],
        prompt: str,
        max_new_tokens: int,
    ) -> dict[str, Any]:
        messages = self._build_messages(video_paths, prompt)
        inputs = self._prepare_inputs(messages)

        generate_kwargs: dict[str, Any] = {
            "max_new_tokens": max_new_tokens,
            "repetition_penalty": self.repetition_penalty,
        }
        if self.greedy:
            generate_kwargs["do_sample"] = False
        else:
            generate_kwargs.update(
                {
                    "do_sample": True,
                    "temperature": self.temperature,
                    "top_p": self.top_p,
                    "top_k": self.top_k,
                }
            )

        with self.torch.inference_mode():
            generated_ids = self.model.generate(**inputs, **generate_kwargs)
        generated_ids_trimmed = [
            out_ids[len(in_ids) :]
            for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
        ]
        output_text = self.processor.batch_decode(
            generated_ids_trimmed,
            skip_special_tokens=True,
            clean_up_tokenization_spaces=False,
        )[0]
        if self.torch.cuda.is_available():
            self.torch.cuda.empty_cache()
        return parse_json_payload(output_text)


def create_uuid_temp_dir(
    uuid: str,
    temp_root: str | None,
    keep_temp: bool,
) -> tuple[Path, tempfile.TemporaryDirectory[str] | None]:
    if keep_temp:
        path = Path(tempfile.mkdtemp(prefix=f"multi_event_{uuid}_", dir=temp_root))
        logging.info("Keeping temp files for %s in %s", uuid, path)
        return path, None
    manager = tempfile.TemporaryDirectory(prefix=f"multi_event_{uuid}_", dir=temp_root)
    return Path(manager.name), manager


def write_ffmpeg_concat_list(list_path: Path, segments: Iterable[SegmentInfo]) -> None:
    lines = []
    for segment in segments:
        escaped = str(segment.path).replace("\\", "\\\\").replace("'", r"'\''")
        lines.append(f"file '{escaped}'")
    list_path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def stitch_segments(
    uuid: str,
    segments: list[SegmentInfo],
    temp_dir: Path,
    ffmpeg_bin: str,
) -> Path:
    concat_list = temp_dir / f"{uuid}_concat.txt"
    stitched_video = temp_dir / f"{uuid}_stitched.mp4"
    write_ffmpeg_concat_list(concat_list, segments)

    copy_cmd = [
        ffmpeg_bin,
        "-hide_banner",
        "-loglevel",
        "error",
        "-y",
        "-f",
        "concat",
        "-safe",
        "0",
        "-i",
        str(concat_list),
        "-c",
        "copy",
        str(stitched_video),
    ]
    copy_result = subprocess.run(copy_cmd, capture_output=True, text=True)
    if copy_result.returncode == 0 and stitched_video.exists():
        return stitched_video

    reencode_cmd = [
        ffmpeg_bin,
        "-hide_banner",
        "-loglevel",
        "error",
        "-y",
        "-f",
        "concat",
        "-safe",
        "0",
        "-i",
        str(concat_list),
        "-an",
        "-c:v",
        "libx264",
        "-pix_fmt",
        "yuv420p",
        str(stitched_video),
    ]
    reencode_result = subprocess.run(reencode_cmd, capture_output=True, text=True)
    if reencode_result.returncode == 0 and stitched_video.exists():
        return stitched_video

    raise RuntimeError(
        "ffmpeg failed to stitch the segments.\n"
        f"copy stderr: {copy_result.stderr.strip()}\n"
        f"reencode stderr: {reencode_result.stderr.strip()}"
    )


def build_global_video_inputs(
    uuid: str,
    segments: list[SegmentInfo],
    temp_dir: Path,
    ffmpeg_bin: str,
) -> list[Path]:
    if len(segments) == 1:
        return [segments[0].path]
    try:
        return [stitch_segments(uuid, segments, temp_dir, ffmpeg_bin)]
    except Exception as exc:
        logging.warning(
            "Falling back to ordered segment inputs for %s because stitching failed: %s",
            uuid,
            exc,
        )
        return [segment.path for segment in segments]


def coerce_context_payload(payload: dict[str, Any]) -> dict[str, Any]:
    subject_registry = []
    for item in payload.get("subject_registry", []):
        if not isinstance(item, dict):
            continue
        term = normalize_whitespace(item.get("term"))
        if not term:
            continue
        subject_registry.append(
            {
                "subject_id": normalize_whitespace(item.get("subject_id")) or f"subject_{len(subject_registry) + 1}",
                "category": normalize_subject_category(item.get("category")),
                "term": term,
                "profile": normalize_whitespace(item.get("profile")),
            }
        )
    return {
        "is_sports": coerce_bool(payload.get("is_sports")),
        "has_animal": coerce_bool(payload.get("has_animal")),
        "is_cooking": coerce_bool(payload.get("is_cooking")),
        "subject_registry": subject_registry,
        "stable_scene_summary": normalize_whitespace(payload.get("stable_scene_summary")),
        "stable_camera_summary": normalize_whitespace(payload.get("stable_camera_summary")),
    }


def event_camera_output_needs_repair(payload: dict[str, Any]) -> bool:
    has_camera_motion = coerce_bool(payload.get("has_camera_motion"))
    camera_motion_desc = normalize_camera_motion_surface_text(payload.get("camera_motion_desc"))
    return has_camera_motion and not camera_motion_desc


def coerce_event_payload(
    payload: dict[str, Any],
    segment: SegmentInfo,
    event_index: int,
    context: dict[str, Any],
) -> dict[str, Any]:
    camera_motion_label = normalize_camera_motion_label(payload.get("camera_motion_label"))
    camera_motion_desc = normalize_camera_motion_surface_text(payload.get("camera_motion_desc"))
    has_camera_motion = coerce_bool(payload.get("has_camera_motion"))
    if camera_motion_label:
        if not camera_motion_desc:
            camera_motion_desc = describe_camera_motion(camera_motion_label)
        has_camera_motion = True
    elif has_camera_motion and camera_motion_desc:
        pass
    elif has_camera_motion:
        camera_motion_desc = "The framing remains relatively stable while the camera follows the subject."
    else:
        camera_motion_desc = ""
        has_camera_motion = False

    has_focus_shift = None if event_index == 1 else coerce_optional_bool(payload.get("has_focus_shift"))
    if event_index != 1 and has_camera_motion:
        has_focus_shift = False
    if event_index != 1 and has_focus_shift is None:
        has_focus_shift = False

    event = {
        "event_id": event_index,
        "start_time": round(segment.start_time, 3),
        "end_time": round(segment.end_time, 3),
        "duration": round(segment.duration, 3),
        "caption": normalize_whitespace(payload.get("caption")),
        "filename": segment.path.name,
        "has_appearance": coerce_bool(payload.get("has_appearance")),
        "has_disappearance": coerce_bool(payload.get("has_disappearance")),
        "has_camera_motion": has_camera_motion,
        "has_focus_shift": has_focus_shift,
        "is_multi_person": coerce_bool(payload.get("is_multi_person")),
        "has_environmental_change": coerce_bool(payload.get("has_environmental_change")),
        "has_occlusion": coerce_bool(payload.get("has_occlusion")),
        "has_lighting_change": coerce_bool(payload.get("has_lighting_change")),
        "has_animal": coerce_bool(payload.get("has_animal"), default=context.get("has_animal", False)),
        "is_cooking": coerce_bool(payload.get("is_cooking"), default=context.get("is_cooking", False)),
        "focus_object": normalize_focus_object(payload.get("focus_object"), context.get("subject_registry", [])),
        "camera_motion_desc": camera_motion_desc,
        "camera_motion_label": camera_motion_label,
    }

    if not event["caption"]:
        raise ValueError(f"Model returned an empty caption for {segment.path.name}")
    if event["has_camera_motion"]:
        event["has_environmental_change"] = True
    return event


def global_camera_output_needs_repair(payload: dict[str, Any], events: list[dict[str, Any]]) -> bool:
    prefix = normalize_global_camera_prefix(payload.get("camera_movement_prefix"))
    label = normalize_camera_motion_label(payload.get("camera_movement_label"))
    description = normalize_camera_movement_description_output(
        payload.get("camera_movement_description")
    )
    if not description:
        return True
    if prefix and global_camera_conflicts_with_events(prefix, label, events):
        return True
    return False


def subject_profile_needs_repair(payload: dict[str, Any]) -> bool:
    return not normalize_whitespace(payload.get("subject_profile"))


def coerce_global_payload(
    payload: dict[str, Any],
    context: dict[str, Any],
    events: list[dict[str, Any]],
) -> dict[str, Any]:
    derived_prefix, derived_label, derived_description = derive_global_camera_from_events(events)
    prefix = normalize_global_camera_prefix(payload.get("camera_movement_prefix"))
    label = normalize_camera_motion_label(payload.get("camera_movement_label"))
    model_description = normalize_camera_movement_description_output(
        payload.get("camera_movement_description")
    )
    if not prefix:
        prefix, label = derived_prefix, derived_label
    elif global_camera_conflicts_with_events(prefix, label, events):
        prefix, label = derived_prefix, derived_label
    elif prefix in {"static", "near-static"}:
        label = ""
    description = model_description or derived_description
    if not description:
        if prefix in {"static", "near-static"}:
            description = "The camera remains essentially static throughout the video."
        elif label:
            description = describe_camera_motion(label, scope="global")
        else:
            description = "The camera behavior remains visually consistent across the video."

    global_prompt = {
        "background": normalize_whitespace(payload.get("background")),
        "theme": normalize_whitespace(payload.get("theme")),
        "style": normalize_whitespace(payload.get("style")),
        "shot_type": normalize_whitespace(payload.get("shot_type")),
        "camera_movement": compose_global_camera_movement(prefix, label, description),
        "lighting": normalize_whitespace(payload.get("lighting")),
        "atmosphere": normalize_whitespace(payload.get("atmosphere")),
        "subject_profile": normalize_whitespace(payload.get("subject_profile")),
        "short_caption": normalize_whitespace(payload.get("short_caption")),
        "middle_caption": normalize_whitespace(payload.get("middle_caption")),
        "long_caption": normalize_whitespace(payload.get("long_caption")),
        "detailed_caption": normalize_whitespace(payload.get("detailed_caption")),
    }
    if not global_prompt["subject_profile"]:
        global_prompt["subject_profile"] = fallback_subject_profile(context.get("subject_registry", []))
    return global_prompt


def validate_caption_lengths(uuid: str, global_prompt: dict[str, Any]) -> None:
    ranges = {
        "short_caption": (10, 35),
        "middle_caption": (55, 120),
        "long_caption": (140, 260),
        "detailed_caption": (450, 900),
    }
    for key, (low, high) in ranges.items():
        count = word_count(global_prompt.get(key, ""))
        if count and not (low <= count <= high):
            logging.warning(
                "%s %s has %d words, outside the expected range %d-%d.",
                uuid,
                key,
                count,
                low,
                high,
            )


def annotate_uuid(
    uuid: str,
    segments: list[SegmentInfo],
    runner: Qwen3VLRunner,
    camera_motion_guide: str,
    args: argparse.Namespace,
) -> dict[str, Any]:
    temp_dir, temp_manager = create_uuid_temp_dir(uuid, args.temp_dir, args.keep_temp)
    try:
        global_video_inputs = build_global_video_inputs(
            uuid=uuid,
            segments=segments,
            temp_dir=temp_dir,
            ffmpeg_bin=args.ffmpeg_bin,
        )

        logging.info("Running global context pass for %s", uuid)
        context_payload = runner.generate_json(
            video_paths=global_video_inputs,
            prompt=build_global_context_prompt(
                uuid=uuid,
                total_events=len(segments),
                camera_motion_guide=camera_motion_guide,
            ),
            max_new_tokens=args.context_max_new_tokens,
        )
        context = coerce_context_payload(context_payload)

        events: list[dict[str, Any]] = []
        previous_caption: str | None = None
        previous_focus_object: str | None = None
        for event_index, segment in enumerate(segments, start=1):
            logging.info(
                "Running event pass for %s (%d/%d): %s",
                uuid,
                event_index,
                len(segments),
                segment.path.name,
            )
            event_payload = runner.generate_json(
                video_paths=[segment.path],
                prompt=build_event_prompt(
                    uuid=uuid,
                    event_index=event_index,
                    total_events=len(segments),
                    previous_caption=previous_caption,
                    previous_focus_object=previous_focus_object,
                    context=context,
                    camera_motion_guide=camera_motion_guide,
                ),
                max_new_tokens=args.event_max_new_tokens,
            )
            if event_camera_output_needs_repair(event_payload):
                for attempt in (1, 2):
                    logging.info(
                        "Repairing event camera fields for %s event %d (attempt %d)",
                        uuid,
                        event_index,
                        attempt,
                    )
                    try:
                        repaired_camera_payload = runner.generate_json(
                            video_paths=[segment.path],
                            prompt=build_event_camera_motion_repair_prompt(
                                uuid=uuid,
                                event_index=event_index,
                                total_events=len(segments),
                                draft_payload=event_payload,
                                context=context,
                                camera_motion_guide=camera_motion_guide,
                                attempt=attempt,
                            ),
                            max_new_tokens=min(args.event_max_new_tokens, 512),
                        )
                        for key in ("has_camera_motion", "camera_motion_label", "camera_motion_desc"):
                            if key in repaired_camera_payload:
                                event_payload[key] = repaired_camera_payload[key]
                        if not event_camera_output_needs_repair(event_payload):
                            break
                    except Exception as exc:
                        logging.warning(
                            "Event camera repair failed for %s event %d on attempt %d: %s",
                            uuid,
                            event_index,
                            attempt,
                            exc,
                        )
            event = coerce_event_payload(
                payload=event_payload,
                segment=segment,
                event_index=event_index,
                context=context,
            )
            if previous_caption and not context.get("is_sports", False):
                if normalized_caption_key(event["caption"]) == normalized_caption_key(previous_caption):
                    for attempt in (1, 2):
                        logging.info(
                            "Refining near-duplicate event caption for %s event %d (attempt %d)",
                            uuid,
                            event_index,
                            attempt,
                        )
                        try:
                            refined_payload = runner.generate_json(
                                video_paths=[segment.path],
                                prompt=build_event_caption_refine_prompt(
                                    uuid=uuid,
                                    event_index=event_index,
                                    total_events=len(segments),
                                    previous_caption=previous_caption,
                                    draft_caption=event["caption"],
                                    context=context,
                                    attempt=attempt,
                                ),
                                max_new_tokens=min(args.event_max_new_tokens, 768),
                            )
                            refined_caption = normalize_whitespace(refined_payload.get("caption"))
                            if refined_caption:
                                event["caption"] = refined_caption
                            if normalized_caption_key(event["caption"]) != normalized_caption_key(previous_caption):
                                break
                        except Exception as exc:
                            logging.warning(
                                "Caption refinement failed for %s event %d on attempt %d: %s",
                                uuid,
                                event_index,
                                attempt,
                                exc,
                            )
                event["caption"] = disambiguate_repeated_caption(
                    current_caption=event["caption"],
                    previous_caption=previous_caption,
                    event=event,
                )
            events.append(event)
            previous_caption = event["caption"]
            previous_focus_object = event["focus_object"]

        logging.info("Running final global prompt pass for %s", uuid)
        global_payload = runner.generate_json(
            video_paths=global_video_inputs,
            prompt=build_global_refine_prompt(
                uuid=uuid,
                context=context,
                events=events,
            ),
            max_new_tokens=args.global_max_new_tokens,
        )
        derived_prefix, derived_label, derived_description = derive_global_camera_from_events(events)
        if global_camera_output_needs_repair(global_payload, events):
            for attempt in (1, 2):
                logging.info(
                    "Repairing global camera fields for %s (attempt %d)",
                    uuid,
                    attempt,
                )
                try:
                    repaired_global_camera_payload = runner.generate_json(
                        video_paths=global_video_inputs,
                        prompt=build_global_camera_repair_prompt(
                            uuid=uuid,
                            context=context,
                            events=events,
                            draft_payload=global_payload,
                            derived_prefix=derived_prefix,
                            derived_label=derived_label,
                            derived_description=derived_description,
                            attempt=attempt,
                        ),
                        max_new_tokens=min(args.global_max_new_tokens, 768),
                    )
                    for key in (
                        "camera_movement_prefix",
                        "camera_movement_label",
                        "camera_movement_description",
                    ):
                        if key in repaired_global_camera_payload:
                            global_payload[key] = repaired_global_camera_payload[key]
                    if not global_camera_output_needs_repair(global_payload, events):
                        break
                except Exception as exc:
                    logging.warning(
                        "Global camera repair failed for %s on attempt %d: %s",
                        uuid,
                        attempt,
                        exc,
                    )
        if subject_profile_needs_repair(global_payload):
            for attempt in (1, 2):
                logging.info(
                    "Repairing subject_profile for %s (attempt %d)",
                    uuid,
                    attempt,
                )
                try:
                    repaired_subject_profile_payload = runner.generate_json(
                        video_paths=global_video_inputs,
                        prompt=build_subject_profile_repair_prompt(
                            uuid=uuid,
                            context=context,
                            events=events,
                            draft_payload=global_payload,
                            attempt=attempt,
                        ),
                        max_new_tokens=min(args.global_max_new_tokens, 768),
                    )
                    if "subject_profile" in repaired_subject_profile_payload:
                        global_payload["subject_profile"] = repaired_subject_profile_payload["subject_profile"]
                    if not subject_profile_needs_repair(global_payload):
                        break
                except Exception as exc:
                    logging.warning(
                        "subject_profile repair failed for %s on attempt %d: %s",
                        uuid,
                        attempt,
                        exc,
                    )
        global_prompt = coerce_global_payload(global_payload, context=context, events=events)
        validate_caption_lengths(uuid, global_prompt)

        return {
            "uuid": uuid,
            "total_events": len(events),
            "global_prompt": global_prompt,
            "events": events,
        }
    finally:
        if temp_manager is not None:
            temp_manager.cleanup()


def main() -> int:
    args = parse_args()
    setup_logging(args.log_level)

    input_path = Path(args.input).expanduser().resolve()
    output_path = Path(args.output).expanduser().resolve()
    model_path = Path(args.model_path).expanduser().resolve()
    camera_motion_guide_path = Path(args.camera_motion_guide).expanduser().resolve()

    suffixes = parse_video_suffixes(args.video_suffixes)
    allow_uuids = load_uuid_allow_list(args.uuids, args.uuids_file)
    camera_motion_guide = load_camera_motion_guide(camera_motion_guide_path)

    groups = collect_segment_groups(
        input_path=input_path,
        recursive=args.recursive,
        suffixes=suffixes,
        allow_uuids=allow_uuids,
    )
    if not groups:
        raise RuntimeError("No valid segmented videos were found.")

    uuids = list(groups)
    if args.limit is not None:
        uuids = uuids[: args.limit]

    pending_for_print = uuids
    if args.resume and output_path.exists():
        existing_videos_for_print = load_existing_output(
            output_path=output_path,
            overwrite=False,
            resume=True,
        )
        pending_for_print = [uuid for uuid in uuids if uuid not in existing_videos_for_print]

    if args.print_uuids_json:
        print(json.dumps(pending_for_print, ensure_ascii=False))
        return 0

    uuids = apply_uuid_shard(
        uuids=pending_for_print if args.resume else uuids,
        num_shards=args.num_shards,
        shard_index=args.shard_index,
    )

    logging.info("Discovered %d UUID groups.", len(uuids))
    for uuid in uuids:
        logging.info("UUID %s has %d segments.", uuid, len(groups[uuid]))

    if args.dry_run:
        logging.info("Dry run complete. No model was loaded.")
        return 0

    existing_videos = load_existing_output(
        output_path=output_path,
        overwrite=args.overwrite,
        resume=args.resume,
    )
    pending_uuids = [uuid for uuid in uuids if uuid not in existing_videos]
    if not pending_uuids:
        logging.info("Nothing to do. All requested UUIDs already exist in %s.", output_path)
        return 0

    if not model_path.exists():
        raise FileNotFoundError(
            f"Model path does not exist: {model_path}. Point --model-path to the local Qwen3-VL checkpoint directory."
        )

    logging.info("Loading model from %s", model_path)
    runner = Qwen3VLRunner(
        model_path=model_path,
        device_map=args.device_map,
        dtype=args.dtype,
        attn_implementation=args.attn_implementation,
        trust_remote_code=args.trust_remote_code,
        fps=args.fps,
        max_pixels=args.max_pixels,
        temperature=args.temperature,
        top_p=args.top_p,
        top_k=args.top_k,
        repetition_penalty=args.repetition_penalty,
        greedy=args.greedy,
    )

    failures: list[str] = []
    for index, uuid in enumerate(pending_uuids, start=1):
        logging.info("Processing UUID %s (%d/%d)", uuid, index, len(pending_uuids))
        try:
            video_payload = annotate_uuid(
                uuid=uuid,
                segments=groups[uuid],
                runner=runner,
                camera_motion_guide=camera_motion_guide,
                args=args,
            )
            existing_videos[uuid] = video_payload
            write_output(output_path, existing_videos)
            logging.info("Wrote %s", output_path)
        except Exception as exc:
            if not args.continue_on_error:
                raise
            logging.exception("Failed to process %s: %s", uuid, exc)
            failures.append(uuid)

    if failures:
        logging.error("Completed with failures: %s", ", ".join(failures))
        return 1
    logging.info("Finished %d UUIDs.", len(pending_uuids))
    return 0


if __name__ == "__main__":
    sys.exit(main())