Video Voiceover

"""English→Chinese dubbing — the render engine behind `recap.py --edit-mode dub`.

Invoked by the orchestrator (NOT run by hand). It replaces the original English speech with a
faithful Chinese translation spoken in the ORIGINAL speaker's cloned voice
(`mimo-v2.5-tts-voiceclone`, same MIMO_API_KEY, pure stdlib + ffmpeg, no GPU). This differs
from recap/解说, which overlays Chinese commentary on ducked original audio.

Division of labour (the same as recap): CODE does only the mechanical parts; the AGENT does all
the judgment. So there are NO text heuristics here (no sentence-splitting, hook-dedup, or junk
filters) — those are exactly the things an LLM does better, and trying to do them in code is
brittle and case-by-case. Two stages around an agent-authored pause:

  --stage prepare : extract audio → English ASR (timed windows) → pull one reference clip →
                    write dub_transcript.json + dub_brief.md, then stop.
  --stage render  : read the agent's dub_script.json ([{"start", "zh"}] on the source timeline)
                    → clone the original voice per line → time-fit each to its slot (anchored at
                    its start; only compress when it would overrun the next line — never a global
                    speed-up, so the voice never finishes ahead of the picture) → full-replace
                    the audio track → mux → dub_<name>.mp4.
"""
import argparse
import base64
import json
import re
import wave
from pathlib import Path

from lib import (
    CONFIG,
    get_video_duration,
    log,
    mimo_asr_api_call,
    mimo_tts_api_call,
    run_cmd,
)

CLONE_MODEL = "mimo-v2.5-tts-voiceclone"
CLONE_SR = 24000  # mimo voiceclone returns 24kHz mono PCM16 wav
ASR_MIME = "audio/wav"
ATEMPO_CAP = 2.0  # max compression before we trim instead (atempo>2 also sounds rushed)


# ── ffmpeg / wav helpers ─────────────────────────────────────────────

def _ffmpeg_extract_wav(video, out_wav, sr=16000):
    run_cmd(["ffmpeg", "-y", "-i", str(video), "-vn", "-ar", str(sr), "-ac", "1", str(out_wav)])


def _cut_wav(src_wav, out_wav, start, dur, sr=16000):
    run_cmd(["ffmpeg", "-y", "-i", str(src_wav), "-ss", str(start), "-t", str(dur),
             "-ar", str(sr), "-ac", "1", str(out_wav)])


def _atempo_chain(factor):
    """ffmpeg atempo only accepts 0.5–2.0 per stage; chain for larger factors."""
    factor = max(0.5, factor)
    stages, remaining = [], factor
    while remaining > 2.0:
        stages.append("atempo=2.0")
        remaining /= 2.0
    stages.append(f"atempo={remaining:.4f}")
    return ",".join(stages)


def _wav_frames(path):
    with wave.open(str(path), "rb") as w:
        return w.getframerate(), w.getnchannels(), w.readframes(w.getnframes())


# ── ASR (timed windows) ──────────────────────────────────────────────

def _run_asr(wav_path, lang="en"):
    raw = Path(wav_path).read_bytes()
    if not raw:
        return ""
    b64 = base64.b64encode(raw).decode("ascii")
    payload = {
        "model": CONFIG.get("mimo_asr_model", "mimo-v2.5-asr"),
        "messages": [{"role": "user", "content": [
            {"type": "input_audio", "input_audio": {"data": f"data:{ASR_MIME};base64,{b64}"}},
        ]}],
        "asr_options": {"language": lang},
    }
    resp = mimo_asr_api_call(payload)
    try:
        text = str(resp["choices"][0]["message"]["content"] or "").strip()
    except (KeyError, IndexError, TypeError):
        return ""
    return re.sub(r"<[^>]{1,20}>", "", text).strip()  # strip ASR markers like "<chinese>"


def _asr_windows(audio_wav, segs_dir, duration, window):
    """Transcribe fixed-time windows → [{start, end, text}]. Coarse timing anchors for the agent;
    the agent does the sentence/segment judgment, not this code."""
    windows = []
    start, idx = 0.0, 0
    while start < duration:
        end = min(start + window, duration)
        seg_wav = segs_dir / f"asr_{idx:03d}.wav"
        _cut_wav(audio_wav, seg_wav, start, end - start)
        text = _run_asr(seg_wav)
        if text:
            windows.append({"start": round(start, 2), "end": round(end, 2), "text": text})
        log(f"  ASR {start:.0f}-{end:.0f}s: {len(text)} chars")
        start, idx = end, idx + 1
    return windows


