Video Recap

#!/usr/bin/env python3
"""Environment doctor for the video-recap skill bundle.

The whole pipeline runs on ffmpeg + a single MiMo API key (ASR + VLM + TTS all use MiMo).
"""
from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any, cast

from lib import CONFIG


SCRIPT_DIR = Path(__file__).resolve().parent
DEGRADED_GROUP = "warnings/degraded"


def _command_path(name: str) -> str | None:
    """Return resolved command path, accepting absolute/relative executable paths."""
    if not name:
        return None
    if os.path.sep in name or (os.path.altsep and os.path.altsep in name):
        path = Path(name).expanduser()
        return str(path) if path.exists() and os.access(path, os.X_OK) else None
    return shutil.which(name)


def _run(cmd: list[str], *, timeout: int = 20) -> subprocess.CompletedProcess[str]:
    return subprocess.run(cmd, text=True, capture_output=True, timeout=timeout)


def _ffmpeg_filters() -> set[str]:
    ffmpeg = _command_path("ffmpeg")
    if not ffmpeg:
        return set()
    try:
        result = _run([ffmpeg, "-hide_banner", "-filters"], timeout=20)
    except (OSError, subprocess.SubprocessError):
        return set()
    if result.returncode != 0:
        return set()
    filters = set()
    for line in result.stdout.splitlines():
        parts = line.split()
        if len(parts) >= 2 and parts[0] and parts[0][0] in ".TSCAPN|":
            filters.add(parts[1])
    return filters


def ffmpeg_has_subtitles_filter() -> bool:
    """True when this ffmpeg can burn subtitles — its filter list includes the libass
    `subtitles` filter. The render burns even the .ass file through `subtitles=` (see
    video-assemble assemble.py:_subtitle_burn_filter), so this — not the `ass` filter — is
    the exact capability `--burn-subtitles` needs. Reused by the orchestrator preflight
    (recap.py) to fail fast before any API spend."""
    return "subtitles" in _ffmpeg_filters()


def _asr_status() -> dict[str, object]:
    configured = bool(CONFIG.get("mimo_asr_api_key"))
    return {
        "configured": configured,
        "available": configured,
        "mimo_asr_model": str(CONFIG.get("mimo_asr_model") or ""),
        "mimo_asr_api_url": str(CONFIG.get("mimo_asr_api_url") or ""),
        "mimo_asr_api_url_source": CONFIG.get("mimo_asr_api_url_source", "default"),
        "mimo_asr_language": str(CONFIG.get("mimo_asr_language") or "auto"),
        "mimo_asr_api_key_source": CONFIG.get("mimo_asr_api_key_source", "MIMO_API_KEY"),
        "note": "ASR uses MiMo (mimo-v2.5-asr); set MIMO_API_KEY, or run with --skip-asr.",
    }


def _capability(name: str, summary: str, *, detail: str = "", action: str = "") -> dict[str, str]:
    item = {"name": name, "summary": summary}
    if detail:
        item["detail"] = detail
    if action:
        item["action"] = action
    return item


def _build_capability_menu(checks: dict[str, object]) -> dict[str, list[dict[str, str]]]:
    """Human-ready preflight summary grouped by what can run, what blocks, and what degrades.

    This is intentionally a small rollup over the existing `checks` tree. It does not replace
    the raw machine checks, install anything, or introduce provider ranking.
    """
    system = cast(dict[str, Any], checks["system_tools"])
    api = cast(dict[str, Any], checks["api_config"])
    asr = cast(dict[str, Any], checks["asr"])
    tts = cast(dict[str, Any], checks["tts"])

    menu: dict[str, list[dict[str, str]]] = {
        "ready": [],
        "blocked": [],
        DEGRADED_GROUP: [],
        "optional_upgrades": [],
    }

    ffmpeg_ready = bool(system.get("ffmpeg"))
    ffprobe_ready = bool(system.get("ffprobe"))
    subtitles_ready = bool(system.get("burn_subtitles_ready"))
    api_key_set = bool(api.get("api_key_set"))
    asr_ready = bool(asr.get("available"))
    tts_ready = bool(tts.get("available"))
    vlm_ready = bool(api.get("mimo_video_configured"))
    normal_core_ready = ffmpeg_ready and ffprobe_ready and api_key_set and vlm_ready and tts_ready

    if ffmpeg_ready and ffprobe_ready:
        menu["ready"].append(
            _capability(
                "core_media_tools",
                "ffmpeg and ffprobe are available",
                detail="Local probing, cutting, rendering, and duration checks can run.",
            )
        )
    else:
        if not ffmpeg_ready:
            menu["blocked"].append(
                _capability("ffmpeg", "Missing ffmpeg", action="Install ffmpeg before running the recap pipeline.")
            )
        if not ffprobe_ready:
            menu["blocked"].append(
                _capability("ffprobe", "Missing ffprobe", action="Install ffprobe before running media probing/export.")
            )

    if api_key_set:
        menu["ready"].append(
            _capability(
                "mimo_credentials",
                "MiMo API key is configured",
                detail=f"Source: {api.get('api_key_source')}",
            )
        )
    else:
        menu["blocked"].append(
            _capability(
                "mimo_credentials",
                "Missing MIMO_API_KEY",
                action="Set MIMO_API_KEY; the default ASR / VLM / TTS path depends on it.",
            )
        )

    if vlm_ready:
        menu["ready"].append(
            _capability(
                "mimo_vlm",
                "MiMo VLM/video understanding is configured",
                detail=f"Model: {api.get('vlm_model')}",
            )
        )
    elif api_key_set:
        menu["blocked"].append(
            _capability(
                "mimo_vlm",
                "MiMo VLM/video understanding is not configured",
                action="Set MIMO_VIDEO_API_KEY or the shared MIMO_API_KEY before video understanding.",
            )
        )

    if tts_ready:
        menu["ready"].append(
            _capability(
                "mimo_tts",
                "MiMo TTS is configured",
                detail=f"Voice: {tts.get('mimo_tts_voice')}; model: {tts.get('mimo_tts_model')}",
            )
        )
    elif api_key_set:
        menu["blocked"].append(
            _capability(
                "mimo_tts",
                "MiMo TTS is not configured",
                action="Set MIMO_TTS_API_KEY or the shared MIMO_API_KEY before voiceover.",
            )
        )

    if asr_ready:
        menu["ready"].append(
            _capability(
                "mimo_asr",
                "MiMo ASR is configured",
                detail=f"Language: {asr.get('mimo_asr_language')}; model: {asr.get('mimo_asr_model')}",
            )
        )
    else:
        menu[DEGRADED_GROUP].append(
            _capability(
                "mimo_asr",
                "ASR is unavailable; run only with --skip-asr",
                action=str(asr.get("note") or "Set MIMO_ASR_API_KEY or MIMO_API_KEY to enable ASR."),
            )
        )

    if subtitles_ready:
        menu["ready"].append(
            _capability("subtitle_burn", "Subtitle burn-in is available", detail="ffmpeg has the subtitles/libass filter.")
        )
    elif ffmpeg_ready:
        menu[DEGRADED_GROUP].append(
            _capability(
                "subtitle_burn",
                "Subtitle burn-in is unavailable",
                action="Use --no-burn-subtitles or install an ffmpeg build with the subtitles/libass filter.",
            )
        )

    if not normal_core_ready:
        menu["blocked"].append(
            _capability(
                "default_recap_pipeline",
                "Default recap run is blocked",
                detail="Resolve the blocking items above before a normal run.",
            )
        )
    elif asr_ready and subtitles_ready:
        menu["ready"].append(
            _capability("default_recap_pipeline", "Default recap run is ready", detail="ASR, VLM, TTS, and media tools are configured.")
        )
    else:
        actions = []
        if not asr_ready:
            actions.append("run with --skip-asr")
        if not subtitles_ready:
            actions.append("run with --no-burn-subtitles")
        menu[DEGRADED_GROUP].append(
            _capability(
                "recap_degraded_mode",
                "Recap can run only in an explicit degraded mode",
                detail="; ".join(actions),
            )
        )

    menu["optional_upgrades"].append(
        _capability(
            "jianying_export",
            "Editable JianYing draft export can be requested with --export-jianying",
            detail="No JianYing install is required to write the draft; ffprobe improves media metadata.",
        )
    )
    if subtitles_ready:
        menu["optional_upgrades"].append(
            _capability(
                "burned_subtitles",
                "Burned subtitles are available and enabled by default",
                action="Use --no-burn-subtitles if you prefer external subtitle files.",
            )
        )

    return menu


def build_report() -> dict[str, object]:
    filters = _ffmpeg_filters()
    ffmpeg_path = _command_path("ffmpeg") or ""
    ffprobe_path = _command_path("ffprobe") or ""
    mimo_video_configured = bool(CONFIG.get("mimo_video_api_key"))
    mimo_tts_configured = bool(CONFIG.get("mimo_tts_api_key"))
    subtitle_filter = "subtitles" in filters
    ass_filter = "ass" in filters
    checks: dict[str, object] = {
        "system_tools": {
            "ffmpeg": bool(ffmpeg_path),
            "ffmpeg_path": ffmpeg_path,
            "ffprobe": bool(ffprobe_path),
            "ffprobe_path": ffprobe_path,
            "ffmpeg_subtitles_filter": subtitle_filter,
            "ffmpeg_ass_filter": ass_filter,
            "burn_subtitles_ready": bool(ffmpeg_path and subtitle_filter),
        },
        "tts": {
            "mimo_tts_configured": mimo_tts_configured,
            "mimo_tts_api_url": CONFIG.get("mimo_tts_api_url"),
            "mimo_tts_api_url_source": CONFIG.get("mimo_tts_api_url_source", "default"),
            "mimo_tts_model": CONFIG.get("mimo_tts_model"),
            "mimo_tts_model_source": CONFIG.get("mimo_tts_model_source", "default"),
            "mimo_tts_voice": CONFIG.get("mimo_tts_voice"),
            "mimo_tts_voice_source": CONFIG.get("mimo_tts_voice_source", "default"),
            "available": mimo_tts_configured,
        },
        "asr": _asr_status(),
        "api_config": {
            "api_provider": CONFIG.get("api_provider", "mimo"),
            "api_url": str(CONFIG.get("api_url") or ""),
            "api_url_source": CONFIG.get("api_url_source", "default"),
            "api_key_source": CONFIG.get("api_key_source", "MIMO_API_KEY"),
            "api_key_set": bool(CONFIG.get("api_key")),
            "vlm_model": CONFIG.get("vlm_model"),
            "vlm_model_source": CONFIG.get("vlm_model_source", "default"),
            "vlm_workers": CONFIG.get("vlm_workers"),
            "mimo_video_configured": mimo_video_configured,
            "mimo_video_api_url": CONFIG.get("mimo_video_api_url"),
            "mimo_video_model": CONFIG.get("mimo_video_model"),
            "mimo_video_model_source": CONFIG.get("mimo_video_model_source", "default"),
        },
        "python": {
            "executable": sys.executable,
            "version": sys.version.split()[0],
        },
    }

    failures: list[str] = []
    warnings: list[str] = []
    tools = cast(dict[str, Any], checks["system_tools"])
    api_config = cast(dict[str, Any], checks["api_config"])
    asr_check = cast(dict[str, Any], checks["asr"])
    for name in ("ffmpeg", "ffprobe"):
        if not tools.get(name):
            failures.append(f"Missing system tool: {name}")
    if tools.get("ffmpeg") and not tools.get("ffmpeg_subtitles_filter"):
        warnings.append("ffmpeg lacks subtitles/libass filter; --burn-subtitles will fail")
    if not api_config.get("api_key_set"):
        failures.append("MIMO_API_KEY is not set; ASR / VLM / TTS all require a MiMo key")
    if not asr_check.get("available"):
        warnings.append("ASR not configured (MIMO_API_KEY); pipeline can run with --skip-asr")
    capability_menu = _build_capability_menu(checks)
    return {
        "ok": not failures,
        "repo_root": str(SCRIPT_DIR.parents[2]),
        "checks": checks,
        "capability_menu": capability_menu,
        "failures": failures,
        "warnings": warnings,
    }


