Video Script

"""Self-contained config + utilities for this skill (no cross-skill imports).
Merged from the shared core; reads the same env vars as the rest of the bundle."""
import json
import hashlib
import os
import re
import subprocess
import time
import urllib.request
import urllib.error
from pathlib import Path
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime


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

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)

def run_cmd(cmd, **kwargs):
    """运行命令，返回 CompletedProcess"""
    if isinstance(cmd, list):
        display_parts = []
        for part in cmd:
            text = str(part)
            display_parts.append(text if len(text) <= 240 else text[:237] + "...")
        display = " ".join(display_parts)
    else:
        display = str(cmd)
        if len(display) > 2000:
            display = display[:1997] + "..."
    log(f"运行: {display}")
    return subprocess.run(cmd, capture_output=True, text=True, **kwargs)

def get_video_duration(video_path):
    """获取视频时长（秒）"""
    cmd = ["ffprobe", "-v", "quiet", "-show_entries", "format=duration",
           "-of", "csv=p=0", str(video_path)]
    result = run_cmd(cmd)
    if result.returncode != 0:
        return 0.0
    try:
        return float(result.stdout.strip())
    except (TypeError, ValueError):
        return 0.0

def stable_json_dumps(value):
    """Serialize values deterministically for non-secret cache fingerprints."""
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)

def stable_hash(value):
    """Return an md5 digest for deterministic JSON-serializable values."""
    return hashlib.md5(stable_json_dumps(value).encode("utf-8")).hexdigest()

def file_fingerprint(path, chunk_size=1024 * 1024):
    """Return a full-content fingerprint for cache-correct identity checks.

    This intentionally avoids mtime/path so copied videos or JSON artifacts can
    be reused when their bytes are identical, while any byte change invalidates
    the cache even if timestamps, size, head, or tail bytes are misleading.
    """
    h = hashlib.sha256()
    with open(os.fspath(path), "rb") as f:
        for chunk in iter(lambda: f.read(chunk_size), b""):
            h.update(chunk)
    return h.hexdigest()
def video_fingerprint(video_path):
    """Full video content fingerprint used as the root pipeline asset print."""
    return file_fingerprint(video_path)

def step_cache_key(video_path, step_name, params_fingerprint=""):
    """Build a cache key from video content, step name and step parameters."""
    params_digest = params_fingerprint
    if not isinstance(params_digest, str):
        params_digest = stable_hash(params_digest)
    payload = f"{video_fingerprint(video_path)}_{step_name}_{params_digest}"
    return hashlib.md5(payload.encode("utf-8")).hexdigest()

def _retry_after_seconds(value, fallback):
    """Parse Retry-After seconds or HTTP-date; return fallback on malformed input."""
    if not value:
        return fallback
    try:
        return max(fallback, max(0, int(value)))
    except (TypeError, ValueError):
        pass
    try:
        retry_at = parsedate_to_datetime(value)
        if retry_at.tzinfo is None:
            retry_at = retry_at.replace(tzinfo=timezone.utc)
        return max(fallback, max(0, int((retry_at - datetime.now(timezone.utc)).total_seconds())))
    except (TypeError, ValueError, IndexError, OverflowError):
        return fallback

def _api_headers(api_provider=None, api_url=None, api_key=None):
    """Build MiMo auth headers (OpenAI-compatible chat/completions with an api-key header)."""
    del api_provider, api_url  # MiMo is the only provider; signature kept for call sites
    key = CONFIG.get("api_key", "") if api_key is None else api_key
    return {
        "Content-Type": "application/json",
        "User-Agent": "video-recap/1.0",
        "api-key": key,
    }

def _prepare_api_payload(payload, api_provider=None, api_url=None):
    """Normalize payload fields for MiMo's OpenAI-compatible chat/completions API."""
    del api_provider, api_url
    normalized = dict(payload)
    if "max_tokens" in normalized and "max_completion_tokens" not in normalized:
        normalized["max_completion_tokens"] = normalized.pop("max_tokens")
    model = str(normalized.get("model") or "")
    if (
        CONFIG.get("mimo_disable_thinking", True)
        and not model.endswith(("-tts", "-asr"))
        and "thinking" not in normalized
    ):
        # MiMo V2.5 may spend small max_completion_tokens budgets on reasoning_content.
        # The recap pipeline needs visible text, so disable thinking unless set explicitly.
        normalized["thinking"] = {"type": "disabled"}
    return normalized

