#!/usr/bin/env python3
"""Deterministic, source-locked bilingual subtitle preparation and rendering.

The source SRT is archived byte-for-byte.  Source cue text is parsed into an
immutable ledger and every rendered segment references whole ledger cues.  A
translation file can provide only an id, the locked source hash, and the
target-language text; it cannot provide an editable copy of the source.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import sys
import tempfile
import unicodedata
from typing import Any, Iterable, Iterator, Sequence


SCHEMA_VERSION = 1
PIPELINE_VERSION = "1.1"
TRANSLATION_CONTRACT_VERSION = 4
DEFAULT_TARGET_LANGUAGE = "zh-CN"
ARCHIVE_NAME = "source.original.srt"
MANIFEST_NAME = "subtitle-manifest.json"
VALIDATION_NAME = "validation.json"
TRANSLATION_INPUT_DIR = "translation-input"
TRANSLATION_OUTPUT_DIR = "translation-output"
DEFAULT_FONT = "MiSans"
DEFAULT_FONT_WEIGHT = 700
TRANSLATION_ENGINE = "active_codex_default_gpt"
TRANSLATION_BATCH_SIZE = 80
TRANSLATION_CONTEXT_SEGMENTS = 2
ASS_WORD_JOINER = "\u2060"
# Both languages get the same pixel budget: 68 half-width columns at font 42
# and 62 columns (31 CJK characters) at font 46 are each about 1430 of the
# 1760 available PlayRes pixels on 16:9 video.
SOURCE_WRAP_COLUMNS = 68
TARGET_WRAP_COLUMNS = 62
SOURCE_FONT_SIZE = 42
TARGET_FONT_SIZE = 46
PORTRAIT_SOURCE_FONT_SIZE = 36
PORTRAIT_TARGET_FONT_SIZE = 40
ASS_PLAY_RES_Y = 1080
ASS_MARGIN_X = 80
ASS_BOTTOM_MARGIN = 50
PORTRAIT_BOTTOM_MARGIN = 120
DEFAULT_VIDEO_SIZE = (1920, 1080)


class PipelineError(RuntimeError):
    """A user-facing validation or pipeline failure."""


class NoDialogueError(PipelineError):
    """The source subtitle contains only non-dialogue annotations."""


_TIME_RE = re.compile(
    r"^(?P<h>\d+):(?P<m>[0-5]\d):(?P<s>[0-5]\d)[,.](?P<ms>\d{1,3})$"
)
_TIMING_RE = re.compile(
    r"^\s*(?P<start>\d+:[0-5]\d:[0-5]\d[,.]\d{1,3})"
    r"\s*-->\s*"
    r"(?P<end>\d+:[0-5]\d:[0-5]\d[,.]\d{1,3})"
    r"(?P<settings>.*)$"
)


def _sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _canonical_json_bytes(value: Any) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def _sha256_json(value: Any) -> str:
    return _sha256_bytes(_canonical_json_bytes(value))


def _json_bytes(value: Any) -> bytes:
    return (
        json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    ).encode("utf-8")


def _atomic_write(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    temporary = Path(temporary_name)
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    except BaseException:
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass
        raise


def _read_json(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise PipelineError(f"file not found: {path}") from exc
    except json.JSONDecodeError as exc:
        raise PipelineError(f"invalid JSON in {path}: {exc}") from exc


def _parse_timestamp(value: str) -> int:
    match = _TIME_RE.fullmatch(value)
    if not match:
        raise PipelineError(f"invalid SRT timestamp: {value!r}")
    milliseconds = int(match.group("ms").ljust(3, "0"))
    return (
        int(match.group("h")) * 3_600_000
        + int(match.group("m")) * 60_000
        + int(match.group("s")) * 1_000
        + milliseconds
    )


def parse_srt_bytes(raw: bytes) -> list[dict[str, Any]]:
    """Parse UTF-8 SRT while preserving every cue text code point.

    SRT record separators and CRLF/LF encoding are structural.  The original
    bytes are independently archived and SHA-256 locked.
    """

    try:
        decoded = raw.decode("utf-8-sig")
    except UnicodeDecodeError as exc:
        raise PipelineError("source SRT must be UTF-8 or UTF-8 with BOM") from exc

    normalized = decoded.replace("\r\n", "\n").replace("\r", "\n")
    body = normalized.strip("\n")
    if not body:
        raise PipelineError("source SRT is empty")

    blocks = re.split(r"\n[ \t]*\n+", body)
    parsed: list[dict[str, Any]] = []
    previous_start = -1
    for position, block in enumerate(blocks, start=1):
        lines = block.split("\n")
        timing_index: int | None = None
        timing_match: re.Match[str] | None = None
        for candidate in range(min(2, len(lines))):
            match = _TIMING_RE.fullmatch(lines[candidate])
            if match:
                timing_index = candidate
                timing_match = match
                break
        if timing_index is None or timing_match is None:
            raise PipelineError(f"SRT cue {position} has no valid timing line")

        original_index = lines[0] if timing_index == 1 else None
        text = "\n".join(lines[timing_index + 1 :])
        if text == "":
            raise PipelineError(f"SRT cue {position} has empty source text")
        start_ms = _parse_timestamp(timing_match.group("start"))
        end_ms = _parse_timestamp(timing_match.group("end"))
        if end_ms <= start_ms:
            raise PipelineError(f"SRT cue {position} has non-positive duration")
        if start_ms < previous_start:
            raise PipelineError("source SRT cue start times are not monotonic")
        previous_start = start_ms
        parsed.append(
            {
                "position": position,
                "original_index": original_index,
                "timing_line": lines[timing_index],
                "start_ms": start_ms,
                "end_ms": end_ms,
                "settings": timing_match.group("settings"),
                "text": text,
            }
        )
    return parsed


# Corner brackets 「」『』 are Japanese quotation marks around real dialogue,
# so they are deliberately not treated as annotation brackets.
_ANNOTATION_BRACKETS = re.compile(r"[\[(【（〔][^\[\]()【】（）〔〕]*[\])】）〕]")
_MUSIC_NOTES = re.compile(r"[♪♫♬♩🎵🎶]+")


def is_non_dialogue_annotation(text: str) -> bool:
    """True when an entire cue is a sound description such as [Music].

    YouTube captions interleave cues like ``[Music]``, ``[Applause]``,
    ``【音乐】``, ``（拍手）`` or bare music notes with the dialogue. Whole
    annotation cues carry nothing to translate or display. Cues that mix an
    annotation with dialogue are kept untouched, because displayed source text
    must never be edited.
    """

    remainder = _MUSIC_NOTES.sub(" ", _ANNOTATION_BRACKETS.sub(" ", text))
    if remainder == text:
        return False
    return not any(character.isalnum() for character in remainder)


def _build_cue_ledger(parsed: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
    cues: list[dict[str, Any]] = []
    for raw_cue in parsed:
        if is_non_dialogue_annotation(raw_cue["text"]):
            continue
        payload = {
            "position": raw_cue["position"],
            "original_index": raw_cue["original_index"],
            "timing_line": raw_cue["timing_line"],
            "start_ms": raw_cue["start_ms"],
            "end_ms": raw_cue["end_ms"],
            "settings": raw_cue["settings"],
            "text": raw_cue["text"],
        }
        cue_sha256 = _sha256_json(payload)
        cues.append(
            {
                **payload,
                "id": f"cue-{raw_cue['position']:06d}-{cue_sha256[:12]}",
                "text_sha256": _sha256_bytes(raw_cue["text"].encode("utf-8")),
                "cue_sha256": cue_sha256,
            }
        )
    return cues


def _cluster_spans(text: str) -> Iterator[tuple[int, int, str]]:
    """Yield practical extended grapheme clusters using only the stdlib."""

    index = 0
    length = len(text)
    while index < length:
        start = index
        index += 1
        while index < length:
            codepoint = ord(text[index])
            if (
                unicodedata.combining(text[index])
                or 0xFE00 <= codepoint <= 0xFE0F
                or 0xE0100 <= codepoint <= 0xE01EF
                or 0x1F3FB <= codepoint <= 0x1F3FF
            ):
                index += 1
                continue
            if text[index] == "\u200d" and index + 1 < length:
                index += 2
                continue
            break
        yield start, index, text[start:index]


def _cluster_width(cluster: str) -> int:
    if cluster == "\t":
        return 4
    if cluster == "\n":
        return 0
    widths: list[int] = []
    for character in cluster:
        if character == "\u200d" or unicodedata.combining(character):
            continue
        if unicodedata.category(character) in {"Cf", "Mn", "Me"}:
            continue
        widths.append(2 if unicodedata.east_asian_width(character) in {"W", "F"} else 1)
    return max(widths, default=0) if "\u200d" in cluster else sum(widths)


def _display_width(text: str) -> int:
    return sum(_cluster_width(cluster) for _, _, cluster in _cluster_spans(text))


def _wrap_single_line_exact(text: str, max_columns: int) -> list[str]:
    if not text:
        return [""]
    clusters = list(_cluster_spans(text))
    pieces: list[str] = []
    start_cluster = 0
    while start_cluster < len(clusters):
        width = 0
        index = start_cluster
        last_space_boundary: int | None = None
        while index < len(clusters):
            cluster = clusters[index][2]
            next_width = _cluster_width(cluster)
            if index > start_cluster and width + next_width > max_columns:
                break
            width += next_width
            index += 1
            if cluster.isspace():
                last_space_boundary = index
            if width > max_columns and index == start_cluster + 1:
                break
        if index >= len(clusters):
            pieces.append(text[clusters[start_cluster][0] :])
            break
        cut_cluster = (
            last_space_boundary
            if last_space_boundary is not None and last_space_boundary > start_cluster
            else index
        )
        if cut_cluster <= start_cluster:
            cut_cluster = start_cluster + 1
        cut_codepoint = clusters[cut_cluster - 1][1]
        pieces.append(text[clusters[start_cluster][0] : cut_codepoint])
        start_cluster = cut_cluster
    return pieces


def wrap_layout_chunks(text: str, max_columns: int) -> list[str]:
    """Return exact source chunks separated only by renderer-added line breaks.

    ``''.join(result)`` is guaranteed to equal ``text`` exactly.  Renderers use
    ``'\n'.join(result)``; therefore wrapping never replaces or removes a source
    character.
    """

    if max_columns < 1:
        raise ValueError("max_columns must be positive")
    chunks = [""]
    cursor = 0
    while cursor <= len(text):
        newline = text.find("\n", cursor)
        if newline < 0:
            content = text[cursor:]
            delimiter = ""
            at_end = True
        else:
            content = text[cursor:newline]
            delimiter = "\n"
            at_end = False
        pieces = _wrap_single_line_exact(content, max_columns)
        chunks[-1] += pieces[0]
        chunks.extend(pieces[1:])
        chunks[-1] += delimiter
        if at_end:
            break
        cursor = newline + 1
    if "".join(chunks) != text:
        raise AssertionError("layout wrapping changed source text")
    return chunks


def _render_wrapped(text: str, max_columns: int) -> str:
    return "\n".join(wrap_layout_chunks(text, max_columns))


_CHINESE_TARGET_BASES = {"zh", "zho", "chi", "cmn", "yue", "wuu"}


def _language_base_code(language: str) -> str:
    return language.strip().lower().replace("_", "-").split("-", 1)[0]


def normalize_target_caption(text: str, target_language: str) -> str:
    """Apply the house style without changing the immutable source subtitle.

    Only Chinese targets have a house style: full-width commas and periods
    become a single space inside a cue and disappear at its edges, while
    ASCII punctuation stays available for model numbers, URLs, code, and
    foreign names. Other target languages keep their native punctuation.
    """

    if _language_base_code(target_language) in _CHINESE_TARGET_BASES:
        return re.sub(r"\s*[，。]+\s*", " ", text).strip()
    return text.strip()


_SENTENCE_END = re.compile(r"[.!?。！？…][\"'”’)）\]】》]*\s*$")


def _smart_group_indices(cues: Sequence[dict[str, Any]]) -> list[list[int]]:
    """Group whole adjacent cues into sentence-aligned display segments.

    A group always closes at end-of-sentence punctuation, so one caption never
    spans two sentences and a sentence is split only when it exceeds the
    duration or width budget. Cues are never split or rewritten.
    """

    groups: list[list[int]] = []
    current: list[int] = []
    for index, cue in enumerate(cues):
        if current:
            first = cues[current[0]]
            previous = cues[current[-1]]
            gap = cue["start_ms"] - previous["end_ms"]
            combined_span = cue["end_ms"] - first["start_ms"]
            combined_text = " ".join(
                cues[item]["text"].replace("\n", " ") for item in [*current, index]
            )
            joins_sentence = (
                not _SENTENCE_END.search(previous["text"])
                and -1_000 <= gap <= 750
                and combined_span <= 7_000
                and _display_width(combined_text) <= 84
            )
            if not joins_sentence:
                groups.append(current)
                current = []
        current.append(index)
    if current:
        groups.append(current)
    return groups


def _segment_payload(cues: Sequence[dict[str, Any]], cue_indices: Sequence[int]) -> dict[str, Any]:
    selected = [cues[index] for index in cue_indices]
    return {
        "cue_ids": [cue["id"] for cue in selected],
        "cue_sha256": [cue["cue_sha256"] for cue in selected],
        "start_ms": selected[0]["start_ms"],
        "end_ms": max(cue["end_ms"] for cue in selected),
    }


def _build_segments(
    cues: Sequence[dict[str, Any]], segment_mode: str
) -> list[dict[str, Any]]:
    if segment_mode == "preserve":
        groups = [[index] for index in range(len(cues))]
    elif segment_mode == "smart":
        groups = _smart_group_indices(cues)
    else:
        raise PipelineError(f"unsupported segment mode: {segment_mode}")

    segments: list[dict[str, Any]] = []
    for position, group in enumerate(groups, start=1):
        payload = _segment_payload(cues, group)
        source_sha256 = _sha256_json(payload)
        segments.append(
            {
                "position": position,
                "id": f"seg-{position:06d}-{source_sha256[:12]}",
                "start_ms": payload["start_ms"],
                "end_ms": payload["end_ms"],
                "cue_ids": payload["cue_ids"],
                "source_sha256": source_sha256,
            }
        )
    if segment_mode == "smart":
        for current, following in zip(segments, segments[1:]):
            if current["end_ms"] > following["start_ms"]:
                current["end_ms"] = max(
                    current["start_ms"] + 10, following["start_ms"]
                )
    return segments


def _cue_map(manifest: dict[str, Any]) -> dict[str, dict[str, Any]]:
    return {cue["id"]: cue for cue in manifest["cues"]}


def _segment_source_text(
    segment: dict[str, Any], cue_by_id: dict[str, dict[str, Any]]
) -> str:
    return "\n".join(cue_by_id[cue_id]["text"] for cue_id in segment["cue_ids"])


def _translation_item(
    segment: dict[str, Any], cue_by_id: dict[str, dict[str, Any]]
) -> dict[str, str]:
    return {
        "id": segment["id"],
        "source_sha256": segment["source_sha256"],
        "source": _segment_source_text(segment, cue_by_id),
    }


def _compact_translation_item(
    segment: dict[str, Any], cue_by_id: dict[str, dict[str, Any]]
) -> dict[str, str]:
    return {
        "id": segment["id"],
        "source": _segment_source_text(segment, cue_by_id),
    }


def _write_translation_batches(
    work_dir: Path,
    source_language: str,
    target_language: str,
    cues: Sequence[dict[str, Any]],
    segments: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
    input_dir = work_dir / TRANSLATION_INPUT_DIR
    input_dir.mkdir(parents=True, exist_ok=True)
    for stale in input_dir.glob("batch-*.json"):
        stale.unlink()

    cue_by_id = {cue["id"]: cue for cue in cues}
    batches: list[dict[str, Any]] = []
    # Bounded ordered batches keep one model pass small enough to answer
    # completely; read-only neighbor context preserves terminology, pronouns,
    # and sentence flow across batch edges.
    for batch_number, start in enumerate(
        range(0, len(segments), TRANSLATION_BATCH_SIZE), start=1
    ):
        selected = list(segments[start : start + TRANSLATION_BATCH_SIZE])
        end = start + len(selected)
        before = list(segments[max(0, start - TRANSLATION_CONTEXT_SEGMENTS) : start])
        after = list(segments[end : end + TRANSLATION_CONTEXT_SEGMENTS])
        payload = {
            "translation_contract_version": TRANSLATION_CONTRACT_VERSION,
            "source_language": source_language,
            "target_language": target_language,
            "context": {
                "before": [_compact_translation_item(item, cue_by_id) for item in before],
                "after": [_compact_translation_item(item, cue_by_id) for item in after],
            },
            "items": [_compact_translation_item(item, cue_by_id) for item in selected],
            "output_fields": ["id", "translation"],
        }
        path = (input_dir / f"batch-{batch_number:04d}.json").resolve()
        encoded = _canonical_json_bytes(payload) + b"\n"
        _atomic_write(path, encoded)
        batches.append(
            {
                "path": str(path),
                "sha256": _sha256_bytes(encoded),
                "segment_ids": [segment["id"] for segment in selected],
            }
        )
    return batches


def _validated_video_size(video_size: tuple[int, int] | None) -> tuple[int, int]:
    if video_size is None:
        return DEFAULT_VIDEO_SIZE
    try:
        width, height = int(video_size[0]), int(video_size[1])
    except (TypeError, ValueError, IndexError) as exc:
        raise PipelineError("video size must be two positive integers") from exc
    if width <= 0 or height <= 0:
        raise PipelineError("video size must be two positive integers")
    return width, height


def prepare(
    source_srt: Path,
    work_dir: Path,
    source_language: str,
    segment_mode: str = "preserve",
    video_size: tuple[int, int] | None = None,
    target_language: str = DEFAULT_TARGET_LANGUAGE,
) -> Path:
    source_srt = source_srt.expanduser().resolve()
    work_dir = work_dir.expanduser().resolve()
    width, height = _validated_video_size(video_size)
    if not source_language.strip():
        raise PipelineError("--source-language cannot be empty")
    target_language = target_language.strip()
    if not target_language or not re.fullmatch(r"[A-Za-z]{2,3}(?:-[A-Za-z0-9]{1,8})*", target_language):
        raise PipelineError("--target-language must be a language tag such as zh-CN, ja, or fr")
    try:
        raw = source_srt.read_bytes()
    except FileNotFoundError as exc:
        raise PipelineError(f"source SRT not found: {source_srt}") from exc

    parsed = parse_srt_bytes(raw)
    cues = _build_cue_ledger(parsed)
    if not cues:
        raise NoDialogueError(
            "source SRT contains only non-dialogue annotations such as [Music]; "
            "there is nothing to translate"
        )
    # Translation units retain every original cue. Display segmentation is
    # derived separately and is applied only after the complete translation.
    segments = _build_segments(cues, "preserve")
    render_segments = _build_segments(cues, segment_mode)
    work_dir.mkdir(parents=True, exist_ok=True)
    archive = (work_dir / ARCHIVE_NAME).resolve()
    if archive.exists():
        if archive.read_bytes() != raw:
            raise PipelineError(
                "write-once source archive already exists with different bytes"
            )
    elif archive != source_srt:
        _atomic_write(archive, raw)
    else:
        raise PipelineError("source archive unexpectedly disappeared while preparing")
    if archive.read_bytes() != raw:
        raise PipelineError("source archive is not byte-for-byte identical")

    batches = _write_translation_batches(
        work_dir, source_language.strip(), target_language, cues, segments
    )
    translation_output_dir = (work_dir / TRANSLATION_OUTPUT_DIR).resolve()
    translation_output_dir.mkdir(parents=True, exist_ok=True)
    manifest = {
        "schema_version": SCHEMA_VERSION,
        "pipeline_version": PIPELINE_VERSION,
        "translation_contract_version": TRANSLATION_CONTRACT_VERSION,
        "source_language": source_language.strip(),
        "target_language": target_language,
        "segment_mode": segment_mode,
        "video_size": {"width": width, "height": height},
        "source": {
            "original_path": str(source_srt),
            "archive_path": str(archive),
            "sha256": _sha256_bytes(raw),
            "size_bytes": len(raw),
            "encoding": "utf-8-sig" if raw.startswith(b"\xef\xbb\xbf") else "utf-8",
        },
        "cues": cues,
        "source_ledger_sha256": _sha256_json(cues),
        "segments": segments,
        "segment_ledger_sha256": _sha256_json(segments),
        "render_segments": render_segments,
        "render_segment_ledger_sha256": _sha256_json(render_segments),
        "translation_batches": batches,
        "translation_output_dir": str(translation_output_dir),
    }
    manifest_path = (work_dir / MANIFEST_NAME).resolve()
    _atomic_write(manifest_path, _json_bytes(manifest))
    validate_manifest(manifest_path)
    return manifest_path


def validate_manifest(manifest_path: Path) -> dict[str, Any]:
    manifest_path = manifest_path.expanduser().resolve()
    manifest = _read_json(manifest_path)
    if not isinstance(manifest, dict):
        raise PipelineError("subtitle manifest root must be an object")
    if manifest.get("schema_version") != SCHEMA_VERSION:
        raise PipelineError("unsupported subtitle manifest schema version")
    if manifest.get("pipeline_version") != PIPELINE_VERSION:
        raise PipelineError("unsupported subtitle pipeline version")

    source = manifest.get("source")
    if not isinstance(source, dict) or not isinstance(source.get("archive_path"), str):
        raise PipelineError("manifest source archive is missing")
    archive = Path(source["archive_path"]).expanduser()
    try:
        raw = archive.read_bytes()
    except FileNotFoundError as exc:
        raise PipelineError(f"locked source archive not found: {archive}") from exc
    if _sha256_bytes(raw) != source.get("sha256"):
        raise PipelineError("locked source archive SHA-256 mismatch")
    if len(raw) != source.get("size_bytes"):
        raise PipelineError("locked source archive size mismatch")

    expected_cues = _build_cue_ledger(parse_srt_bytes(raw))
    if manifest.get("cues") != expected_cues:
        raise PipelineError("manifest cue ledger differs from locked source SRT")
    if manifest.get("source_ledger_sha256") != _sha256_json(expected_cues):
        raise PipelineError("source cue ledger SHA-256 mismatch")

    segment_mode = manifest.get("segment_mode")
    contract_version = manifest.get("translation_contract_version", 1)
    expected_segments = _build_segments(
        expected_cues, "preserve" if contract_version >= 3 else segment_mode
    )
    if manifest.get("segments") != expected_segments:
        raise PipelineError("segment ledger/provenance differs from locked source cues")
    if manifest.get("segment_ledger_sha256") != _sha256_json(expected_segments):
        raise PipelineError("segment ledger SHA-256 mismatch")

    if contract_version >= 3:
        expected_render_segments = _build_segments(expected_cues, segment_mode)
        if manifest.get("render_segments") != expected_render_segments:
            raise PipelineError("render segment ledger differs from locked source cues")
        if manifest.get("render_segment_ledger_sha256") != _sha256_json(expected_render_segments):
            raise PipelineError("render segment ledger SHA-256 mismatch")

    covered = [cue_id for segment in expected_segments for cue_id in segment["cue_ids"]]
    expected_order = [cue["id"] for cue in expected_cues]
    if covered != expected_order:
        raise PipelineError("segments do not cover source cues exactly once and in order")

    if contract_version not in {1, 2, 3, TRANSLATION_CONTRACT_VERSION}:
        raise PipelineError("unsupported translation contract version")
    batches = manifest.get("translation_batches")
    if not isinstance(batches, list) or not batches:
        raise PipelineError("manifest translation batches are missing")
    cue_by_id = {cue["id"]: cue for cue in expected_cues}
    segment_by_id = {segment["id"]: segment for segment in expected_segments}
    segment_position = {
        segment["id"]: index for index, segment in enumerate(expected_segments)
    }
    batched_ids: list[str] = []
    for batch in batches:
        if not isinstance(batch, dict) or not isinstance(batch.get("path"), str):
            raise PipelineError("invalid translation batch record")
        path = Path(batch["path"])
        try:
            data = path.read_bytes()
        except FileNotFoundError as exc:
            raise PipelineError(f"translation input batch not found: {path}") from exc
        if _sha256_bytes(data) != batch.get("sha256"):
            raise PipelineError(f"translation input batch SHA-256 mismatch: {path}")
        payload = _read_json(path)
        if not isinstance(payload, dict) or not isinstance(payload.get("items"), list):
            raise PipelineError(f"invalid translation input batch: {path}")
        if payload.get("source_language") != manifest.get("source_language") or payload.get("target_language") != manifest.get("target_language"):
            raise PipelineError(f"translation input contract mismatch: {path}")
        if contract_version == 1:
            expected_contract = {
                "engine": TRANSLATION_ENGINE,
                "external_translation_service_allowed": False,
                "local_inference_allowed": False,
            }
            if payload.get("task") != "translate_subtitles_to_simplified_chinese" or payload.get("execution_contract") != expected_contract:
                raise PipelineError(f"translation input contract mismatch: {path}")
        elif payload.get("translation_contract_version") != contract_version or payload.get("output_fields") != ["id", _translation_field(manifest)]:
            raise PipelineError(f"translation input contract mismatch: {path}")
        item_ids = [item.get("id") for item in payload["items"] if isinstance(item, dict)]
        if item_ids != batch.get("segment_ids"):
            raise PipelineError(f"translation batch segment IDs mismatch: {path}")
        if any(segment_id not in segment_by_id for segment_id in item_ids):
            raise PipelineError(f"translation batch has an unknown segment ID: {path}")
        item_builder = _translation_item if contract_version == 1 else _compact_translation_item
        expected_items = [item_builder(segment_by_id[segment_id], cue_by_id) for segment_id in item_ids]
        if payload["items"] != expected_items:
            raise PipelineError(f"translation batch source text/hash was altered: {path}")
        positions = [segment_position[segment_id] for segment_id in item_ids]
        if positions != list(range(positions[0], positions[0] + len(positions))):
            raise PipelineError(f"translation batch segment order is not contiguous: {path}")
        context = payload.get("read_only_context" if contract_version == 1 else "context")
        if not isinstance(context, dict) or set(context) != {"before", "after"}:
            raise PipelineError(f"translation read-only context is invalid: {path}")
        first_position = positions[0]
        after_position = positions[-1] + 1
        expected_before = [
            item_builder(segment, cue_by_id)
            for segment in expected_segments[
                max(0, first_position - TRANSLATION_CONTEXT_SEGMENTS) : first_position
            ]
        ]
        expected_after = [
            item_builder(segment, cue_by_id)
            for segment in expected_segments[
                after_position : after_position + TRANSLATION_CONTEXT_SEGMENTS
            ]
        ]
        if context["before"] != expected_before or context["after"] != expected_after:
            raise PipelineError(f"translation read-only context was altered: {path}")
        batched_ids.extend(item_ids)
    if batched_ids != [segment["id"] for segment in expected_segments]:
        raise PipelineError("translation batches do not cover segments exactly once")
    return manifest


def _translation_field(manifest: dict[str, Any]) -> str:
    """Output field name: generic since contract v4, zh_cn on legacy manifests."""

    contract_version = manifest.get("translation_contract_version", 1)
    return "translation" if isinstance(contract_version, int) and contract_version >= 4 else "zh_cn"


def _translation_records_from_root(
    root: Any, path: Path, field: str = "translation"
) -> list[dict[str, Any]]:
    if isinstance(root, list):
        records = root
    elif isinstance(root, dict) and set(root) == {"translations"}:
        records = root["translations"]
    elif isinstance(root, dict) and set(root) in (
        {"id", field},
        {"id", "source_sha256", field},
    ):
        records = [root]
    else:
        raise PipelineError(
            f"{path} must be a translation list, a translations object, or one strict item"
        )
    if not isinstance(records, list):
        raise PipelineError(f"translations in {path} must be an array")
    output: list[dict[str, Any]] = []
    for index, record in enumerate(records, start=1):
        if not isinstance(record, dict):
            raise PipelineError(f"translation {index} in {path} must be an object")
        allowed_sets = ({"id", field}, {"id", "source_sha256", field})
        if set(record) not in allowed_sets:
            allowed = {"id", field}
            extra = sorted(set(record) - {"id", "source_sha256", field})
            missing = sorted(allowed - set(record))
            raise PipelineError(
                f"translation {index} in {path} has forbidden/missing fields "
                f"(extra={extra}, missing={missing})"
            )
        if not all(isinstance(value, str) for value in record.values()):
            raise PipelineError(f"translation {index} in {path} fields must be strings")
        translated = record[field]
        if not translated.strip():
            raise PipelineError(f"translation {record['id']} in {path} is empty")
        if any(unicodedata.category(character) == "Cc" for character in translated):
            raise PipelineError(
                f"translation {record['id']} in {path} contains a control character"
            )
        output.append(record)
    return output


def load_translations(
    manifest: dict[str, Any], translations_dir: Path
) -> dict[str, str]:
    translations_dir = translations_dir.expanduser().resolve()
    if not translations_dir.is_dir():
        raise PipelineError(f"translations directory not found: {translations_dir}")
    files = sorted(translations_dir.glob("*.json"))
    if not files:
        raise PipelineError(f"no translation JSON files found in {translations_dir}")

    field = _translation_field(manifest)
    expected = {segment["id"]: segment for segment in manifest["segments"]}
    collected: dict[str, str] = {}
    for path in files:
        for record in _translation_records_from_root(_read_json(path), path, field):
            segment_id = record["id"]
            if segment_id in collected:
                raise PipelineError(f"duplicate translation ID: {segment_id}")
            if segment_id not in expected:
                raise PipelineError(f"extra translation ID: {segment_id}")
            if "source_sha256" in record and record["source_sha256"] != expected[segment_id]["source_sha256"]:
                raise PipelineError(f"source SHA-256 mismatch for translation {segment_id}")
            collected[segment_id] = record[field]

    missing = [segment_id for segment_id in expected if segment_id not in collected]
    if missing:
        raise PipelineError(f"missing translations: {', '.join(missing)}")
    return collected


def next_translation_batch(manifest_path: Path) -> dict[str, Any]:
    """Return one pending compact batch without exposing the full manifest."""

    manifest = validate_manifest(manifest_path)
    output_dir = Path(manifest["translation_output_dir"])
    output_dir.mkdir(parents=True, exist_ok=True)
    batches = manifest["translation_batches"]
    pending: list[tuple[dict[str, Any], Path]] = []
    for batch in batches:
        input_path = Path(batch["path"])
        output_path = output_dir / input_path.name
        if not output_path.exists():
            pending.append((batch, output_path))
            continue
        records = _translation_records_from_root(
            _read_json(output_path), output_path, _translation_field(manifest)
        )
        if [record["id"] for record in records] != batch["segment_ids"]:
            raise PipelineError(f"translation output IDs mismatch: {output_path}")
    if not pending:
        return {"done": True, "remaining": 0, "translations_dir": str(output_dir)}
    batch, output_path = pending[0]
    input_path = Path(batch["path"])
    return {
        "done": False,
        "remaining": len(pending),
        "input_path": str(input_path),
        "output_path": str(output_path),
        "batch": _read_json(input_path),
    }


def _srt_timestamp(milliseconds: int) -> str:
    hours, remainder = divmod(milliseconds, 3_600_000)
    minutes, remainder = divmod(remainder, 60_000)
    seconds, millis = divmod(remainder, 1_000)
    return f"{hours:02d}:{minutes:02d}:{seconds:02d},{millis:03d}"


def _ass_timestamp(milliseconds: int, *, end: bool = False) -> str:
    centiseconds = (milliseconds + 9) // 10 if end else milliseconds // 10
    hours, remainder = divmod(centiseconds, 360_000)
    minutes, remainder = divmod(remainder, 6_000)
    seconds, centis = divmod(remainder, 100)
    return f"{hours}:{minutes:02d}:{seconds:02d}.{centis:02d}"


def ass_escape(text: str) -> str:
    """Losslessly encode untrusted visible text for an ASS Dialogue field."""

    # Mirror FFmpeg's official ff_ass_bprint_text_event strategy: guard every
    # literal backslash with U+2060 WORD JOINER so sequences such as \N, \n and
    # \h cannot become ASS commands; encode an opening brace as ``\{{}`` so it
    # cannot begin an override block; and reserve ``\N`` for layout newlines.
    output: list[str] = []
    for character in text:
        if character == "\\":
            output.append("\\" + ASS_WORD_JOINER)
        elif character == "{":
            output.append(r"\{{}")
        elif character == "\n":
            output.append(r"\N")
        else:
            output.append(character)
    return "".join(output)


def ass_unescape_for_validation(encoded: str) -> str:
    """Strict inverse of :func:`ass_escape` for source-integrity validation."""

    output: list[str] = []
    index = 0
    while index < len(encoded):
        if encoded.startswith("\\" + ASS_WORD_JOINER, index):
            output.append("\\")
            index += 2
            continue
        if encoded.startswith(r"\{{}", index):
            output.append("{")
            index += 4
            continue
        if encoded.startswith(r"\N", index):
            output.append("\n")
            index += 2
            continue
        if encoded[index] == "\\":
            raise PipelineError("ASS text contains an unguarded backslash")
        output.append(encoded[index])
        index += 1
    return "".join(output)


def _render_srt(entries: Iterable[tuple[int, int, str]]) -> str:
    blocks: list[str] = []
    for index, (start_ms, end_ms, text) in enumerate(entries, start=1):
        blocks.append(
            f"{index}\n{_srt_timestamp(start_ms)} --> {_srt_timestamp(end_ms)}\n{text}"
        )
    return "\n\n".join(blocks) + "\n"


def _validate_font(font: str) -> str:
    if not font.strip():
        raise PipelineError("font name cannot be empty")
    if any(character in font for character in "\r\n,"):
        raise PipelineError("font name contains an unsafe ASS header character")
    return font.strip()


def _render_segments(manifest: dict[str, Any]) -> list[dict[str, Any]]:
    return manifest.get("render_segments", manifest["segments"])


def _display_translation(
    manifest: dict[str, Any], segment: dict[str, Any], translations: dict[str, str]
) -> str:
    if "render_segments" not in manifest:
        return translations[segment["id"]]
    translation_by_cue: dict[str, str] = {}
    for unit in manifest["segments"]:
        if len(unit["cue_ids"]) != 1:
            raise PipelineError("post-translation segmentation requires one cue per unit")
        translation_by_cue[unit["cue_ids"][0]] = translations[unit["id"]]
    return " ".join(translation_by_cue[cue_id].strip() for cue_id in segment["cue_ids"])


def _manifest_video_size(manifest: dict[str, Any]) -> tuple[int, int]:
    size = manifest.get("video_size")
    if size is None:
        return DEFAULT_VIDEO_SIZE
    if not isinstance(size, dict):
        raise PipelineError("manifest video_size must be an object")
    return _validated_video_size((size.get("width"), size.get("height")))


def _ass_layout(manifest: dict[str, Any]) -> dict[str, int]:
    """Derive PlayRes and wrap widths from the video aspect ratio.

    PlayResY is fixed and PlayResX follows the display aspect, so libass scales
    fonts and positions isotropically on portrait and landscape video alike.
    Wrap widths shrink with the available horizontal space and are capped at
    the 16:9 house-style limits.
    """

    width, height = _manifest_video_size(manifest)
    play_res_y = ASS_PLAY_RES_Y
    play_res_x = max(320, round(play_res_y * width / height))
    portrait = width < height
    source_font_size = PORTRAIT_SOURCE_FONT_SIZE if portrait else SOURCE_FONT_SIZE
    target_font_size = PORTRAIT_TARGET_FONT_SIZE if portrait else TARGET_FONT_SIZE
    bottom_margin = PORTRAIT_BOTTOM_MARGIN if portrait else ASS_BOTTOM_MARGIN
    available = max(160, play_res_x - 2 * ASS_MARGIN_X)
    return {
        "play_res_x": play_res_x,
        "play_res_y": play_res_y,
        "source_font_size": source_font_size,
        "target_font_size": target_font_size,
        "bottom_margin": bottom_margin,
        "source_columns": max(12, min(SOURCE_WRAP_COLUMNS, 2 * available // source_font_size)),
        "target_columns": max(8, min(TARGET_WRAP_COLUMNS, 2 * available // target_font_size)),
        "position_x": play_res_x // 2,
        "position_y": play_res_y - bottom_margin,
    }


def _render_ass(
    manifest: dict[str, Any], translations: dict[str, str], font: str,
    suppressed_segment_ids: set[str] | None = None,
) -> str:
    font = _validate_font(font)
    cue_by_id = _cue_map(manifest)
    layout = _ass_layout(manifest)
    header = f"""[Script Info]