def _status_icon(ok: bool, *, warning: bool = False) -> str:
    if ok:
        return "✓"
    return "!" if warning else "✗"


def _print_human(report: dict[str, object]) -> None:
    checks = cast(dict[str, Any], report["checks"])
    print("video-recap doctor")
    print(f"Repo root: {report['repo_root']}")

    system = cast(dict[str, Any], checks["system_tools"])
    print("\n[system]")
    print(f"{_status_icon(bool(system.get('ffmpeg')))} ffmpeg: {system.get('ffmpeg_path') or 'not found'}")
    print(f"{_status_icon(bool(system.get('ffprobe')))} ffprobe: {system.get('ffprobe_path') or 'not found'}")
    print(
        f"{_status_icon(bool(system.get('ffmpeg_subtitles_filter')), warning=True)} "
        f"ffmpeg subtitles/libass filter: "
        f"{'available' if system.get('ffmpeg_subtitles_filter') else 'missing'}"
    )

    api = cast(dict[str, Any], checks["api_config"])
    print("\n[api]")
    print(f"✓ API provider: {api.get('api_provider')}")
    print(f"✓ API URL: {api.get('api_url')} (source: {api.get('api_url_source')})")
    print(
        f"{_status_icon(bool(api.get('api_key_set')))} "
        f"{api.get('api_key_source')}: {'set' if api.get('api_key_set') else 'not set'}"
    )
    print(f"✓ VLM model: {api.get('vlm_model')} (source: {api.get('vlm_model_source')})")
    print(f"✓ VLM_WORKERS: {api.get('vlm_workers')}")

    asr = cast(dict[str, Any], checks["asr"])
    print("\n[asr]")
    print(
        f"{_status_icon(bool(asr.get('available')), warning=True)} "
        f"MiMo ASR: {'configured' if asr.get('available') else 'not configured'} "
        f"(key: {asr.get('mimo_asr_api_key_source')})"
    )
    print(f"✓ ASR model: {asr.get('mimo_asr_model')}")
    print(f"✓ ASR API URL: {asr.get('mimo_asr_api_url')} (source: {asr.get('mimo_asr_api_url_source')})")
    print(f"✓ ASR language: {asr.get('mimo_asr_language')}")
    if not asr.get("available"):
        print(f"  note: {asr.get('note')}")

    tts = cast(dict[str, Any], checks["tts"])
    print("\n[tts]")
    print(
        f"{_status_icon(bool(tts.get('available')))} MiMo TTS: "
        f"{'configured' if tts.get('mimo_tts_configured') else 'not configured'}"
    )
    print(f"✓ TTS model: {tts.get('mimo_tts_model')} (source: {tts.get('mimo_tts_model_source')})")
    print(f"✓ TTS voice: {tts.get('mimo_tts_voice')} (source: {tts.get('mimo_tts_voice_source')})")
    print(f"✓ TTS API URL: {tts.get('mimo_tts_api_url')} (source: {tts.get('mimo_tts_api_url_source')})")

    menu = cast(dict[str, list[dict[str, str]]], report.get("capability_menu") or {})
    print("\n[capability menu]")
    for group in ("ready", "blocked", DEGRADED_GROUP, "optional_upgrades"):
        print(f"{group}:")
        items = menu.get(group) or []
        if not items:
            print("  - none")
            continue
        for item in items:
            line = f"  - {item.get('name')}: {item.get('summary')}"
            if item.get("detail"):
                line += f" ({item['detail']})"
            print(line)
            if item.get("action"):
                print(f"    action: {item['action']}")

    warnings = cast(list[str], report.get("warnings") or [])
    failures = cast(list[str], report.get("failures") or [])
    if warnings:
        print("\nWarnings:")
        for warning in warnings:
            print(f"- {warning}")
    if failures:
        print("\nStatus: FAILED")
        for failure in failures:
            print(f"- {failure}")
    else:
        print("\nStatus: OK")


def main() -> int:
    parser = argparse.ArgumentParser(description="Check video-recap runtime prerequisites.")
    parser.add_argument("--json", action="store_true", help="Print machine-readable JSON")
    args = parser.parse_args()

    report = build_report()
    if args.json:
        print(json.dumps(report, ensure_ascii=False, indent=2))
        return 0 if report["ok"] else 1

    _print_human(report)
    return 0 if report["ok"] else 1


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

"""Self-contained config + log for the video-recap orchestrator (no cross-skill imports)."""
import os
from pathlib import Path


# ── 配置 ──────────────────────────────────────────────────────────────

DEFAULT_MIMO_API_URL = "https://api.xiaomimimo.com/v1"
DEFAULT_MIMO_TOKEN_PLAN_CLUSTER = "cn"
MIMO_TOKEN_PLAN_API_URLS = {
    "cn": "https://token-plan-cn.xiaomimimo.com/v1",
    "sgp": "https://token-plan-sgp.xiaomimimo.com/v1",
    "ams": "https://token-plan-ams.xiaomimimo.com/v1",
}
DEFAULT_MIMO_MODEL = "mimo-v2.5"          # VLM / chat (vision understanding)
DEFAULT_MIMO_ASR_MODEL = "mimo-v2.5-asr"  # speech-to-text
DEFAULT_MIMO_TTS_MODEL = "mimo-v2.5-tts"  # text-to-speech


def normalize_api_url(raw_url):
    """Normalize a MiMo (OpenAI-compatible) base URL or chat/completions endpoint."""
    url = (raw_url or DEFAULT_MIMO_API_URL).rstrip("/")
    if url.endswith("/chat/completions"):
        return url
    return f"{url}/chat/completions"


def is_mimo_token_plan_key(api_key):
    """Return True for Xiaomi MiMo Token Plan keys, which use token-plan base URLs."""
    return str(api_key or "").strip().startswith("tp-")


def default_mimo_api_url(api_key="", cluster=None):
    """Pick the correct MiMo base URL for pay-as-you-go vs Token Plan keys.

    MiMo uses independent credentials for pay-as-you-go (`sk-*`) and Token Plan
    (`tp-*`). Token Plan keys must be sent to the Token Plan cluster base URL,
    not the pay-as-you-go `api.xiaomimimo.com` endpoint.
    """
    if is_mimo_token_plan_key(api_key):
        cluster_name = (cluster or os.environ.get("MIMO_TOKEN_PLAN_CLUSTER") or DEFAULT_MIMO_TOKEN_PLAN_CLUSTER)
        cluster_name = str(cluster_name).strip().lower()
        return MIMO_TOKEN_PLAN_API_URLS.get(cluster_name, MIMO_TOKEN_PLAN_API_URLS[DEFAULT_MIMO_TOKEN_PLAN_CLUSTER])
    return DEFAULT_MIMO_API_URL


def env_int(name, default, *, minimum=None):
    """Read an integer env var; ignore malformed values instead of crashing import."""
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    try:
        value = int(raw)
    except (TypeError, ValueError):
        return default
    if minimum is not None:
        value = max(minimum, value)
    return value


def env_bool(name, default=False):
    """Read common boolean env var forms."""
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    return raw.strip().lower() in {"1", "true", "yes", "y", "on"}


def env_float(name, default, *, minimum=None):
    """Read a float env var; ignore malformed values instead of crashing import."""
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return default
    if minimum is not None:
        value = max(minimum, value)
    return value


# Single MiMo credential powers ASR + VLM + TTS. Per-capability overrides
# (MIMO_VIDEO_API_KEY / MIMO_TTS_API_KEY / MIMO_ASR_API_KEY and their *_API_URL forms)
# are optional and fall back to MIMO_API_KEY / MIMO_API_URL. Token-Plan keys (tp-*) auto-
# route to the Token-Plan cluster base URL; pay-as-you-go keys use api.xiaomimimo.com.
_mimo_api_key = os.environ.get("MIMO_API_KEY", "")
_mimo_video_api_key = os.environ.get("MIMO_VIDEO_API_KEY", "") or _mimo_api_key
_mimo_tts_api_key = os.environ.get("MIMO_TTS_API_KEY", "") or _mimo_api_key
_mimo_asr_api_key = os.environ.get("MIMO_ASR_API_KEY", "") or _mimo_api_key
_raw_api_url = os.environ.get("MIMO_API_URL") or default_mimo_api_url(_mimo_api_key)
_raw_mimo_video_api_url = (
    os.environ.get("MIMO_VIDEO_API_URL")
    or os.environ.get("MIMO_API_URL")
    or default_mimo_api_url(_mimo_video_api_key)
)
_raw_mimo_tts_api_url = (
    os.environ.get("MIMO_TTS_API_URL")
    or os.environ.get("MIMO_API_URL")
    or default_mimo_api_url(_mimo_tts_api_key)
)
_raw_mimo_asr_api_url = (
    os.environ.get("MIMO_ASR_API_URL")
    or os.environ.get("MIMO_API_URL")
    or default_mimo_api_url(_mimo_asr_api_key)
)