# ── clone TTS + time-fit + mix ───────────────────────────────────────

def _clone_tts(text, ref_b64, out_wav):
    payload = {
        "model": CLONE_MODEL,
        "messages": [
            {"role": "user", "content": "自然、清晰，保持原说话人的音色与节奏，语气平稳。"},
            {"role": "assistant", "content": text},
        ],
        "audio": {"format": "wav", "voice": f"data:audio/wav;base64,{ref_b64}"},
    }
    resp = mimo_tts_api_call(payload)
    data = resp["choices"][0]["message"]["audio"]["data"]
    Path(out_wav).write_bytes(base64.b64decode(data))


def _time_fit(raw_wav, fitted_wav, room_seconds):
    """Anchor-at-start fit: only compress when the dub would overrun `room_seconds` (the gap until
    the next line). Never globally speed up; a short dub keeps the natural pause. Returns the
    fitted duration."""
    dur = get_video_duration(raw_wav)
    if dur <= 0:
        return 0.0
    if dur <= room_seconds + 0.05:
        run_cmd(["ffmpeg", "-y", "-i", str(raw_wav), "-ar", str(CLONE_SR), "-ac", "1", str(fitted_wav)])
        return dur
    factor = dur / room_seconds
    if factor <= ATEMPO_CAP:
        run_cmd(["ffmpeg", "-y", "-i", str(raw_wav), "-filter:a", _atempo_chain(factor),
                 "-ar", str(CLONE_SR), "-ac", "1", str(fitted_wav)])
        return get_video_duration(fitted_wav)
    run_cmd(["ffmpeg", "-y", "-i", str(raw_wav),
             "-filter:a", f"{_atempo_chain(ATEMPO_CAP)},atrim=0:{room_seconds:.3f},"
                          f"afade=t=out:st={max(0, room_seconds - 0.15):.3f}:d=0.15",
             "-ar", str(CLONE_SR), "-ac", "1", str(fitted_wav)])
    return get_video_duration(fitted_wav)