; Generated by subtitle_pipeline.py from a SHA-256 locked source ledger.
; Typeface: MiSans Bold by Xiaomi. https://hyperos.mi.com/font/zh/download/
ScriptType: v4.00+
PlayResX: {layout['play_res_x']}
PlayResY: {layout['play_res_y']}
WrapStyle: 2
ScaledBorderAndShadow: yes

[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Bilingual,{font},{layout['target_font_size']},&H00FFFFFF,&H000000FF,&H00000000,&H64000000,-1,0,0,0,100,100,0,0,1,2,1,2,{ASS_MARGIN_X},{ASS_MARGIN_X},{layout['bottom_margin']},1
Style: BilingualBox,{font},{layout['target_font_size']},&HFF000000,&HFF000000,&H78000000,&H78000000,-1,0,0,0,100,100,0,0,4,8,0,2,{ASS_MARGIN_X},{ASS_MARGIN_X},{layout['bottom_margin']},1

[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
    # One bottom-anchored event stacks source directly above the translation,
    # so the pair hugs the bottom margin and the two languages can never
    # overlap. The box layer uses libass BorderStyle=4: one translucent panel
    # behind the whole event. Per-line BorderStyle=3 boxes would overlap
    # between adjacent lines and double-paint the translucent color.
    anchor = (
        rf"{{\an2\pos({layout['position_x']},{layout['position_y']})"
        rf"\fs{layout['source_font_size']}}}"
    )
    target_language = str(manifest.get("target_language") or DEFAULT_TARGET_LANGUAGE)
    dialogue: list[str] = []
    for segment in _render_segments(manifest):
        if suppressed_segment_ids and any(
            segment_id in suppressed_segment_ids for segment_id in segment["cue_ids"]
        ):
            continue
        source_exact = _segment_source_text(segment, cue_by_id)
        target_exact = normalize_target_caption(
            _display_translation(manifest, segment, translations), target_language
        )
        source = "\n".join(wrap_layout_chunks(source_exact, layout["source_columns"]))
        target = "\n".join(wrap_layout_chunks(target_exact, layout["target_columns"]))
        escaped_source = ass_escape(source)
        escaped_target = ass_escape(target)
        if ass_unescape_for_validation(escaped_source) != source:
            raise PipelineError(f"ASS source escape round-trip failed for {segment['id']}")
        if ass_unescape_for_validation(escaped_target) != target:
            raise PipelineError(f"ASS translation escape round-trip failed for {segment['id']}")
        start = _ass_timestamp(segment["start_ms"])
        end_ms = max(segment["end_ms"], segment["start_ms"] + 10)
        end = _ass_timestamp(end_ms, end=True)
        box_text = (
            f"{anchor}{escaped_source}"
            rf"\N{{\fs{layout['target_font_size']}}}{escaped_target}"
        )
        text = (
            f"{anchor}{escaped_source}"
            rf"\N{{\fs{layout['target_font_size']}\1c&H00FFFF&}}{escaped_target}"
        )
        dialogue.append(f"Dialogue: 0,{start},{end},BilingualBox,,0,0,0,,{box_text}")
        dialogue.append(f"Dialogue: 1,{start},{end},Bilingual,,0,0,0,,{text}")
    return header + "\n".join(dialogue) + "\n"


def _target_srt_name(manifest: dict[str, Any]) -> str:
    target_language = str(manifest.get("target_language") or DEFAULT_TARGET_LANGUAGE)
    safe = re.sub(r"[^A-Za-z0-9_-]+", "_", target_language).strip("_-") or "target"
    return f"{safe}.srt"


def _expected_outputs(
    manifest: dict[str, Any], translations: dict[str, str], font: str,
    suppressed_segment_ids: set[str] | None = None,
) -> dict[str, bytes]:
    cue_by_id = _cue_map(manifest)
    layout = _ass_layout(manifest)
    target_language = str(manifest.get("target_language") or DEFAULT_TARGET_LANGUAGE)
    source_entries: list[tuple[int, int, str]] = []
    target_entries: list[tuple[int, int, str]] = []
    bilingual_entries: list[tuple[int, int, str]] = []
    for segment in _render_segments(manifest):
        if suppressed_segment_ids and any(
            segment_id in suppressed_segment_ids for segment_id in segment["cue_ids"]
        ):
            continue
        source_exact = _segment_source_text(segment, cue_by_id)
        source_chunks = wrap_layout_chunks(source_exact, layout["source_columns"])
        if "".join(source_chunks) != source_exact:
            raise PipelineError(f"source wrapping changed {segment['id']}")
        source_layout = "\n".join(source_chunks)
        target_exact = normalize_target_caption(
            _display_translation(manifest, segment, translations), target_language
        )
        target_chunks = wrap_layout_chunks(target_exact, layout["target_columns"])
        if "".join(target_chunks) != target_exact:
            raise PipelineError(f"translation wrapping changed {segment['id']}")
        target_layout = "\n".join(target_chunks)
        timing = (segment["start_ms"], segment["end_ms"])
        source_entries.append((*timing, source_layout))
        target_entries.append((*timing, target_layout))
        bilingual_entries.append((*timing, f"{source_layout}\n{target_layout}"))
    return {
        "source.srt": _render_srt(source_entries).encode("utf-8"),
        _target_srt_name(manifest): _render_srt(target_entries).encode("utf-8"),
        "bilingual.srt": _render_srt(bilingual_entries).encode("utf-8"),
        "bilingual.ass": _render_ass(
            manifest, translations, font, suppressed_segment_ids
        ).encode("utf-8"),
    }


def _check_outputs(
    output_dir: Path,
    expected: dict[str, bytes],
    manifest_path: Path,
) -> dict[str, str]:
    checksums: dict[str, str] = {}
    for name, expected_bytes in expected.items():
        path = output_dir / name
        try:
            actual = path.read_bytes()
        except FileNotFoundError as exc:
            raise PipelineError(f"rendered subtitle is missing: {path}") from exc
        if actual != expected_bytes:
            raise PipelineError(f"rendered subtitle differs from locked expectation: {path}")
        checksums[name] = _sha256_bytes(actual)

    rendered_manifest = output_dir / MANIFEST_NAME
    try:
        copied = rendered_manifest.read_bytes()
    except FileNotFoundError as exc:
        raise PipelineError(f"rendered manifest is missing: {rendered_manifest}") from exc
    original = manifest_path.read_bytes()
    if copied != original:
        raise PipelineError("rendered manifest differs from locked input manifest")
    checksums[MANIFEST_NAME] = _sha256_bytes(copied)
    return checksums


def _scientific_review_binding(
    review_report: Path | None,
    manifest_path: Path,
    translations_dir: Path,
    translations: dict[str, str],
) -> dict[str, Any] | None:
    if review_report is None:
        return None
    review_report = review_report.expanduser().resolve()
    value = _read_json(review_report)
    if not isinstance(value, dict) or value.get("status") != "complete":
        raise PipelineError("scientific review report is incomplete or malformed")
    if value.get("human_expert_reviewed") is not False:
        raise PipelineError("scientific review report must not claim human expert review")
    if value.get("subtitle_manifest_sha256") != _sha256_bytes(manifest_path.read_bytes()):
        raise PipelineError("scientific review report is bound to a different subtitle manifest")
    declared_dir = value.get("reviewed_translations_dir")
    if not isinstance(declared_dir, str) or Path(declared_dir).expanduser().resolve() != translations_dir.expanduser().resolve():
        raise PipelineError("render translations do not match the scientific review report")
    if value.get("reviewed_translation_sha256") != _sha256_json(translations):
        raise PipelineError("reviewed translations changed after scientific review")
    counts = value.get("counts")
    if not isinstance(counts, dict) or counts.get("total") != len(translations):
        raise PipelineError("scientific review report segment count is invalid")
    suppressed = value.get("suppressed_segment_ids", [])
    if not isinstance(suppressed, list) or not all(isinstance(segment_id, str) for segment_id in suppressed):
        raise PipelineError("scientific review suppression list is malformed")
    return {
        "report_path": str(review_report),
        "report_sha256": _sha256_bytes(review_report.read_bytes()),
        "method": value.get("review_method"),
        "human_expert_reviewed": False,
        "counts": counts,
        "disclosure": value.get("disclosure"),
        "suppressed_segment_ids": suppressed,
    }


def _validation_report(
    manifest: dict[str, Any],
    checksums: dict[str, str],
    font: str,
    scientific_review: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return {
        "schema_version": SCHEMA_VERSION,
        "structurally_valid": True,
        "validation_scope": "structural_source_integrity",
        "translation_quality_reviewed": scientific_review is not None,
        "scientific_review": scientific_review,
        "target_language": manifest.get("target_language") or DEFAULT_TARGET_LANGUAGE,
        "source_sha256": manifest["source"]["sha256"],
        "source_ledger_sha256": manifest["source_ledger_sha256"],
        "segment_ledger_sha256": manifest["segment_ledger_sha256"],
        "segment_count": len(manifest["segments"]),
        "translation_count": len(manifest["segments"]),
        "font": _validate_font(font),
        "font_weight": DEFAULT_FONT_WEIGHT,
        "outputs": checksums,
        "invariants": {
            "raw_source_sha256_locked": True,
            "source_cues_exact_and_ordered": True,
            "segment_provenance_exact_and_ordered": True,
            "translations_complete_and_hash_matched": True,
            "ass_escape_round_trip": True,
            "rendered_outputs_match_expectation": True,
        },
    }


def render(
    manifest_path: Path,
    translations_dir: Path,
    output_dir: Path,
    font: str = DEFAULT_FONT,
    scientific_review_report: Path | None = None,
) -> Path:
    manifest_path = manifest_path.expanduser().resolve()
    output_dir = output_dir.expanduser().resolve()
    manifest = validate_manifest(manifest_path)
    translations = load_translations(manifest, translations_dir)
    scientific_review = _scientific_review_binding(
        scientific_review_report, manifest_path, translations_dir, translations
    )
    suppressed_cue_ids = None
    if scientific_review:
        suppressed_review_ids = set(scientific_review.get("suppressed_segment_ids", []))
        suppressed_cue_ids = {
            cue_id
            for translation_segment in manifest["segments"]
            if translation_segment["id"] in suppressed_review_ids
            for cue_id in translation_segment["cue_ids"]
        }
    expected = _expected_outputs(
        manifest,
        translations,
        font,
        suppressed_cue_ids,
    )
    output_dir.mkdir(parents=True, exist_ok=True)
    for name, data in expected.items():
        _atomic_write(output_dir / name, data)
    destination_manifest = output_dir / MANIFEST_NAME
    if destination_manifest.resolve() != manifest_path:
        _atomic_write(destination_manifest, manifest_path.read_bytes())
    checksums = _check_outputs(output_dir, expected, manifest_path)
    report = _validation_report(manifest, checksums, font, scientific_review)
    report_path = output_dir / VALIDATION_NAME
    _atomic_write(report_path, _json_bytes(report))
    return report_path


def validate(
    manifest_path: Path,
    translations_dir: Path,
    output_dir: Path,
    font: str = DEFAULT_FONT,
    scientific_review_report: Path | None = None,
) -> Path:
    manifest_path = manifest_path.expanduser().resolve()
    output_dir = output_dir.expanduser().resolve()
    manifest = validate_manifest(manifest_path)
    translations = load_translations(manifest, translations_dir)
    scientific_review = _scientific_review_binding(
        scientific_review_report, manifest_path, translations_dir, translations
    )
    expected = _expected_outputs(manifest, translations, font)
    checksums = _check_outputs(output_dir, expected, manifest_path)
    report = _validation_report(manifest, checksums, font, scientific_review)
    report_path = output_dir / VALIDATION_NAME
    _atomic_write(report_path, _json_bytes(report))
    return report_path


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Prepare, render, and validate source-locked bilingual subtitles."
    )
    commands = parser.add_subparsers(dest="command", required=True)

    prepare_parser = commands.add_parser("prepare", help="lock and segment a source SRT")
    prepare_parser.add_argument("source_srt", type=Path)
    prepare_parser.add_argument("--work-dir", type=Path, required=True)
    prepare_parser.add_argument("--source-language", required=True)
    prepare_parser.add_argument(
        "--segment-mode", choices=("preserve", "smart"), default="preserve"
    )
    prepare_parser.add_argument(
        "--video-size",
        metavar="WIDTHxHEIGHT",
        help="video display size used for caption layout (default: 1920x1080)",
    )
    prepare_parser.add_argument(
        "--target-language",
        default=DEFAULT_TARGET_LANGUAGE,
        help="translation target language tag, e.g. zh-CN (default), ja, or fr",
    )

    next_parser = commands.add_parser(
        "next-batch", help="print only the next pending compact translation batch"
    )
    next_parser.add_argument("--manifest", type=Path, required=True)

    for name, help_text in (
        ("render", "render bilingual subtitle artifacts"),
        ("validate", "validate existing bilingual subtitle artifacts"),
    ):
        command = commands.add_parser(name, help=help_text)
        command.add_argument("--manifest", type=Path, required=True)
        command.add_argument("--translations-dir", type=Path, required=True)
        command.add_argument("--output-dir", type=Path, required=True)
        command.add_argument(
            "--font",
            default=DEFAULT_FONT,
            help="ASS font family (default: MiSans; subtitle styles use weight 700/Bold)",
        )
        command.add_argument(
            "--scientific-review-report",
            type=Path,
            help="checksum-bound scientific review report for the reviewed translations",
        )
    return parser


def _parse_video_size(value: str | None) -> tuple[int, int] | None:
    if value is None:
        return None
    match = re.fullmatch(r"\s*(\d+)\s*[xX×]\s*(\d+)\s*", value)
    if not match:
        raise PipelineError("--video-size must look like 1920x1080")
    return int(match.group(1)), int(match.group(2))


def main(argv: Sequence[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    try:
        if args.command == "prepare":
            result = prepare(
                args.source_srt,
                args.work_dir,
                args.source_language,
                args.segment_mode,
                _parse_video_size(args.video_size),
                args.target_language,
            )
            payload = {"ok": True, "manifest": str(result)}
        elif args.command == "next-batch":
            payload = {"ok": True, **next_translation_batch(args.manifest)}
        elif args.command == "render":
            result = render(
                args.manifest,
                args.translations_dir,
                args.output_dir,
                args.font,
                args.scientific_review_report,
            )
            payload = {"ok": True, "validation": str(result)}
        else:
            result = validate(
                args.manifest,
                args.translations_dir,
                args.output_dir,
                args.font,
                args.scientific_review_report,
            )
            payload = {"ok": True, "validation": str(result)}
        print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
        return 0
    except (PipelineError, OSError, UnicodeError) as exc:
        print(f"subtitle pipeline error: {exc}", file=sys.stderr)
        return 2


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