CONFIG = {
    "api_provider": "mimo",
    "api_provider_source": "default",
    "api_url": normalize_api_url(_raw_api_url),
    "api_url_source": "env" if os.environ.get("MIMO_API_URL") else "default",
    "api_key": _mimo_api_key,
    "api_key_source": "MIMO_API_KEY",
    "mimo_api_url": normalize_api_url(_raw_api_url),
    "mimo_api_url_source": "env" if os.environ.get("MIMO_API_URL") else "default",
    "mimo_api_key": _mimo_api_key,
    "mimo_api_key_source": "MIMO_API_KEY",
    "mimo_video_api_url": normalize_api_url(_raw_mimo_video_api_url),
    "mimo_video_api_url_source": "env" if (
        os.environ.get("MIMO_VIDEO_API_URL") or os.environ.get("MIMO_API_URL")
    ) else "default",
    "mimo_video_api_key": _mimo_video_api_key,
    "mimo_video_api_key_source": "MIMO_VIDEO_API_KEY" if os.environ.get("MIMO_VIDEO_API_KEY") else "MIMO_API_KEY",
    "mimo_tts_api_url": normalize_api_url(_raw_mimo_tts_api_url),
    "mimo_tts_api_url_source": "env" if (
        os.environ.get("MIMO_TTS_API_URL") or os.environ.get("MIMO_API_URL")
    ) else "default",
    "mimo_tts_api_key": _mimo_tts_api_key,
    "mimo_tts_api_key_source": "MIMO_TTS_API_KEY" if os.environ.get("MIMO_TTS_API_KEY") else "MIMO_API_KEY",
    "mimo_asr_api_url": normalize_api_url(_raw_mimo_asr_api_url),
    "mimo_asr_api_url_source": "env" if (
        os.environ.get("MIMO_ASR_API_URL") or os.environ.get("MIMO_API_URL")
    ) else "default",
    "mimo_asr_api_key": _mimo_asr_api_key,
    "mimo_asr_api_key_source": "MIMO_ASR_API_KEY" if os.environ.get("MIMO_ASR_API_KEY") else "MIMO_API_KEY",
    "mimo_model": os.environ.get("MIMO_MODEL", DEFAULT_MIMO_MODEL),
    "mimo_model_source": "env" if os.environ.get("MIMO_MODEL") else "default",
    "mimo_video_model": os.environ.get("MIMO_VIDEO_MODEL") or os.environ.get("MIMO_MODEL", DEFAULT_MIMO_MODEL),
    "mimo_video_model_source": "env" if (
        os.environ.get("MIMO_VIDEO_MODEL") or os.environ.get("MIMO_MODEL")
    ) else "default",
    "vlm_model": os.environ.get("MIMO_MODEL", DEFAULT_MIMO_MODEL),
    "vlm_model_source": "env" if os.environ.get("MIMO_MODEL") else "default",
    "mimo_asr_model": os.environ.get("MIMO_ASR_MODEL", DEFAULT_MIMO_ASR_MODEL),
    "mimo_asr_model_source": "env" if os.environ.get("MIMO_ASR_MODEL") else "default",
    "mimo_asr_language": os.environ.get("MIMO_ASR_LANGUAGE", "auto"),  # auto | zh | en
    "mimo_asr_base64_max_mb": env_float("MIMO_ASR_BASE64_MAX_MB", 10.0, minimum=1.0),
    # ASR 分段窗口秒数。越小 → 长视频的对白时间戳越精细（默认 15s）。旧值 180s 会把 >3min
    # 视频的对白塌缩成一个时间戳，既让 brief 无法定位对白，又触发 detect.py 的粗粒度跳过，
    # 使 overlaps_speech/安静窗口判断失真。代价是更多 ASR 调用；ASR 慢时可调大。
    "asr_segment_seconds": env_float("ASR_SEGMENT_SECONDS", 15.0, minimum=5.0),
    "scene_threshold": 0.1,
    "scene_threshold_source": "default",
    "mimo_tts_model": os.environ.get("MIMO_TTS_MODEL", DEFAULT_MIMO_TTS_MODEL),
    "mimo_tts_model_source": "env" if os.environ.get("MIMO_TTS_MODEL") else "default",
    "mimo_tts_voice": os.environ.get("MIMO_TTS_VOICE", "冰糖"),
    "mimo_tts_voice_source": "env" if os.environ.get("MIMO_TTS_VOICE") else "default",
    "mimo_tts_style": os.environ.get(
        "MIMO_TTS_STYLE",
        "自然、清晰、有感染力，像在给观众讲故事；随剧情起伏，该紧张时紧张、该动情时动情，不平铺直叙。",
    ),
    "mimo_tts_style_source": "env" if os.environ.get("MIMO_TTS_STYLE") else "default",
    "mimo_media_resolution": os.environ.get("MIMO_MEDIA_RESOLUTION", "default"),
    "mimo_media_resolution_source": "env" if os.environ.get("MIMO_MEDIA_RESOLUTION") else "default",
    "mimo_video_overview": env_bool("MIMO_VIDEO_OVERVIEW", False),  # opt-in (--mimo-video-overview / =1); when on it becomes the PRIMARY per-scene description, frames stay the anchor/fallback
    "mimo_video_overview_source": "env" if os.environ.get("MIMO_VIDEO_OVERVIEW") else "default",
    "mimo_video_fps": env_float("MIMO_VIDEO_FPS", 3.0, minimum=0.1),
    "mimo_video_fps_source": "env" if os.environ.get("MIMO_VIDEO_FPS") else "default",
    "mimo_video_chunk_max_seconds": env_float("MIMO_VIDEO_CHUNK_MAX_SECONDS", 20.0, minimum=1.0),
    "mimo_video_chunk_min_seconds": env_float("MIMO_VIDEO_CHUNK_MIN_SECONDS", 1.0, minimum=0.2),
    "mimo_video_chunk_timeout": env_int("MIMO_VIDEO_CHUNK_TIMEOUT", 180, minimum=1),
    "mimo_video_base64_max_mb": env_float("MIMO_VIDEO_BASE64_MAX_MB", 45.0, minimum=1.0),
    # Per-scene frame VLM sampling — scale frames with scene length instead of a hard cap of 6
    "vlm_seconds_per_frame": env_float("VLM_SECONDS_PER_FRAME", 4.0, minimum=0.5),
    "vlm_max_frames": env_int("VLM_MAX_FRAMES", 16, minimum=3),
    "vlm_max_tokens": env_int("VLM_MAX_TOKENS", 1500, minimum=200),
    "mimo_video_prompt": os.environ.get(
        "MIMO_VIDEO_PROMPT",
        "请用中文分析这个视频分片的主要人物、场景变化、关键动作、情绪走向和剧情冲突，"
        "重点提取适合写短视频解说的故事线索。不要泛泛复述画面，要标出对后续写稿有用的信息。",
    ),
    "mimo_disable_thinking": env_bool("MIMO_DISABLE_THINKING", True),
    "mimo_disable_thinking_source": "env" if os.environ.get("MIMO_DISABLE_THINKING") else "default",
    "fps": 0,  # 0 = 自动（≤60s→2fps, ≤5min→1.5fps, >5min→1fps）
    # TTS 语速（字符/秒）。实测 mimo-tts 冰糖音色中位 ~3.9 字/秒，可用 SPEECH_RATE 覆盖
    # 生成解说时使用 speech_rate * safety_margin 作为约束
    "speech_rate": env_float("SPEECH_RATE", 3.9, minimum=0.5),  # 旧值 3.5 系统性偏低 ~10-17%
    "speech_safety_margin": env_float("SPEECH_SAFETY_MARGIN", 0.85, minimum=0.1),  # 保守系数：TTS 实际语速有 ±20% 波动
    # Block-coverage lint thresholds — promoted from inline .get() literals to real CONFIG keys (tunable; defaults unchanged)
    "narration_coverage_target": 0.7,   # aim ~70% narrated:original (7:3)
    "narration_coverage_min": 0.5,      # below this coverage → under_narrated
    "narration_block_seconds": 9.0,     # block cadence used to derive target block count
    "original_block_min_seconds": 2.5,  # a deliberate original-audio gap must be at least this long
    "narration_block_min_chars": 16,    # below this avg block size → fragmented_beats
    "fade_ms": 300,  # TTS fade-in/fade-out 时长(ms)
    "breath_ms": 250,  # 段间呼吸空间(ms)；block recap 块内连贯、块间留原声呼吸
    # Legacy single-pass cut mapping density fields; current writing uses block coverage controls below.
    "target_segments_per_minute": 9.6,   # legacy single-pass cut mapping report only; block recap uses narration_coverage_*
    "min_segments_per_minute": 6.24,     # legacy single-pass cut mapping report only
    "max_narration_gap_seconds": 11.0,   # legacy single-pass cut mapping report only
    "ducking_mode": "fixed",  # fixed | sidechaincompress | none
    "ducking_threshold": 0.15,
    "ducking_ratio": 3,
    "ducking_attack": 10,
    "ducking_release": 300,
    "ducking_level_sc": 2.0,
    "ducking_makeup": 1.2,
    "ducking_narr_weight": 1.5,
    "ducking_orig_volume": env_float("DUCKING_ORIG_VOLUME", 0.3, minimum=0.0),  # 解说时原声基准音量
    "zone_ducking_volume": 0.12,    # 解说时原声压低到的音量
    "zone_fade_seconds": 0.5,      # 解说/原声切换的淡入淡出时长(秒)
    "idle_orig_volume": env_float("IDLE_ORIG_VOLUME", 1.0, minimum=0.0),  # 解说间隙(无旁白)时的原声音量，铺底避免顿挫
    "duck_fade_seconds": env_float("DUCK_FADE_SECONDS", 0.3, minimum=0.0),  # 原声 ducking 过渡淡入淡出(秒)
    "bgm_path": os.environ.get("BGM_PATH", "").strip(),  # 背景音乐文件(可选)，留空则不加 BGM
    "source_video": os.environ.get("SOURCE_VIDEO", "").strip(),  # 剪辑模式下的原始视频(可选)，用于时间线/剪映导出引用原片片段
    "export_jianying": env_bool("EXPORT_JIANYING", False),  # 渲染后可选导出剪映草稿(默认关；与核心解耦)
    "jianying_draft_dir": os.environ.get("JIANYING_DRAFT_DIR", "").strip(),  # 剪映草稿输出父目录(留空=work_dir)
    "jianying_bundle_media": env_bool("JIANYING_BUNDLE_MEDIA", True),  # 默认开：macOS 剪映沙箱读不到外部路径，须把素材拷进草稿目录
    "bgm_volume": env_float("BGM_VOLUME", 0.18, minimum=0.0),  # BGM 铺底音量
    "bgm_ducking_volume": env_float("BGM_DUCKING_VOLUME", 0.10, minimum=0.0),  # 旁白时 BGM 压低到的音量
    "narration_speed": env_float("NARRATION_SPEED", 1.3, minimum=0.5),  # 解说整体提速(atempo)，默认偏快适配短视频；长片可设 1.0
    "mask_source_subtitles": env_bool("MASK_SOURCE_SUBTITLES", True),  # 遮挡原片烧录字幕（默认开；无烧录字幕素材设 false）
    "source_subtitle_mask_ratio": env_float("SOURCE_SUBTITLE_MASK_RATIO", 0.14, minimum=0.0),  # 底部遮挡比例
    "narration_delay_seconds": 1.5,  # 解说延迟放置秒数，让画面先出现再解说
    "narration_tail_pad_seconds": 0.1,  # 解说尾部最少留白；短 slot 会自动压低 delay 避免截断
    "quiet_overlap_min_ratio": 0.8,  # 解说段至少多少比例落在安静窗口内才标记为非对白重叠
    "visual_beat_max_seconds": 18.0,  # 单段解说超过该时长且跨多个帧锚点时给 lint 提醒
    "visual_beat_max_facts": 3,  # 单段解说最多建议覆盖的 frame_facts 锚点数量
    "asr_chunk_min_chars": env_int("ASR_CHUNK_MIN_CHARS", 500, minimum=1),  # brief 中 ASR 写作分块最小字数/词数
    "asr_chunk_max_chars": env_int("ASR_CHUNK_MAX_CHARS", 800, minimum=1),  # brief 中 ASR 写作分块最大字数/词数
    "speech_ducking_volume": env_float("SPEECH_DUCKING_VOLUME", 0.2, minimum=0.0),    # 解说与对白重叠时原声音量
    "silence_noise_threshold": "-25dB",  # ffmpeg silencedetect 噪声阈值
    "silence_min_duration": 0.3,     # 静音最短持续秒数
    "quiet_window_min": 1.0,         # 可放解说的安静窗口最短秒数
    "silence_merge_gap": 0.5,        # 相邻静音段间隔<此值时合并
    "scene_merge_min": 4.0,         # 场景合并最短时长，<此值的场景合并到相邻场景
    "scene_junk_filter": env_bool("SCENE_JUNK_FILTER", True),  # 过滤连续黑/白帧无效过渡场景
    "scene_junk_dark_luma": env_float("SCENE_JUNK_DARK_LUMA", 8.0, minimum=0.0),
    "scene_junk_bright_luma": env_float("SCENE_JUNK_BRIGHT_LUMA", 245.0, minimum=0.0),
    "scene_junk_pixel_ratio": env_float("SCENE_JUNK_PIXEL_RATIO", 0.995, minimum=0.0),
    "context_info": "",              # 额外上下文（节目名、角色名等）
    "context_info_source": "default",
    "fps_source": "default",
    "style": "纪录片",               # 解说风格（resume 时随 run_settings 持久化/恢复）
    "style_source": "default",
    "tts_dynamic_params": True,  # 启用动态语速调节
    "vlm_workers": env_int("VLM_WORKERS", 8, minimum=1),  # VLM 并行分析线程数
    "tts_workers": env_int("TTS_WORKERS", 4, minimum=1),  # TTS 并行合成线程数
    "tts_timeout": env_int("TTS_TIMEOUT", 90, minimum=1),  # 单段 TTS 命令超时秒数
    "tts_retries": env_int("TTS_RETRIES", 3, minimum=1),  # 单段 TTS 失败重试次数
    "allow_partial_tts": env_bool("ALLOW_PARTIAL_TTS", False),
    "edit_mode": os.environ.get("EDIT_MODE", "full"),  # full | cut
    "edit_mode_source": "env" if os.environ.get("EDIT_MODE") else "default",
    "target_duration": os.environ.get("TARGET_DURATION", ""),  # cut 模式目标成片时长，如 10m
    "target_duration_source": "env" if os.environ.get("TARGET_DURATION") else "default",
    "clip_padding": env_float("CLIP_PADDING", 0.0, minimum=0.0),  # cut 模式片段两端扩展秒数
    "clip_padding_source": "env" if os.environ.get("CLIP_PADDING") else "default",
    "allow_clip_overlap": env_bool("ALLOW_CLIP_OVERLAP", False),  # cut 模式是否允许重复/重叠使用原片
    "burn_subtitles": env_bool("BURN_SUBTITLES", True),  # 烧录解说字幕（默认开；遮挡原字幕后需自带字幕，否则字幕区空白）
    "force_video_reencode": env_bool("FORCE_VIDEO_REENCODE", False),  # 组装时重编码视频，修复部分容器时间戳问题
    # 成片末端整体响度归一（默认混音偏轻，归一后更接近常见短视频响度；样片约 -11.9，默认取更安全的 -14）
    "final_loudnorm": env_bool("FINAL_LOUDNORM", True),  # 组装末端做一次整体响度归一
    "target_lufs": env_float("TARGET_LUFS", -14.0),       # 目标综合响度 (LUFS)
    "target_true_peak": env_float("TARGET_TRUE_PEAK", -1.0),  # 目标真峰值 (dBTP)
    "target_lra": env_float("TARGET_LRA", 11.0),          # 目标响度范围 (LU)
    "subtitle_font_name": os.environ.get("SUBTITLE_FONT_NAME", "Arial"),
    "subtitle_font_size": env_int("SUBTITLE_FONT_SIZE", 42, minimum=8),
    "subtitle_primary_color": os.environ.get("SUBTITLE_PRIMARY_COLOR", "&H00FFFFFF"),
    "subtitle_outline_color": os.environ.get("SUBTITLE_OUTLINE_COLOR", "&H00000000"),
    "subtitle_outline": env_float("SUBTITLE_OUTLINE", 2.0, minimum=0.0),
    "subtitle_shadow": env_float("SUBTITLE_SHADOW", 1.0, minimum=0.0),
    "subtitle_margin_v": env_int("SUBTITLE_MARGIN_V", 48, minimum=0),
    "subtitle_margin_l": env_int("SUBTITLE_MARGIN_L", 40, minimum=0),
    "subtitle_margin_r": env_int("SUBTITLE_MARGIN_R", 40, minimum=0),
    "subtitle_alignment": env_int("SUBTITLE_ALIGNMENT", 2, minimum=1),
    "subtitle_max_chars": env_int("SUBTITLE_MAX_CHARS", 20, minimum=6),
    "subtitle_play_res_x": env_int("SUBTITLE_PLAY_RES_X", 1280, minimum=1),
    "subtitle_play_res_y": env_int("SUBTITLE_PLAY_RES_Y", 720, minimum=1),
}