def _build_dub_track(lines, duration, out_wav):
    canvas = bytearray(int(duration * CLONE_SR) * 2)  # 16-bit mono silence
    for ln in lines:
        fw = ln.get("fitted_wav")
        if not fw or not Path(fw).exists():
            continue
        sr, ch, frames = _wav_frames(fw)
        if sr != CLONE_SR or ch != 1:
            continue
        off = int(ln["start"] * CLONE_SR) * 2
        end = min(off + len(frames), len(canvas))
        canvas[off:end] = frames[: end - off]
    with wave.open(str(out_wav), "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(CLONE_SR)
        w.writeframes(bytes(canvas))


def _mux(video, dub_wav, out_video):
    run_cmd(["ffmpeg", "-y", "-i", str(video), "-i", str(dub_wav),
             "-map", "0:v:0", "-map", "1:a:0", "-c:v", "copy",
             "-af", "loudnorm=I=-16:TP=-1.5:LRA=11", "-c:a", "aac", "-b:a", "192k",
             "-shortest", str(out_video)])


# ── stages ───────────────────────────────────────────────────────────

def _ref_window(duration, ref_start, ref_dur):
    start = max(0.0, min(ref_start, max(0.0, duration - 2.0)))
    return start, max(2.0, min(ref_dur, duration - start))


def stage_prepare(video, work, asr_window, ref_start, ref_dur):
    work.mkdir(parents=True, exist_ok=True)
    segs_dir = work / "dub_asr"
    segs_dir.mkdir(exist_ok=True)
    duration = get_video_duration(video)
    log(f"[dub:prepare] video {duration:.1f}s")

    audio_wav = work / "dub_source.wav"
    _ffmpeg_extract_wav(video, audio_wav)
    windows = _asr_windows(audio_wav, segs_dir, duration, asr_window)
    if not windows:
        raise SystemExit("[dub] no speech transcribed")
    log(f"[dub:prepare] {len(windows)} ASR windows")

    rs, rd = _ref_window(duration, ref_start, ref_dur)
    _cut_wav(audio_wav, work / "dub_reference.wav", rs, rd)

    (work / "dub_transcript.json").write_text(
        json.dumps({"video": str(video), "duration": duration, "windows": windows},
                   ensure_ascii=False, indent=2), encoding="utf-8")
    (work / "dub_brief.md").write_text(_brief_md(windows, duration), encoding="utf-8")
    print(json.dumps({"status": "dub_prepared", "windows": len(windows),
                      "brief": str(work / "dub_brief.md")}, ensure_ascii=False))


def _brief_md(windows, duration):
    lines = [
        "# 配音翻译任务（dub）",
        "",
        f"视频时长 {duration:.1f}s。下面是英文原声的分窗转写（时间戳偏粗，仅作锚点）。",
        "把它**忠实翻译并配音**，写入 `dub_script.json`——核心是**和原视频节奏一致**，不是做解说、"
        "也不是做精简版。",
        "",
        "要点：",
        "1. 逐句忠实翻译，原声说什么就配什么；**不要删内容、不要合并、不要自作主张去掉开场 hook**；"
        "原声重复，配音也跟着重复。",
        "2. 每句放在它在原声里被说出的时间，`start`/`end` 取那句话的起止——这样配音才跟着原声节奏走"
        "（该停顿处自然留白）。",
        "3. 译文忠实精简，能在自己的 `start`→`end` 区间内用正常语速说完（约 5 字/秒）。",
        "4. 只有完全被截断、没法说完整的残句，才取其中说得完整的部分。",
        "",
        "输出 `dub_script.json` = `[{\"start\": 起秒, \"end\": 止秒, \"zh\": \"中文台词\"}, ...]`，按 start "
        "升序，[start,end] 是该句在原声里的时间区间，相邻句不要重叠。",
        "",
        "## 英文原声转写（按时间窗）",
        "",
        "| 起–止 | 英文 |",
        "|---|---|",
    ]
    for w in windows:
        lines.append(f"| {w['start']}–{w['end']}s | {w['text'].replace('|', '/')} |")
    return "\n".join(lines) + "\n"


def stage_render(video, work, ref_start, ref_dur):
    transcript = json.loads((work / "dub_transcript.json").read_text(encoding="utf-8"))
    duration = transcript["duration"]
    script = json.loads((work / "dub_script.json").read_text(encoding="utf-8"))
    lines = sorted(({"start": float(d["start"]),
                     "end": float(d["end"]) if d.get("end") is not None else None,
                     "zh": str(d.get("zh", "")).strip()}
                    for d in script if str(d.get("zh", "")).strip()),
                   key=lambda x: x["start"])
    if not lines:
        raise SystemExit("[dub] dub_script.json has no lines")

    ref_wav = work / "dub_reference.wav"
    if not ref_wav.exists():
        rs, rd = _ref_window(duration, ref_start, ref_dur)
        _cut_wav(work / "dub_source.wav", ref_wav, rs, rd)
    ref_b64 = base64.b64encode(ref_wav.read_bytes()).decode("ascii")

    tts_dir = work / "dub_tts"
    tts_dir.mkdir(exist_ok=True)
    log("[dub:render] clone-TTS + time-fit per line…")
    for i, ln in enumerate(lines):
        nxt = lines[i + 1]["start"] if i + 1 < len(lines) else duration
        slot_end = ln["end"] if ln["end"] is not None else nxt
        # fit to the original utterance's span (rhythm-faithful), but never overrun the next line
        room = max(0.4, min(slot_end, nxt) - ln["start"])
        raw = tts_dir / f"line_{i:03d}_raw.wav"
        fitted = tts_dir / f"line_{i:03d}.wav"
        _clone_tts(ln["zh"], ref_b64, raw)
        ln["fitted_wav"] = str(fitted)
        ln["fitted_dur"] = round(_time_fit(raw, fitted, room), 2)
        ln["room"] = round(room, 2)
        log(f"  line {i}: {ln['start']:.1f}s fit={ln['fitted_dur']}s/room {ln['room']}s «{ln['zh'][:18]}»")

    dub_wav = work / "dub_track.wav"
    _build_dub_track(lines, duration, dub_wav)
    out_video = work / f"dub_{Path(video).stem}.mp4"
    _mux(video, dub_wav, out_video)
    (work / "dub_manifest.json").write_text(
        json.dumps({"video": str(video), "duration": duration, "lines": lines},
                   ensure_ascii=False, indent=2), encoding="utf-8")
    log(f"[dub:render] done → {out_video}")
    print(json.dumps({"status": "dubbed", "output": str(out_video), "lines": len(lines)},
                     ensure_ascii=False))


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--stage", choices=["prepare", "render"], required=True)
    ap.add_argument("--video", required=True)
    ap.add_argument("--work-dir", required=True)
    ap.add_argument("--asr-window", type=float, default=6.0)
    ap.add_argument("--ref-start", type=float, default=2.0)
    ap.add_argument("--ref-dur", type=float, default=10.0)
    args = ap.parse_args()
    video, work = Path(args.video), Path(args.work_dir)
    if args.stage == "prepare":
        stage_prepare(video, work, args.asr_window, args.ref_start, args.ref_dur)
    else:
        stage_render(video, work, args.ref_start, args.ref_dur)


if __name__ == "__main__":
    main()

"""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

def _text_char_count(text):
    """计算文本的有效字数（去除标点和空白，这些不占 TTS 朗读时间）。"""
    return len(re.sub(r'[，。！？、；：…“”‘’《》〈〉\s"\'「」『』（）()【】\[\]—～·,.!?;:\\-]', '', text or ""))


def _truncate_at_sentence(text, max_chars):
    """在句子边界截断，不产生残句。max_chars 按有效字符计（不含标点空白）。"""
    if _text_char_count(text) <= max_chars:
        return text
    eff = 0
    cutoff = len(text)
    for i, ch in enumerate(text):
        eff += 1 if _text_char_count(ch) else 0
        if eff > max_chars:
            cutoff = i + 1
            break
    idx = max(text[:cutoff].rfind(sep) for sep in ['。', '！', '？', '!', '?'])
    if idx > 0:
        return text[:idx + 1]
    idx = max(text[:cutoff].rfind(sep) for sep in ['，', '、', '；', ','])
    if idx > 3:
        return text[:idx] + '。'
    return ""

import base64
import os
import re
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from lib import CONFIG
from lib import log, mimo_tts_api_call, get_video_duration
from lib import _truncate_at_sentence, stable_hash, file_fingerprint

SUPPORTED_TTS_ENGINES = {"mimo-tts"}
TTS_CACHE_VERSION = 1


def _parse_rate_offset(rate_str):
    """'+5%' -> 0.05, '-3%' -> -0.03, '+0%' -> 0.0"""
    m = re.match(r'([+-])(\d+)%', rate_str)
    if m:
        return float(m.group(1) + m.group(2)) / 100.0
    return 0.0


def _compute_tts_params(text, narration, seg_index):
    """根据内容特征和位置计算 TTS 语速/音高参数"""
    rate = "+5%"
    pitch = "+0Hz"
    total = len(narration)
    # 位置相关
    if seg_index == 0:
        rate = "+5%"       # 开头稍快，抓住注意力
    elif seg_index >= total - 1:
        rate = "-5%"       # 结尾放慢，收束感
    elif seg_index >= total - 2:
        rate = "-2%"       # 倒数第二段略慢

    # 内容相关
    has_exclamation = any(c in text for c in "！!")
    has_question = "？" in text or "?" in text
    has_ellipsis = "……" in text or "..." in text

    if has_exclamation:
        rate = "+8%"
        pitch = "+3Hz"    # 感叹句加速+微升调
    elif has_question:
        pitch = "+5Hz"    # 疑问句升调
    elif has_ellipsis:
        rate = "-3%"      # 省略号（悬念/犹豫）放慢

    # 长文本稍快
    if len(text) > 35 and not has_ellipsis:
        rate = max(rate, "+6%", key=lambda x: int(x.rstrip('%+-')))

    return rate, pitch

def _clean_narration_text(text):
    """清理解说文本中 TTS 不应读出的内容"""
    if not text:
        return text
    # 移除 markdown 格式标记
    text = re.sub(r'\*\*(.+?)\*\*', r'\1', text)  # **bold** → bold
    text = re.sub(r'\*(.+?)\*', r'\1', text)       # *italic* → italic
    text = re.sub(r'「(.+?)」', r'\1', text)        # 「quote」 → quote
    text = re.sub(r'」|「|『|』', '', text)
    # 移除方括号标注（[climax]、[suspense] 等舞台指示）
    text = re.sub(r'\[[^\]]*\]', '', text)
    # 移除圆括号标注（（旁白）、（转场）等）
    text = re.sub(r'[（(][^）)]*[）)]', '', text)
    # 规范化省略号和重复标点
    text = re.sub(r'\.{3,}|…{2,}', '……', text)
    text = re.sub(r'……+', '……', text)
    text = re.sub(r'([。！？，；：])\1+', r'\1', text)  # 重复标点 → 单个
    # 移除 emoji
    text = re.sub(r'[\U0001F600-\U0001F64F\U0001F300-\U0001F5FF\U0001F680-\U0001F6FF'
                  r'\U0001F1E0-\U0001F1FF\U00002702-\U000027B0\U0000FE00-\U0000FEFF]', '', text)
    # 清理多余空白
    text = re.sub(r'\s+', ' ', text).strip()
    return text


def _synthesize_segment(i, seg, narration, tts_dir, engine):
    """合成单个 TTS 段（线程安全），支持 resume 跳过已有文件"""
    prepared = _prepare_tts_segment(i, seg, narration, tts_dir, engine)
    if prepared is None:
        return None
    text, output_wav, rate, pitch, cache_key = prepared
    cached = _reuse_tts_segment_cache(i, seg, output_wav, text, rate, cache_key)
    if cached:
        return cached

    _run_tts_engine(engine, text, output_wav, rate=rate, pitch=pitch, emotion=seg.get("emotion"))

    dur = _get_audio_duration(output_wav)
    seg_slot = seg["end"] - seg["start"]
    seg_pause = seg.get("pause_after_ms", CONFIG.get("breath_ms", 250)) / 1000
    available = max(0.5, seg_slot - seg_pause)
    # assemble speeds every segment up by narration_speed (global atempo) BEFORE placement, so the
    # slot actually holds raw_dur / narration_speed; fold that in (plus the local per-segment adjust
    # headroom atempo_max) or a 1.3x-authored BLOCK gets needlessly truncated into a clipped fragment.
    narration_speed = float(CONFIG.get("narration_speed", 1.0) or 1.0)
    raw_budget = available * narration_speed * 1.2
    if dur > raw_budget and len(text) > 5:
        chars_per_sec = len(text) / dur if dur > 0 else 3.0
        target_chars = max(5, int(raw_budget * chars_per_sec) - 1)
        truncated = _truncate_at_sentence(text, target_chars)
        if truncated and len(truncated) >= 5 and truncated != text:
            log(f"  段 {i+1}: 解说超出片段时长，自动截断 {len(text)}→{len(truncated)} 字以适配（建议在解说里改写得更短）")
            text = truncated
            _run_tts_engine(engine, text, output_wav, rate=rate, pitch=pitch, emotion=seg.get("emotion"))
            dur = _get_audio_duration(output_wav)

    _write_tts_segment_cache(output_wav, cache_key, text, dur, _parse_rate_offset(rate))
    return _build_tts_segment_result(i, seg, text, output_wav, dur, _parse_rate_offset(rate))


def _build_tts_segment_result(index, seg, text, output_wav, duration, rate_offset):
    result = {
        "index": index,
        "start": seg["start"],
        "end": seg["end"],
        "narration": text,
        "audio_path": str(output_wav),
        "audio_duration": duration,
        "tts_rate_offset": rate_offset,
        "pause_after_ms": seg.get("pause_after_ms", CONFIG.get("breath_ms", 250)),
        "overlaps_speech": seg.get("overlaps_speech", True),
    }
    for optional_key in ("source_start", "source_end", "source_clip_id", "emotion"):
        if optional_key in seg:
            result[optional_key] = seg[optional_key]
    return result


def synthesize_tts(narration, work_dir):
    """合成解说音频（并行）"""
    tts_dir = work_dir / "tts_segments"
    tts_dir.mkdir(exist_ok=True)

    if not narration:
        raise RuntimeError("narration.json 没有可配音的解说段，已中止以避免生成无解说视频")

    cache_engine = "mimo-tts"
    cached_segments = []
    needs_fresh = False
    prepared_count = 0
    for i, seg in enumerate(narration):
        prepared = _prepare_tts_segment(i, seg, narration, tts_dir, cache_engine)
        if prepared is None:
            continue
        prepared_count += 1
        text, output_wav, rate, _pitch, cache_key = prepared
        cached = _reuse_tts_segment_cache(i, seg, output_wav, text, rate, cache_key)
        if not cached:
            needs_fresh = True
            break
        cached_segments.append(cached)
    if prepared_count == 0:
        raise RuntimeError("narration.json 没有可配音的有效文本，已中止以避免生成无解说视频")
    if not needs_fresh:
        cached_segments.sort(key=lambda x: x["index"])
        log(f"TTS 引擎: {cache_engine} (cache)")
        return cached_segments, cache_engine

    engine = resolve_tts_engine()
    if engine == "mimo-tts" and not CONFIG.get("mimo_tts_api_key"):
        key_name = CONFIG.get("mimo_tts_api_key_source", "MIMO_API_KEY")
        raise RuntimeError(f"请设置 {key_name} 环境变量用于 MiMo TTS")

    log(f"TTS 引擎: {engine}")

    segments = []
    failures = []
    max_workers = max(1, min(len(narration), CONFIG.get("tts_workers", 4)))

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(_synthesize_segment, i, seg, narration, tts_dir, engine): i
            for i, seg in enumerate(narration)
        }
        for future in as_completed(futures):
            try:
                result = future.result()
            except Exception as e:
                i = futures[future]
                failures.append((i, str(e)))
                log(f"  TTS 段 {i+1} 失败: {e}")
                continue
            if result:
                segments.append(result)
                log(f"  段 {result['index']+1}: {result['audio_duration']:.1f}s - {result['narration'][:25]}...")

    segments.sort(key=lambda x: x["index"])
    if failures and not CONFIG.get("allow_partial_tts", False):
        sample = "; ".join(f"段 {i+1}: {msg}" for i, msg in failures[:3])
        raise RuntimeError(
            f"TTS 失败 {len(failures)}/{len(narration)} 段，已中止以避免生成缺解说的视频。"
            f"示例: {sample}。如确需继续，可设置 ALLOW_PARTIAL_TTS=1 或 --allow-partial-tts。"
        )
    if failures:
        log(f"警告: TTS 部分失败 {len(failures)}/{len(narration)} 段，继续生成部分解说")
    if not segments:
        raise RuntimeError("TTS 没有生成任何有效解说音频，已中止以避免生成无解说视频")
    return segments, engine


def _run_tts_engine(engine, text, output_wav, rate="+0%", pitch="+0Hz", emotion=None):
    """Run one TTS engine with retry and remove partial files after failures."""
    retries = max(1, CONFIG.get("tts_retries", 3))
    last_error = None

    for attempt in range(1, retries + 1):
        try:
            _cleanup_partial_tts_outputs(output_wav)
            if engine == "mimo-tts":
                _tts_mimo(text, output_wav, rate=rate, pitch=pitch, emotion=emotion)
            else:
                raise RuntimeError(
                    f"不支持的 TTS 引擎: {engine}。当前仅支持 mimo-tts。"
                )

            dur = _get_audio_duration(output_wav)
            if dur <= 0:
                raise RuntimeError(f"{engine} 输出音频时长无效")
            return
        except Exception as exc:
            last_error = exc
            _cleanup_partial_tts_outputs(output_wav)
            if attempt < retries:
                wait = min(2 ** (attempt - 1), 8)
                log(f"  TTS 重试 {attempt+1}/{retries}: {exc}，等待 {wait}s")
                time.sleep(wait)

    raise RuntimeError(f"{engine} 合成失败: {last_error}") from last_error


def _cleanup_partial_tts_outputs(output_wav):
    """Remove stale partial media files before/after a failed TTS attempt."""
    wav_path = Path(output_wav)
    mp3_path = wav_path.with_suffix(".mp3")
    cache_path = str(_tts_segment_cache_path(output_wav))
    for path in (str(wav_path), str(mp3_path), cache_path):
        try:
            if os.path.exists(path):
                os.remove(path)
        except OSError:
            pass


def _tts_segment_cache_path(output_wav):
    return Path(str(output_wav) + ".cache.json")


def _prepare_tts_segment(index, seg, narration, tts_dir, engine):
    text = _clean_narration_text(seg["narration"])
    if not text or not text.strip():
        return None
    output_wav = tts_dir / f"narr_{index:03d}.wav"
    if CONFIG.get("tts_dynamic_params", True):
        rate, pitch = _compute_tts_params(text, narration, index)
    else:
        rate, pitch = "+0%", "+0Hz"
    cache_key = _tts_segment_cache_key(engine, index, seg, text, rate, pitch)
    return text, output_wav, rate, pitch, cache_key


def _reuse_tts_segment_cache(index, seg, output_wav, source_text, rate, cache_key):
    cached = _load_tts_segment_cache(output_wav, cache_key)
    if not cached:
        return None
    existing_dur = _get_audio_duration(output_wav)
    if existing_dur <= 0:
        return None
    spoken_text = str(cached.get("spoken_text") or source_text)
    rate_offset = float(cached.get("tts_rate_offset", _parse_rate_offset(rate)) or 0.0)
    log(f"  段 {index+1}: 复用已有 ({existing_dur:.1f}s)")
    return _build_tts_segment_result(index, seg, spoken_text, output_wav, existing_dur, rate_offset)


def _tts_segment_cache_key(engine, index, seg, source_text, rate, pitch):
    """Fingerprint the exact inputs that make a cached segment safe to reuse."""
    pause = seg.get("pause_after_ms", CONFIG.get("breath_ms", 250))
    try:
        pause = int(pause)
    except (TypeError, ValueError):
        pause = CONFIG.get("breath_ms", 250)
    payload = {
        "version": TTS_CACHE_VERSION,
        "engine": engine,
        "source_text": source_text,
        "segment_index": int(index),
        "start": round(float(seg.get("start", 0.0)), 3),
        "end": round(float(seg.get("end", 0.0)), 3),
        "pause_after_ms": pause,
        "rate": rate,
        "pitch": pitch,
        "emotion": (str(seg.get("emotion")).strip() if seg.get("emotion") else ""),
        "settings": tts_settings_fingerprint(engine),
    }
    return stable_hash(payload)


def _load_tts_segment_cache(output_wav, cache_key):
    """Return cache metadata only when the sidecar proves the WAV matches narration."""
    if not output_wav.exists():
        return None
    cache_path = _tts_segment_cache_path(output_wav)
    if not cache_path.exists():
        return None
    try:
        import json
        data = json.loads(cache_path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    if not isinstance(data, dict) or data.get("cache_key") != cache_key:
        return None
    try:
        if data.get("audio_fingerprint") != file_fingerprint(output_wav):
            return None
    except OSError:
        return None
    return data


def _write_tts_segment_cache(output_wav, cache_key, spoken_text, duration, rate_offset):
    """Persist non-secret provenance for safe per-segment TTS reuse."""
    try:
        import json
        _tts_segment_cache_path(output_wav).write_text(
            json.dumps({
                "version": TTS_CACHE_VERSION,
                "cache_key": cache_key,
                "audio_fingerprint": file_fingerprint(output_wav),
                "spoken_text": spoken_text,
                "audio_duration": duration,
                "tts_rate_offset": rate_offset,
            }, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
    except OSError as exc:
        log(f"  TTS 缓存元数据写入失败（忽略）: {exc}")


def _detect_tts_engine():
    """MiMo TTS is the only engine; require a MiMo key."""
    if CONFIG.get("mimo_tts_api_key"):
        return "mimo-tts"
    key_name = CONFIG.get("mimo_tts_api_key_source", "MIMO_API_KEY")
    raise RuntimeError(f"没有可用的 TTS 引擎：请设置 {key_name}（MiMo TTS 需要）。")


def resolve_tts_engine(prefer_existing=None):
    """Resolve the TTS engine. MiMo TTS (mimo-v2.5-tts) is the only engine.

    `prefer_existing` lets an assemble-only rerun reuse already-generated audio
    even when no fresh MiMo key is configured.
    """
    try:
        return _detect_tts_engine()
    except RuntimeError:
        if prefer_existing in SUPPORTED_TTS_ENGINES:
            return prefer_existing
        raise


def tts_settings_fingerprint(engine=None):
    """Return non-secret TTS settings that materially affect generated audio."""
    resolved = engine or resolve_tts_engine()
    return {
        "engine": resolved,
        "tts_dynamic_params": bool(CONFIG.get("tts_dynamic_params", True)),
        "mimo_tts_api_url": CONFIG.get("mimo_tts_api_url"),
        "mimo_tts_model": CONFIG.get("mimo_tts_model"),
        "mimo_tts_voice": CONFIG.get("mimo_tts_voice"),
        "mimo_tts_style": CONFIG.get("mimo_tts_style"),
        "narration_speed": float(CONFIG.get("narration_speed", 1.0) or 1.0),
    }


def _mimo_tts_style_instruction(rate="+0%", pitch="+0Hz", emotion=None):
    style = CONFIG.get("mimo_tts_style") or "自然、清晰、适合中文视频解说。"
    emo = str(emotion).strip() if emotion else ""
    if emo:
        tone = f"用「{emo}」的情绪和语气演绎这句解说，代入感强、有起伏，不要平铺直叙。"
    else:
        tone = "语气有感染力、有起伏，像在给观众讲故事，不要平淡机械。"
    rate_offset = _parse_rate_offset(rate)
    if rate_offset >= 0.06:
        speed = "语速略快，但吐字保持清楚。"
    elif rate_offset <= -0.03:
        speed = "语速略慢，适当停顿，保留收束感。"
    else:
        speed = "语速中等，节奏稳定。"
    pitch_hint = "疑问句或情绪抬升处可自然微升调。" if pitch and pitch != "+0Hz" else "音调自然。"
    return f"{style} {tone} {speed} {pitch_hint}"


def _tts_mimo(text, output_path, rate="+0%", pitch="+0Hz", emotion=None):
    """使用 Xiaomi MiMo-V2.5-TTS 合成，返回 wav 音频。

    MiMo-v2.5-tts 是 instruct-TTS：user 消息里的自然语言指令控制整句的情绪/语气/语速。
    每段 narration 的 `emotion` 标签即写进该指令，让解说有起伏、不机械。"""
    payload = {
        "model": CONFIG.get("mimo_tts_model", "mimo-v2.5-tts"),
        "messages": [
            {"role": "user", "content": _mimo_tts_style_instruction(rate, pitch, emotion)},
            {"role": "assistant", "content": text},
        ],
        "audio": {
            "format": "wav",
            "voice": CONFIG.get("mimo_tts_voice", "mimo_default"),
        },
    }
    resp = mimo_tts_api_call(payload)
    try:
        audio_data = resp["choices"][0]["message"]["audio"]["data"]
    except (KeyError, IndexError, TypeError) as exc:
        raise RuntimeError("MiMo-TTS 响应缺少 audio.data") from exc
    try:
        output_path.write_bytes(base64.b64decode(audio_data))
    except (TypeError, ValueError) as exc:
        raise RuntimeError("MiMo-TTS 返回的 audio.data 不是有效 base64") from exc


def _get_audio_duration(audio_path):
    """获取音频文件时长（复用 common.get_video_duration 的 ffprobe 探测）。"""
    return get_video_duration(audio_path)


# ── Step 7: 视频组装 ─────────────────────────────────────────────────


def main():
    import argparse
    import json
    from pathlib import Path
    ap = argparse.ArgumentParser(
        description="video-voiceover: synthesize narration audio segments from narration.json.")
    ap.add_argument("--work-dir", required=True)
    ap.add_argument("--narration", default=None,
                    help="narration json (default: <work-dir>/narration.json; pass narration_mapped.json explicitly for legacy cut runs)")
    ap.add_argument("--mimo-voice", default=None, help="MiMo TTS voice name")
    ap.add_argument("--allow-partial-tts", action="store_true",
                    help="allow output when some narration segments fail TTS")
    args = ap.parse_args()
    work_dir = Path(args.work_dir)
    if args.mimo_voice:
        CONFIG["mimo_tts_voice"] = args.mimo_voice
    if args.allow_partial_tts:
        CONFIG["allow_partial_tts"] = True
    if args.narration:
        narration_path = Path(args.narration)
    else:
        # Canonical cut mode is cut-first/narrate-second: narration.json is already on the
        # output timeline. A stale legacy narration_mapped.json in the same work_dir must not
        # silently override it; legacy direct-cut callers can still pass --narration explicitly.
        narration_path = work_dir / "narration.json"
    narration = json.loads(narration_path.read_text(encoding="utf-8"))
    tts_segments, engine_used = synthesize_tts(narration, work_dir)
    (work_dir / "tts_meta.json").write_text(
        json.dumps({"segments": tts_segments, "engine": engine_used, "narration": narration_path.name},
                   ensure_ascii=False, indent=2), encoding="utf-8")
    log(f"配音完成: {len(tts_segments)} 段, 引擎 {engine_used}")
    print(json.dumps({"status": "voiced", "segments": len(tts_segments), "engine": engine_used,
                      "tts_meta": str(work_dir / "tts_meta.json")}, ensure_ascii=False))


if __name__ == "__main__":
    main()