def _mimo_endpoint(kind):
    """Return per-capability MiMo endpoint settings (video understanding / TTS / ASR)."""
    by_kind = {
        "video": ("mimo_video_api_url", "mimo_video_api_key", "mimo_video_api_key_source"),
        "tts": ("mimo_tts_api_url", "mimo_tts_api_key", "mimo_tts_api_key_source"),
        "asr": ("mimo_asr_api_url", "mimo_asr_api_key", "mimo_asr_api_key_source"),
    }
    if kind not in by_kind:
        raise ValueError(f"Unsupported MiMo endpoint kind: {kind}")
    url_key, key_key, src_key = by_kind[kind]
    return {
        "api_url": CONFIG.get(url_key) or CONFIG.get("mimo_api_url"),
        "api_key": CONFIG.get(key_key) or CONFIG.get("mimo_api_key"),
        "api_key_source": CONFIG.get(src_key, "MIMO_API_KEY"),
    }

def _call_mimo_endpoint(kind, payload, max_retries=10):
    settings = _mimo_endpoint(kind)
    return api_call(
        payload,
        max_retries=max_retries,
        api_provider="mimo",
        api_url=settings["api_url"],
        api_key=settings["api_key"],
        api_key_source=settings["api_key_source"],
    )

def mimo_video_api_call(payload, max_retries=10):
    """Call the MiMo video-understanding endpoint."""
    return _call_mimo_endpoint("video", payload, max_retries=max_retries)

def mimo_tts_api_call(payload, max_retries=10):
    """Call the MiMo TTS endpoint."""
    return _call_mimo_endpoint("tts", payload, max_retries=max_retries)

def mimo_asr_api_call(payload, max_retries=10):
    """Call the MiMo speech-recognition (ASR) endpoint."""
    return _call_mimo_endpoint("asr", payload, max_retries=max_retries)