SCRIPT_DIR = Path(__file__).parent
PROMPTS_DIR = SCRIPT_DIR.parent / "references"

def log(msg):
    print(f"[video-recap] {msg}", flush=True)

#!/usr/bin/env python3
"""video-recap inspect — advisory, read-only orientation over a recap work_dir.

Pure stdlib. Reads ONLY the JSON artifacts already in work_dir (no ffmpeg, no frame
reads, no video probing, no new deps, no cross-skill import). Every missing or malformed
artifact degrades to a clear human message — never a traceback.

Two subcommands:
  state     summarize the work_dir: source video + fingerprint, full|cut mode, which stage
            artifacts are present vs missing, which file is the NEXT pause the pipeline waits
            on, stale-manifest risk, and storyboard path(s) if present.
  clip-map  read clip_plan_validated.json and map a queried window between the OUTPUT and
            SOURCE timelines using the SAME forward affine map cut.py uses
            (output = clip.output_start + (src - clip.source_start), clamped to the clip),
            reimplemented locally. Flags cross-clip boundaries and cut-out gaps.

Output: markdown by default; --json for machine-readable; --compact (default ON) truncates
long free text — pass --full to keep it.
"""
import argparse
import json
from pathlib import Path


# --- artifact catalog --------------------------------------------------------
# Stage artifacts probed by `state`. Order = rough pipeline order.
_UNDERSTANDING_ARTIFACTS = [
    "scenes.json",
    "asr_result.json",
    "silence_periods.json",
    "vlm_analysis.json",
    "understanding_index.json",
    "agent_narration_brief.md",
]
_CUT_ARTIFACTS = [
    "clip_plan.json",
    "clip_plan_validated.json",
    "edited_source.mp4",
]
_SCRIPT_ARTIFACTS = [
    "narration.json",
    "narration_lint.json",
    "narration_review.json",
    "original_subtitles.json",
]
_RENDER_ARTIFACTS = [
    "tts_meta.json",
    "narration_mapped.json",
    "timeline.json",
    "subtitles.srt",
    "subtitles.ass",
    "assembly_manifest.json",
]
_STORYBOARD_ARTIFACTS = [
    "storyboard/source_storyboard.json",
    "storyboard/edited_storyboard.json",
]
# Forward-compat: if a write-side state file ever lands, prefer it as the state source.
_FORWARD_STATE_FILES = ["manifest.json", "task_state.json"]

_COMPACT_TEXT_LIMIT = 80


def _truncate(text, compact):
    text = str(text or "").strip()
    if not compact or len(text) <= _COMPACT_TEXT_LIMIT:
        return text
    return text[: _COMPACT_TEXT_LIMIT - 1] + "…"


def _load_json(path):
    """Return (data, error). error is a human string when the file is missing/malformed."""
    path = Path(path)
    if not path.exists():
        return None, f"{path.name} 不存在"
    try:
        return json.loads(path.read_text(encoding="utf-8")), None
    except ValueError:
        return None, f"{path.name} 不是合法 JSON（可能写坏了）"
    except OSError as exc:
        return None, f"{path.name} 读取失败: {exc}"


