Video Understanding

import base64
import json
import os
import time
from pathlib import Path

from lib import CONFIG
from lib import log, run_cmd, get_video_duration, mimo_asr_api_call, file_fingerprint

# ── Step 3: ASR 转录（MiMo mimo-v2.5-asr，云端 API）────────────────────────

_ASR_AUDIO_MIME = "audio/wav"


def _load_name_glossary(work_dir):
    """从 background_research.json 收集已知人名（characters 键 + character_details 键及别名）。

    返回去重后、长度 >=2 的人名列表（按长度降序，长名优先匹配）。文件缺失或无名字时返回 []。
    """
    research_path = Path(work_dir) / "background_research.json"
    if not research_path.exists():
        return []
    try:
        data = json.loads(research_path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return []
    if not isinstance(data, dict):
        return []

    names = set()
    characters = data.get("characters")
    if isinstance(characters, dict):
        names.update(characters.keys())
    details = data.get("character_details")
    if isinstance(details, dict):
        for key, val in details.items():
            names.add(key)
            if isinstance(val, dict):
                aliases = val.get("aliases")
                if isinstance(aliases, list):
                    names.update(a for a in aliases if isinstance(a, str))

    cleaned = {n for n in names if isinstance(n, str) and len(n) >= 2}
    return sorted(cleaned, key=len, reverse=True)


def _correct_text_with_glossary(text, names):
    """用人名表修正 ASR 同音字错误（如 叶青眉 → 叶轻眉）。

    对每个已知人名，扫描文本中所有等长窗口；若某窗口与人名恰好相差一个字符（仅一处不同），
    则替换为该人名。严格约束为「恰好一字之差」，避免过度纠正（叶轻风 与 叶轻眉 也是一字之差，
    但这正是限制为单字替换的边界——只有当窗口本身不是任何已知人名时才会被改写）。
    """
    if not text or not names:
        return text
    name_set = set(names)
    for name in names:
        n = len(name)
        if n < 2 or len(text) < n:
            continue
        i = 0
        while i <= len(text) - n:
            window = text[i:i + n]
            # Only rewrite a one-char-off window when it is NOT itself a known name —
            # otherwise a distinct real name one char away (叶轻风 vs 叶轻眉) would be corrupted.
            if window != name and window not in name_set and _one_char_diff(window, name):
                text = text[:i] + name + text[i + n:]
                i += n
            else:
                i += 1
    return text


def _one_char_diff(a, b):
    """两个等长字符串是否恰好相差一个字符（仅一处不同）。"""
    if len(a) != len(b):
        return False
    diff = 0
    for ca, cb in zip(a, b):
        if ca != cb:
            diff += 1
            if diff > 1:
                return False
    return diff == 1


def _apply_glossary_corrections(segments, work_dir):
    """对已转录的 segments 就地应用人名表修正；无人名表时为 no-op。"""
    names = _load_name_glossary(work_dir)
    if not names:
        return segments
    for seg in segments:
        original = seg.get("text") or ""
        corrected = _correct_text_with_glossary(original, names)
        if corrected != original:
            seg["text"] = corrected
    return segments


def _audio_meta_path(work_dir):
    return Path(work_dir) / "audio.wav.meta.json"


def _write_audio_meta(work_dir, video_path):
    _audio_meta_path(work_dir).write_text(
        json.dumps({
            "schema_version": 1,
            "source_video_fingerprint": file_fingerprint(video_path),
            "audio": "audio.wav",
        }, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


def transcribe_audio(video_path, work_dir):
    """提取音频并用 MiMo ASR 分段转录，通过分段合成时间戳。"""
    asr_file = work_dir / "asr_result.json"

    if not CONFIG.get("mimo_asr_api_key"):
        key_name = CONFIG.get("mimo_asr_api_key_source", "MIMO_API_KEY")
        log(f"ASR 跳过：未设置 {key_name}（MiMo ASR 需要；VLM/TTS 也需要同一个 key）。"
            f"如不需要对白可加 --skip-asr")
        asr_file.write_text(json.dumps([], ensure_ascii=False, indent=2), encoding="utf-8")
        return []

    # 提取音频
    audio_wav = work_dir / "audio.wav"
    cmd = ["ffmpeg", "-y", "-i", str(video_path), "-vn",
           "-ar", "16000", "-ac", "1", str(audio_wav)]
    result = run_cmd(cmd)
    if result.returncode != 0:
        raise RuntimeError(f"音频提取失败: {result.stderr}")
    _write_audio_meta(work_dir, video_path)

    # 获取音频时长
    duration = get_video_duration(video_path)
    if duration <= 0:
        # ffprobe 失败时不再伪造 180s 时长，否则会向 asr_result.json 写入虚构时间戳
        log("ASR 警告: 无法获取音频时长（ffprobe 失败），跳过 ASR 转录")
        asr_file.write_text(json.dumps([], ensure_ascii=False, indent=2), encoding="utf-8")
        return []

    segments_dir = work_dir / "audio_segments"
    segments_dir.mkdir(exist_ok=True)

    segment_length = max(5, int(CONFIG.get("asr_segment_seconds", 30) or 30))
    if duration <= segment_length:
        # 短音频，整段转录
        text = _run_asr(audio_wav)
        asr_result = [{"start": 0.0, "end": round(duration, 2), "text": text}]
    else:
        # 长音频，分段转录（更细的窗口 → 更精细的对白时间戳）
        asr_result = _segment_and_transcribe(audio_wav, segments_dir, duration, segment_length)

    # 用 background_research.json 的人名表修正 ASR 同音字错误（如 叶青眉 → 叶轻眉）；无人名表时为 no-op
    _apply_glossary_corrections(asr_result, work_dir)

    # 保存
    asr_file.write_text(json.dumps(asr_result, ensure_ascii=False, indent=2), encoding="utf-8")

    total_text = " ".join(s["text"] for s in asr_result if s["text"])
    empty = sum(1 for s in asr_result if not s["text"])
    suffix = f"（{empty} 段无文本：静音/切分失败/超限被跳过）" if empty else ""
    log(f"ASR 转录完成: {len(asr_result)} 段, 共 {len(total_text)} 字{suffix}")
    return asr_result


def _run_asr(wav_path):
    """用 MiMo ASR (mimo-v2.5-asr) 转录单个 wav 文件，返回纯文本。

    音频以 base64 data-URI 放进 OpenAI 风格的 chat/completions 消息里，转写文本回到
    choices[0].message.content。API/响应结构失败会抛错，避免把瞬时失败缓存成空转写；
    只有无音频、超体积等确定不可发送的片段返回空串。
    """
    try:
        raw = Path(wav_path).read_bytes()
    except OSError as e:
        log(f"ASR 警告: 无法读取音频 {wav_path}: {e}")
        return ""
    if not raw:
        return ""

    b64 = base64.b64encode(raw).decode("ascii")
    max_b64_bytes = int(float(CONFIG.get("mimo_asr_base64_max_mb", 10.0)) * 1024 * 1024)
    if len(b64) > max_b64_bytes:
        log(f"ASR 警告: 分片 base64 体积 {len(b64) / 1024 / 1024:.1f}MB 超过 MiMo 上限 "
            f"{CONFIG.get('mimo_asr_base64_max_mb')}MB，跳过该段；可调小 ASR_SEGMENT_SECONDS")
        return ""

    payload = {
        "model": CONFIG.get("mimo_asr_model", "mimo-v2.5-asr"),
        "messages": [{
            "role": "user",
            "content": [{
                "type": "input_audio",
                "input_audio": {"data": f"data:{_ASR_AUDIO_MIME};base64,{b64}"},
            }],
        }],
        "asr_options": {"language": CONFIG.get("mimo_asr_language", "auto")},
    }
    try:
        resp = mimo_asr_api_call(payload)
    except Exception as e:
        raise RuntimeError(f"MiMo ASR 调用失败: {e}") from e
    try:
        return str(resp["choices"][0]["message"]["content"] or "").strip()
    except (KeyError, IndexError, TypeError):
        raise RuntimeError(f"MiMo ASR 返回结构异常: {json.dumps(resp, ensure_ascii=False)[:200]}")


def _segment_and_transcribe(audio_wav, segments_dir, total_duration, segment_length=None):
    """分段转录长音频"""
    if segment_length is None:
        segment_length = max(5, int(CONFIG.get("asr_segment_seconds", 30) or 30))
    # 长视频 ASR 是顺序调用；可选节流让调用间隔开，降低踩到集群限流的频率（默认 0=不节流）
    try:
        throttle = max(0.0, float(os.environ.get("ASR_THROTTLE_SECONDS", "0") or 0))
    except ValueError:
        throttle = 0.0
    results = []

    for i, start in enumerate(range(0, int(total_duration), segment_length)):
        if throttle and i:
            time.sleep(throttle)
        end = min(start + segment_length, total_duration)
        seg_wav = segments_dir / f"seg_{i:03d}.wav"

        cmd = ["ffmpeg", "-y", "-i", str(audio_wav),
               "-ss", str(start), "-to", str(end),
               "-ar", "16000", "-ac", "1", str(seg_wav)]
        cut = run_cmd(cmd)
        if cut.returncode != 0:
            # 切分失败时不要对磁盘上的陈旧/残缺音频转录，否则会得到错位文本
            log(f"  段 {i+1}: 切分失败，跳过转录 ({cut.stderr.strip()[:200]})")
            text = ""
        else:
            text = _run_asr(seg_wav)
        results.append({
            "start": round(start, 2),
            "end": round(end, 2),
            "text": text,
        })
        log(f"  段 {i+1}: {start:.0f}s-{end:.0f}s => {len(text)} 字")

    return results

#!/usr/bin/env python3
"""video-understanding consolidation / 整理 (index build-up).

Optional, synchronous, in-pipeline. Two independent LLM passes over the video's own signal:
  - Pass B (index, default): roll up per-scene vlm_analysis into a global
    character / relationship / plot index -> understanding_index.json (+ .md).
  - Pass A (asr cleanup, opt-in): clean/punctuate/lightly speaker-attribute the raw run-on
    ASR text -> asr_clean.json. Timing is preserved BY CONSTRUCTION: the model returns cleaned
    TEXT per segment index only; each segment's original start/end is re-attached here, so a
    cleanup pass can never shift the timing-bearing spans downstream chunking depends on.

Both passes degrade gracefully (a chat-API failure logs and is skipped) and are idempotent
(a fresh artifact is reused). Mirrors review.py's pure-seam + thin-driver shape so it is
unit-testable with a mocked api_call. NON-required: the pipeline runs unchanged without it.
"""
import argparse
import hashlib
import json
import re
from pathlib import Path

from lib import CONFIG, log, api_call
from understand import _fresh  # canonical freshness helper (same skill, safe import)

# Shared tolerance for the per-segment span check. The brief-side gate inlines the SAME
# literal (it cannot import this module without breaking the brief/narration byte-parity).
# Keep both in sync. Consolidate preserves spans exactly, so this only guards hand-edited files.
_ASR_SPAN_TOL = 0.05

CLEAN_PROMPT = """你在清洗中文视频的 ASR 逐段转写。对【每一段】做：补标点、修明显同音/错别字、（能判断时）在句首轻标说话人，让长段连读文本变成清晰可读的句子。
铁律：
- 不要合并或拆分段落，输出段数必须与输入完全一致，顺序一致。
- 不要改时间，不要输出 start/end（时间由程序保留）。
- 只清洗 text，不要增删事实、不要脑补画面。
只返回 JSON：{"segments":[{"i":0,"text":"清洗后的文本","speaker":"可选说话人"}, ...]}，i 为输入段的下标。"""

INDEX_PROMPT = """你在根据逐场景画面分析，为一个视频建立【全局理解索引】，供后续写解说词时保持人物/关系/主线一致。
只依据给到的画面证据（场景描述 + 帧实动作），不要脑补画面之外的剧情。
只返回 JSON：
{"characters":[{"name":"角色名或外观指代","description":"身份/特征"}],
 "relationships":[{"a":"角色","b":"角色","relation":"关系"}],
 "plot_points":["按时间顺序的关键剧情节点"],
 "entities":["重要物件/地点/线索"]}"""


# ── pure seams (no I/O; unit-testable) ────────────────────────────────────────

def build_clean_messages(asr_result):
    lines = []
    for i, seg in enumerate(asr_result or []):
        if not isinstance(seg, dict):
            continue
        lines.append(f'{i}. {str(seg.get("text", "")).strip()}')
    user = f"{CLEAN_PROMPT}\n\n## 逐段转写（共 {len(lines)} 段）\n" + "\n".join(lines)
    return [{"role": "user", "content": user}]


def parse_clean_response(text, asr_result):
    """Zip cleaned TEXT onto the original segments (start/end preserved by construction).
    Returns the original asr_result UNCHANGED on any parse / shape / count problem."""
    base = [s for s in (asr_result or []) if isinstance(s, dict)]
    if not base:
        return asr_result
    data = _extract_json(text)
    if not isinstance(data, dict):
        return asr_result
    segs = data.get("segments")
    if not isinstance(segs, list) or len(segs) != len(base):
        return asr_result  # count mismatch -> reject wholesale (idempotent no-op)
    by_index = {}
    for item in segs:
        if isinstance(item, dict) and isinstance(item.get("i"), int):
            by_index[item["i"]] = item
    if len(by_index) != len(base):
        return asr_result
    out = []
    for i, seg in enumerate(base):
        cleaned = by_index.get(i, {})
        new_text = str(cleaned.get("text", "")).strip() or str(seg.get("text", ""))
        merged = {"start": seg.get("start"), "end": seg.get("end"), "text": new_text}
        speaker = str(cleaned.get("speaker", "")).strip()
        if speaker:
            merged["speaker"] = speaker
        out.append(merged)
    return out


def build_index_messages(vlm_analysis):
    lines = []
    for scene in (vlm_analysis or []):
        if not isinstance(scene, dict):
            continue
        sid = scene.get("scene_id", "?")
        start = float(scene.get("start", 0) or 0)
        end = float(scene.get("end", 0) or 0)
        desc = str(scene.get("description", "")).strip().replace("\n", " ")
        facts = scene.get("frame_facts")
        fact_txt = ""
        if isinstance(facts, dict) and facts:  # frame_facts is a DICT {ts: [actions]}
            actions = []
            for ts in sorted(facts.keys(), key=lambda x: float(x)):
                vals = facts[ts]
                actions.extend(vals if isinstance(vals, list) else [str(vals)])
            if actions:
                fact_txt = " | 帧实: " + "；".join(a for a in actions[:6] if a)
        lines.append(f"[场景{sid} {start:.0f}-{end:.0f}s] {desc}{fact_txt}")
    user = f"{INDEX_PROMPT}\n\n## 逐场景画面分析（共 {len(lines)} 段）\n" + "\n".join(lines)
    return [{"role": "user", "content": user}]


def parse_index_response(text):
    data = _extract_json(text)
    if not isinstance(data, dict):
        return {"characters": [], "relationships": [], "plot_points": [], "entities": []}
    out = {}
    for key in ("characters", "relationships", "plot_points", "entities"):
        val = data.get(key)
        out[key] = val if isinstance(val, list) else []
    return out


def format_index_md(index):
    out = ["# Understanding index (from consolidate.py)", ""]
    chars = index.get("characters") or []
    if chars:
        out.append("## Characters")
        for c in chars:
            if isinstance(c, dict):
                out.append(f"- **{c.get('name', '?')}** — {c.get('description', '')}".rstrip())
            else:
                out.append(f"- {c}")
        out.append("")
    rels = index.get("relationships") or []
    if rels:
        out.append("## Relationships")
        for r in rels:
            if isinstance(r, dict):
                out.append(f"- {r.get('a', '?')} — {r.get('relation', '?')} — {r.get('b', '?')}")
            else:
                out.append(f"- {r}")
        out.append("")
    plot = index.get("plot_points") or []
    if plot:
        out.append("## Plot spine")
        out.extend(f"{i+1}. {p}" for i, p in enumerate(plot))
        out.append("")
    ents = index.get("entities") or []
    if ents:
        out.append("## Entities")
        out.extend(f"- {e}" for e in ents)
        out.append("")
    return "\n".join(out).rstrip() + "\n"


def _extract_json(text):
    raw = str(text or "")
    fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
    candidate = fence.group(1) if fence else raw
    if not fence:
        first, last = candidate.find("{"), candidate.rfind("}")
        if first != -1 and last > first:
            candidate = candidate[first:last + 1]
    try:
        return json.loads(candidate)
    except ValueError:
        return None


def _asr_source_md5(work_dir):
    """Provenance: md5 of the on-disk asr_result.json BYTES (writer + reader hash the same thing)."""
    path = Path(work_dir) / "asr_result.json"
    return hashlib.md5(path.read_bytes()).hexdigest() if path.exists() else ""


def _vlm_source_md5(work_dir):
    """Provenance: md5 of the on-disk vlm_analysis.json bytes."""
    path = Path(work_dir) / "vlm_analysis.json"
    return hashlib.md5(path.read_bytes()).hexdigest() if path.exists() else ""


def _index_meta_path(work_dir):
    return Path(work_dir) / "understanding_index.json.meta.json"


def _prompt_fingerprint(prompt):
    return hashlib.md5(str(prompt or "").encode("utf-8")).hexdigest()


def _write_index_meta(work_dir, vlm_analysis):
    _index_meta_path(work_dir).write_text(json.dumps({
        "schema_version": 1,
        "source_md5": _vlm_source_md5(work_dir),
        "scene_count": len([s for s in (vlm_analysis or []) if isinstance(s, dict)]),
        "model": CONFIG.get("vlm_model", ""),
        "prompt_md5": _prompt_fingerprint(INDEX_PROMPT),
    }, ensure_ascii=False, indent=2), encoding="utf-8")


def _index_cache_matches(work_dir, vlm_analysis):
    meta_path = _index_meta_path(work_dir)
    if not meta_path.exists():
        return False
    try:
        meta = json.loads(meta_path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return False
    return (
        isinstance(meta, dict)
        and meta.get("source_md5") == _vlm_source_md5(work_dir)
        and meta.get("scene_count") == len([s for s in (vlm_analysis or []) if isinstance(s, dict)])
        and meta.get("model") == CONFIG.get("vlm_model", "")
        and meta.get("prompt_md5") == _prompt_fingerprint(INDEX_PROMPT)
    )


# ── thin drivers (I/O + api_call) ─────────────────────────────────────────────

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 consolidate_transcript(work_dir):
    work_dir = Path(work_dir)
    asr_result = _load(work_dir, "asr_result.json")
    if not asr_result:
        log("consolidate(asr): 无 asr_result.json，跳过")
        return None
    out_path = work_dir / "asr_clean.json"
    if _fresh(out_path, work_dir / "asr_result.json"):
        existing = _load(work_dir, "asr_clean.json") or {}
        if (
            existing.get("source_md5") == _asr_source_md5(work_dir)
            and existing.get("model") == CONFIG.get("vlm_model", "")
            and existing.get("prompt_md5") == _prompt_fingerprint(CLEAN_PROMPT)
        ):
            log("consolidate(asr): asr_clean.json 已最新，跳过")
            return existing
    resp = api_call({"model": CONFIG.get("vlm_model", ""),
                     "messages": build_clean_messages(asr_result),
                     "max_tokens": 4000, "temperature": 0.2})
    content = _response_text(resp)
    segments = parse_clean_response(content, asr_result)
    payload = {
        "source_md5": _asr_source_md5(work_dir),
        "model": CONFIG.get("vlm_model", ""),
        "prompt_md5": _prompt_fingerprint(CLEAN_PROMPT),
        "segments": segments,
    }
    out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    log(f"consolidate(asr): 写出 asr_clean.json（{len(segments)} 段）")
    return payload


def consolidate_index(work_dir):
    work_dir = Path(work_dir)
    vlm_analysis = _load(work_dir, "vlm_analysis.json")
    if not vlm_analysis:
        log("consolidate(index): 无 vlm_analysis.json，跳过")
        return None
    out_path = work_dir / "understanding_index.json"
    if _fresh(out_path, work_dir / "vlm_analysis.json") and _index_cache_matches(work_dir, vlm_analysis):
        log("consolidate(index): understanding_index.json 已最新，跳过")
        return _load(work_dir, "understanding_index.json")
    resp = api_call({"model": CONFIG.get("vlm_model", ""),
                     "messages": build_index_messages(vlm_analysis),
                     "max_tokens": 2500, "temperature": 0.2})
    index = parse_index_response(_response_text(resp))
    out_path.write_text(json.dumps(index, ensure_ascii=False, indent=2), encoding="utf-8")
    _write_index_meta(work_dir, vlm_analysis)
    (work_dir / "understanding_index.md").write_text(format_index_md(index), encoding="utf-8")
    log(f"consolidate(index): 写出 understanding_index.json（角色 {len(index['characters'])}）")
    return index


def consolidate(work_dir, do_asr=False, do_index=True):
    """Default = index-only (Pass B, zero timing risk). Pass A (asr) is opt-in."""
    result = {}
    if do_index:
        try:
            result["index"] = consolidate_index(work_dir)
        except Exception as e:
            log(f"consolidate(index) 跳过（忽略）: {e}")
    if do_asr:
        try:
            result["asr_clean"] = consolidate_transcript(work_dir)
        except Exception as e:
            log(f"consolidate(asr) 跳过（忽略）: {e}")
    return result


def _response_text(resp):
    try:
        return resp["choices"][0]["message"]["content"]
    except (KeyError, IndexError, TypeError):
        log("consolidate: API 返回结构异常")
        return ""


def main():
    ap = argparse.ArgumentParser(description="Consolidate the understanding index (and optionally clean ASR).")
    ap.add_argument("--work-dir", required=True)
    ap.add_argument("--asr", action="store_true", help="also run Pass A (ASR cleanup)")
    ap.add_argument("--no-index", action="store_true", help="skip Pass B (index)")
    args = ap.parse_args()
    res = consolidate(args.work_dir, do_asr=args.asr, do_index=not args.no_index)
    print(json.dumps({"status": "consolidated",
                      "index": bool(res.get("index")),
                      "asr_clean": bool(res.get("asr_clean"))}, ensure_ascii=False))


if __name__ == "__main__":
    main()

import json
import os
import re
import subprocess
from pathlib import Path

from lib import CONFIG
from lib import log, run_cmd, get_video_duration, file_fingerprint

# ── Step 2: 场景检测 ──────────────────────────────────────────────────

def detect_scenes(video_path, work_dir, threshold=None):
    """使用 ffmpeg scdet 滤镜检测场景切换"""
    threshold = CONFIG["scene_threshold"] if threshold is None else threshold
    scdet_threshold = int(threshold * 100)

    cmd = ["ffmpeg", "-i", str(video_path),
           "-vf", f"scdet=threshold={scdet_threshold}",
           "-f", "null", "-"]
    result = run_cmd(cmd)
    if result.returncode != 0:
        raise RuntimeError(f"场景检测失败: {result.stderr}")

    # 解析 lavfi.scd.time 和 lavfi.scd.score
    times = []
    for line in result.stderr.split("\n"):
        match = re.search(r"lavfi\.scd\.time[:=]\s*(\S+)", line)
        if match:
            times.append(float(match.group(1)))

    if not times:
        # 没有检测到场景切换，整个视频作为一个场景
        duration = get_video_duration(video_path)
        scenes = [{"start": 0.0, "end": duration}]
        log(f"未检测到场景切换，整个视频作为一个场景 ({duration:.1f}s)")
    else:
        scenes = []
        prev = 0.0
        for t in times:
            scenes.append({"start": round(prev, 2), "end": round(t, 2)})
            prev = t
        # 最后一个场景到视频结束
        duration = get_video_duration(video_path)
        scenes.append({"start": round(prev, 2), "end": round(duration, 2)})

    log(f"检测到 {len(scenes)} 个场景")

    # 先过滤黑/白帧过渡场景，再合并短场景
    # （合并会保留短场景的 start，若黑场被并入长场景，单点采样会误删整段，故顺序在前）
    if CONFIG.get("scene_junk_filter", True):
        scenes = _filter_junk_scenes(scenes, video_path)
    # 合并短场景（< 3s 合并到相邻场景）
    scenes = _merge_short_scenes(scenes, min_duration=CONFIG.get("scene_merge_min", 4.0))

    # 保存
    scenes_file = work_dir / "scenes.json"
    scenes_file.write_text(json.dumps(scenes, ensure_ascii=False, indent=2), encoding="utf-8")
    for i, s in enumerate(scenes):
        log(f"  场景 {i+1}: {s['start']:.1f}s - {s['end']:.1f}s ({s['end']-s['start']:.1f}s)")

    return scenes


def _merge_short_scenes(scenes, min_duration=4.0):
    """合并过短的场景到相邻场景"""
    if len(scenes) <= 1:
        return scenes
    merged = [scenes[0]]
    for s in scenes[1:]:
        prev = merged[-1]
        # 如果前一个场景太短，合并到当前
        if prev["end"] - prev["start"] < min_duration:
            merged[-1] = {"start": prev["start"], "end": s["end"]}
        # 如果当前场景太短，合并到前一个
        elif s["end"] - s["start"] < min_duration:
            merged[-1] = {"start": prev["start"], "end": s["end"]}
        else:
            merged.append(s)
    # 如果最后一个太短，已在前一步合并
    log(f"合并短场景后: {len(scenes)} → {len(merged)} 个场景")
    return merged


def _sample_frame_luma(video_path, timestamp, sample_size=64):
    """Extract one frame via ffmpeg and return luma values without extra deps."""
    cmd = [
        "ffmpeg",
        "-hide_banner",
        "-loglevel",
        "error",
        "-ss",
        f"{max(0.0, float(timestamp)):.3f}",
        "-i",
        str(video_path),
        "-frames:v",
        "1",
        "-vf",
        f"scale={sample_size}:{sample_size},format=rgb24",
        "-f",
        "rawvideo",
        "-",
    ]
    result = subprocess.run(cmd, capture_output=True)
    if result.returncode != 0 or not result.stdout:
        raise RuntimeError(result.stderr.decode("utf-8", errors="replace")[:300])
    data = result.stdout
    return [
        0.2126 * data[i] + 0.7152 * data[i + 1] + 0.0722 * data[i + 2]
        for i in range(0, len(data) - 2, 3)
    ]


def _is_junk_scene(video_path, timestamp, threshold_dark=None, threshold_bright=None):
    """Return True for near-black or near-white scene-start frames."""
    threshold_dark = CONFIG.get("scene_junk_dark_luma", 8.0) if threshold_dark is None else threshold_dark
    threshold_bright = CONFIG.get("scene_junk_bright_luma", 245.0) if threshold_bright is None else threshold_bright
    pixel_ratio = min(1.0, max(0.0, float(CONFIG.get("scene_junk_pixel_ratio", 0.995))))
    try:
        lumas = _sample_frame_luma(video_path, timestamp)
    except Exception as exc:
        log(f"场景亮度采样失败，保留场景 {timestamp:.1f}s: {exc}")
        return False
    if not lumas:
        return False
    avg_luma = sum(lumas) / len(lumas)
    dark_ratio = sum(1 for value in lumas if value <= threshold_dark) / len(lumas)
    bright_ratio = sum(1 for value in lumas if value >= threshold_bright) / len(lumas)
    return (
        avg_luma <= threshold_dark and dark_ratio >= pixel_ratio
    ) or (
        avg_luma >= threshold_bright and bright_ratio >= pixel_ratio
    )


def _filter_junk_scenes(scenes, video_path):
    """Filter black/white transition scenes while never deleting the whole video."""
    if len(scenes) <= 1:
        return scenes
    filtered = []
    removed = []
    for scene in scenes:
        start = float(scene["start"])
        end = float(scene["end"])
        # 多点采样：起始、中点、结尾各探一帧；只有全部为垃圾帧才删除，
        # 含任意非垃圾帧的场景必须保留（避免短黑场并入长真实场景后被整段误删）
        probe_times = [
            min(end, start + 0.1),
            (start + end) / 2.0,
            max(start, end - 0.1),
        ]
        if all(_is_junk_scene(video_path, t) for t in probe_times):
            removed.append(scene)
        else:
            filtered.append(scene)
    if not filtered:
        log("场景黑/白帧过滤会删除全部场景，已放弃过滤")
        return scenes
    if removed:
        log(f"过滤黑/白帧过渡场景: {len(scenes)} → {len(filtered)}")
    return filtered



# ── Step 3.5: 静音检测 ─────────────────────────────────────────────

def detect_silence_periods(video_path, work_dir, asr_result=None):
    """用 ffmpeg silencedetect 检测安静时段，作为解说插入的候选窗口"""
    audio_path = work_dir / "audio.wav"
    if not _audio_cache_matches(audio_path, video_path):
        if audio_path.exists():
            audio_path.unlink()
        # 提取到临时文件，成功后原子移动到位，避免被中断的 -y 运行留下半截 audio.wav
        tmp_path = work_dir / "audio.wav.tmp"
        extract = run_cmd([
            "ffmpeg", "-y", "-i", str(video_path),
            "-vn", "-ar", "16000", "-ac", "1",
            "-f", "wav", str(tmp_path)  # .tmp extension hides the format from ffmpeg; state it
        ])
        if extract.returncode != 0 or not tmp_path.exists():
            log(f"音频提取失败，无法检测静音窗口（视频可能无音轨）: {extract.stderr}")
            if tmp_path.exists():
                tmp_path.unlink()
            return []
        os.replace(str(tmp_path), str(audio_path))
        _write_audio_meta(work_dir, video_path)

    noise = CONFIG["silence_noise_threshold"]
    min_dur = CONFIG["silence_min_duration"]
    cmd = ["ffmpeg", "-i", str(audio_path),
           "-af", f"silencedetect=noise={noise}:d={min_dur}",
           "-f", "null", "-"]
    result = run_cmd(cmd, timeout=120)
    if result.returncode != 0:
        log(f"静音检测失败: {result.stderr}")
        return []
    output = result.stderr

    # 解析 silence_start / silence_end
    starts = [float(m) for m in re.findall(r'silence_start:\s*([\d.]+)', output)]
    ends = [float(m) for m in re.findall(r'silence_end:\s*([\d.]+)', output)]

    # 配对所有静音段（不过滤时长，后面合并后再过滤）
    raw_periods = []
    for s, e in zip(starts, ends):
        raw_periods.append({"start": round(s, 2), "end": round(e, 2),
                            "duration": round(e - s, 2)})
    # 末尾静音
    if len(starts) > len(ends):
        dur_start = starts[len(ends)]
        total_dur = get_video_duration(str(audio_path))
        raw_periods.append({"start": round(dur_start, 2), "end": round(total_dur, 2),
                            "duration": round(total_dur - dur_start, 2)})

    # 合并相邻静音段（间隔 < merge_gap 的合并为一个大窗口）
    merge_gap = CONFIG.get("silence_merge_gap", 0.5)
    merged = []
    for rp in sorted(raw_periods, key=lambda x: x["start"]):
        if merged and rp["start"] - merged[-1]["end"] < merge_gap:
            merged[-1]["end"] = rp["end"]
            merged[-1]["duration"] = round(merged[-1]["end"] - merged[-1]["start"], 2)
        else:
            merged.append({"start": rp["start"], "end": rp["end"],
                           "duration": rp["duration"], "has_speech": False})

    # 过滤最短时长
    quiet_min = CONFIG["quiet_window_min"]
    periods = [p for p in merged if p["duration"] >= quiet_min]

    # 与 ASR 交叉验证：标记有语音的窗口
    # 跳过条件：ASR 段太粗（无法精确判断语音位置）
    # 1. 段数少(<=5)且覆盖>80%视频时长 → 时间戳不可靠
    # 2. 覆盖率>150% → 时间戳明显异常
    # 3. 平均段长过大 → 粒度太粗，无法判断哪些窗口有语音
    #    阈值 45s 给默认 ASR 窗口(asr_segment_seconds=30s)留出余量，使交叉验证重新生效；
    #    粗粒度(如 180s 旧窗口)仍会被正确跳过。
    if asr_result:
        video_dur = get_video_duration(str(audio_path))
        asr_coverage = sum(seg.get("end", 0) - seg.get("start", 0) for seg in asr_result)
        avg_seg_dur = asr_coverage / len(asr_result) if asr_result else 0
        skip_cross_check = (
            (len(asr_result) <= 5 and asr_coverage > video_dur * 0.8) or
            asr_coverage > video_dur * 1.5 or
            avg_seg_dur > 45
        )
        if not skip_cross_check:
            for qp in periods:
                for seg in asr_result:
                    seg_s = seg.get("start", 0)
                    seg_e = seg.get("end", 0)
                    overlap = min(qp["end"], seg_e) - max(qp["start"], seg_s)
                    if overlap > qp["duration"] * 0.3:
                        qp["has_speech"] = True
                        break

    # 保存
    (work_dir / "silence_periods.json").write_text(
        json.dumps(periods, ensure_ascii=False, indent=2), encoding="utf-8")
    log(f"检测到 {len(periods)} 个安静窗口 (≥{quiet_min}s)")
    for qp in periods:
        flag = " [有语音]" if qp["has_speech"] else ""
        log(f"  {qp['start']:.1f}s-{qp['end']:.1f}s ({qp['duration']:.1f}s){flag}")
    return periods


def _audio_meta_path(work_dir):
    return Path(work_dir) / "audio.wav.meta.json"


def _audio_cache_matches(audio_path, video_path):
    audio_path = Path(audio_path)
    if not audio_path.exists():
        return False
    meta_path = _audio_meta_path(audio_path.parent)
    if not meta_path.exists():
        return False
    try:
        meta = json.loads(meta_path.read_text(encoding="utf-8"))
    except (OSError, ValueError, TypeError):
        return False
    try:
        expected = file_fingerprint(video_path)
    except OSError:
        return False
    return meta.get("source_video_fingerprint") == expected


def _write_audio_meta(work_dir, video_path):
    _audio_meta_path(work_dir).write_text(
        json.dumps({
            "schema_version": 1,
            "source_video_fingerprint": file_fingerprint(video_path),
            "audio": "audio.wav",
        }, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

from lib import CONFIG
from lib import log, run_cmd

# ── Step 1: 帧提取 ───────────────────────────────────────────────────

def extract_frames(video_path, work_dir, fps=None):
    """提取视频帧"""
    fps = CONFIG["fps"] if fps is None else fps
    if fps <= 0:
        raise ValueError("fps 必须大于 0；完整 pipeline 会自动计算 fps")
    frames_dir = work_dir / "frames"
    frames_dir.mkdir(exist_ok=True)

    # 清理上一次（可能更高 fps）残留的帧，避免陈旧帧泄漏进本次结果
    for stale in frames_dir.glob("frame_*.jpg"):
        stale.unlink()

    output_pattern = str(frames_dir / "frame_%05d.jpg")
    cmd = ["ffmpeg", "-y", "-i", str(video_path),
           "-vf", f"fps={fps}", "-q:v", "2", output_pattern]
    result = run_cmd(cmd)
    if result.returncode != 0:
        raise RuntimeError(f"帧提取失败: {result.stderr}")

    frames = sorted(frames_dir.glob("frame_*.jpg"))
    log(f"提取了 {len(frames)} 帧 ({fps}fps)")
    return frames

"""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）
    # Storyboard contact sheets (advisory): generated only in video-understanding (owns frames+fps).
    # Kept ONLY here, matching the bundle convention that each lib.py carries the keys ITS skill uses
    # (video-cut's lib.py is deliberately minimal); no other skill reads storyboard_*.
    "storyboard": env_bool("STORYBOARD", True),  # generate source/edited storyboard contact sheets
    "storyboard_max_tiles": env_int("STORYBOARD_MAX_TILES", 30, minimum=1),  # cap tiles per sheet for legibility
    "storyboard_columns": env_int("STORYBOARD_COLUMNS", 6, minimum=1),  # tile grid columns
    # 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，带重试。

    长视频理解会发出数百次 VLM/ASR 调用，集群的 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