def api_call(payload, max_retries=8, *, api_provider=None, api_url=None, api_key=None, api_key_source=None):
    """调用 OpenAI-compatible API，带重试。

    集群的 429 限流是常态而非错误，所以重试更耐心（更多次数 + 退避封顶 60s + 遵从 Retry-After），
    避免一次瞬时限流就中止整个阶段。配额窗口常以分钟计，所以 429 在没有 Retry-After 时也至少等 10s。
    """
    endpoint = normalize_api_url(api_url if api_url is not None else CONFIG["api_url"])
    headers = _api_headers(api_provider=api_provider, api_url=endpoint, api_key=api_key)
    data = json.dumps(_prepare_api_payload(payload, api_provider=api_provider, api_url=endpoint)).encode("utf-8")

    for attempt in range(max_retries):
        try:
            req = urllib.request.Request(endpoint, data=data, headers=headers)
            with urllib.request.urlopen(req, timeout=300) as resp:
                result = json.loads(resp.read().decode("utf-8"))
                return result
        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8", errors="replace")[:500]
            wait = min(2 ** attempt, 60)
            if e.code == 429:
                retry_after = e.headers.get("Retry-After")
                wait = _retry_after_seconds(retry_after, max(wait, 10))
                log(f"API 速率限制 (尝试 {attempt+1}/{max_retries}), 等待 {wait}s")
            elif e.code == 401:
                key_name = api_key_source or CONFIG.get("api_key_source", "MIMO_API_KEY")
                raise RuntimeError(f"API 认证失败 (401)。请检查 {key_name} 和 API URL 是否匹配。")
            elif e.code == 403:
                hint = "API 访问被拒绝 (403)。"
                if "1010" in body or "cloudflare" in body.lower():
                    hint += "IP 被 Cloudflare 限流，请等待几分钟后重试。"
                    raise RuntimeError(hint)
                hint += "请检查 API key 权限和 API URL 设置。"
                raise RuntimeError(hint)
            elif e.code == 405:
                raise RuntimeError("API 端点不可用 (405)，可能被 WAF 拦截。请检查 MIMO_API_URL 或稍后重试。")
            elif e.code == 503:
                log(f"API 服务暂不可用 (503)，等待 {wait}s (尝试 {attempt+1}/{max_retries})")
            elif e.code == 524:
                # Cloudflare 超时：服务端处理超时，需要更长退避
                wait = max(wait, 4 * (attempt + 1))
                log(f"API 超时 (524)，等待 {wait}s (尝试 {attempt+1}/{max_retries})")
            else:
                log(f"API 调用失败 (尝试 {attempt+1}/{max_retries}): HTTP {e.code} — {body}")
            if attempt < max_retries - 1:
                time.sleep(wait)
            else:
                raise RuntimeError(f"API 调用失败 {max_retries} 次: HTTP {e.code} — {body}")
        except (urllib.error.URLError, Exception) as e:
            wait = min(2 ** attempt, 60)
            log(f"API 调用失败 (尝试 {attempt+1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                log(f"等待 {wait}s 后重试...")
                time.sleep(wait)
            else:
                raise RuntimeError(f"API 调用失败 {max_retries} 次: {e}")

def load_prompt(name):
    """加载 prompt 模板"""
    path = PROMPTS_DIR / "prompt-templates.md"
    if not path.exists():
        return None
    content = path.read_text(encoding="utf-8")
    # 用 ### NAME 和 ### 分隔提取对应 prompt
    pattern = rf"### {name}\s*\n(.*?)(?=\n### |\Z)"
    m = re.search(pattern, content, re.DOTALL)
    return m.group(1).strip() if m else None

#!/usr/bin/env python3
"""video-script narration reviewer (LLM-as-judge).

A separate *quality* pass over an agent-written narration.json — distinct from the
mechanical validate.py (which only checks budget/timing/density). The reviewer reads the
draft against the visual/ASR grounding and the writing rubric, then returns severity-rated
findings (hallucination, weak hook, no throughline, narrating-the-picture, density, pacing,
cliché, incomplete sentence). It does NOT edit narration.json — the writer revises and re-runs.

Output: narration_review.json (structured) + narration_review.md (readable). Advisory:
verdict REVISE means "fix and re-review"; it never blocks (validate.py is the hard gate).
"""
import argparse
import json
import math
import re
from pathlib import Path

from lib import CONFIG, log, api_call, stable_hash

CATEGORIES = [
    "hallucination", "weak_hook", "no_throughline", "narrating_picture",
    "density", "pacing", "cliche", "incomplete", "disjoint_handoff", "other",
]

# Categories whose findings are allowed to keep severity=error and thus gate strict
# mode. Everything else is craft/subjective and is clamped to at most "warning".
FACTUAL_CATEGORIES = {"hallucination", "incomplete"}

RUBRIC = """你是中文视频解说稿的严格评审。依据以下规则审阅草稿，只指出真实问题，宁缺毋滥：
1. 反幻觉（最重要）：解说里的人物、动作、因果、关系只要能由「画面证据」(frame_facts/描述) 或「对白」(ASR) 或「背景资料」(background_research) 任一支撑，即不算幻觉。只有与全部可得证据都矛盾的论断才是 severity=error, category=hallucination，并指出与哪条证据冲突。若与背景资料一致但画面/对白里看不到（合理推断、非矛盾），最多 severity=suggestion，不要判 error。
2. 钩子：开头 1-2 段要制造悬念/利害，不是交代场景。弱钩子 → weak_hook。
3. 主线：应有一条贯穿主线（目标/关系/悬念），每段推进它，不要每个场景从头讲。缺主线 → no_throughline。
4. 给信息而非念画面：观众看得见动作表情；解说要讲动机/关系/潜台词/剧情意义。复述画面 → narrating_picture。
5. 密度/节奏：连续铺底、短句、相邻段不要断太久；过疏/过密/拖沓 → density 或 pacing。
6. 去废词：删空泛形容（"危机四伏""震撼人心"）→ cliche。
7. 完整句子：半句话/未收尾 → incomplete。
8. 段落衔接：解说块要为随后的原声留白铺垫，下一块要承接原声刚呈现的内容；若两块各说各的、原声进来接不上 → disjoint_handoff。
只返回 JSON（不要额外解释），格式：
{"verdict":"REVISE|OK","summary":"一两句总体判断","findings":[{"segment":<草稿段号(从0起)或null表示整体>,"severity":"error|warning|suggestion","category":"<上面类别之一>","issue":"问题","fix":"具体改法"}]}"""



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




def _load_cut_clip_spans(work_dir):
    """Load explicit source→output spans from the validated cut plan only.

    `cut_output` review compares output-time narration to output-time evidence. A raw
    clip_plan can be pre-padding/pre-snap and may omit output spans, so using it would
    look grounded while disagreeing with edited_source.mp4. Missing/stale validated data
    should fail this advisory stage and let the orchestrator fail-open visibly.
    """
    work_dir = Path(work_dir)
    plan = _load(work_dir, "clip_plan_validated.json")
    if not isinstance(plan, dict):
        return None
    raw_plan = _load(work_dir, "clip_plan.json")
    if raw_plan is not None and plan.get("raw_plan_fingerprint") != stable_hash(raw_plan):
        return None
    clips = plan.get("clips")
    if not isinstance(clips, list):
        return None
    spans = []
    for clip in clips:
        if not isinstance(clip, dict):
            continue
        if not all(key in clip for key in ("source_start", "source_end", "output_start", "output_end")):
            return None
        try:
            source_start = float(clip["source_start"])
            source_end = float(clip["source_end"])
            output_start = float(clip["output_start"])
            output_end = float(clip["output_end"])
        except (TypeError, ValueError):
            return None
        values = (source_start, source_end, output_start, output_end)
        if not all(math.isfinite(value) for value in values):
            return None
        if source_end <= source_start or output_end <= output_start:
            return None
        spans.append({
            "source_start": source_start,
            "source_end": source_end,
            "output_start": output_start,
            "output_end": output_end,
        })
    return spans or None


def _source_output_overlaps(start, end, spans):
    overlaps = []
    for span in spans or []:
        source_start = max(start, span["source_start"])
        source_end = min(end, span["source_end"])
        if source_end <= source_start:
            continue
        output_start = span["output_start"] + (source_start - span["source_start"])
        output_end = span["output_start"] + (source_end - span["source_start"])
        overlaps.append({
            "source_start": source_start,
            "source_end": source_end,
            "output_start": output_start,
            "output_end": output_end,
        })
    return overlaps


def _map_source_range_to_output(start, end, spans):
    return [(o["output_start"], o["output_end"]) for o in _source_output_overlaps(start, end, spans)]


def _remap_frame_facts(frame_facts, overlap):
    if not isinstance(frame_facts, dict):
        return frame_facts
    out = {}
    for raw_ts, vals in frame_facts.items():
        try:
            ts = float(raw_ts)
        except (TypeError, ValueError):
            continue
        if not (overlap["source_start"] <= ts <= overlap["source_end"]):
            continue
        out_ts = overlap["output_start"] + (ts - overlap["source_start"])
        out[f"{out_ts:.3f}"] = vals
    return out


def remap_grounding_to_output_timeline(vlm_analysis, asr_result, clip_spans):
    """Return VLM/ASR grounding clipped/remapped from source time to cut output time."""
    if not clip_spans:
        return vlm_analysis or [], asr_result or []

    remapped_scenes = []
    for scene in vlm_analysis or []:
        if not isinstance(scene, dict):
            continue
        try:
            start = float(scene.get("start", 0))
            end = float(scene.get("end", start))
        except (TypeError, ValueError):
            continue
        overlaps = _source_output_overlaps(start, end, clip_spans)
        for part_idx, overlap in enumerate(overlaps):
            item = dict(scene)
            item["start"] = round(overlap["output_start"], 3)
            item["end"] = round(overlap["output_end"], 3)
            item["frame_facts"] = _remap_frame_facts(scene.get("frame_facts"), overlap)
            if len(overlaps) > 1:
                item["scene_id"] = f"{scene.get('scene_id', '?')}.{part_idx}"
            remapped_scenes.append(item)

    remapped_asr = []
    for seg in asr_result or []:
        if not isinstance(seg, dict):
            continue
        try:
            start = float(seg.get("start", 0))
            end = float(seg.get("end", start))
        except (TypeError, ValueError):
            continue
        text = str(seg.get("text", "")).strip()
        if not text:
            continue
        for overlap in _source_output_overlaps(start, end, clip_spans):
            item = dict(seg)
            item["start"] = round(overlap["output_start"], 3)
            item["end"] = round(overlap["output_end"], 3)
            remapped_asr.append(item)

    remapped_scenes.sort(key=lambda x: (float(x.get("start", 0)), float(x.get("end", 0))))
    remapped_asr.sort(key=lambda x: (float(x.get("start", 0)), float(x.get("end", 0))))
    return remapped_scenes, remapped_asr


def _scene_grounding(vlm_analysis, limit=60):
    lines = []
    for scene in (vlm_analysis or [])[:limit]:
        if not isinstance(scene, dict):
            continue
        sid = scene.get("scene_id", "?")
        start = scene.get("start", 0)
        end = scene.get("end", 0)
        desc = str(scene.get("description", "")).strip().replace("\n", " ")
        facts = scene.get("frame_facts")
        fact_txt = ""
        picks = []
        if isinstance(facts, dict):  # canonical shape: {"<ts>": ["action", ...]} (vlm.py)
            def fact_sort_key(value):
                try:
                    return (0, float(value))
                except (TypeError, ValueError):
                    return (1, str(value))

            for ts in sorted(facts.keys(), key=fact_sort_key):
                vals = facts[ts]
                picks.extend(vals if isinstance(vals, list) else [str(vals)])
        elif isinstance(facts, list):  # defensive: legacy list shape
            for f in facts:
                picks.append(str(f.get("fact", f.get("text", ""))).strip() if isinstance(f, dict) else str(f).strip())
        if picks:
            fact_txt = " | 帧实: " + "；".join(p for p in picks[:4] if p)
        lines.append(f"[场景{sid} {float(start):.0f}-{float(end):.0f}s] {desc}{fact_txt}")
    return "\n".join(lines)


def _asr_grounding(asr_result, limit=80):
    lines = []
    for seg in (asr_result or [])[:limit]:
        if not isinstance(seg, dict):
            continue
        text = str(seg.get("text", "")).strip()
        if not text:
            continue
        lines.append(f"[{float(seg.get('start', 0)):.0f}-{float(seg.get('end', 0)):.0f}s] {text}")
    return "\n".join(lines)


def _clip_text(text, limit):
    value = re.sub(r"\s+", " ", str(text or "")).strip()
    return value[:limit]


def _load_review_research_context(work_dir):
    path = Path(work_dir) / "background_research.json"
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (ValueError, OSError):
        return {}
    return data if isinstance(data, dict) else {}


def _format_review_research_context(research, limit=1200):
    """Compact background_research.json for the quality reviewer.

    Background research is valid grounding alongside visual/ASR: a claim it supports is
    not a hallucination; only claims contradicting all available evidence are errors.
    """
    if not isinstance(research, dict) or not research:
        return ""
    lines = []
    for key, label in (
        ("synopsis", "Synopsis"),
        ("episode_context", "Episode context"),
        ("worldbuilding", "Worldbuilding"),
    ):
        value = _clip_text(research.get(key), 500)
        if value:
            lines.append(f"- {label}: {value}")

    characters = research.get("characters")
    if isinstance(characters, dict) and characters:
        lines.append("- Characters:")
        for name, desc in list(characters.items())[:12]:
            clean_name = _clip_text(name, 60)
            clean_desc = _clip_text(desc, 160)
            if clean_name:
                lines.append(f"    - {clean_name}：{clean_desc}")

    details = research.get("character_details")
    if isinstance(details, dict) and details:
        lines.append("- Character details:")
        for name, info in list(details.items())[:8]:
            if not isinstance(info, dict):
                continue
            bits = []
            aliases = info.get("aliases")
            if isinstance(aliases, list) and aliases:
                bits.append("别名 " + "/".join(_clip_text(alias, 40) for alias in aliases[:4] if _clip_text(alias, 40)))
            role = _clip_text(info.get("role"), 80)
            if role:
                bits.append(role)
            rels = info.get("relationships")
            if isinstance(rels, list) and rels:
                bits.append("；".join(_clip_text(rel, 80) for rel in rels[:4] if _clip_text(rel, 80)))
            clean_name = _clip_text(name, 60)
            if clean_name and bits:
                lines.append(f"    - {clean_name}：{'；'.join(bits)}")

    arcs = research.get("plot_arcs")
    if isinstance(arcs, list) and arcs:
        lines.append("- Plot arcs:")
        for arc in arcs[:8]:
            if isinstance(arc, dict):
                name = _clip_text(arc.get("name"), 80)
                desc = _clip_text(arc.get("description"), 180)
                status = _clip_text(arc.get("status"), 40)
                if name or desc:
                    tail = f" [{status}]" if status else ""
                    lines.append(f"    - {name}：{desc}{tail}")
            else:
                val = _clip_text(arc, 180)
                if val:
                    lines.append(f"    - {val}")

    text = "\n".join(lines).strip()
    return text[:limit]


def _format_draft(narration):
    lines = []
    for i, seg in enumerate(narration or []):
        if not isinstance(seg, dict):
            continue
        start = seg.get("start", 0)
        end = seg.get("end", 0)
        text = str(seg.get("narration", "")).strip()
        overlap = seg.get("overlaps_speech")
        tag = "" if overlap is None else (" [盖原声]" if overlap else " [静音槽]")
        lines.append(f"{i}. [{float(start):.1f}-{float(end):.1f}s]{tag} {text}")
    return "\n".join(lines)


def build_review_messages(narration, vlm_analysis, asr_result, work_dir=None, research_context=None):
    """Pure: assemble the reviewer chat messages (testable without the API)."""
    grounding = _scene_grounding(vlm_analysis)
    dialogue = _asr_grounding(asr_result)
    draft = _format_draft(narration)
    if research_context is None and work_dir is not None:
        research_context = _format_review_research_context(_load_review_research_context(work_dir))
    user = (
        f"{RUBRIC}\n\n"
        f"## 背景资料（与画面/对白并列的有效依据：被其支撑的事实不算幻觉，仅与全部证据矛盾才算）\n{research_context or '(无)'}\n\n"
        f"## 画面证据（场景描述 + 帧实）\n{grounding or '(无)'}\n\n"
        f"## 对白（ASR）\n{dialogue or '(无对白/静音视频)'}\n\n"
        f"## 解说草稿（共 {len([s for s in (narration or []) if isinstance(s, dict)])} 段）\n{draft or '(空)'}\n"
    )
    return [{"role": "user", "content": user}]


def parse_review_response(text):
    """Pure: robustly extract the reviewer JSON; fall back to a REVISE-unknown shell."""
    raw = str(text or "")
    candidate = raw
    fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
    if fence:
        candidate = fence.group(1)
    else:
        first, last = raw.find("{"), raw.rfind("}")
        if first != -1 and last > first:
            candidate = raw[first:last + 1]
    try:
        data = json.loads(candidate)
    except ValueError:
        return {"verdict": "REVISE", "summary": "评审输出无法解析为 JSON，请人工检查。",
                "findings": [], "parse_error": True, "raw": raw[:2000]}
    verdict = str(data.get("verdict", "REVISE")).upper()
    if verdict not in ("REVISE", "OK"):
        verdict = "REVISE"
    findings = []
    for f in data.get("findings", []) or []:
        if not isinstance(f, dict):
            continue
        sev = str(f.get("severity", "warning")).lower()
        if sev not in ("error", "warning", "suggestion"):
            sev = "warning"
        cat = str(f.get("category", "other")).lower()
        if cat not in CATEGORIES:
            cat = "other"
        # Only factual defects may gate strict mode (severity=error). Craft findings
        # (weak_hook, narrating_picture, cliche, disjoint_handoff, ...) are advisory:
        # clamp them to at most "warning" so they never block on subjective judgement.
        if cat not in FACTUAL_CATEGORIES and sev == "error":
            sev = "warning"
        findings.append({
            "segment": f.get("segment"),
            "severity": sev,
            "category": cat,
            "issue": str(f.get("issue", "")).strip(),
            "fix": str(f.get("fix", "")).strip(),
        })
    return {"verdict": verdict, "summary": str(data.get("summary", "")).strip(), "findings": findings}


def format_review_md(review):
    order = {"error": 0, "warning": 1, "suggestion": 2}
    findings = sorted(review.get("findings", []), key=lambda f: order.get(f["severity"], 3))
    counts = {s: sum(1 for f in findings if f["severity"] == s) for s in ("error", "warning", "suggestion")}
    out = [
        "# Narration review",
        "",
        f"Verdict: **{review.get('verdict', 'REVISE')}**  "
        f"(errors {counts['error']}, warnings {counts['warning']}, suggestions {counts['suggestion']})",
        "",
        review.get("summary", "") or "_(no summary)_",
        "",
        "## Findings",
    ]
    if not findings:
        out.append("- (none)")
    for f in findings:
        seg = "整体" if f["segment"] is None else f"段 {f['segment']}"
        out.append(f"- **[{f['severity']}/{f['category']}] {seg}** — {f['issue']}")
        if f["fix"]:
            out.append(f"  - 改法: {f['fix']}")
    return "\n".join(out) + "\n"


def review_narration(work_dir, *, timeline="source"):
    work_dir = Path(work_dir)
    narration = _load(work_dir, "narration.json")
    if narration is None:
        raise SystemExit(f"缺少 {work_dir / 'narration.json'}；先写解说草稿再评审")
    vlm_analysis = _load(work_dir, "vlm_analysis.json") or []
    asr_result = _load(work_dir, "asr_result.json") or []
    if timeline == "cut_output":
        spans = _load_cut_clip_spans(work_dir)
        if not spans:
            raise SystemExit("cut_output review requires fresh clip_plan_validated.json with explicit source/output spans")
        vlm_analysis, asr_result = remap_grounding_to_output_timeline(vlm_analysis, asr_result, spans)
    elif timeline != "source":
        raise SystemExit(f"unknown review timeline: {timeline}")

    messages = build_review_messages(narration, vlm_analysis, asr_result, work_dir=work_dir)
    resp = api_call({
        "model": CONFIG.get("vlm_model", ""),
        "messages": messages,
        "max_tokens": 2000,
        "temperature": 0,
        "seed": 7,
    })
    content = ""
    try:
        content = resp["choices"][0]["message"]["content"]
    except (KeyError, IndexError, TypeError):
        log("评审 API 返回结构异常")
    review = parse_review_response(content)

    (work_dir / "narration_review.json").write_text(
        json.dumps(review, ensure_ascii=False, indent=2), encoding="utf-8")
    (work_dir / "narration_review.md").write_text(format_review_md(review), encoding="utf-8")
    n_err = sum(1 for f in review["findings"] if f["severity"] == "error")
    log(f"解说评审完成: {review['verdict']} | {len(review['findings'])} 条意见（error {n_err}）")
    return review


def _auto_timeline(work_dir):
    """Default the grounding timeline so a manual `review.py --work-dir` matches what the
    orchestrator does: cut_output when narration.json is in the cut OUTPUT timeline, else
    source. Without this, reviewing a cut narration on the default 'source' timeline compares
    OUTPUT-time narration against SOURCE-time evidence and floods false-positive 'hallucination'
    findings (and the inverse flood for a legacy source-time narration mis-read as cut_output).

    Detection is authoritative-first: the orchestrator records the run's edit_mode in
    recap_run_manifest.json. In orchestrated cut mode narration.json is OUTPUT time; in full
    mode it is SOURCE time. Trusting edit_mode is correct even when stale cut artifacts from a
    prior run linger in a reused work_dir. Only when no manifest is present (standalone review
    or a hand-built work_dir) do we fall back to artifact sniffing — and even then the legacy
    direct video-cut single-pass path writes a SOURCE-time narration.json alongside a separate
    output-time narration_mapped.json, so its presence pins us back to source."""
    work_dir = Path(work_dir)
    manifest = work_dir / "recap_run_manifest.json"
    if manifest.exists():
        try:
            mode = json.loads(manifest.read_text(encoding="utf-8")).get("settings", {}).get("edit_mode")
        except (ValueError, OSError):
            mode = None
        if mode == "cut":
            return "cut_output"
        if mode:  # "full" or any non-cut mode → narration.json is source time
            return "source"
    has_cut = (work_dir / "clip_plan_validated.json").exists() and (work_dir / "edited_source.mp4").exists()
    if has_cut and not (work_dir / "narration_mapped.json").exists():
        return "cut_output"
    return "source"


def main():
    ap = argparse.ArgumentParser(description="Review an agent-written narration.json for quality (LLM-as-judge).")
    ap.add_argument("--work-dir", required=True)
    ap.add_argument("--timeline", choices=["source", "cut_output"], default=None,
                    help="grounding timeline for narration.json; DEFAULT auto-detects cut_output when a "
                         "validated cut (clip_plan_validated.json + edited_source.mp4) is present, else source. "
                         "cut_output remaps source VLM/ASR to the cut output timeline via clip_plan_validated.json")
    args = ap.parse_args()
    timeline = args.timeline or _auto_timeline(args.work_dir)
    if args.timeline is None and timeline != "source":
        log(f"评审 grounding 时间轴自动判定为 {timeline}（检测到已校验的剪辑产物）")
    review = review_narration(args.work_dir, timeline=timeline)
    print(json.dumps({
        "status": "reviewed",
        "verdict": review["verdict"],
        "findings": len(review["findings"]),
        "review": str(Path(args.work_dir) / "narration_review.md"),
    }, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""video-script validation entrypoint.

Validate (and, in full mode, time-align) an agent-written narration.json against the
understanding index produced by video-understanding. Writes narration_lint.json and,
in full mode, rewrites narration.json with quiet-window alignment applied.
"""
import argparse
import json
import math
from pathlib import Path

from lib import CONFIG, log, stable_hash
from narration import (
    validate_narration_or_raise,
    _validate_narration_budget,
    _align_narration_to_quiet,
)


def _load(path):
    path = Path(path)
    return json.loads(path.read_text(encoding="utf-8")) if path.exists() else None


def _load_cut_clip_plan(work_dir):
    raw_plan = Path(work_dir) / "clip_plan.json"
    validated_plan = Path(work_dir) / "clip_plan_validated.json"
    if not validated_plan.exists():
        return _load(raw_plan)
    if not raw_plan.exists():
        return _load(validated_plan)

    raw = _load(raw_plan)
    validated = _load(validated_plan)
    if (
        isinstance(validated, dict)
        and validated.get("raw_plan_fingerprint") == stable_hash(raw)
    ):
        return validated
    # The orchestrator validates before video-cut refreshes clip_plan_validated.json.
    # Without a matching raw-plan provenance fingerprint, lint against the current
    # raw plan even when mtimes are equal or misleading.
    return raw




def _validate_output_timeline_bounds(narration, output_duration, tolerance=0.05):
    """Hard-gate cut_output narration against the rendered output timeline.

    cut_output narration is authored in edited_source.mp4 time. If any segment falls outside
    that media duration, fail before TTS/render instead of spending time on unusable audio.
    """
    try:
        duration = float(output_duration)
    except (TypeError, ValueError):
        raise SystemExit(f"output_duration must be numeric, got {output_duration!r}")
    if not math.isfinite(duration) or duration <= 0:
        raise SystemExit(f"output_duration must be finite and positive, got output_duration={duration:.3f}")
    if not isinstance(narration, list):
        return

    problems = []
    for idx, seg in enumerate(narration):
        if not isinstance(seg, dict):
            continue
        try:
            start = float(seg.get("start"))
            end = float(seg.get("end"))
        except (TypeError, ValueError):
            continue
        if not math.isfinite(start) or not math.isfinite(end):
            problems.append(f"segment {idx} has non-finite time [{start!r},{end!r}]")
            continue
        if end <= -tolerance or start >= duration + tolerance:
            problems.append(
                f"segment {idx} [{start:.3f},{end:.3f}] fully outside output_duration={duration:.3f}"
            )
            continue
        if start < -tolerance:
            problems.append(
                f"segment {idx} start={start:.3f} before output timeline (output_duration={duration:.3f})"
            )
        if end > duration + tolerance:
            problems.append(
                f"segment {idx} end={end:.3f} exceeds output_duration={duration:.3f}"
            )
    if problems:
        raise SystemExit("cut_output narration exceeds rendered output timeline: " + "; ".join(problems))

def main():
    ap = argparse.ArgumentParser(description="Validate + align agent-written narration.json.")
    ap.add_argument("--work-dir", required=True)
    ap.add_argument("--mode", default="full", choices=["full", "cut", "cut_output"])
    ap.add_argument("--output-duration", type=float, default=None,
                    help="cut_output: rendered edited_source.mp4 duration in seconds")
    args = ap.parse_args()

    work_dir = Path(args.work_dir)
    CONFIG["edit_mode"] = args.mode
    narration_path = work_dir / "narration.json"
    narration = _load(narration_path)
    if narration is None:
        raise SystemExit(f"缺少 {narration_path}；请先按 video-script 规则写解说词")
    vlm_analysis = _load(work_dir / "vlm_analysis.json")
    silence_periods = _load(work_dir / "silence_periods.json") or []
    if args.mode == "cut_output":
        # Two-pass cut: narration is authored in OUTPUT time against edited_source.mp4 — there is
        # no clip_plan to fall into and no source-time scene/quiet data to align to. Lint timing /
        # budget / overlap / density on the output timeline only; never realign or rewrite it.
        validate_narration_or_raise(narration, None, clip_plan=None, mode="full", work_dir=work_dir)
        if args.output_duration is None:
            raise SystemExit("--output-duration is required when --mode cut_output")
        _validate_output_timeline_bounds(narration, args.output_duration)
    elif args.mode == "cut":
        clip_plan = _load_cut_clip_plan(work_dir)
        validate_narration_or_raise(narration, vlm_analysis, clip_plan=clip_plan, mode="cut", work_dir=work_dir)
        narration = _validate_narration_budget(narration, vlm_analysis)
    else:
        validate_narration_or_raise(narration, vlm_analysis, clip_plan=None, mode="full", work_dir=work_dir)
        narration = _align_narration_to_quiet(narration, vlm_analysis, silence_periods)
        narration_path.write_text(json.dumps(narration, ensure_ascii=False, indent=2), encoding="utf-8")

    log(f"解说词验证完成: {len(narration)} 段")
    print(json.dumps({"status": "validated", "segments": len(narration),
                      "lint": str(work_dir / "narration_lint.json")}, ensure_ascii=False))


if __name__ == "__main__":
    main()