def _fmt_seconds(value):
    """mm:ss for a numeric seconds value, or the raw value when not numeric."""
    try:
        total = float(value)
    except (TypeError, ValueError):
        return str(value)
    sign = "-" if total < 0 else ""
    total = abs(total)
    m = int(total // 60)
    s = total % 60
    return f"{sign}{m:02d}:{s:05.2f}"


# --- source video discovery --------------------------------------------------
def _discover_source(work_dir):
    """Find the source video path + fingerprint by reading recap artifacts, most authoritative
    first. Returns a dict {path, fingerprint, origin} with None values + origin="unknown" when
    nothing records it. Never raises."""
    # 1. recap_run_manifest.json — canonical: written before the first pause with both fields.
    data, _ = _load_json(Path(work_dir) / "recap_run_manifest.json")
    if isinstance(data, dict) and data.get("source_video"):
        return {
            "path": data.get("source_video"),
            "fingerprint": data.get("source_video_fingerprint"),
            "origin": "recap_run_manifest.json",
        }
    # 2. assembly_manifest.json — late stage; carries input_video (+ source_video in cut mode).
    data, _ = _load_json(Path(work_dir) / "assembly_manifest.json")
    if isinstance(data, dict) and (data.get("source_video") or data.get("input_video")):
        return {
            "path": data.get("source_video") or data.get("input_video"),
            "fingerprint": data.get("source_video_fingerprint"),
            "origin": "assembly_manifest.json",
        }
    # 3. edited_source.mp4.meta.json — cut mode; fingerprint only, no path.
    data, _ = _load_json(Path(work_dir) / "edited_source.mp4.meta.json")
    if isinstance(data, dict) and data.get("source_video_fingerprint"):
        return {
            "path": None,
            "fingerprint": data.get("source_video_fingerprint"),
            "origin": "edited_source.mp4.meta.json",
        }
    return {"path": None, "fingerprint": None, "origin": "unknown"}


# --- clip plan loading + forward affine map ----------------------------------
def _clip_entries(plan):
    """Normalize a clip plan (dict-with-clips or bare list) to the list of clip dicts."""
    if isinstance(plan, dict):
        entries = plan.get("clips", [])
    elif isinstance(plan, list):
        entries = plan
    else:
        return None
    return entries if isinstance(entries, list) else None


def _normalize_clips(entries):
    """Coerce raw clip entries to {clip_id, source_start, source_end, output_start, output_end}.

    Mirrors assemble._output_clip_spans / _build_video_clips field access: source_start/source_end
    fall back to start/end; when output_start/output_end are absent they are derived with the same
    forward cursor cut.py uses (sum of prior kept-clip durations). Bad rows are skipped, not fatal.
    """
    clips, cursor = [], 0.0
    for idx, c in enumerate(entries or []):
        if not isinstance(c, dict):
            continue
        try:
            ss = float(c.get("source_start", c.get("start")))
            se = float(c.get("source_end", c.get("end")))
        except (TypeError, ValueError):
            continue
        if se - ss <= 0:
            continue
        out_s, out_e = c.get("output_start"), c.get("output_end")
        if out_s is None or out_e is None:
            out_s, out_e = cursor, cursor + (se - ss)
            cursor += se - ss
        else:
            try:
                out_s, out_e = float(out_s), float(out_e)
            except (TypeError, ValueError):
                out_s, out_e = cursor, cursor + (se - ss)
                cursor += se - ss
            else:
                cursor = max(cursor, out_e)
        clips.append({
            "clip_id": c.get("clip_id", idx),
            "source_start": ss,
            "source_end": se,
            "output_start": out_s,
            "output_end": out_e,
            "reason": str(c.get("reason", c.get("note", ""))).strip(),
        })
    return clips


def _source_to_output(src, clip):
    """The forward affine map cut.py:319 uses, clamped to the clip's output span."""
    mapped = clip["output_start"] + (float(src) - clip["source_start"])
    return round(max(clip["output_start"], min(mapped, clip["output_end"])), 3)


def _output_to_source(out, clip):
    """Inverse of the forward affine map (same slope, clamped to the clip's source span)."""
    mapped = clip["source_start"] + (float(out) - clip["output_start"])
    return round(max(clip["source_start"], min(mapped, clip["source_end"])), 3)


def _overlap(a0, a1, b0, b1):
    """Intersection [lo, hi] of two closed ranges, or None when they do not overlap.

    A zero-width query (a0 == a1, e.g. --output-start 10 --output-end 10) is treated as a
    POINT lookup: it returns (a0, a0) when the point lies within [b0, b1], so a point that sits
    inside a clip is reported instead of silently falling through to "not in any clip".
    """
    if a0 == a1:
        return (a0, a0) if b0 <= a0 <= b1 else None
    lo, hi = max(a0, b0), min(a1, b1)
    return (lo, hi) if hi > lo else None


# --- state subcommand --------------------------------------------------------
def _present(work_dir, name):
    return (Path(work_dir) / name).exists()


def _detect_mode(work_dir):
    """cut when any cut artifact is present, else full."""
    if any(_present(work_dir, n) for n in _CUT_ARTIFACTS):
        return "cut"
    return "full"


def _next_pause(work_dir, mode):
    """The artifact the pipeline is currently waiting on the agent to write, or None when the
    next-needed input is already present. Mirrors recap.py's two-pause cut flow / single-pause full
    flow purely from file presence (advisory; the orchestrator owns the real decision)."""
    if mode == "cut":
        if not _present(work_dir, "clip_plan.json"):
            return ("clip_plan.json", "pass1: 写剪辑计划（只写 clip_plan.json）")
        if not _present(work_dir, "narration.json"):
            return ("narration.json", "pass2: 对着剪好的成片用 OUTPUT 时间轴写解说")
        return None
    if not _present(work_dir, "narration.json"):
        return ("narration.json", "写解说 narration.json")
    return None


def _stale_manifest_note(work_dir, mode):
    """Advisory stale-manifest risks read purely from JSON (no fingerprint recompute — that would
    need the source bytes). Surfaces the two desync traps recap.py guards: a cut narration written
    against an older clip_plan, and a missing run manifest."""
    notes = []
    if not _present(work_dir, "recap_run_manifest.json"):
        notes.append("缺少 recap_run_manifest.json：无法证明 work_dir 属于当前视频/参数（resume 会被拒绝）。")
    if mode == "cut" and _present(work_dir, "narration.json"):
        ledger, _ = _load_json(Path(work_dir) / "recap_phase.json")
        if isinstance(ledger, dict):
            recorded = ledger.get("clip_plan_fingerprint")
            if recorded is None:
                notes.append("recap_phase.json 未记录 clip_plan_fingerprint：无法判断 narration 是否对当前剪辑写的。")
        else:
            notes.append("有 narration.json 但缺少 recap_phase.json：无法判断解说是否对当前剪辑写的（可能 stale）。")
    return notes


def _present_storyboards(work_dir):
    return [n for n in _STORYBOARD_ARTIFACTS if _present(work_dir, n)]


def cmd_state(work_dir, compact):
    work_dir = Path(work_dir)
    if not work_dir.exists():
        return {"error": f"work_dir 不存在: {work_dir}"}

    forward = [n for n in _FORWARD_STATE_FILES if _present(work_dir, n)]
    mode = _detect_mode(work_dir)
    source = _discover_source(work_dir)

    groups = {
        "understanding": _UNDERSTANDING_ARTIFACTS,
        "cut": _CUT_ARTIFACTS,
        "script": _SCRIPT_ARTIFACTS,
        "render": _RENDER_ARTIFACTS,
    }
    artifacts = {}
    for group, names in groups.items():
        artifacts[group] = {
            "present": [n for n in names if _present(work_dir, n)],
            "missing": [n for n in names if not _present(work_dir, n)],
        }

    pause = _next_pause(work_dir, mode)
    return {
        "work_dir": str(work_dir),
        "mode": mode,
        "forward_state_files": forward,
        "source_video": source,
        "next_pause": (
            {"artifact": pause[0], "hint": pause[1]} if pause else None
        ),
        "artifacts": artifacts,
        "storyboards": _present_storyboards(work_dir),
        "stale_manifest_notes": _stale_manifest_note(work_dir, mode),
    }


def _render_state_md(state, compact):
    if "error" in state:
        return state["error"]
    lines = [f"# recap work_dir 状态: {state['work_dir']}", ""]
    if state["forward_state_files"]:
        lines.append(f"状态来源（write-side manifest）: {', '.join(state['forward_state_files'])}")
    lines.append(f"模式: **{state['mode']}**")
    src = state["source_video"]
    path = src["path"] or "unknown"
    fp = src["fingerprint"]
    fp_short = (fp[:12] + "…") if isinstance(fp, str) and len(fp) > 12 else (fp or "unknown")
    lines.append(f"源视频: {_truncate(path, compact)}  (fp {fp_short}, 来源 {src['origin']})")
    lines.append("")

    if state["next_pause"]:
        np = state["next_pause"]
        lines.append(f"下一步暂停 → 等待写入 **{np['artifact']}**  ({np['hint']})")
    else:
        lines.append("下一步暂停 → 无（所需输入已就绪，可继续 voiceover/assemble）")
    lines.append("")

    lines.append("## 各阶段产物")
    for group, label in (("understanding", "理解"), ("cut", "剪辑"),
                         ("script", "解说"), ("render", "渲染")):
        info = state["artifacts"][group]
        lines.append(f"### {label}")
        lines.append(f"  有: {', '.join(info['present']) or '（无）'}")
        lines.append(f"  缺: {', '.join(info['missing']) or '（无）'}")
    lines.append("")

    if state["storyboards"]:
        lines.append("## storyboard")
        for s in state["storyboards"]:
            lines.append(f"  - {s}")
        lines.append("")

    lines.append("## stale-manifest 风险")
    if state["stale_manifest_notes"]:
        for n in state["stale_manifest_notes"]:
            lines.append(f"  ⚠️ {n}")
    else:
        lines.append("  无明显 stale 风险。")
    return "\n".join(lines)


# --- clip-map subcommand -----------------------------------------------------
def cmd_clip_map(work_dir, output_start, output_end, source_start, source_end, compact):
    work_dir = Path(work_dir)
    validated = work_dir / "clip_plan_validated.json"
    if not validated.exists():
        return {"error": "clip_plan_validated.json 不存在：这不是 cut 运行，或剪辑计划尚未被 cut.py 校验。"
                         "（full 模式没有源↔输出映射；cut 模式请先跑 video-cut。）"}
    plan, err = _load_json(validated)
    if err:
        return {"error": err}
    entries = _clip_entries(plan)
    if entries is None:
        return {"error": "clip_plan_validated.json 结构异常：既不是 clips 数组也不是 {\"clips\": [...]}。"}
    clips = _normalize_clips(entries)
    if not clips:
        return {"error": "clip_plan_validated.json 没有有效的 clip（每段需要 source_start/source_end）。"}

    have_output = output_start is not None or output_end is not None
    have_source = source_start is not None or source_end is not None
    if not have_output and not have_source:
        return {"error": "请用 --output-start/--output-end 或 --source-start/--source-end 指定要查询的窗口。"}

    result = {
        "work_dir": str(work_dir),
        "clip_count": len(clips),
        "source_span": [clips[0]["source_start"], clips[-1]["source_end"]],
        "output_span": [clips[0]["output_start"], clips[-1]["output_end"]],
        "queries": [],
    }
    if have_output:
        result["queries"].append(
            _map_output_window(output_start, output_end, clips, compact))
    if have_source:
        result["queries"].append(
            _map_source_window(source_start, source_end, clips, compact))
    return result


def _map_output_window(out_start, out_end, clips, compact):
    """Map a queried OUTPUT window to its SOURCE window(s), one per touched clip."""
    out_lo = clips[0]["output_start"] if out_start is None else float(out_start)
    out_hi = clips[-1]["output_end"] if out_end is None else float(out_end)
    if out_hi < out_lo:
        out_lo, out_hi = out_hi, out_lo
    segments = []
    for clip in clips:
        ov = _overlap(out_lo, out_hi, clip["output_start"], clip["output_end"])
        if not ov:
            continue
        segments.append({
            "clip_id": clip["clip_id"],
            "output": [round(ov[0], 3), round(ov[1], 3)],
            "source": [_output_to_source(ov[0], clip), _output_to_source(ov[1], clip)],
            "reason": _truncate(clip.get("reason"), compact),
        })
    return {
        "direction": "output→source",
        "query": [round(out_lo, 3), round(out_hi, 3)],
        "clips_touched": [s["clip_id"] for s in segments],
        "cross_clip_boundary": len(segments) > 1,
        "segments": segments,
        # An OUTPUT window can't fall outside any clip (output is contiguous), so no cut-out gaps.
        "cut_out_source_gaps": [],
    }


def _map_source_window(src_start, src_end, clips, compact):
    """Map a queried SOURCE window to its OUTPUT window(s), flagging source ranges in no clip."""
    src_lo = clips[0]["source_start"] if src_start is None else float(src_start)
    src_hi = clips[-1]["source_end"] if src_end is None else float(src_end)
    if src_hi < src_lo:
        src_lo, src_hi = src_hi, src_lo
    segments = []
    covered = []
    for clip in clips:
        ov = _overlap(src_lo, src_hi, clip["source_start"], clip["source_end"])
        if not ov:
            continue
        covered.append(ov)
        segments.append({
            "clip_id": clip["clip_id"],
            "source": [round(ov[0], 3), round(ov[1], 3)],
            "output": [_source_to_output(ov[0], clip), _source_to_output(ov[1], clip)],
            "reason": _truncate(clip.get("reason"), compact),
        })
    # Cut-out gaps: parts of [src_lo, src_hi] covered by NO kept clip (footage that was cut away).
    gaps = []
    cursor = src_lo
    for lo, hi in sorted(covered):
        if lo > cursor:
            gaps.append([round(cursor, 3), round(lo, 3)])
        cursor = max(cursor, hi)
    if cursor < src_hi:
        gaps.append([round(cursor, 3), round(src_hi, 3)])
    return {
        "direction": "source→output",
        "query": [round(src_lo, 3), round(src_hi, 3)],
        "clips_touched": [s["clip_id"] for s in segments],
        "cross_clip_boundary": len(segments) > 1,
        "segments": segments,
        "cut_out_source_gaps": gaps,
    }


def _render_clip_map_md(result, compact):
    if "error" in result:
        return result["error"]
    lines = [f"# clip-map: {result['work_dir']}", ""]
    lines.append(
        f"{result['clip_count']} clips · "
        f"源 {_fmt_seconds(result['source_span'][0])}–{_fmt_seconds(result['source_span'][1])} · "
        f"输出 {_fmt_seconds(result['output_span'][0])}–{_fmt_seconds(result['output_span'][1])}")
    for q in result["queries"]:
        lines.append("")
        lines.append(f"## {q['direction']}  查询 "
                     f"{_fmt_seconds(q['query'][0])}–{_fmt_seconds(q['query'][1])}")
        if q["cross_clip_boundary"]:
            lines.append(f"  ⚠️ 跨剪辑边界（涉及 clip {q['clips_touched']}）")
        if not q["segments"]:
            lines.append("  （该窗口不落在任何保留片段内）")
        for s in q["segments"]:
            src = f"{_fmt_seconds(s['source'][0])}–{_fmt_seconds(s['source'][1])}"
            out = f"{_fmt_seconds(s['output'][0])}–{_fmt_seconds(s['output'][1])}"
            reason = f"  〔{s['reason']}〕" if s.get("reason") else ""
            lines.append(f"  clip {s['clip_id']}: 源 {src}  ↔  输出 {out}{reason}")
        if q["cut_out_source_gaps"]:
            lines.append("  ✂️ 被剪掉的源区间（不在成片里）:")
            for g in q["cut_out_source_gaps"]:
                lines.append(f"     {_fmt_seconds(g[0])}–{_fmt_seconds(g[1])}")
    return "\n".join(lines)


# --- CLI ---------------------------------------------------------------------
def main(argv=None):
    parser = argparse.ArgumentParser(
        prog="recap_inspect.py",
        description="Advisory read-only inspection of a video-recap work_dir (pure JSON).")
    parser.add_argument("--work-dir", required=True, help="recap work_dir to inspect")
    parser.add_argument("--json", action="store_true", help="machine-readable JSON output")
    compact = parser.add_mutually_exclusive_group()
    compact.add_argument("--compact", dest="compact", action="store_true", default=True,
                         help="truncate long free text to keep agent context small (default ON)")
    compact.add_argument("--full", dest="compact", action="store_false",
                         help="do not truncate long free text")
    sub = parser.add_subparsers(dest="command", required=True)

    sub.add_parser("state", help="summarize work_dir + the next pause the pipeline is waiting on")

    cm = sub.add_parser("clip-map", help="map a window between OUTPUT and SOURCE timelines")
    cm.add_argument("--output-start", type=float, default=None)
    cm.add_argument("--output-end", type=float, default=None)
    cm.add_argument("--source-start", type=float, default=None)
    cm.add_argument("--source-end", type=float, default=None)

    args = parser.parse_args(argv)

    if args.command == "state":
        result = cmd_state(args.work_dir, args.compact)
        rendered = _render_state_md(result, args.compact)
    else:
        result = cmd_clip_map(args.work_dir, args.output_start, args.output_end,
                              args.source_start, args.source_end, args.compact)
        rendered = _render_clip_map_md(result, args.compact)

    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(rendered)
    return 0


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

#!/usr/bin/env python3
"""video-recap orchestrator.

Chains the independent video-* stage skills into a full narrated recap:

  video-understanding  ->  (agent writes narration.json per video-script)  ->
  [video-cut]  ->  video-voiceover  ->  video-assemble

Each stage is a self-contained sibling skill invoked as a subprocess; they communicate
only through JSON/MP4 artifacts in the shared work_dir. Resume by rerunning the same
command after writing narration.json; Phase B verifies a run manifest before reusing
work_dir artifacts.
"""
import argparse
import hashlib
import json
import math
import os
import shlex
import subprocess
import sys
from pathlib import Path

from doctor import ffmpeg_has_subtitles_filter

BUNDLE = Path(__file__).resolve().parents[2]  # the skills/ directory
RUN_MANIFEST = "recap_run_manifest.json"
ASSEMBLY_MANIFEST = "assembly_manifest.json"
PHASE_LEDGER = "recap_phase.json"


def _entry(skill, script):
    return BUNDLE / skill / "scripts" / script


def _run(skill, script, *cli_args):
    cmd = [sys.executable, str(_entry(skill, script)), *map(str, cli_args)]
    print(f"[video-recap] ▶ {skill}/{script}", flush=True)
    res = subprocess.run(cmd)
    if res.returncode != 0:
        raise SystemExit(f"{skill}/{script} 失败 (exit {res.returncode})")


def _env_bool(name, default=False):
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    return raw.strip().lower() in {"1", "true", "yes", "y", "on"}


def _read_video_duration_or_raise(path):
    """Return media duration via ffprobe, or hard-fail before downstream TTS/render."""
    path = Path(path)
    cmd = [
        "ffprobe", "-v", "quiet", "-show_entries", "format=duration",
        "-of", "csv=p=0", str(path),
    ]
    res = subprocess.run(cmd, capture_output=True, text=True)
    if res.returncode != 0:
        detail = (res.stderr or res.stdout or "ffprobe failed").strip()
        raise SystemExit(f"无法读取成片时长: {path} ({detail})")
    try:
        duration = float(res.stdout.strip())
    except (TypeError, ValueError):
        raise SystemExit(f"无法读取成片时长: {path} (ffprobe 输出无效: {res.stdout!r})")
    if not math.isfinite(duration) or duration <= 0:
        raise SystemExit(f"无法读取成片时长: {path} (duration={duration:.3f})")
    return duration


def _review_narration_enabled(args):
    if getattr(args, "review_narration", None) is not None:
        return bool(args.review_narration)
    return _env_bool("REVIEW_NARRATION", True)


def _require_narration_review(args):
    if getattr(args, "require_narration_review", False):
        return True
    return _env_bool("REQUIRE_NARRATION_REVIEW", False)


def _review_result_status(work_dir):
    data = _load_json(Path(work_dir) / "narration_review.json")
    if not isinstance(data, dict):
        return {"ok": False, "reason": "missing or invalid narration_review.json"}
    findings = [f for f in (data.get("findings") or []) if isinstance(f, dict)]
    n_err = sum(1 for f in findings if f.get("severity") == "error")
    if data.get("parse_error"):
        return {"ok": False, "reason": "parse_error", "review": data, "errors": n_err}
    if n_err:
        return {"ok": False, "reason": f"error {n_err}", "review": data, "errors": n_err}
    if str(data.get("verdict", "")).upper() == "REVISE" and n_err:
        return {"ok": False, "reason": f"error {n_err}", "review": data, "errors": n_err}
    return {"ok": True, "reason": "ok", "review": data, "errors": n_err}


def _clear_narration_review_artifacts(work_dir):
    """Remove prior review artifacts before a fresh pre-TTS review run.

    The review is allowed to fail open in advisory mode, but completion output and
    strict gating must never accidentally trust a stale narration_review.* from an
    earlier run.
    """
    for name in ("narration_review.json", "narration_review.md"):
        try:
            (Path(work_dir) / name).unlink()
        except FileNotFoundError:
            pass


def _run_narration_review(work_dir, args, *, timeline="source"):
    """Run quality review before TTS.

    Default mode remains advisory/fail-open. Strict mode
    (`--require-narration-review` or REQUIRE_NARRATION_REVIEW) hard-fails before
    TTS when review is unavailable, unparsable, or reports error findings.
    Returns True only when review.py completed, so completion messages do not
    point at stale review artifacts after opt-out/fail-open runs.
    """
    strict = _require_narration_review(args)
    if not _review_narration_enabled(args) and not strict:
        return False
    try:
        _clear_narration_review_artifacts(work_dir)
        # Always pin the grounding timeline explicitly so the orchestrated review never falls
        # through to review.py's auto-detect (which could flip on stale cut artifacts left in a
        # reused full-mode work_dir).
        rargs = ["--work-dir", work_dir, "--timeline", timeline]
        _run("video-script", "review.py", *rargs)
    except SystemExit as exc:
        if strict:
            raise SystemExit(f"严格解说评审失败，已阻止 TTS: {exc}")
        print(f"[video-recap] ⚠️ 建议性评审失败，继续执行 TTS: {exc}", flush=True)
        return False

    status = _review_result_status(work_dir)
    if strict and not status["ok"]:
        raise SystemExit(f"严格解说评审未通过，已阻止 TTS: {status['reason']}")
    if strict:
        print("[video-recap] ✅ 严格解说评审通过，继续 TTS", flush=True)
    return True


def _file_fingerprint(path, chunk_size=1024 * 1024):
    h = hashlib.sha256()
    with Path(path).open("rb") as f:
        for chunk in iter(lambda: f.read(chunk_size), b""):
            h.update(chunk)
    return h.hexdigest()


def _run_manifest_payload(video, args):
    return {
        "schema_version": 1,
        "source_video": str(Path(video).resolve()),
        "source_video_fingerprint": _file_fingerprint(video),
        "settings": {
            "context": args.context,
            "scene_threshold": args.scene_threshold,
            "style": args.style,
            "edit_mode": args.edit_mode,
            "target_duration": args.target_duration,
            "skip_asr": bool(args.skip_asr),
            "mimo_video_overview": bool(args.mimo_video_overview),
            "consolidate": bool(args.consolidate),
            "consolidate_asr": bool(args.consolidate_asr),
        },
    }


def _write_run_manifest(work_dir, video, args):
    payload = _run_manifest_payload(video, args)
    (work_dir / RUN_MANIFEST).write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def _load_run_manifest(work_dir):
    path = Path(work_dir) / RUN_MANIFEST
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError, TypeError):
        return None


def _load_json(path):
    try:
        return json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, ValueError, TypeError):
        return None


def _burn_subtitles_intended(args):
    """Effective burn-subtitles state at orchestrator level. Mirrors video-assemble's
    CONFIG default `env_bool("BURN_SUBTITLES", True)` (burn is ON by default); an explicit
    CLI flag (--burn-subtitles / --no-burn-subtitles) overrides the env."""
    if getattr(args, "burn_subtitles", None) is not None:
        return bool(args.burn_subtitles)
    raw = os.environ.get("BURN_SUBTITLES")
    if raw is None or raw == "":
        return True
    return raw.strip().lower() in {"1", "true", "yes", "y", "on"}


def _ffmpeg_present_but_cannot_burn():
    """True only when ffmpeg EXISTS but lacks the libass `subtitles` filter — the specific
    "subtitle-burn environment unsupported" case. Returns False when ffmpeg is absent
    entirely: that is a more fundamental problem that surfaces at the first stage (understand
    calls ffprobe/ffmpeg) and is reported by doctor, so this guard stays narrow — and does
    not fire in mocked, ffmpeg-less test environments."""
    import shutil
    if shutil.which("ffmpeg") is None:
        return False
    return not ffmpeg_has_subtitles_filter()


def _preflight_burn_subtitles(args):
    """Fail fast BEFORE any understanding/VLM/ASR/TTS spend when subtitle burn-in is on but
    this ffmpeg can't burn it. Without it the run only dies at the final assemble
    `-vf subtitles=` step — after the whole expensive pipeline has run."""
    if not _burn_subtitles_intended(args):
        return
    if _ffmpeg_present_but_cannot_burn():
        raise SystemExit(
            "字幕烧录已开启，但当前 ffmpeg 不支持 subtitles/libass 滤镜，整条流程会跑到最后渲染才失败。\n"
            "  解决其一：(1) 安装带 libass 的 ffmpeg；(2) 加 --no-burn-subtitles 关闭烧录"
            "（仍输出 .srt 外挂字幕）。\n"
            "  自检：python3 skills/video-recap/scripts/doctor.py")


def _print_narration_review_pointer(work_dir, *, review_ran=True):
    """Surface the advisory narration review produced by this run, if any.

    Review is optional/fail-open. Avoid surfacing a stale narration_review.md from an
    older run when review was disabled or failed before producing fresh artifacts.
    """
    if not review_ran:
        return
    review_md = Path(work_dir) / "narration_review.md"
    if not review_md.exists():
        return
    data = _load_json(Path(work_dir) / "narration_review.json")
    if isinstance(data, dict):
        findings = [f for f in (data.get("findings") or []) if isinstance(f, dict)]
        n_err = sum(1 for f in findings if f.get("severity") == "error")
        tag = str(data.get("verdict") or "见文件")
        print(f"[video-recap] 📋 解说评审（建议性，不拦截）: {tag} · "
              f"{len(findings)} 条意见（error {n_err}）→ {review_md}")
    else:
        print(f"[video-recap] 📋 解说评审意见 → {review_md}")


def _settings_for_compare(settings):
    """Settings that, if changed, invalidate reusing an existing work_dir on resume.

    `consolidate`/`consolidate_asr` are EXCLUDED: they only ADD an optional understanding
    artifact and never re-run Phase A on a Phase-B resume, so a stored manifest carrying the
    old default (or missing the key entirely, pre-dating it) must still resume — otherwise
    flipping `--consolidate`'s default ON would hard-fail every in-flight work_dir.
    """
    s = dict(settings or {})
    s.pop("consolidate", None)
    s.pop("consolidate_asr", None)
    return s


def _manifest_mismatches(work_dir, video, args):
    expected = _run_manifest_payload(video, args)
    actual = _load_run_manifest(work_dir)
    if not actual:
        return ["缺少或无法读取 recap_run_manifest.json；不能证明 work_dir 属于当前视频/参数"]
    mismatches = []
    for key in ("source_video", "source_video_fingerprint"):
        if actual.get(key) != expected.get(key):
            mismatches.append(f"{key}: expected {expected.get(key)!r}, got {actual.get(key)!r}")
    if _settings_for_compare(actual.get("settings")) != _settings_for_compare(expected.get("settings")):
        mismatches.append("settings: 当前 CLI/env 参数与 Phase A manifest 不匹配")
    return mismatches


def _read_assembly_output(work_dir):
    manifest = _load_json(Path(work_dir) / ASSEMBLY_MANIFEST)
    if isinstance(manifest, dict) and manifest.get("final_output"):
        return Path(manifest["final_output"])
    return None


def _file_md5(path):
    path = Path(path)
    return hashlib.md5(path.read_bytes()).hexdigest() if path.exists() else None


def _read_phase_ledger(work_dir):
    """Phase ledger (cut mode): which artifacts exist and the clip_plan/narration they match.

    Lets resume be driven by recorded phase state rather than bare file existence — the
    prerequisite for the cut-first/narrate-second two-pause flow, and the guard that keeps a
    narration written for one clip_plan from silently driving a different cut into TTS.
    """
    ledger = _load_json(Path(work_dir) / PHASE_LEDGER)
    return ledger if isinstance(ledger, dict) else None


def _write_phase_ledger(work_dir, **fields):
    ledger = _read_phase_ledger(work_dir) or {}
    ledger.update(fields)
    (Path(work_dir) / PHASE_LEDGER).write_text(
        json.dumps(ledger, ensure_ascii=False, indent=2), encoding="utf-8")
    return ledger


def _cut_narration_is_stale(ledger, current_clip_plan_fp):
    """Two-pass cut: the narration is authored against the rendered cut shown at the A2 pause,
    i.e. against the clip_plan recorded in the ledger. If clip_plan changed since (a re-cut)
    while that narration is still present, it describes the OLD cut — stale."""
    if not ledger:
        return False
    recorded_cp = ledger.get("clip_plan_fingerprint")
    return bool(recorded_cp is not None and recorded_cp != current_clip_plan_fp)


def _continuation_command(video, work_dir, args):
    parts = [sys.executable, str(_entry("video-recap", "recap.py")), str(video), "--work-dir", str(work_dir)]
    if args.context:
        parts += ["--context", args.context]
    if args.scene_threshold is not None:
        parts += ["--scene-threshold", str(args.scene_threshold)]
    if args.style != "纪录片":
        parts += ["--style", args.style]
    if args.edit_mode != "full":
        parts += ["--edit-mode", args.edit_mode]
    if args.target_duration:
        parts += ["--target-duration", args.target_duration]
    if getattr(args, "allow_sparse_cut", False):
        parts.append("--allow-sparse-cut")
    if args.skip_asr:
        parts.append("--skip-asr")
    if args.mimo_video_overview:
        parts.append("--mimo-video-overview")
    if not args.consolidate:  # default is ON now; only the opt-out needs to round-trip
        parts.append("--no-consolidate")
    if args.consolidate_asr:
        parts.append("--consolidate-asr")
    if getattr(args, "mimo_tts_voice", None):
        parts += ["--mimo-tts-voice", args.mimo_tts_voice]
    if getattr(args, "allow_partial_tts", False):
        parts.append("--allow-partial-tts")
    if getattr(args, "burn_subtitles", None) is not None:
        parts.append("--burn-subtitles" if args.burn_subtitles else "--no-burn-subtitles")
    if getattr(args, "output_dir", None):
        parts += ["--output-dir", args.output_dir]
    if getattr(args, "export_jianying", False):
        parts.append("--export-jianying")
    if getattr(args, "jianying_bundle_media", False):
        parts.append("--jianying-bundle-media")
    if getattr(args, "jianying_no_bundle_media", False):
        parts.append("--jianying-no-bundle-media")
    if getattr(args, "review_narration", None) is not None:
        parts.append("--review-narration" if args.review_narration else "--no-review-narration")
    if getattr(args, "require_narration_review", False):
        parts.append("--require-narration-review")
    return " ".join(shlex.quote(part) for part in parts)


def main():
    ap = argparse.ArgumentParser(description="Full video recap orchestrator (video-* skill bundle).")
    ap.add_argument("video", nargs="?")
    ap.add_argument("--work-dir", default=None)
    ap.add_argument("--context", default="")
    ap.add_argument("--scene-threshold", type=float, default=None)
    ap.add_argument("--style", default="纪录片")
    ap.add_argument("--edit-mode", default=os.environ.get("EDIT_MODE", "full"), choices=["full", "cut", "dub"])
    ap.add_argument("--target-duration", default=os.environ.get("TARGET_DURATION") or None)
    ap.add_argument("--allow-sparse-cut", action="store_true",
                    help="cut mode: accept a sparse/heavily-dropped narration mapping instead of failing the cut preflight")
    ap.add_argument("--skip-asr", action="store_true")
    ap.add_argument("--mimo-video-overview", action="store_true")
    ap.add_argument("--consolidate", action=argparse.BooleanOptionalAction, default=True,
                    help="build the understanding story index (Pass B); default ON, --no-consolidate to skip")
    ap.add_argument("--consolidate-asr", action="store_true", help="also clean ASR (Pass A)")
    ap.add_argument("--mimo-tts-voice", default=None, help="MiMo TTS voice")
    ap.add_argument("--allow-partial-tts", action="store_true",
                    help="allow video-voiceover to continue when some narration segments fail TTS")
    ap.add_argument("--burn-subtitles", action=argparse.BooleanOptionalAction, default=None,
                    help="burn narration subtitles into the video (default on; --no-burn-subtitles to disable)")
    ap.add_argument("--review-narration", action=argparse.BooleanOptionalAction, default=None,
                    help="run advisory narration quality review before TTS (default on; fail-open)")
    ap.add_argument("--require-narration-review", action="store_true",
                    help="make narration review a strict pre-TTS gate (also REQUIRE_NARRATION_REVIEW=1)")
    ap.add_argument("--output-dir", default=None)
    ap.add_argument("--export-jianying", action="store_true",
                    help="also export an OPTIONAL 剪映/JianYing draft (decoupled; never required)")
    ap.add_argument("--jianying-bundle-media", action="store_true",
                    help="copy media into the 剪映 draft (default on; portable to another machine)")
    ap.add_argument("--jianying-no-bundle-media", action="store_true",
                    help="reference media in place instead of copying it into the draft")
    ap.add_argument("--doctor", action="store_true")
    args = ap.parse_args()

    if args.doctor:
        _run("video-recap", "doctor.py")
        return
    if not args.video:
        ap.error("video is required (unless --doctor)")

    # Fail fast before any expensive understanding/VLM/ASR/TTS work if the run will burn
    # subtitles but this ffmpeg can't (otherwise it only blows up at the final render).
    _preflight_burn_subtitles(args)

    video = Path(args.video).resolve()
    work_dir = Path(args.work_dir).resolve() if args.work_dir else video.parent / f"work_dir_{video.stem}"
    work_dir.mkdir(parents=True, exist_ok=True)
    cut = args.edit_mode == "cut"
    narration_json = work_dir / "narration.json"
    clip_plan_json = work_dir / "clip_plan.json"

    edited_source = work_dir / "edited_source.mp4"

    def _understand():
        uargs = [str(video), "--work-dir", str(work_dir), "--style", args.style]
        if args.context:
            uargs += ["--context", args.context]
        if args.scene_threshold is not None:
            uargs += ["--scene-threshold", str(args.scene_threshold)]
        if args.edit_mode:
            uargs += ["--edit-mode", args.edit_mode]
        if args.target_duration:
            uargs += ["--target-duration", args.target_duration]
        if args.skip_asr:
            uargs.append("--skip-asr")
        if args.mimo_video_overview:
            uargs.append("--mimo-video-overview")
        uargs.append("--consolidate" if args.consolidate else "--no-consolidate")
        if args.consolidate_asr:
            uargs.append("--consolidate-asr")
        _run("video-understanding", "understand.py", *uargs)

    inspect_py = _entry("video-recap", "recap_inspect.py")

    def _pause(need_text, inspect_hint=None):
        brief = work_dir / "agent_narration_brief.md"
        cont = _continuation_command(video, work_dir, args)
        print("=" * 50)
        # The brief fires a research directive only when the substrate is thin/empty and no
        # background_research.json exists yet; amplify it so the agent researches BEFORE writing.
        if brief.exists() and "Research the story FIRST" in brief.read_text(encoding="utf-8"):
            print("[video-recap] ⚑ 理解素材偏薄：先按 brief 顶部「Research the story FIRST」调研并写 "
                  "background_research.json，再写解说，避免看图说话。")
        print(f"[video-recap] ⏸  阅读 {brief}（按 video-script 规则）后写入 {need_text}")
        if inspect_hint:
            print(f"[video-recap]    先核对状态/时间轴（建议性）: {inspect_hint}")
        print(f"[video-recap]    写完后重跑继续: {cont}")
        print("=" * 50)

    def _reject_stale_manifest():
        mismatches = _manifest_mismatches(work_dir, video, args)
        if mismatches:
            details = "\n  - ".join(mismatches)
            raise SystemExit(
                "work_dir 与当前 recap 输入不匹配，拒绝复用既有 narration/clip_plan；"
                "请使用新的 --work-dir，或删除旧产物后重新运行 Phase A。\n"
                f"  - {details}")

    if args.edit_mode == "dub":
        # Dub mode: EN→ZH translation-dub in the original cloned voice (replaces speech, not
        # overlay). One pause: prepare (ASR + sentence-seg + reference) -> agent writes the
        # Chinese translation (dub_script.json) -> render (clone TTS + full-replace mux).
        dub_script = work_dir / "dub_script.json"
        if not dub_script.exists():
            _run("video-voiceover", "dub.py", "--stage", "prepare",
                 "--video", str(video), "--work-dir", str(work_dir))
            _write_run_manifest(work_dir, video, args)
            cont = _continuation_command(video, work_dir, args)
            print("=" * 50)
            print(f"[video-recap] ⏸  阅读 {work_dir / 'dub_brief.md'}，把英文原声转写切分并翻译成中文，写入 {dub_script}")
            print('[video-recap]    格式 [{"start": 起秒, "end": 止秒, "zh": "译文"}]（按 start 升序）；逐句忠实、跟原声节奏一致、保留原音色')
            print(f"[video-recap]    写完后重跑继续: {cont}")
            print("=" * 50)
            return
        _reject_stale_manifest()
        _run("video-voiceover", "dub.py", "--stage", "render",
             "--video", str(video), "--work-dir", str(work_dir))
        print(f"[video-recap] ✅ 配音完成: {work_dir / ('dub_' + video.stem + '.mp4')}")
        return

    if not cut:
        # Full mode: a single pause (understand -> agent writes narration.json -> produce).
        if not narration_json.exists():
            _understand()
            _write_run_manifest(work_dir, video, args)
            _pause(f"{narration_json}",
                   inspect_hint=f"python3 {inspect_py} --work-dir {work_dir} state")
            return
        _reject_stale_manifest()
        _run("video-script", "validate.py", "--work-dir", work_dir, "--mode", "full")
        narration_for_tts = narration_json
        assemble_video_path = video
    else:
        # Cut mode: cut-first / narrate-second (two pauses), so narration is authored against the
        # REAL output timeline — map_narration_to_clips is never used and cannot drop/clamp/desync.
        if not clip_plan_json.exists():
            # PASS 1: understand -> agent writes clip_plan.json ONLY.
            _understand()
            _write_run_manifest(work_dir, video, args)
            _pause(f"{clip_plan_json}（只写剪辑计划；解说下一步对着剪好的成片写）",
                   inspect_hint=f"python3 {inspect_py} --work-dir {work_dir} state")
            return
        _reject_stale_manifest()
        cp_fp = _file_md5(clip_plan_json)
        # Render the cut from clip_plan (no narration mapping — narration is OUTPUT-time).
        crender = [str(video), "--work-dir", str(work_dir), "--no-narration-map"]
        if args.target_duration:
            crender += ["--target-duration", args.target_duration]
        _run("video-cut", "cut.py", *crender)
        if not narration_json.exists():
            # PASS 2: rebuild the brief (now an OUTPUT-timeline variant) and pause for narration.
            _understand()
            _write_phase_ledger(work_dir, clip_plan_fingerprint=cp_fp, edited_source_rendered=True)
            _pause(f"{narration_json}（用成片 OUTPUT 时间轴写解说，对着 {edited_source}）",
                   inspect_hint=(f"python3 {inspect_py} --work-dir {work_dir} "
                                 "clip-map --output-start <s> --output-end <e>  # 核对输出↔原片时间轴"))
            return
        if _cut_narration_is_stale(_read_phase_ledger(work_dir), cp_fp):
            raise SystemExit(
                "clip_plan.json 已改变，但 narration.json 仍是对旧剪辑写的，会与剪后画面对不上。"
                "请删除 narration.json，重跑后按新成片重新写解说。")
        _write_phase_ledger(work_dir, clip_plan_fingerprint=cp_fp,
                            narration_fingerprint=_file_md5(narration_json), narration_written=True)
        output_duration = _read_video_duration_or_raise(edited_source)
        _run("video-script", "validate.py", "--work-dir", work_dir, "--mode", "cut_output",
             "--output-duration", f"{output_duration:.3f}")
        narration_for_tts = narration_json
        assemble_video_path = edited_source
    review_ran = _run_narration_review(work_dir, args, timeline="cut_output" if cut else "source")
    vargs = ["--work-dir", str(work_dir), "--narration", str(narration_for_tts)]
    if args.mimo_tts_voice:
        vargs += ["--mimo-voice", args.mimo_tts_voice]
    if args.allow_partial_tts:
        vargs.append("--allow-partial-tts")
    _run("video-voiceover", "voiceover.py", *vargs)

    aargs = [str(assemble_video_path), "--work-dir", str(work_dir), "--recap-stem", video.stem]
    if args.output_dir:
        aargs += ["--output-dir", args.output_dir]
    if args.burn_subtitles is not None:
        aargs.append("--burn-subtitles" if args.burn_subtitles else "--no-burn-subtitles")
    # env-only burn intent (BURN_SUBTITLES) is propagated implicitly: assemble re-derives it
    # via the same env_bool default the preflight used, so the two agree by shared env.
    if cut:
        # let the timeline / 剪映 export reference the original clips, not edited_source.mp4
        aargs += ["--source-video", str(video)]
    if args.export_jianying:  # env EXPORT_JIANYING is honored by assemble.py itself
        aargs.append("--export-jianying")
    if args.jianying_bundle_media:
        aargs.append("--jianying-bundle-media")
    if args.jianying_no_bundle_media:
        aargs.append("--jianying-no-bundle-media")
    _run("video-assemble", "assemble.py", *aargs)

    final_dir = Path(args.output_dir) if args.output_dir else work_dir.parent
    final_output = _read_assembly_output(work_dir) or (final_dir / ("recap_" + video.stem + ".mp4"))
    print(f"[video-recap] ✅ 完成: {final_output}")
    _print_narration_review_pointer(work_dir, review_ran=review_ran)


if __name__ == "__main__":
    main()