#!/usr/bin/env python3
"""
视频渲染工具 - 接收 DSL + TemplateBinding，补齐素材，调用 Remotion 渲染

默认行为:
  读取 DSL 与 TemplateBinding，解析缺失素材，调用原子 Skills 生成，
  编译时间线，最终调用 Remotion 渲染引擎输出视频。

用法:
  python render_video.py --dsl video.dsl.json --template-id screen-walkthrough
  python render_video.py --dsl video.dsl.json --template-id screen-walkthrough --resolve-only
  python render_video.py --render-plan video.render-plan.json

环境变量:
  PRIV_TOKEN                     - PrivToken（素材生成需要）
  MM_API_BASE_URL                - API 根地址
  REMOTION_OUTPUT_DIR            - 渲染输出目录（默认: tempfile.mkdtemp，按调用隔离；调试时可显式指定固定路径）
  ASSET_CACHE_DIR                - 素材缓存目录（默认: ./.asset-cache/）
  REMOTION_SKIP_VENDOR_CHROME    - 设为 1 则不从 OSS 拉取 Chrome Headless
  REMOTION_FORCE_VENDOR_CHROME   - 设为 1 则强制重新下载并解压（同 REMOTION_FORCE_VENDOR_ZIP）
  REMOTION_CHROME_VENDOR_JSON    - vendor-zip-urls.json 路径（默认在 chrome-headless-vendor-template 下）
  REMOTION_CHROME_VENDOR_DOWNLOAD_TIMEOUT - 下载超时秒数（默认 3600）
  REMOTION_CHROME_SHARE_DIR      - 浏览器 zip 共享目录（默认 /tmp/agent-share/chrome-headless，仅存 VERSION + *.zip）
"""

import argparse
import atexit
import builtins
import json
import os
import re
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from typing import Optional

# ── Cross-skill dependency ────────────────────────────────────────────────
# Reach into `skills/template-registry/` — that directory is the de-facto home
# for shared Python code in ab-skill (timeline-compilation, registry loader,
# render-job HTTP client, etc.). Naming it `_SHARED_LIB_DIR` reflects its
# actual role: it hosts much more than template lookup (see AGENTS.md
# "skills/template-registry/scripts/" note).
#
# If signatures of the functions below change, also update:
#   - skills/template-registry/video_dsl/runtime/timeline_compiler.py (the source)
#   - any other caller discoverable via `grep -r "split_subtitle\|segment_narration"`
# ───────────────────────────────────────────────────────────────────────────
_SHARED_LIB_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "template-registry")
# Single sys.path setup for both cross-skill import surfaces of the shared lib:
#   - <shared>            for `video_dsl.runtime.*`
#   - <shared>/scripts    for `registry_loader`, `match_template`,
#                          `render_job_client`, `template_paths`
# Each deeper function used to repeat its own sys.path.insert; consolidating
# here keeps the module's import side-effects in one place and matches
# Python's "set up sys.path once at module top" idiom.
sys.path.insert(0, _SHARED_LIB_DIR)
sys.path.insert(0, os.path.join(_SHARED_LIB_DIR, "scripts"))
from video_dsl.runtime.timeline_compiler import (
    split_subtitle,
    split_subtitle_from_lines,
    segment_narration,
)

# 字幕末尾标点去除（与 timeline_compiler 中的 _strip_trailing_punct 同逻辑）
_TRAILING_PUNCT_RE = re.compile(r"[。！？；，,、：:．.…]+$")

def _strip_subtitle_trailing_punct(text: str) -> str:
    """去掉字幕段末尾的标点符号，让画面更干净。"""
    return _TRAILING_PUNCT_RE.sub("", text)


# ── 字幕段二次平衡（切长 + 合短） ─────────────────────────────────────

_SUB_MAX_CHARS = 18  # 超过此长度的段尝试二次切分
_SUB_MIN_CHARS = 7   # 短于此长度的段尝试合并到相邻段
_SUB_SPLIT_RE = re.compile(r"(?<=[，,、；;])")  # 二次切分点：顿号/逗号/分号后


def _rebalance_subtitle_segments(segs: list[dict]) -> list[dict]:
    """对 TTS 返回的字幕段做二次平衡：
    1. 过长的段（> _SUB_MAX_CHARS）在逗号/顿号处切开，时间按字符比例分配
    2. 过短的段（< _SUB_MIN_CHARS）合并到前一段（共享时间窗口）
    """
    if not segs:
        return segs

    # Phase 1: 切长
    expanded: list[dict] = []
    for seg in segs:
        text = seg["text"]
        if len(text) <= _SUB_MAX_CHARS:
            expanded.append(seg)
            continue
        # 尝试在逗号/顿号处切分
        parts = _SUB_SPLIT_RE.split(text)
        parts = [p for p in parts if p.strip()]
        if len(parts) <= 1:
            # 没有合适的切分点，保持原样
            expanded.append(seg)
            continue
        # 按字符比例分配时间
        total_chars = max(sum(len(p) for p in parts), 1)
        total_frames = seg["endFrame"] - seg["startFrame"]
        cur_frame = seg["startFrame"]
        for p in parts:
            ratio = len(p) / total_chars
            frames = max(int(total_frames * ratio), 1)
            expanded.append({
                "text": _strip_subtitle_trailing_punct(p.strip()),
                "startFrame": cur_frame,
                "endFrame": cur_frame + frames,
            })
            cur_frame += frames
        # 修正最后一段的 endFrame 对齐
        if expanded:
            expanded[-1]["endFrame"] = seg["endFrame"]

    # Phase 2: 合短（把过短的段合并到前一段）
    if len(expanded) <= 1:
        return expanded
    merged: list[dict] = [expanded[0]]
    for seg in expanded[1:]:
        if len(seg["text"]) < _SUB_MIN_CHARS and merged:
            # 合并到前一段：文本拼接，endFrame 取后者
            merged[-1]["text"] = merged[-1]["text"] + seg["text"]
            merged[-1]["endFrame"] = seg["endFrame"]
        else:
            merged.append(seg)
    # 最后一段如果太短也合并
    if len(merged) > 1 and len(merged[-1]["text"]) < _SUB_MIN_CHARS:
        merged[-2]["text"] = merged[-2]["text"] + merged[-1]["text"]
        merged[-2]["endFrame"] = merged[-1]["endFrame"]
        merged.pop()

    return merged

# remotion-renderer 位于 monorepo 的 apps/ab-render/ 目录。脚本按相对位置推断
# (skills/render-video/scripts/ → ../../../../apps/ab-render)。
# 若 skill 被单独 clone 或路径不同，通过 REMOTION_RENDERER_DIR 环境变量覆盖。
_DEFAULT_RENDERER_DIR = os.path.abspath(
    os.path.join(os.path.dirname(__file__), "..", "..", "..", "..", "apps", "ab-render")
)
REMOTION_RENDERER_DIR = os.environ.get("REMOTION_RENDERER_DIR", _DEFAULT_RENDERER_DIR)
# 默认按调用创建独立 tempdir，避免多用户并发时落盘文件互相覆盖。
# 调试场景可显式设置 REMOTION_OUTPUT_DIR 指向固定路径。
OUTPUT_DIR = os.environ.get("REMOTION_OUTPUT_DIR") or tempfile.mkdtemp(prefix="ab-render-")
ASSET_CACHE_DIR = os.environ.get("ASSET_CACHE_DIR", "./.asset-cache")

RESOLUTION_MAP = {
    "16:9": {"1080p": (1920, 1080), "720p": (1280, 720), "4k": (3840, 2160)},
    "9:16": {"1080p": (1080, 1920), "720p": (720, 1280), "4k": (2160, 3840)},
    "1:1": {"1080p": (1080, 1080), "720p": (720, 720), "4k": (2160, 2160)},
    "4:3": {"1080p": (1440, 1080), "720p": (960, 720), "4k": (2880, 2160)},
    "3:4": {"1080p": (1080, 1440), "720p": (720, 960), "4k": (2160, 2880)},
}


def _narration_speed(narration: dict) -> Optional[float]:
    """Read a narration block's speech-rate multiplier, or None when unset.

    Tolerant on purpose: the DSL is hand-editable, and a malformed speed must
    not take down a render that would otherwise be fine — it falls back to the
    voice's own pace. 0 and negatives are treated as unset for the same reason
    ab-api does (`speed: 0` there means "follow the global setting").
    """
    if not isinstance(narration, dict):
        return None
    raw = narration.get("speed")
    if isinstance(raw, bool) or not isinstance(raw, (int, float)):
        return None
    return float(raw) if raw > 0 else None


def _narration_language_boost(narration: dict) -> Optional[str]:
    """Read a narration block's language hint, or None when unset.

    Same tolerance as _narration_speed: a malformed value must not take down an
    otherwise fine render. The service applies its own default when we send
    nothing, so "unset" is always a safe outcome — this flag exists to pin a
    language when auto-detection is not good enough (e.g. an English-only
    template whose narration still carries a Chinese brand name).
    """
    if not isinstance(narration, dict):
        return None
    raw = narration.get("languageBoost")
    if not isinstance(raw, str) or not raw.strip():
        return None
    return raw.strip()


def extract_narration_lines(narration: dict) -> Optional[tuple[list[str], int, list[Optional[float]]]]:
    """If narration uses the structured {intro, items, outro} form, return
    (lines, intro_line_count, at_sec_list). `at_sec_list` is parallel to
    `lines` and contains the per-line `atSec` hint (video-timeline offset
    in seconds) when the DSL author specified one, or None otherwise.
    Returns None if the narration only uses the flat `text` form.
    """
    if not narration:
        return None
    items = narration.get("items")
    if not items or not isinstance(items, list):
        return None
    intro = (narration.get("intro") or "").strip()
    outro = (narration.get("outro") or "").strip()

    def _coerce_item(x) -> tuple[str, Optional[float]]:
        if isinstance(x, dict):
            text = str(x.get("text", "")).strip()
            at = x.get("atSec")
            at_val = float(at) if isinstance(at, (int, float)) else None
            return text, at_val
        return str(x).strip(), None

    cleaned: list[tuple[str, Optional[float]]] = [
        (t, a) for (t, a) in (_coerce_item(x) for x in items) if t
    ]
    if not cleaned:
        return None
    lines: list[str] = []
    at_secs: list[Optional[float]] = []
    intro_lines = 0
    if intro:
        lines.append(intro)
        at_secs.append(None)
        intro_lines = 1
    for text, at in cleaned:
        lines.append(text)
        at_secs.append(at)
    if outro:
        lines.append(outro)
        at_secs.append(None)
    return lines, intro_lines, at_secs

_SKILLS_BASE_DIR = os.environ.get(
    "SKILLS_BASE_DIR",
    os.path.join(os.path.dirname(__file__), "..", "..")
)


def _resolve_ab_skill_cli_path() -> str | None:
    """Locate the compiled ab-skill CLI (dist/cli.js).

    Search order:
      1. AB_SKILL_CLI_PATH env (ab-agent always sets this)
      2. <_SKILLS_BASE_DIR>/../dist/cli.js     (ab-skill source layout)
      3. <_SKILLS_BASE_DIR>/../skills-cli/cli.js (ab-agent bundled layout)

    Returns None when nothing is found. The caller's error message then
    says exactly what's missing.
    """
    explicit = os.environ.get("AB_SKILL_CLI_PATH", "").strip()
    if explicit and os.path.exists(explicit):
        return explicit
    candidates = [
        os.path.normpath(os.path.join(_SKILLS_BASE_DIR, "..", "dist", "cli.js")),
        os.path.normpath(os.path.join(_SKILLS_BASE_DIR, "..", "skills-cli", "cli.js")),
    ]
    for c in candidates:
        if os.path.exists(c):
            return c
    return None


def build_skill_command(skill_name: str) -> list[str]:
    """Return the argv prefix to invoke sibling skill `skill_name`.

    Reads <_SKILLS_BASE_DIR>/<skill_name>/skill.json once and dispatches by
    entry.type:
      - python           → [sys.executable, "<skillDir>/<scriptPath>"]
      - http / builtin   → [node, "<ab-skill-cli>", skill_name]

    Centralizing the dispatch here means the 4 resolve_asset_* functions
    only need to append their per-call --flag value pairs to the prefix —
    they don't have to know whether the sibling is Python or TS.

    Raises RuntimeError with a precise message when skill.json / script
    file / CLI binary is missing, so subprocess.run gets a useful failure
    rather than a generic "file not found".
    """
    skill_dir = os.path.join(_SKILLS_BASE_DIR, skill_name)
    skill_json_path = os.path.join(skill_dir, "skill.json")
    if not os.path.exists(skill_json_path):
        raise RuntimeError(f"skill.json not found: {skill_json_path}")
    with open(skill_json_path, "r", encoding="utf-8") as f:
        meta = json.load(f)

    entry = meta.get("entry")
    if not entry and meta.get("scriptPath"):
        entry = {"type": "python", "scriptPath": meta["scriptPath"]}
    if not isinstance(entry, dict):
        raise RuntimeError(f"skill {skill_name}: missing entry/scriptPath in skill.json")

    entry_type = entry.get("type")
    if entry_type == "python":
        script_rel = entry.get("scriptPath")
        if not script_rel:
            raise RuntimeError(f"skill {skill_name}: entry.scriptPath empty")
        script_abs = os.path.join(skill_dir, script_rel)
        if not os.path.exists(script_abs):
            raise RuntimeError(f"Script not found: {script_abs}")
        return [sys.executable, script_abs]

    if entry_type in ("http", "builtin"):
        cli_path = _resolve_ab_skill_cli_path()
        if not cli_path:
            raise RuntimeError(
                f"skill {skill_name} has entry.type={entry_type} but ab-skill CLI not found. "
                f"Set AB_SKILL_CLI_PATH or rebuild via `npm run install-skills` in ab-agent."
            )
        return ["node", cli_path, skill_name]

    raise RuntimeError(f"skill {skill_name}: unknown entry.type={entry_type!r}")


def resolve_remotion_entry(template_id: str, aspect_ratio: str) -> str:
    """Look up the top-level Remotion composition for (templateId, aspectRatio).

    Reads ``remotionEntry[aspect_ratio]`` from the template's registry entry.
    The registry is the canonical source (matches what ab-render's manifest
    consumes); the previous implementation read a local ``template.json`` via
    a ``TEMPLATES_DIR`` env override that pointed at a directory which no
    longer exists in the repo — so the function always fell back to the
    hard-coded defaults regardless of what the template declared. Switching
    to the registry preserves the fallback for templates that don't declare
    ``remotionEntry`` and lets ones that DO declare it take effect (e.g.
    screen-walkthrough's 9:16 entry).

    Falls back to ``MainVideo16x9`` / ``MainVideo`` when the template isn't
    found, the registry can't be loaded, or the aspect ratio isn't keyed.

    ``include_all_statuses=True`` —— 状态门控是**选型策略**，不是渲染策略。
    它要回答的是"列表里该不该出现这个模板、agent 该不该挑它"，一旦模板已经被选定并
    通过校验，渲染就必须读到它真实的定义。把门控带到这里的后果是静默换成另一个
    composition / 丢掉封面，成片与模板声明不符，而日志里只字未提。
    「能不能被选中」由 template-registry 的 ``--include-beta`` 与
    ``dsl_validator._check_template_status_gate`` 负责把关。
    """
    if template_id:
        try:
            from registry_loader import get_template  # type: ignore
            tpl = get_template(template_id, include_all_statuses=True)
            if tpl:
                entry_map = tpl.get("remotionEntry") or {}
                hit = entry_map.get(aspect_ratio)
                if isinstance(hit, str) and hit:
                    return hit
        except Exception:
            pass
    return "MainVideo16x9" if aspect_ratio == "16:9" else "MainVideo"


def resolve_cover_composition_id(template_id: str) -> Optional[str]:
    """Look up the cover compositionId for a template via the registry.

    Each template's `compositions[]` array may contain at most one entry with
    `slot == "cover"`. We return its compositionId, or None if the template
    doesn't ship a cover.

    Reading from the registry rather than the template.json file directly so
    behaviour matches what ab-render's manifest exposes (the registry is the
    aggregated truth used by both ab-render and ab-skill).

    ``include_all_statuses=True`` 的理由同 ``resolve_remotion_entry``：门控管选型，
    不管渲染。带上门控的话 beta 模板的封面会无声无息地不渲。
    """
    if not template_id:
        return None
    try:
        # sys.path setup happens once at module top — see header.
        from registry_loader import get_template  # type: ignore

        tpl = get_template(template_id, include_all_statuses=True)
        if not tpl:
            return None
        for comp in tpl.get("compositions", []) or []:
            if comp.get("slot") == "cover":
                cid = comp.get("compositionId")
                if isinstance(cid, str) and cid:
                    return cid
    except Exception:
        pass
    return None


def now_iso():
    return datetime.now(timezone.utc).isoformat()


def LogPrint(*args, sep=" ", end="\n", file=None, flush=False):
    """带本地时间戳的 stderr/stdout 日志，格式 yyyyMMdd HHmmss:SSS（毫秒）。"""
    now = datetime.now()
    stamp = now.strftime("%Y%m%d %H%M%S") + f":{now.microsecond // 1000:03d}"
    if file is None:
        file = sys.stdout
    message = sep.join(str(a) for a in args)
    builtins.print(f"[{stamp}] {message}", end=end, file=file, flush=flush)


# ─── 扣费汇总 ────────────────────────────────────────────────────────────────
# 一条视频管线的花费分散在多处：每个 gen-image / gen-voice / gen-video /
# gen-digital-human 子进程各扣一次，渲染再通过 saveManifest 扣一次。子进程是
# capture_output 起的，它们自己打的 💳 页脚会被吞掉，所以这里把子进程 stdout 里的
# billing 事件抓出来累加，最后统一报一次总账——否则用户跑完一整条管线，只知道视频
# 好了，不知道花了多少积分。
_BILLING: dict = {"credits": 0, "balance": None}


def absorb_child_billing(stdout: str) -> None:
    """从子 skill 的 stdout 里收集 `__progress__` billing 事件（见 CLI src/billing.ts）。"""
    for line in (stdout or "").splitlines():
        line = line.strip()
        if not line.startswith("{") or "__progress__" not in line:
            continue
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        if not isinstance(event, dict) or event.get("phase") != "billing":
            continue
        credits = event.get("credits")
        if not isinstance(credits, (int, float)) or credits <= 0:
            continue
        _BILLING["credits"] += int(credits)
        balance = event.get("balance")
        if isinstance(balance, (int, float)):
            _BILLING["balance"] = int(balance)


def print_billing_footer() -> None:
    """收尾时报一次总账。通过 atexit 注册，所以 sys.exit / 中途失败也会打印
    ——已经发生的扣费不该因为后面某一步失败就不告知用户。

    走 stdout 而不是 stderr：ab-agent 交给 LLM 的是 `result.stdout || result.stderr`
    （mcp-tools.ts），只写 stderr 在托管 agent 里等于不可见。
    """
    credits = _BILLING["credits"]
    balance = _BILLING["balance"]
    try:
        import render_job_client  # type: ignore

        job_billing = render_job_client.billing_summary()
        if job_billing.get("credits"):
            credits += int(job_billing["credits"])
            # saveManifest（渲染扣费）在管线里排最后，它带回的余额最新。
            if job_billing.get("balance") is not None:
                balance = int(job_billing["balance"])
    except Exception:  # noqa: BLE001 — 报账失败绝不能影响已完成的渲染
        pass
    if credits <= 0:
        return
    suffix = f" · balance {balance:,}" if balance is not None else ""
    builtins.print(f"\n💳 Charged {credits:,} credits{suffix}", flush=True)


def sync_chrome_headless_vendor(renderer_dir: str, render_plan: dict) -> None:
    """Chrome Headless vendor 同步——实现已抽到独立模块 ``_chrome_vendor``。

    保留这个 thin wrapper 是为了：
      1. ``render_with_local_cli`` 现有调用点无需改名；
      2. 让 render_video.py 自身只关心"渲染调度"；vendor 部署细节留给独立模块。
    """
    from _chrome_vendor import sync_chrome_headless_vendor as _sync
    _sync(renderer_dir, render_plan)


def load_json(path: str) -> dict:
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def save_json(data: dict, path: str):
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def validate_dsl(dsl: dict) -> list:
    """Integrity DSL validation — delegates to the unified validator.

    Historical rule set (preserved verbatim by ``validate_integrity``):
      version + scene count + assetId duplicates + ``scene.audio.narration
      .assetRef`` / ``scene.visuals.background.assetRef`` reference integrity
      + structured narration items count vs templateData. ``meta.title`` and
      ``global`` presence are intentionally NOT enforced here — by the time
      a DSL reaches render_video those have already been gated upstream, and
      enforcing them again would change historical behavior.

    ``severity="warning"`` 的条目**不阻断**：它们描述的是"能渲染出来但有一处会静默
    退化"（如黑板版 highlight 不是 title 的子串 → 标题退回单色）。把它们和硬错误混在
    同一个 fatal 列表里，作者就只能二选一：要么被无关的软问题挡住，要么把整条校验关掉。
    警告改为打印到 stderr，退出码不受影响。
    """
    from video_dsl.runtime.dsl_validator import (  # noqa: E402
        validate_integrity,
        errors_as_strings,
    )
    found = validate_integrity(dsl)
    warnings = [e for e in found if getattr(e, "severity", "error") != "error"]
    if warnings:
        LogPrint("⚠️  DSL warnings (not blocking):", file=sys.stderr)
        for w in errors_as_strings(warnings):
            LogPrint(f"   - {w}", file=sys.stderr)
    return errors_as_strings([e for e in found if getattr(e, "severity", "error") == "error"])


def resolve_dimensions(dsl: dict) -> tuple:
    ratio = dsl.get("global", {}).get("aspectRatio", "16:9")
    resolution = dsl.get("global", {}).get("resolution", "1080p")
    dims = RESOLUTION_MAP.get(ratio, RESOLUTION_MAP["16:9"])
    return dims.get(resolution, dims.get("1080p", (1920, 1080)))


def validate_and_fix_render_plan(render_plan: dict) -> list:
    """Validate RenderPlan integrity and auto-fix recoverable issues.

    Checks performed:
      1. subtitleSegments frame numbers within scene [startFrame, endFrame]
      2. Video assets have non-null duration (required for screen-walkthrough)
      3. Timeline frame continuity (prev.endFrame == next.startFrame)
      4. All generated assets have a URL

    Returns list of warning messages (empty = all good).
    Mutates render_plan in-place to fix issues.
    """
    warnings: list[str] = []
    fps = render_plan.get("renderConfig", {}).get("fps", 30)

    # ── Check 1: subtitle frame bounds ────────────────────────────────────
    for entry in render_plan.get("timeline", []):
        scene_id = entry.get("sceneId", "?")
        scene_start = entry.get("startFrame", 0)
        scene_end = entry.get("endFrame", scene_start + entry.get("durationFrames", 0))

        for seg in entry.get("subtitleSegments", []):
            fixed = False
            if seg["startFrame"] < scene_start:
                warnings.append(
                    f"[fix] {scene_id}: subtitle startFrame {seg['startFrame']} < scene start {scene_start}, clamped"
                )
                seg["startFrame"] = scene_start
                fixed = True
            if seg["endFrame"] > scene_end:
                warnings.append(
                    f"[fix] {scene_id}: subtitle endFrame {seg['endFrame']} > scene end {scene_end}, clamped"
                )
                seg["endFrame"] = scene_end
                fixed = True
            if seg["startFrame"] >= seg["endFrame"]:
                # Degenerate segment after clamping — give it at least 1 frame
                seg["endFrame"] = min(seg["startFrame"] + max(1, int(fps * 0.5)), scene_end)

    # ── Check 2: video asset duration ─────────────────────────────────────
    # Whether the template needs every video asset to declare a non-null
    # duration is a TEMPLATE CAPABILITY, not a property of ab-skill. The
    # capability lives in template.json under ``capabilities.needsVideoDuration``;
    # ab-skill simply reads + applies it. New templates that need this guarantee
    # only have to declare the field — they don't have to touch this code.
    template_id = render_plan.get("templateId", "")
    needs_video_duration = False
    if template_id:
        try:
            from registry_loader import get_template  # type: ignore
            # include_all_statuses：门控管选型不管渲染（理由见 resolve_remotion_entry）。
            _tpl = get_template(template_id, include_all_statuses=True)
            needs_video_duration = bool(
                ((_tpl or {}).get("capabilities") or {}).get("needsVideoDuration")
            )
        except Exception:
            needs_video_duration = False

    for asset in render_plan.get("resolvedAssets", []):
        if asset.get("type") != "video":
            continue
        if asset.get("duration") is None and needs_video_duration:
            # Fallback: use the scene duration of whichever scene references this asset
            # via props.videoAssetId / backgroundAssetId / imageAssetId.
            fallback_dur = None
            for entry in render_plan.get("timeline", []):
                props = entry.get("props") or {}
                refs = (
                    props.get("videoAssetId"),
                    props.get("backgroundAssetId"),
                    props.get("imageAssetId"),
                )
                if asset.get("assetId") in refs:
                    fallback_dur = entry.get("durationFrames", 150) / fps
                    break
            if fallback_dur:
                asset["duration"] = fallback_dur
                warnings.append(
                    f"[fix] asset {asset.get('assetId')}: duration was null, set to scene duration {fallback_dur:.1f}s"
                )

    # ── Check 3: timeline frame continuity ────────────────────────────────
    timeline = render_plan.get("timeline", [])
    for i in range(1, len(timeline)):
        prev_end = timeline[i - 1].get("endFrame", 0)
        curr_start = timeline[i].get("startFrame", 0)
        if curr_start != prev_end:
            warnings.append(
                f"[fix] timeline gap: {timeline[i-1].get('sceneId')}.endFrame={prev_end} != "
                f"{timeline[i].get('sceneId')}.startFrame={curr_start}, correcting"
            )
            # Shift current and all subsequent scenes
            offset = prev_end - curr_start
            for j in range(i, len(timeline)):
                timeline[j]["startFrame"] += offset
                timeline[j]["endFrame"] += offset
                timeline[j]["startTime"] = round(timeline[j]["startFrame"] / fps, 2)
                timeline[j]["endTime"] = round(timeline[j]["endFrame"] / fps, 2)
                for seg in timeline[j].get("subtitleSegments", []):
                    seg["startFrame"] += offset
                    seg["endFrame"] += offset
            # Update total
            last = timeline[-1]
            render_plan["renderConfig"]["totalFrames"] = last["endFrame"]
            render_plan["renderConfig"]["totalDuration"] = round(last["endFrame"] / fps, 2)
            break  # re-check from start would be needed for multiple gaps, but rare

    # ── Check 4: generated assets have URL ────────────────────────────────
    for asset in render_plan.get("resolvedAssets", []):
        if asset.get("status") == "generated" and not asset.get("url"):
            warnings.append(
                f"[warn] asset {asset.get('assetId')}: status=generated but url is empty"
            )

    # Log warnings into render_plan
    if warnings:
        for w in warnings:
            render_plan.setdefault("logs", []).append({
                "phase": "validate-fix",
                "message": w,
                "timestamp": now_iso(),
            })
        LogPrint(f"⚠️  RenderPlan validation found {len(warnings)} issue(s) (auto-fixed):", file=sys.stderr)
        for w in warnings:
            LogPrint(f"   {w}", file=sys.stderr)

    return warnings


def _build_bgm_props(dsl: dict) -> dict:
    """Extract BGM config from DSL global settings and return as Remotion props."""
    bgm = dsl.get("global", {}).get("bgm", {})
    if bgm.get("enabled") and bgm.get("url"):
        return {"bgm": {"url": bgm["url"], "volume": bgm.get("volume", 0.15)}}
    return {}


def _probe_video_duration(url: str):
    """Backwards-compat wrapper — implementation lives in ``_video_probe`` module."""
    from _video_probe import probe_video_duration
    return probe_video_duration(url)


def build_render_plan(dsl: dict, binding: dict) -> dict:
    """Build a RenderPlan from DSL and TemplateBinding."""
    fps = dsl.get("global", {}).get("fps", 30)
    width, height = resolve_dimensions(dsl)

    binding_map = {}
    for b in binding.get("bindings", []):
        binding_map[b["sceneId"]] = b

    # Pre-step: copy narration text from scenes[].audio.narration.text down
    # into the matching audio asset's payload, so resolve_asset_audio (which
    # only sees the asset, not the scene) can call gen-voice with the right
    # text. narration.text is the single source of truth — DSL authors do
    # NOT need to populate audio asset payload.text themselves.
    #
    # Structured narration form ({intro, items, outro}) is also expanded here
    # so adjust_timeline_to_audio can later auto-derive highlightMap from
    # per-line TTS timestamps.
    dsl_assets_by_id = {a["assetId"]: a for a in dsl.get("assets", [])}
    global_narration = (dsl.get("global") or {}).get("narration") or {}
    global_speed = _narration_speed(global_narration)
    global_language_boost = _narration_language_boost(global_narration)
    for scene in dsl.get("scenes", []):
        narration = (scene.get("audio") or {}).get("narration") or {}
        ref = narration.get("assetRef")
        asset = dsl_assets_by_id.get(ref)
        if not asset:
            continue
        payload = asset.setdefault("payload", {})
        # Speech rate follows the same route as the narration text: single source
        # of truth on the DSL, copied down here because resolve_asset_audio only
        # ever sees the asset. Scene-level overrides global (mirrors the narration
        # editor's "行级覆盖 > 全局"); neither set = gen-voice's own 1.0 default.
        speed = _narration_speed(narration)
        if speed is None:
            speed = global_speed
        if speed is not None:
            payload["speed"] = speed
        # 语种提示走与语速完全相同的路线（场景覆盖全局，都不写就交给服务端默认）：
        # resolve_asset_audio 只看得见 asset，看不见 scene。
        language_boost = _narration_language_boost(narration)
        if language_boost is None:
            language_boost = global_language_boost
        if language_boost is not None:
            payload["languageBoost"] = language_boost
        extracted = extract_narration_lines(narration)
        if extracted:
            lines, intro_lines, _ = extracted
            payload["narrationItems"] = lines
            payload["narrationIntroLines"] = intro_lines
            payload["text"] = "\n".join(lines)
        else:
            text = narration.get("text", "")
            if text:
                payload["text"] = text

    assets_by_id = {}
    for asset in dsl.get("assets", []):
        # url 优先读顶层字段，其次 fallback 到 payload.url（兼容 source=url 写法）
        asset_url = asset.get("url", "") or (asset.get("payload") or {}).get("url", "")
        assets_by_id[asset["assetId"]] = {
            "assetId": asset["assetId"],
            "type": asset.get("type", ""),
            "source": asset.get("source", "existing"),
            "status": "pending" if asset.get("status") in ("planned", "missing") else asset.get("status", "pending"),
            "url": asset_url,
            "localPath": asset.get("localPath", ""),
            "duration": asset.get("duration"),
            "width": asset.get("width"),
            "height": asset.get("height"),
            "mimeType": asset.get("mimeType", ""),
            "generatedBy": {},
            "retryCount": 0,
            "maxRetries": 3,
        }

    # ── Probe duration for existing video assets without duration ──────
    # screen-walkthrough's adaptStrategy needs videoDurationSec to avoid
    # falling back to static-fallback. Probe via partial download + ffprobe
    # or fall back to a heuristic based on Content-Length.
    for aid, asset in assets_by_id.items():
        if asset["type"] == "video" and asset["source"] == "existing" and not asset.get("duration") and asset.get("url"):
            probed = _probe_video_duration(asset["url"])
            if probed:
                asset["duration"] = probed

    timeline = []
    current_frame = 0

    for idx, scene in enumerate(dsl.get("scenes", [])):
        scene_id = scene.get("id") or scene.get("sceneId") or f"scene-{idx:03d}"
        duration = scene.get("duration", 5)
        duration_frames = int(duration * fps)

        scene_binding = binding_map.get(scene_id, {})

        # P2.3: 不再产出 entry.layers。背景视觉资产、旁白音频、文本图层这些
        # 信息全都通过 propExtractors → binding.props 显式传给模板（例如
        # backgroundAssetId / narrationAssetId / titleText），模板用 props 即可。

        narration = (scene.get("audio") or {}).get("narration") or {}

        narration_text = narration.get("text", "")
        extracted = extract_narration_lines(narration)
        subtitle_at_sec: list[Optional[float]] = []
        if extracted:
            # Honour the authored line boundaries exactly (no secondary
            # comma-split, no short-fragment merge) so the fallback
            # subtitle count equals len(lines) — which is what
            # _auto_highlight_map expects when TTS timestamps are absent.
            lines, _intro_lines, subtitle_at_sec = extracted
            local_subs = split_subtitle_from_lines(lines, duration_frames, fps)
        else:
            local_subs = split_subtitle(narration_text, duration_frames, fps)
        subtitle_segments = [
            {
                "text": s["text"],
                "startFrame": s["startFrame"] + current_frame,
                "endFrame": s["endFrame"] + current_frame,
            }
            for s in local_subs
        ]

        transition_config = dsl.get("transitions", {})
        trans_type = transition_config.get("default", "fade")
        trans_dur = int(transition_config.get("duration", 0.5) * fps)

        # 首场景不做 fade-in 转场，避免开头黑屏
        scene_trans_type = "cut" if idx == 0 else trans_type
        scene_trans_dur = 0 if idx == 0 else trans_dur

        # 透传 scene.customPayload.minDurationSec 到 entry.minDurationFrames，
        # 供 adjust_timeline_to_audio 在 audio-driven 计算时取下限（不让 scene
        # 被旁白时长拍短）。
        custom_payload = scene.get("customPayload") or {}
        min_dur_sec = custom_payload.get("minDurationSec")
        min_dur_frames = (
            int(round(float(min_dur_sec) * fps))
            if isinstance(min_dur_sec, (int, float)) and min_dur_sec > 0
            else 0
        )
        # 透传 customPayload.tailPadSec 到 entry.tailPadFrames，覆盖
        # adjust_timeline_to_audio 默认的 1.5s 旁白尾巴。设 0 = 没有尾巴，
        # scene 贴音频结束立刻切走。未指定时（None）走全局默认 1.5s。
        tail_pad_sec = custom_payload.get("tailPadSec")
        tail_pad_frames = (
            int(round(float(tail_pad_sec) * fps))
            if isinstance(tail_pad_sec, (int, float)) and tail_pad_sec >= 0
            else None
        )

        entry = {
            "sceneId": scene_id,
            "startFrame": current_frame,
            "endFrame": current_frame + duration_frames,
            "durationFrames": duration_frames,
            "minDurationFrames": min_dur_frames,
            "tailPadFrames": tail_pad_frames,
            "startTime": round(current_frame / fps, 2),
            "endTime": round((current_frame + duration_frames) / fps, 2),
            "compositionId": scene_binding.get("compositionId", "GenericScene"),
            "props": scene_binding.get("props", {}),
            "subtitleSegments": subtitle_segments,
            "subtitleAtSec": subtitle_at_sec,
            "transition": {"type": scene_trans_type, "durationFrames": scene_trans_dur},
        }
        timeline.append(entry)
        current_frame += duration_frames

    total_frames = current_frame

    return {
        "version": "v1alpha1",
        "createdAt": now_iso(),
        "status": "planning",
        "templateId": binding.get("templateId", ""),
        # 渲染时常用的 DSL 摘要字段，避免下游再去 dsl 全文里捞。
        # 不复用 DSL 的 meta（dsl 是单一事实来源），只挑必要的几个供 UI/上传使用。
        "title": dsl.get("meta", {}).get("title", ""),
        "targetDuration": dsl.get("meta", {}).get("targetDuration"),
        "resolvedAssets": list(assets_by_id.values()),
        "timeline": timeline,
        "renderConfig": {
            "width": width,
            "height": height,
            "fps": fps,
            "totalFrames": total_frames,
            "totalDuration": round(total_frames / fps, 2),
            "codec": "h264",
            "crf": 18,
            "outputFormat": "mp4",
        },
        "remotionProps": {
            "compositionId": resolve_remotion_entry(
                binding.get("templateId", ""),
                dsl.get("global", {}).get("aspectRatio", "9:16"),
            ),
            "inputProps": {
                "globalTypography": binding.get("globalOverrides", {}).get("typography", {}),
                "motionPreset": binding.get("globalOverrides", {}).get("motionPreset", "smooth"),
                "colorScheme": binding.get("globalOverrides", {}).get("colorScheme", []),
                **({"variantId": binding["variantId"]} if binding.get("variantId") else {}),
                **(_build_bgm_props(dsl)),
            },
        },
        "errors": [],
        "logs": [
            {"phase": "validate", "message": "DSL schema validation passed", "timestamp": now_iso()},
            {"phase": "template-bind", "message": f"Using template: {binding.get('templateId', 'unknown')}", "timestamp": now_iso()},
        ],
    }


def resolve_asset_image(asset: dict, private_token: str, timeout: int) -> dict:
    """Resolve a single image asset by calling gen-image skill."""
    payload = asset.get("payload", {}) if "payload" not in asset else asset["payload"]

    try:
        cmd = build_skill_command("gen-image")
    except RuntimeError as e:
        return {"status": "failed", "error": str(e)}

    cmd.extend(["--prompt", payload.get("prompt", "placeholder image")])
    if payload.get("ratio"):
        cmd.extend(["--size", payload["ratio"]])
    if payload.get("model"):
        cmd.extend(["--model", payload["model"]])
    if private_token:
        cmd.extend(["--priv-token", private_token])

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL)
        absorb_child_billing(result.stdout)
        if result.returncode == 0:
            for line in reversed(result.stdout.strip().split("\n")):
                line = line.strip()
                if line.startswith("http"):
                    return {"status": "generated", "url": line}
        return {"status": "failed", "error": result.stderr.strip()[:1000]}
    except subprocess.TimeoutExpired:
        return {"status": "failed", "error": "Asset generation timed out"}
    except Exception as e:
        return {"status": "failed", "error": str(e)[:1000]}


def resolve_asset_audio(asset: dict, private_token: str, timeout: int) -> dict:
    """Resolve a single audio asset by calling gen-voice skill.

    Pre-processes text with segment_narration() and newlines so Minimax
    returns per-segment timestamps.  Uses --json-output to capture metadata.

    If the payload carries `narrationItems` (structured {intro,items,outro}
    form), those lines are fed verbatim to Minimax so the returned subtitle
    count matches the authored line count — enabling an automatic 1:1
    subtitle → card highlightMap in adjust_timeline_to_audio().
    """
    payload = asset.get("payload", {})

    try:
        cmd = build_skill_command("gen-voice")
    except RuntimeError as e:
        return {"status": "failed", "error": str(e)}

    narration_items = payload.get("narrationItems")
    if isinstance(narration_items, list) and narration_items:
        cleaned = [str(s).strip() for s in narration_items if str(s).strip()]
        tts_text = "\n".join(cleaned)
        if not tts_text:
            return {"status": "failed", "error": "No text provided for TTS"}
    else:
        text = payload.get("text", "")
        if not text:
            return {"status": "failed", "error": "No text provided for TTS"}
        segments = segment_narration(text)
        tts_text = "\n".join(s.strip() for s in segments) if segments else text

    cmd.extend(["--text", tts_text, "--json-output"])
    if payload.get("voiceId"):
        cmd.extend(["--voice-id", payload["voiceId"]])
    # Injected by build_render_plan from the DSL (scene narration > global).
    # Absent = let gen-voice apply its own default rather than pinning 1.0 here.
    speed = _narration_speed(payload)
    if speed is not None:
        cmd.extend(["--speed", str(speed)])
    # 同 speed：不带就**不传**这个 flag，默认值归服务端一处持有。
    language_boost = _narration_language_boost(payload)
    if language_boost is not None:
        cmd.extend(["--language-boost", language_boost])
    if private_token:
        cmd.extend(["--priv-token", private_token])

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL)
        absorb_child_billing(result.stdout)
        if result.returncode == 0:
            for line in reversed(result.stdout.strip().split("\n")):
                line = line.strip()
                if not line:
                    continue
                try:
                    meta = json.loads(line)
                    if meta.get("url"):
                        return {
                            "status": "generated",
                            "url": meta["url"],
                            "audio_length_ms": meta.get("audio_length_ms"),
                            "subtitles": meta.get("subtitles", []),
                        }
                except json.JSONDecodeError:
                    if line.startswith("http"):
                        return {"status": "generated", "url": line}
        return {"status": "failed", "error": result.stderr.strip()[:1000]}
    except subprocess.TimeoutExpired:
        return {"status": "failed", "error": "Asset generation timed out"}
    except Exception as e:
        return {"status": "failed", "error": str(e)[:1000]}


def resolve_asset_video(asset: dict, private_token: str, timeout: int) -> dict:
    """Resolve a single video asset by calling gen-video skill."""
    payload = asset.get("payload", {})

    try:
        cmd = build_skill_command("gen-video")
    except RuntimeError as e:
        return {"status": "failed", "error": str(e)}

    cmd.extend(["--prompt", payload.get("prompt", "")])
    if payload.get("duration"):
        cmd.extend(["--duration", str(int(payload["duration"]))])
    if payload.get("ratio"):
        cmd.extend(["--ratio", payload["ratio"]])
    if payload.get("model"):
        cmd.extend(["--model", payload["model"]])
    if private_token:
        cmd.extend(["--priv-token", private_token])

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL)
        absorb_child_billing(result.stdout)
        if result.returncode == 0:
            for line in reversed(result.stdout.strip().split("\n")):
                line = line.strip()
                if line.startswith("http"):
                    return {"status": "generated", "url": line}
        return {"status": "failed", "error": result.stderr.strip()[:1000]}
    except subprocess.TimeoutExpired:
        return {"status": "failed", "error": "Asset generation timed out"}
    except Exception as e:
        return {"status": "failed", "error": str(e)[:1000]}


def resolve_asset_digital_human(asset: dict, private_token: str, timeout: int) -> dict:
    """Resolve a single digital-human avatar asset by calling gen-digital-human skill."""
    payload = asset.get("payload", {})

    try:
        cmd = build_skill_command("gen-digital-human")
    except RuntimeError as e:
        return {"status": "failed", "error": str(e)}

    if payload.get("avatarId"):
        cmd.extend(["--avatar-id", str(payload["avatarId"])])
    if payload.get("text"):
        cmd.extend(["--text", payload["text"]])
    if payload.get("voiceId"):
        cmd.extend(["--voice-id", payload["voiceId"]])
    if payload.get("audioUrl"):
        cmd.extend(["--audio-url", payload["audioUrl"]])
    if payload.get("source"):
        cmd.extend(["--source", payload["source"]])
    if payload.get("ratio"):
        cmd.extend(["--aspect-ratio", payload["ratio"]])
    if private_token:
        cmd.extend(["--priv-token", private_token])

    dh_timeout = max(timeout, 660)
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=dh_timeout, stdin=subprocess.DEVNULL)
        absorb_child_billing(result.stdout)
        if result.returncode == 0:
            for line in reversed(result.stdout.strip().split("\n")):
                line = line.strip()
                if line.startswith("http"):
                    return {"status": "generated", "url": line}
        return {"status": "failed", "error": result.stderr.strip()[:1000]}
    except subprocess.TimeoutExpired:
        return {"status": "failed", "error": f"Digital human generation timed out after {dh_timeout}s"}
    except Exception as e:
        return {"status": "failed", "error": str(e)[:1000]}


ASSET_RESOLVERS = {
    "gen-image": resolve_asset_image,
    "gen-voice": resolve_asset_audio,
    "gen-video": resolve_asset_video,
    "gen-digital-human": resolve_asset_digital_human,
}


def _resolve_single_asset(asset: dict, render_plan: dict, private_token: str, max_retries: int, timeout: int) -> bool:
    """Resolve one asset. Returns True if generated, False if failed/skipped."""
    source = asset.get("source", "")
    resolver = ASSET_RESOLVERS.get(source)
    if not resolver:
        asset["status"] = "skipped"
        render_plan["logs"].append({
            "phase": "asset-resolve",
            "message": f"No resolver for source '{source}', skipping {asset['assetId']}",
            "timestamp": now_iso(),
        })
        return False

    dsl_asset = None
    for a in render_plan.get("_dsl_assets", []):
        if a.get("assetId") == asset["assetId"]:
            dsl_asset = a
            break

    asset_with_payload = asset.copy()
    if dsl_asset and "payload" in dsl_asset:
        asset_with_payload["payload"] = dsl_asset["payload"]

    # For digital-human assets: inject audioUrl from a resolved TTS narration asset
    if source == "gen-digital-human":
        payload = asset_with_payload.get("payload", {})
        if not payload.get("audioUrl"):
            # Find the first generated TTS audio asset and use its URL
            for ra in render_plan["resolvedAssets"]:
                if ra.get("type") == "audio" and ra.get("source") == "gen-voice" and ra.get("status") == "generated" and ra.get("url"):
                    payload["audioUrl"] = ra["url"]
                    asset_with_payload["payload"] = payload
                    LogPrint(f"   🔗 Injecting audio URL into digital-human asset: {ra['assetId']}", file=sys.stderr)
                    break

    for attempt in range(max_retries):
        asset["retryCount"] = attempt
        asset["generatedBy"] = {
            "skill": source,
            "startedAt": now_iso(),
        }

        LogPrint(f"   🔄 Generating asset {asset['assetId']} (attempt {attempt + 1}/{max_retries})...", file=sys.stderr)
        result = resolver(asset_with_payload, private_token, timeout)

        if result["status"] == "generated":
            asset["status"] = "generated"
            asset["url"] = result["url"]
            asset["generatedBy"]["completedAt"] = now_iso()
            if result.get("audio_length_ms") is not None:
                asset["duration"] = result["audio_length_ms"]
            if result.get("subtitles"):
                asset["ttsSubtitles"] = result["subtitles"]
            # Propagate structured-narration metadata onto the resolved
            # asset so adjust_timeline_to_audio can auto-derive highlightMap.
            src_payload = asset_with_payload.get("payload", {}) or {}
            if isinstance(src_payload.get("narrationItems"), list):
                asset["narrationLineCount"] = len(src_payload["narrationItems"])
                asset["narrationIntroLines"] = int(src_payload.get("narrationIntroLines") or 0)
            LogPrint(f"   ✅ {asset['assetId']} generated", file=sys.stderr)
            return True
        else:
            error_msg = result.get("error", "Unknown error")
            if attempt == max_retries - 1:
                asset["status"] = "failed"
                render_plan["errors"].append({
                    "phase": "asset-resolve",
                    "message": error_msg,
                    "assetId": asset["assetId"],
                    "timestamp": now_iso(),
                })
                LogPrint(f"   ❌ {asset['assetId']} generation failed: {error_msg[:300]}", file=sys.stderr)
                return False
            else:
                LogPrint(f"   ⚠️  {asset['assetId']} retrying...", file=sys.stderr)
                time.sleep(2)
    return False


def apply_stub_urls(render_plan: dict, stub_image_url: str = "", stub_video_url: str = "") -> int:
    """Short-circuit pending image/video assets with a stub URL (test mode, no API cost).

    For every pending asset whose source is gen-image / gen-video and whose type is image
    / video, replace with the stub URL in-place (status → generated). Returns the number
    of assets that were stubbed.
    """
    if not stub_image_url and not stub_video_url:
        return 0

    ts = now_iso()
    stubbed = 0
    for asset in render_plan.get("resolvedAssets", []):
        if asset.get("status") != "pending":
            continue
        atype = asset.get("type", "")
        source = asset.get("source", "")
        url = ""
        if stub_image_url and atype == "image" and source == "gen-image":
            url = stub_image_url
        elif stub_video_url and atype == "video" and source == "gen-video":
            url = stub_video_url
        if not url:
            continue
        asset["status"] = "generated"
        asset["url"] = url
        asset["generatedBy"] = {
            "skill": "stub",
            "startedAt": ts,
            "completedAt": ts,
        }
        stubbed += 1

    if stubbed:
        render_plan.setdefault("logs", []).append({
            "phase": "asset-resolve",
            "message": f"Stub mode: short-circuited {stubbed} asset(s) without calling generation API",
            "timestamp": ts,
        })
    return stubbed


def resolve_assets(render_plan: dict, private_token: str, max_retries: int, timeout: int) -> dict:
    """Resolve all pending assets in the render plan.

    Two-phase resolution:
      Phase 1: resolve non-avatar assets (TTS, images, video) — **in parallel**
      Phase 2: resolve avatar/digital-human assets which may depend on generated audio URLs — serial
    """
    from concurrent.futures import ThreadPoolExecutor, as_completed

    render_plan["status"] = "resolving-assets"
    render_plan["logs"].append({
        "phase": "asset-resolve",
        "message": f"Starting asset resolution for {len(render_plan['resolvedAssets'])} assets",
        "timestamp": now_iso(),
    })

    pending = [a for a in render_plan["resolvedAssets"] if a["status"] == "pending"]
    # Phase 1: resolve non-digital-human assets first (TTS audio needed by avatar)
    phase1 = [a for a in pending if a.get("source") != "gen-digital-human"]
    # Phase 2: resolve digital-human assets (can now use generated audio URLs)
    phase2 = [a for a in pending if a.get("source") == "gen-digital-human"]

    generated = 0
    failed = 0

    # Phase 1: parallel resolution (concurrency configurable via env)
    if phase1:
        default_parallelism = 8
        max_workers = min(
            int(os.environ.get("REMOTION_ASSET_PARALLELISM", str(default_parallelism))),
            len(phase1),
        )
        LogPrint(f"   ⚡ Phase 1: generating {len(phase1)} asset(s) in parallel (concurrency={max_workers})", file=sys.stderr)
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_asset = {
                executor.submit(
                    _resolve_single_asset, asset, render_plan, private_token, max_retries, timeout
                ): asset
                for asset in phase1
            }
            for future in as_completed(future_to_asset):
                asset = future_to_asset[future]
                try:
                    ok = future.result()
                    if ok:
                        generated += 1
                    elif asset["status"] == "failed":
                        failed += 1
                except Exception as exc:
                    asset["status"] = "failed"
                    failed += 1
                    LogPrint(f"   ❌ {asset['assetId']} exception: {exc}", file=sys.stderr)

    # Phase 2: serial resolution (digital-human depends on TTS audio)
    for asset in phase2:
        ok = _resolve_single_asset(asset, render_plan, private_token, max_retries, timeout)
        if ok:
            generated += 1
        elif asset["status"] == "failed":
            failed += 1

    render_plan["status"] = "assets-ready" if failed == 0 else "failed"
    render_plan["logs"].append({
        "phase": "asset-resolve",
        "message": f"Asset resolution complete: {generated} generated, {failed} failed",
        "timestamp": now_iso(),
    })

    return render_plan


def _auto_highlight_map(entry: dict, narration_asset: Optional[dict], render_plan: dict) -> None:
    """Auto-fill ``templateData.highlightMap`` for any template that uses
    structured ``{intro, items, outro}`` narration and ships per-line cards.

    Originally written for html-slide's multi-item slides — that's still the
    primary consumer — but the trigger is capability-driven, not templateId-
    keyed: any template whose binding pushes a ``narrationLineCount`` onto
    the narration asset and whose subtitle segments line up 1-per-line gets
    this auto-fill. Adding a new template that follows the same pattern needs
    NO change to this function.

    Skipped silently when:
      - narration asset has no narrationLineCount (structured form wasn't used)
      - author already supplied a highlightMap (explicit wins)
      - subtitleSegments count ≠ narrationLineCount (emits a warn log)
    """
    if not narration_asset:
        return
    line_count = narration_asset.get("narrationLineCount")
    if not isinstance(line_count, int) or line_count <= 0:
        return

    props = entry.get("props") or {}
    tdata = props.get("templateData")
    if not isinstance(tdata, dict):
        return
    if tdata.get("highlightMap"):
        return  # respect author-provided map

    segs = entry.get("subtitleSegments") or []
    if len(segs) != line_count:
        render_plan.setdefault("logs", []).append({
            "phase": "highlight",
            "level": "warn",
            "message": (
                f"Scene {entry.get('sceneId')}: narrationItems has {line_count} lines "
                f"but got {len(segs)} subtitle segments — skipping auto highlightMap"
            ),
            "timestamp": now_iso(),
        })
        return

    intro_lines = int(narration_asset.get("narrationIntroLines") or 0)
    item_count = line_count - intro_lines
    # outro counts toward line_count too; cap item_count at the authored
    # items-array length so trailing outro lines don't spill into card
    # indices.
    # ⚠️ 这份键表必须与模板侧 utils/slidePayload.ts 的 CARD_ARRAY_KEYS 完全一致：
    # 两处是同一算法的两份实现（这里是生产路径，那边是摊平 binder 形态下的本地兜底）。
    # 2026-09-15 修复：此前这里少了 steps / annotations，于是 step-flow 与
    # code-highlight 拿不到下面那行的上限裁剪，旁白带 outro 行时 highlightMap 会多出
    # 一个指向数组外的下标 —— 那一行播放期间"全部卡片变暗、没有一张点亮"。
    # 顺序也有意义：取第一个命中的键，新键一律追加在末尾。
    # template-library 的 `pnpm validate:contracts` 现在会双向核对这两份表。
    items_key = next(
        (
            k
            for k in ("concepts", "pillars", "eras", "items", "steps", "annotations")
            if isinstance(tdata.get(k), list)
        ),
        None,
    )
    if items_key:
        item_count = min(item_count, len(tdata[items_key]))

    highlight_map = {str(intro_lines + i): i for i in range(item_count) if item_count > 0}
    tdata["highlightMap"] = highlight_map
    props["templateData"] = tdata
    entry["props"] = props

    render_plan.setdefault("logs", []).append({
        "phase": "highlight",
        "message": (
            f"Scene {entry.get('sceneId')}: auto highlightMap={highlight_map} "
            f"(intro={intro_lines}, items={item_count}, total_segments={line_count})"
        ),
        "timestamp": now_iso(),
    })


def adjust_timeline_to_audio(render_plan: dict) -> dict:
    """Post-process: use real TTS audio durations and timestamps to fix
    scene durations and subtitle segments."""
    import math

    fps = render_plan["renderConfig"]["fps"]
    assets_map = {a["assetId"]: a for a in render_plan["resolvedAssets"]}
    current_frame = 0
    adjusted = 0

    for entry in render_plan["timeline"]:
        # narration assetId 现在直接由 propExtractors 注入到 props.narrationAssetId，
        # 不再依赖 entry.layers[type=audio] 这条间接路径。
        narration_asset = None
        nar_id = (entry.get("props") or {}).get("narrationAssetId")
        if nar_id:
            narration_asset = assets_map.get(nar_id)

        old_dur = entry["durationFrames"]

        if narration_asset and narration_asset.get("duration"):
            audio_ms = narration_asset["duration"]
            if audio_ms <= 0:
                # Sanity check: TTS returned 0 or negative duration — keep original estimate
                render_plan.setdefault("logs", []).append({
                    "phase": "audio-adjust",
                    "level": "warn",
                    "message": f"Scene {entry.get('sceneId')}: TTS duration={audio_ms}ms is invalid, keeping estimated {old_dur} frames",
                    "timestamp": now_iso(),
                })
            else:
                audio_frames = math.ceil(audio_ms / 1000.0 * fps)
                # DSL 里 scene.customPayload.tailPadSec 可覆盖默认 1.5s 尾巴
                # （比如开场卡片要紧接下一场，设 0 直接贴音频结束）。
                tail_pad_override = entry.get("tailPadFrames")
                tail_pad = (
                    int(tail_pad_override)
                    if isinstance(tail_pad_override, int) and tail_pad_override >= 0
                    else int(fps * 1.5)
                )
                min_dur = int(fps * 2)     # minimum 2 seconds per scene
                # 用户在 DSL 里指定的 scene 下限（minDurationSec → minDurationFrames），
                # 用于"视频比旁白长"的场景：要把视频完整播完，scene 时长不能被旁白拍短。
                scene_min_dur = int(entry.get("minDurationFrames") or 0)
                new_dur = max(min_dur, scene_min_dur, audio_frames + tail_pad)

                if new_dur != old_dur:
                    entry["durationFrames"] = new_dur
                    adjusted += 1

                tts_subs = narration_asset.get("ttsSubtitles", [])
                if tts_subs:
                    entry["subtitleSegments"] = [
                        {
                            "text": _strip_subtitle_trailing_punct(s["text"].strip()),
                            "startFrame": current_frame + int(s["timeBegin"] / 1000.0 * fps),
                            "endFrame": current_frame + int(s["timeEnd"] / 1000.0 * fps),
                        }
                        for s in tts_subs
                    ]
                else:
                    # No TTS timestamps — re-distribute existing subtitle
                    # lines across the REAL audio duration, by character
                    # length ratio. Avoid going through
                    # split_subtitle_from_lines() because its internal
                    # estimated_audio_frames heuristic can clip
                    # `usable_frames` below the actual audio length.
                    # 字幕跟随真实音频时长分布，撑大 scene（因 minDuration）
                    # 时尾巴静音段无字幕。
                    old_subs = entry.get("subtitleSegments", [])
                    at_sec_list = entry.get("subtitleAtSec") or []
                    # 仅当某条字幕显式给了 atSec（视频时间轴的提示点）
                    # 时，把字幕锚定到对应的视频帧；这样字幕节奏跟随
                    # 画面而不是旁白长度。最后一条字幕会一直显示到
                    # scene 结尾（让"看下效果"等收尾句保持在屏）。
                    has_at_sec = any(
                        isinstance(a, (int, float)) and a is not None
                        for a in at_sec_list
                    ) and len(at_sec_list) == len(old_subs)

                    if has_at_sec and old_subs:
                        scene_dur = entry["durationFrames"]
                        # 算每条字幕的起始帧（相对 scene）。
                        # atSec 缺失时以前一条结束位置为准。
                        starts: list[int] = []
                        for i, s in enumerate(old_subs):
                            at = at_sec_list[i] if i < len(at_sec_list) else None
                            if isinstance(at, (int, float)):
                                f = max(0, int(round(at * fps)))
                            else:
                                f = starts[i - 1] + (fps // 2) if i > 0 else 0
                            # 单调递增，避免后给的 atSec 比前一条小
                            if i > 0 and f <= starts[i - 1]:
                                f = starts[i - 1] + (fps // 2)
                            starts.append(min(f, scene_dur - 1))
                        new_subs = []
                        for i, s in enumerate(old_subs):
                            text = s.get("text", "")
                            start_f = starts[i]
                            # 最后一条延续到 scene 结尾；其余以下一条
                            # 起点为终点。
                            end_f = scene_dur if i == len(old_subs) - 1 else starts[i + 1]
                            new_subs.append({
                                "text": text,
                                "startFrame": current_frame + start_f,
                                "endFrame": current_frame + end_f,
                            })
                        entry["subtitleSegments"] = new_subs
                    elif old_subs:
                        sub_span = audio_frames  # use exact audio length
                        # Use plain char count (markdown ** stripped) for
                        # ratio. Fall back to 1 to avoid div-by-zero.
                        def _plain(t: str) -> int:
                            return len(re.sub(r"\*+", "", t or ""))
                        total_chars = max(sum(_plain(s.get("text", "")) for s in old_subs), 1)
                        new_subs = []
                        cursor = 0
                        for s in old_subs:
                            text = s.get("text", "")
                            ratio = _plain(text) / total_chars
                            seg_frames = max(int(sub_span * ratio), fps // 2)
                            new_subs.append({
                                "text": text,
                                "startFrame": current_frame + cursor,
                                "endFrame": current_frame + cursor + seg_frames,
                            })
                            cursor += seg_frames
                        entry["subtitleSegments"] = new_subs

        _auto_highlight_map(entry, narration_asset, render_plan)

        # Re-anchor subtitles for scenes that weren't handled in the narration block above.
        # This covers scenes without narration or with invalid duration.
        old_scene_start = entry.get("startFrame", 0)
        if not (narration_asset and narration_asset.get("duration") and narration_asset["duration"] > 0):
            # Subtitles weren't touched above — re-anchor if position shifted
            if old_scene_start != current_frame:
                old_subs = entry.get("subtitleSegments", [])
                if old_subs:
                    new_subs = []
                    for s in old_subs:
                        rel_start = s["startFrame"] - old_scene_start
                        rel_end = s["endFrame"] - old_scene_start
                        new_subs.append({
                            "text": s["text"],
                            "startFrame": current_frame + rel_start,
                            "endFrame": current_frame + rel_end,
                        })
                    entry["subtitleSegments"] = new_subs

        entry["startFrame"] = current_frame
        entry["endFrame"] = current_frame + entry["durationFrames"]
        entry["startTime"] = round(current_frame / fps, 2)
        entry["endTime"] = round(entry["endFrame"] / fps, 2)

        # P2.3: 不再维护 entry.layers —— 旧的 layers 数组只是给 ab-render 模板"间接"
        # 找 background asset 用，现已收敛到 propExtractors 写到 props.backgroundAssetId
        # / props.narrationAssetId 等显式字段。新版 build_render_plan 不再写入
        # entry.layers；如果上游真的塞了 layers 进来（旧 RenderPlan 文件被
        # --render-plan 喂进来），干脆扔掉，避免帧号坐标系混乱。
        entry.pop("layers", None)

        # 为 props.slides 中的每张幻灯片分配 durationFrames（Remotion Sequence 必需）
        slides = entry.get("props", {}).get("slides")
        if slides and isinstance(slides, list) and len(slides) > 0:
            scene_dur = entry["durationFrames"]
            per_slide = scene_dur // len(slides)
            remainder = scene_dur - per_slide * len(slides)
            for si, slide in enumerate(slides):
                if not isinstance(slide, dict):
                    continue
                slide["durationFrames"] = per_slide + (1 if si < remainder else 0)

        current_frame += entry["durationFrames"]

    render_plan["renderConfig"]["totalFrames"] = current_frame
    render_plan["renderConfig"]["totalDuration"] = round(current_frame / fps, 2)

    if adjusted:
        render_plan["logs"].append({
            "phase": "audio-adjust",
            "message": f"Adjusted {adjusted} scenes to match TTS audio durations, total {current_frame} frames ({current_frame/fps:.1f}s)",
            "timestamp": now_iso(),
        })
        LogPrint(f"🔧 Adjusted {adjusted} scene(s) to TTS audio duration; total {current_frame/fps:.1f}s", file=sys.stderr)

    return render_plan




def render_with_local_cli(render_plan: dict, output_path: str) -> bool:
    """Call Remotion CLI to render the video locally."""
    render_plan["status"] = "rendering"
    render_plan["logs"].append({
        "phase": "render",
        "message": "Starting Remotion render",
        "timestamp": now_iso(),
    })

    # Absolutise paths so they survive the cwd switch to REMOTION_RENDERER_DIR
    output_path = os.path.abspath(output_path)
    os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)

    # 直接使用 resolvedAssets 中的 HTTPS URL，让 Chrome headless 自行加载外网图片。
    # 若 render-plan.json 曾被本地化处理过（url 以 "/" 开头），从 originalUrl 恢复。
    https_assets = []
    for asset in render_plan.get("resolvedAssets", []):
        asset = asset.copy()
        if asset.get("url", "").startswith("/") and asset.get("originalUrl"):
            asset["url"] = asset["originalUrl"]
        https_assets.append(asset)

    props_path = os.path.abspath(os.path.join(OUTPUT_DIR, "remotion-props.json"))
    save_json({
        "timeline": render_plan["timeline"],
        "renderConfig": render_plan["renderConfig"],
        "resolvedAssets": https_assets,
        **(render_plan.get("remotionProps", {}).get("inputProps", {})),
    }, props_path)

    composition_id = render_plan.get("remotionProps", {}).get("compositionId", "MainVideo")
    config = render_plan["renderConfig"]

    renderer_dir = os.path.abspath(REMOTION_RENDERER_DIR)
    remotion_cli = os.path.join(renderer_dir, "node_modules", ".bin", "remotion")
    if not os.path.exists(remotion_cli):
        LogPrint("⚠️  node_modules not found; running npm install automatically (needs network on first run)...", file=sys.stderr)
        npm_check = subprocess.run(["which", "npm"], capture_output=True)
        if npm_check.returncode != 0:
            LogPrint("❌ npm not found — please install Node.js (18+ required)", file=sys.stderr)
            render_plan["status"] = "failed"
            render_plan["errors"].append({
                "phase": "render",
                "message": "npm not found. Please install Node.js 18+.",
                "timestamp": now_iso(),
            })
            return False
        install_result = subprocess.run(
            ["npm", "install"],
            cwd=renderer_dir,
            capture_output=True,
            text=True,
        )
        if install_result.returncode != 0:
            LogPrint(f"❌ npm install failed:\n{install_result.stderr[:500]}", file=sys.stderr)
            render_plan["status"] = "failed"
            render_plan["errors"].append({
                "phase": "render",
                "message": f"npm install failed: {install_result.stderr[:500]}",
                "timestamp": now_iso(),
            })
            return False
        LogPrint("✅ dependencies installed", file=sys.stderr)
        if not os.path.exists(remotion_cli):
            LogPrint("❌ remotion CLI still missing after npm install", file=sys.stderr)
            render_plan["status"] = "failed"
            render_plan["errors"].append({
                "phase": "render",
                "message": "remotion CLI not found after npm install",
                "timestamp": now_iso(),
            })
            return False

    sync_chrome_headless_vendor(renderer_dir, render_plan)

    cmd = [
        remotion_cli, "render",
        "src/index.ts",
        composition_id,
        output_path,
        "--props", props_path,
        # 不使用 --public-dir，图片直接通过 HTTPS URL 由 Chrome headless 加载。
        "--width", str(config.get("width", 1920)),
        "--height", str(config.get("height", 1080)),
        "--fps", str(config.get("fps", 30)),
        "--codec", config.get("codec", "h264"),
        "--crf", str(config.get("crf", 18)),
        "--timeout", str(config.get("timeoutPerFrame", 120000)),
    ]

    total_frames = config.get("totalFrames", 0)
    LogPrint(f"🎬 Starting Remotion render...", file=sys.stderr)
    LogPrint(f"   Composition: {composition_id}", file=sys.stderr)
    LogPrint(f"   size: {config.get('width')}x{config.get('height')}", file=sys.stderr)
    LogPrint(f"   total frames: {total_frames}", file=sys.stderr)
    LogPrint(f"   output: {output_path}", file=sys.stderr)
    LogPrint(f"   command: {' '.join(cmd)}", file=sys.stderr)
    sys.stderr.flush()

    try:
        # Stream output to stderr so progress is visible in real-time
        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            cwd=renderer_dir,
        )

        last_rendered = 0
        stderr_tail = []
        while True:
            line = proc.stdout.readline()
            if not line and proc.poll() is not None:
                break
            if not line:
                continue
            stderr_tail.append(line)
            if len(stderr_tail) > 50:
                stderr_tail.pop(0)

            # Parse and print compact progress: "Rendered 100/2617"
            stripped = line.strip()
            if stripped.startswith("Rendered "):
                try:
                    parts = stripped.split()
                    frac = parts[1].rstrip(",")  # "100/2617"
                    current = int(frac.split("/")[0])
                    # Print progress every 10% or every 100 frames
                    if total_frames and (current - last_rendered >= max(total_frames // 10, 1)):
                        pct = int(current / total_frames * 100)
                        LogPrint(f"   🎞️  Render progress: {current}/{total_frames} ({pct}%)", file=sys.stderr)
                        sys.stderr.flush()
                        # 结构化进度，供上层解析并转发到前端
                        print(
                            json.dumps({
                                "__progress__": True,
                                "phase": "render",
                                "progress": round(current / total_frames, 4),
                                "current": current,
                                "total": total_frames,
                            }, ensure_ascii=False),
                            flush=True,
                        )
                        last_rendered = current
                except (ValueError, IndexError):
                    pass
            elif "error" in stripped.lower() or stripped.startswith("Error"):
                LogPrint(f"   ⚠️  {stripped[:200]}", file=sys.stderr)
            sys.stderr.flush()

        returncode = proc.wait()

        if returncode == 0:
            render_plan["status"] = "completed"
            render_plan["logs"].append({
                "phase": "render",
                "message": f"Render completed successfully: {output_path}",
                "timestamp": now_iso(),
            })
            return True
        else:
            error_output = "".join(stderr_tail).strip()[-1000:]
            render_plan["status"] = "failed"
            render_plan["errors"].append({
                "phase": "render",
                "message": error_output,
                "timestamp": now_iso(),
            })
            return False
    except subprocess.TimeoutExpired:
        proc.kill()
        render_plan["status"] = "failed"
        render_plan["errors"].append({
            "phase": "render",
            "message": "Remotion render timed out",
            "timestamp": now_iso(),
        })
        return False
    except FileNotFoundError:
        render_plan["status"] = "failed"
        render_plan["errors"].append({
            "phase": "render",
            "message": "Remotion CLI not found",
            "timestamp": now_iso(),
        })
        return False


def _persist_plan_checkpoint(render_plan: dict, job_id: Optional[int], private_token: str) -> None:
    """把当前 RenderPlan 存个档（尽力而为，失败只记日志）。

    存在的唯一理由：远端 taskId 必须在**渲染结束之前**就落库。脚本可能在任何时刻
    被外层超时 SIGTERM，那一刻没有任何机会写盘；只在流程末尾 save_plan 的话，
    taskId 永远存不下来，"续跑"就无从谈起。
    """
    if not job_id:
        return
    try:
        import render_job_client  # type: ignore
        render_job_client.save_plan(job_id, json.dumps(render_plan, ensure_ascii=False), private_token)
        LogPrint(f"   💾 RenderPlan checkpoint saved (jobId={job_id})", file=sys.stderr)
    except Exception as exc:  # noqa: BLE001 — 存档失败不该让渲染本身失败
        LogPrint(f"   ⚠️  RenderPlan checkpoint failed (jobId={job_id}): {exc}", file=sys.stderr)


def render_with_remote_api(
    render_plan: dict,
    output_path: str,
    *,
    private_token: str,
    poll_timeout: float = 1800.0,
    poll_interval: float = 5.0,
    upload_title: Optional[str] = None,
    conversation_id: Optional[str] = None,
    job_id: Optional[int] = None,
) -> bool:
    """Submit render to ab-api /tool/renderVideo and poll until completion.

    On success, writes upload.fileUrl into render_plan and skips local MP4 download.

    job_id 用于中途存档（见 _persist_plan_checkpoint）：有它才能在被强杀后续跑，
    没有则退化成旧行为（一次性，杀了就得重渲）。
    """
    import remote_renderer_client

    render_plan["status"] = "rendering"
    render_plan["renderMode"] = "remote"
    render_plan["logs"].append({
        "phase": "render",
        "message": "Submitting render to remote API",
        "timestamp": now_iso(),
    })

    https_assets = []
    for asset in render_plan.get("resolvedAssets", []):
        asset = asset.copy()
        if asset.get("url", "").startswith("/") and asset.get("originalUrl"):
            asset["url"] = asset["originalUrl"]
        https_assets.append(asset)

    composition_id = render_plan.get("remotionProps", {}).get("compositionId", "MainVideo")
    config = render_plan["renderConfig"]
    input_props = {
        "timeline": render_plan["timeline"],
        "renderConfig": config,
        "resolvedAssets": https_assets,
        **(render_plan.get("remotionProps", {}).get("inputProps", {})),
    }
    effective_title = upload_title or os.path.splitext(os.path.basename(output_path))[0]

    # ─── Cover composition resolution ────────────────────────────────────
    # 优先级：
    #   1) render_plan.remotionProps.coverCompositionId（上游显式指定）
    #   2) registry.compositions[].slot=="cover" 反查（自动匹配）
    # 不传 → ab-render 不渲封面，行为与改造前一致。
    cover_composition_id = (
        render_plan.get("remotionProps", {}).get("coverCompositionId")
    )
    if not cover_composition_id:
        template_id = render_plan.get("templateId", "")
        cover_composition_id = resolve_cover_composition_id(template_id)

    # ─── 私有模板路由 ────────────────────────────────────────────────────
    # 仅【用户私有模板】(isBuiltin=false 且带 sourceOssKey) 源码在 OSS、不在
    # ab-render 启动 bundle，需走动态渲染：presignSource 取临时 GET URL → /renderDraft。
    # 内置模板即使带 sourceOssKey（builtin-template-registry 回填，仅服务
    # derive_template 派生），渲染仍走 ab-render 预构建的 /render，绝不走 /renderDraft，
    # 否则会用 OSS 上的旧快照 + DraftMainVideo 复刻渲染，与生产 MainVideo 漂移。
    # 下面这套 payload 构造在"续跑既有任务"时其实用不上（见后面的 resuming 分支），
    # 所以它的失败不该让续跑也跟着失败 —— 任务还在后端好好渲着。
    pending_resume = bool(str(render_plan.get("remoteTaskId") or "").strip())
    # 先给默认值：presign 失败时下面的分支不会走到，续跑分支仍要读 status_path。
    render_path = "/render"
    status_path = "/renderStatus"

    source_oss_key = None
    template_id = render_plan.get("templateId", "")
    if template_id:
        try:
            from registry_loader import get_template  # type: ignore
            # include_all_statuses：门控管选型不管渲染（理由见 resolve_remotion_entry）。
            # 漏掉它的话私有 beta 模板会丢 sourceOssKey，转而按内置模板渲——渲出来的
            # 是另一个模板的画面。
            _meta = get_template(template_id, include_all_statuses=True)
            if _meta and not _meta.get("isBuiltin"):
                source_oss_key = _meta.get("sourceOssKey")
        except Exception:
            source_oss_key = None

    if source_oss_key:
        import render_job_client  # type: ignore
        tarball_url = None
        try:
            src = render_job_client.presign_template_source(template_id, private_token)
            tarball_url = src.get("tarballUrl")
            if not tarball_url:
                raise RuntimeError(f"presignSource 未返回 tarballUrl: {src}")
        except Exception as exc:
            if pending_resume:
                LogPrint(
                    f"⚠️  presignSource failed ({exc}) — 但有待续跑的任务，忽略继续",
                    file=sys.stderr,
                )
            else:
                LogPrint(f"❌ presignSource failed for private template {template_id}: {exc}", file=sys.stderr)
                render_plan["status"] = "failed"
                render_plan["errors"].append({
                    "phase": "render",
                    "message": f"presignSource failed: {exc}",
                    "timestamp": now_iso(),
                })
                return False
        payload = {
            "tarballUrl": tarball_url,
            "inputProps": input_props,
            "upload": True,
            "uploadTitle": effective_title,
        }
        # 私有模板封面：传 coverCompositionId 让 /renderDraft 渲模板自带 cover 组件
        # （ab-render 注册了 DraftCover），与内置 /render 的 MainCover 对等；
        # 不传则 ab-render 退回截首帧 / 不渲。
        if cover_composition_id:
            payload["coverCompositionId"] = cover_composition_id
        render_path = "/renderDraft"
        status_path = "/renderDraftStatus"
    else:
        payload = {
            "compositionId": composition_id,
            "renderConfig": config,
            "inputProps": input_props,
            "uploadTitle": effective_title,
        }
        if cover_composition_id:
            payload["coverCompositionId"] = cover_composition_id
        render_path = "/render"
        status_path = "/renderStatus"

    total_frames = config.get("totalFrames", 0)
    if pending_resume:
        # 续跑：下面那堆"正在提交"的日志会误导人，跳过
        LogPrint(f"🎬 Remote render task already exists, will resume polling", file=sys.stderr)
    elif source_oss_key:
        LogPrint(f"   Private template (OSS dynamic bundle): {template_id}", file=sys.stderr)
    else:
        LogPrint(f"   Composition: {composition_id}", file=sys.stderr)
    if cover_composition_id:
        LogPrint(f"   Cover: {cover_composition_id}", file=sys.stderr)
    LogPrint(f"   size: {config.get('width')}x{config.get('height')}", file=sys.stderr)
    LogPrint(f"   total frames: {total_frames}", file=sys.stderr)
    sys.stderr.flush()

    # ─── 续跑既有任务 ────────────────────────────────────────────────────
    # 上一次调用可能是被外层超时 SIGTERM 掉的（skill 执行有上限），而远端任务还在
    # ab-render 上跑着。这时重新提交 = 白烧一遍渲染、并且大概率再次撞上同一个上限。
    # 所以先看 plan 里有没有上次留下的 taskId：有就直接续着轮询。
    #
    # 只在"任务不可用"（已失败 / 查不到）时才退回重新提交 —— 轮询超时不算不可用，
    # 那条路走 RemoteRenderTimeout，保留 taskId 交给下一次调用。
    task_id = str(render_plan.get("remoteTaskId") or "").strip()
    if task_id:
        # 提交时用的状态端点存在 plan 里（内置模板 /renderStatus，私有模板
        # /renderDraftStatus）。用存的那个，别依赖这次重新推导的结果。
        status_path = render_plan.get("remoteStatusPath") or status_path
        LogPrint(f"🔄 resuming existing remote task: taskId={task_id}", file=sys.stderr)
        render_plan["logs"].append({
            "phase": "render",
            "message": f"Resuming remote task: {task_id}",
            "timestamp": now_iso(),
        })
    else:
        try:
            task_id = remote_renderer_client.start_render(payload, private_token=private_token, conversation_id=conversation_id, path=render_path)
        except Exception as exc:
            LogPrint(f"❌ remote render submission failed: {exc}", file=sys.stderr)
            render_plan["status"] = "failed"
            render_plan["errors"].append({
                "phase": "render",
                "message": f"remote start_render failed: {exc}",
                "timestamp": now_iso(),
            })
            return False

        LogPrint(f"   ✅ task submitted, taskId={task_id}", file=sys.stderr)
        render_plan["remoteTaskId"] = task_id
        render_plan["remoteStatusPath"] = status_path
        render_plan["remoteStartedAt"] = time.time()
        render_plan["logs"].append({
            "phase": "render",
            "message": f"Remote task submitted: {task_id}",
            "timestamp": now_iso(),
        })
        # **立刻落库**，不等渲染结束。被 SIGTERM 的那一刻脚本没有任何机会写盘，
        # 所以 taskId 必须在这里就进 DB，否则下一次调用无从续起（这正是旧代码
        # 只写内存、`remoteTaskId` 全程没人读的后果）。
        _persist_plan_checkpoint(render_plan, job_id, private_token)

    last_progress = -1.0
    # 续跑时用提交时刻算 elapsed，否则 ETA 会按"刚开始"估，一上来就偏小。
    render_start_time = float(render_plan.get("remoteStartedAt") or time.time())

    def _on_progress(status_data: dict):
        nonlocal last_progress
        progress = float(status_data.get("progress") or 0.0)
        # 门槛设小(2%)：短渲染服务端进度增量小，太大的门槛会把中间档全吞掉，只剩 0%→100%
        if progress - last_progress >= 0.02 or progress >= 1.0:
            pct = int(progress * 100)
            elapsed = time.time() - render_start_time
            eta_remaining = 0.0
            if progress > 0.05 and progress < 1.0:
                eta_total = elapsed / progress
                eta_remaining = eta_total - elapsed
                eta_min = int(eta_remaining) // 60
                eta_sec = int(eta_remaining) % 60
                LogPrint(f"   🎞️  Remote render progress: {pct}% (eta {eta_min}m{eta_sec:02d}s)", file=sys.stderr)
            else:
                LogPrint(f"   🎞️  Remote render progress: {pct}%", file=sys.stderr)
            sys.stderr.flush()
            # 结构化进度，供上层解析并转发到前端
            print(
                json.dumps({
                    "__progress__": True,
                    "phase": "render-remote",
                    "progress": round(progress, 4),
                    "etaSeconds": int(eta_remaining) if eta_remaining > 0 else None,
                    "elapsedSeconds": int(elapsed),
                    "taskId": task_id,
                }, ensure_ascii=False),
                flush=True,
            )
            last_progress = progress

    try:
        result = remote_renderer_client.poll_render(
            task_id,
            private_token=private_token,
            timeout=poll_timeout,
            interval=poll_interval,
            on_progress=_on_progress,
            adaptive_interval=True,
            status_path=status_path,
        )
    except remote_renderer_client.RemoteRenderTimeout as exc:
        # 等不动了，但任务**没失败**。保留 taskId，让下一次调用接着轮询；
        # 状态不写 failed（写了模型会当成"渲染失败"而重新走一遍完整流程）。
        LogPrint(f"⏳ {exc}", file=sys.stderr)
        render_plan["status"] = "rendering"
        _persist_plan_checkpoint(render_plan, job_id, private_token)
        render_plan["logs"].append({
            "phase": "render",
            "message": f"poll timeout, task still running: {task_id}",
            "timestamp": now_iso(),
        })
        print(
            f"\n⏳ 远端渲染仍在进行（taskId={task_id}）。"
            f"用同一个 job_id 再调一次 render_video 即可续上，不会重新渲染。",
            flush=True,
        )
        return False
    except Exception as exc:
        LogPrint(f"❌ remote render polling failed: {exc}", file=sys.stderr)
        render_plan["status"] = "failed"
        # 任务确实不可用了（失败 / 查不到）。清掉 taskId，否则下一次调用会一直
        # 去续一个永远好不了的任务，再也提交不了新的。
        render_plan.pop("remoteTaskId", None)
        render_plan.pop("remoteStatusPath", None)
        render_plan.pop("remoteStartedAt", None)
        _persist_plan_checkpoint(render_plan, job_id, private_token)
        render_plan["errors"].append({
            "phase": "render",
            "message": f"remote poll_render failed: {exc}",
            "timestamp": now_iso(),
        })
        return False

    file_url = (result or {}).get("fileUrl") or (result or {}).get("videoUrl")
    if not file_url:
        LogPrint(f"❌ remote render finished but did not return a fileUrl: {result}", file=sys.stderr)
        render_plan["status"] = "failed"
        render_plan["errors"].append({
            "phase": "render",
            "message": f"remote render returned no fileUrl: {result}",
            "timestamp": now_iso(),
        })
        return False

    # 提取 VOD 元数据
    video_id = (result or {}).get("videoId", "")
    file_id = (result or {}).get("fileId")

    # 提取封面（可选；ab-render 在视频成功后串行渲一帧静态封面）
    # 失败不影响视频本身的成败——cover_url 为空则上游可以选择回退到占位图
    # 或重试。
    cover_url = (result or {}).get("coverUrl")
    cover_error = (result or {}).get("coverError")

    render_plan["status"] = "completed"
    render_plan["upload"] = {
        "fileUrl": file_url,
        "title": effective_title,
        "uploadedAt": now_iso(),
        "source": "remote-renderer",
        "videoId": video_id,
        "fileId": file_id,
    }
    if cover_url:
        render_plan["upload"]["coverUrl"] = cover_url
    if cover_error and not cover_url:
        # 视频成功但封面失败：降级为非致命 log，便于上游看到原因
        render_plan["logs"].append({
            "phase": "render",
            "message": f"Cover render failed (non-fatal): {cover_error}",
            "timestamp": now_iso(),
        })
    render_plan["logs"].append({
        "phase": "render",
        "message": f"Remote render completed: {file_url}",
        "timestamp": now_iso(),
    })
    LogPrint(f"   ✅ remote render finished: {file_url}", file=sys.stderr)
    if cover_url:
        LogPrint(f"   🖼️  cover image: {cover_url}", file=sys.stderr)
    elif cover_composition_id:
        LogPrint(
            f"   ⚠️  封面图未返回（cover_composition_id={cover_composition_id}, "
            f"error={cover_error or 'unknown'}）",
            file=sys.stderr,
        )

    # 如果是 vod:// 地址，轮询后端获取实际播放 URL
    if file_url.startswith("vod://") and file_id and private_token:
        playback_url = _poll_vod_playback_url(file_id, private_token)
        if playback_url:
            render_plan["upload"]["playbackUrl"] = playback_url
            LogPrint(f"   🎬 VOD playback URL: {playback_url}", file=sys.stderr)
        else:
            LogPrint(f"   ⚠️  VOD playback URL not ready yet (fileId={file_id}); query later via /file/get", file=sys.stderr)

    return True


def _poll_vod_playback_url(
    file_id: int,
    private_token: str,
    *,
    max_wait: float = 60.0,
    interval: float = 3.0,
):
    """Backwards-compat wrapper — implementation lives in ``_vod_polling`` module."""
    from _vod_polling import poll_vod_playback_url
    return poll_vod_playback_url(file_id, private_token, max_wait=max_wait, interval=interval)


def auto_bind_template(dsl: dict, template_id: str) -> dict:
    """Auto-generate TemplateBinding from DSL + template-id by importing
    match_template.py's binding logic. Eliminates the need for a separate
    match_template.py invocation step."""
    # template-registry/scripts is on sys.path via the module-top setup.
    try:
        import match_template
    except ImportError as e:
        LogPrint(f"❌ failed to import the match_template module: {e}", file=sys.stderr)
        LogPrint(f"   confirm the shared-lib skill exists at: {_SHARED_LIB_DIR}", file=sys.stderr)
        sys.exit(1)

    templates = match_template.load_registry()
    if not templates:
        LogPrint("❌ no available templates; check the video_dsl/templates/ directory", file=sys.stderr)
        sys.exit(1)

    selected = None
    for tpl in templates:
        if tpl.get("templateId") == template_id:
            selected = tpl
            break
    if not selected:
        available = ", ".join(t.get("templateId", "") for t in templates)
        LogPrint(f"❌ template not found: {template_id} (available: {available})", file=sys.stderr)
        sys.exit(1)

    # 选型门控在这里落地，而不是在上面的查表上：查表刻意用未门控的列表，因为绑定
    # 必须读到模板真实的定义（slotMapping / compositions）。门控管的是"允许不允许
    # 选中它"，而 auto_bind 正是"一个 id 变成本次成片所用模板"的那一刻。
    #
    # dsl_validator 已经在更早的位置报同一件事，但它只看 meta.templateId；手写的
    # DSL 可能没有那个字段，只靠 --template-id 传进来 —— 这里补上那条路。
    try:
        from registry_loader import gated_status  # type: ignore
        _gated = gated_status(template_id)
    except Exception:
        _gated = None
    if _gated:
        LogPrint(
            f"❌ template '{template_id}' exists but its status is '{_gated}', so this "
            f"process is not allowed to select it. Set ENABLE_BETA_TEMPLATES=1 to admit "
            f"beta templates, or promote the template to stable.",
            file=sys.stderr,
        )
        sys.exit(1)

    selected = match_template.load_full_template(selected)
    LogPrint(f"📋 auto-bound template: {selected.get('name', '')} ({template_id})", file=sys.stderr)
    binding = match_template.build_binding(selected, dsl)
    from required_props import validate_required_props
    try:
        validate_required_props(selected, binding.get("bindings", []))
    except ValueError as exc:
        LogPrint(f"❌ {exc}", file=sys.stderr)
        sys.exit(2)
    return binding


def _log_render_plan_summary(render_plan: dict) -> None:
    """Print a one-line identity summary of a loaded RenderPlan.

    The two ways to feed an existing plan (``--render-plan <file>`` and
    ``--job-id <int>``) used to print only the *source* (path / id), never the
    *content*. When a stale ``output/render-plan.json`` from a previous, unrelated
    run got picked up, the mismatch was rendered silently. Surfacing
    templateId / title / duration / scene count here makes a wrong plan obvious
    at a glance before any render time is spent.
    """
    try:
        template_id = render_plan.get("templateId", "?")
        title = render_plan.get("title", "")
        cfg = render_plan.get("renderConfig", {}) or {}
        fps = cfg.get("fps") or 30
        total_frames = cfg.get("totalFrames")
        scenes = len(render_plan.get("timeline", []) or [])
        comp = (render_plan.get("remotionProps", {}) or {}).get("compositionId", "?")
        dur = f"{total_frames / fps:.1f}s/{total_frames}f" if total_frames else "?"
        LogPrint(
            f"   ↳ plan: templateId={template_id} composition={comp} "
            f"duration={dur} scenes={scenes}" + (f' title=\"{title}\"' if title else ""),
            file=sys.stderr,
        )
    except Exception:
        # A summary is a convenience, never a hard dependency of rendering.
        pass


def main():
    # 无论正常结束、sys.exit 还是中途失败，都在最后报一次积分账（无扣费时静默）。
    atexit.register(print_billing_footer)
    parser = argparse.ArgumentParser(
        description="Video render tool — DSL + TemplateBinding → Remotion video.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python render_video.py --dsl video.dsl.json --template-id html-slide
  python render_video.py --dsl video.dsl.json --template-id html-slide --resolve-only
  python render_video.py --render-plan video.render-plan.json
        """,
    )
    parser.add_argument("--dsl", help="Input DSL file path")
    parser.add_argument("--dsl-json", default=None, help="DSL JSON as an inline string (replaces --dsl; no file needed; preferred for multi-user concurrent flows)")
    parser.add_argument("--template-id", default=None, help="Template id (required unless --render-plan / --job-id is used)")
    parser.add_argument("--render-plan", help="Existing RenderPlan file path")
    parser.add_argument("--resolve-only", action="store_true", help="Only resolve assets; do not run the render")
    parser.add_argument("--skip-asset-resolve", action="store_true", help="Skip asset resolution")
    parser.add_argument("--save-render-plan", action="store_true", help="Save the RenderPlan to a file")
    parser.add_argument("--render-plan-output", default=None, help="RenderPlan output path")
    parser.add_argument("--job-id", type=int, default=None, help="Render job id (replaces --render-plan; loads the RenderPlan from the database)")
    parser.add_argument(
        "--save-job",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="Persist RenderPlan / Manifest to the database (on by default — emits a jobId every run); pass --no-save-job to disable. Auto-degrades to off when PRIV_TOKEN is missing.",
    )
    parser.add_argument("--asset-cache-dir", default=ASSET_CACHE_DIR, help="Asset cache directory")
    parser.add_argument("--max-asset-retries", type=int, default=3, help="Max retries per asset")
    parser.add_argument("--asset-timeout", type=int, default=300, help="Per-asset generation timeout (seconds)")
    parser.add_argument("--priv-token", default=None, help="PrivToken")
    parser.add_argument("--upload", action="store_true", default=True, help="Auto-upload to Alibaba Cloud OSS after rendering (on by default)")
    parser.add_argument("--no-upload", action="store_true", help="Skip the upload")
    parser.add_argument("--upload-title", default=None, help="Upload title (defaults to the DSL title or the filename)")
    parser.add_argument(
        "--renderer",
        choices=["local", "remote"],
        # 只有显式等于 local 才本地渲染；空串 / 拼写错误 / 未设一律 remote，
        # 避免 Nacos 里该键为空值时静默掉进 local 模式（本地渲染需容器内自带 renderer）。
        default=(os.environ.get("REMOTION_RENDER_MODE") or "remote").strip().lower(),
        help="Render mode: remote=call the remote renderer service (default); local=local Remotion CLI (needs Node.js 18+)",
    )
    parser.add_argument(
        "--remote-poll-timeout",
        type=float,
        default=float(os.environ.get("REMOTION_REMOTE_POLL_TIMEOUT", "2700")),
        # 实测最长的一支约 30 分钟，加上 ab-render 排队要留余量，所以是 45 分钟。
        # 这个值必须小于 ab-agent 侧 render_video 的 skill 超时（默认 50 分钟），
        # 否则外层先 SIGTERM，这里这套"超时但保留 taskId"的续跑逻辑根本轮不到执行。
        help="Remote-render polling timeout (seconds, default 2700)",
    )
    parser.add_argument(
        "--remote-poll-interval",
        type=float,
        default=float(os.environ.get("REMOTION_REMOTE_POLL_INTERVAL", "2")),
        help="Remote-render polling interval (seconds, default 2)",
    )
    parser.add_argument(
        "--stub-image-url",
        default=None,
        help="Test mode: every source=gen-image asset is short-circuited to this URL, no gen-image API call (env STUB_IMAGE_URL works too, but the explicit CLI flag is preferred to avoid cross-session leakage).",
    )
    parser.add_argument(
        "--stub-video-url",
        default=None,
        help="Test mode: every source=gen-video asset is short-circuited to this URL, no gen-video API call (env STUB_VIDEO_URL works too, but the explicit CLI flag is preferred to avoid cross-session leakage).",
    )

    args = parser.parse_args()

    private_token = args.priv_token or os.environ.get("PRIV_TOKEN", "")
    # The registry loader shares the environment-based auth entry point. Keep
    # explicit CLI credentials consistent with render_job_client's token.
    if args.priv_token:
        os.environ["PRIV_TOKEN"] = private_token

    # --save-job 默认开启，但若没有 PRIV_TOKEN 则自动降级为关闭，避免阻塞无 token 的本地调试。
    if args.save_job and not private_token:
        LogPrint("ℹ️  PRIV_TOKEN not configured; --save-job auto-disabled (will not write to database)", file=sys.stderr)
        args.save_job = False

    if args.job_id is not None and args.job_id > 0:
        # ── Load RenderPlan from the database ─────────────────────────────
        LogPrint(f"📋 loading RenderPlan from database (jobId={args.job_id})...", file=sys.stderr)
        if not private_token:
            LogPrint("❌ --job-id mode requires PRIV_TOKEN or --priv-token", file=sys.stderr)
            sys.exit(1)
        # render_job_client lives in skills/template-registry/scripts/ (the shared
        # location for cross-skill helpers); it is already on sys.path via the
        # module-top setup.
        from render_job_client import get_plan as rjc_get_plan
        try:
            render_plan_str = rjc_get_plan(args.job_id, private_token)
        except RuntimeError as e:
            LogPrint(f"❌ failed to fetch RenderPlan: {e}", file=sys.stderr)
            sys.exit(1)
        render_plan = json.loads(render_plan_str)
        LogPrint(f"✅ RenderPlan loaded (jobId={args.job_id})", file=sys.stderr)
        _log_render_plan_summary(render_plan)
    elif args.render_plan:
        if not os.path.exists(args.render_plan):
            LogPrint(f"❌ file does not exist: {args.render_plan}", file=sys.stderr)
            sys.exit(1)
        render_plan = load_json(args.render_plan)
        LogPrint(f"📋 loaded existing RenderPlan: {args.render_plan}", file=sys.stderr)
        _log_render_plan_summary(render_plan)
        # 陈旧文件防护：output/render-plan.json 等共享文件名常被上一次别的项目
        # 的运行残留覆盖（output/ 是 gitignored 调试目录）。当用户同时给了 --dsl
        # 用以表明"我想渲这个 DSL"，但磁盘上的 plan 比 DSL 还旧时，几乎可以肯定
        # 加载到的是过期 plan —— 明确警告而不是静默渲染错的东西。
        if args.dsl and os.path.exists(args.dsl):
            try:
                if os.path.getmtime(args.render_plan) < os.path.getmtime(args.dsl):
                    LogPrint(
                        f"⚠️  STALE RenderPlan? '{args.render_plan}' is OLDER than the DSL "
                        f"'{args.dsl}'. You may be rendering a leftover plan from a previous "
                        "run. Regenerate with --dsl ... --template-id ... (drop --render-plan) "
                        "if the summary above doesn't match what you expect.",
                        file=sys.stderr,
                    )
            except OSError:
                pass
    else:
        if not args.dsl and not args.dsl_json:
            LogPrint(
                "❌ render_video needs a DSL source. You provided neither --dsl/--dsl-json "
                "nor --render-plan/--job-id.\n"
                "   Typical flow: 1) gen_script → DSL skeleton  2) prepare_video_assets "
                "(returns a job_id)  3) render_video with job_id=<int>.\n"
                "   Or pass the DSL inline: render_video --dsl-json '<json>' --template-id <id>.",
                file=sys.stderr,
            )
            sys.exit(1)

        if not args.template_id:
            LogPrint(
                "❌ render_video needs --template-id when rendering from --dsl/--dsl-json "
                "(binding is computed in-memory from the template id; --binding is no longer supported).",
                file=sys.stderr,
            )
            sys.exit(1)

        # 解析 DSL：优先 --dsl-json（inline），其次 --dsl（文件路径）
        if args.dsl_json:
            try:
                # 规范化 LLM 可能输出的全角/中文标点 → 标准 ASCII，避免 JSON 解析失败
                _dsl_str = args.dsl_json
                # 中文引号 → ASCII 引号
                _dsl_str = _dsl_str.replace('\u201c', '"').replace('\u201d', '"')
                _dsl_str = _dsl_str.replace('\u2018', "'").replace('\u2019', "'")
                # 全角冒号/逗号/括号 → ASCII（最常见的 JSON 结构破坏者）
                _dsl_str = _dsl_str.replace('\uff1a', ':')   # ： → :
                _dsl_str = _dsl_str.replace('\uff0c', ',')   # ， → ,
                _dsl_str = _dsl_str.replace('\uff08', '(')   # （ → (
                _dsl_str = _dsl_str.replace('\uff09', ')')   # ） → )
                _dsl_str = _dsl_str.replace('\u3010', '[')   # 【 → [
                _dsl_str = _dsl_str.replace('\u3011', ']')   # 】 → ]
                _dsl_str = _dsl_str.replace('\uff3b', '[')   # ［ → [
                _dsl_str = _dsl_str.replace('\uff3d', ']')   # ］ → ]
                _dsl_str = _dsl_str.replace('\uff5b', '{')   # ｛ → {
                _dsl_str = _dsl_str.replace('\uff5d', '}')   # ｝ → }
                dsl = json.loads(_dsl_str)
            except json.JSONDecodeError as e:
                LogPrint(f"❌ --dsl-json parse failed: {e}", file=sys.stderr)
                sys.exit(1)
        else:
            if not os.path.exists(args.dsl):
                LogPrint(f"❌ file does not exist: {args.dsl}", file=sys.stderr)
                sys.exit(1)
            dsl = load_json(args.dsl)

        # 自动补全 version 字段（LLM 重构 DSL 时可能遗漏）
        if "version" not in dsl:
            dsl["version"] = "v1alpha1"
            LogPrint("⚠️  DSL missing version field; auto-filled with v1alpha1", file=sys.stderr)

        # 骨架检查：防止 gen_script 原始骨架被直接提交（音频旁白不一致 bug）。
        # agent.ts 已经在 MCP 层做了拦截，这里是第二道保险，兼容绕过 agent 直接调用的场景。
        # narration.text 是唯一权威来源，所以只检查 scenes[].audio.narration。
        _skeleton_offenders: list[str] = []
        _skel_markers = ("[骨架待填充]", "[skeleton placeholder]", "这是需要替换的占位文案")
        for _scene in dsl.get("scenes", []) or []:
            _nar = ((_scene.get("audio") or {}).get("narration") or {})
            _sid = _scene.get("id", "?")
            if _nar.get("needsFill") is True:
                _skeleton_offenders.append(f"scene {_sid}: narration.needsFill=true")
                continue
            _t = _nar.get("text") or ""
            if any(m in _t for m in _skel_markers):
                _skeleton_offenders.append(f"scene {_sid}: narration.text still has the skeleton marker")
        if _skeleton_offenders:
            LogPrint("❌ DSL narration is still the gen_script skeleton and was not replaced with real content; refusing to continue:", file=sys.stderr)
            for _off in _skeleton_offenders[:5]:
                LogPrint(f"   - {_off}", file=sys.stderr)
            if len(_skeleton_offenders) > 5:
                LogPrint(f"   ... ({len(_skeleton_offenders)} total)", file=sys.stderr)
            LogPrint(
                "   Fill audio.narration.text with the real narration for every scene "
                "and remove the needsFill field, then retry.",
                file=sys.stderr,
            )
            sys.exit(2)

        errors = validate_dsl(dsl)
        if errors:
            LogPrint("❌ DSL validation failed:", file=sys.stderr)
            for err in errors:
                LogPrint(f"   - {err}", file=sys.stderr)
            sys.exit(1)

        # binding 由 --template-id 现算，不再接受外部 binding 文件 / JSON。
        # 这是 P1.1 的关键简化：DSL → RenderPlan 之间不再有独立工件，
        # 直接 (DSL + templateId + registry) → RenderPlan。
        # binding 仅在内存中作为中间态使用，不落盘、不入库。需要排查时
        # 直接 print(binding) 即可。
        binding = auto_bind_template(dsl, args.template_id)

        LogPrint(f"📋 Building RenderPlan...", file=sys.stderr)
        render_plan = build_render_plan(dsl, binding)
        render_plan["_dsl_assets"] = dsl.get("assets", [])

    # Job/file mode bypasses auto_bind_template. Validate saved plans too, so a
    # previously accepted placeholder-only plan cannot silently render again.
    if args.job_id or args.render_plan:
        from registry_loader import get_template
        from required_props import validate_required_props
        try:
            template = get_template(render_plan.get("templateId", ""), include_all_statuses=True)
            if not template:
                raise ValueError("Cannot validate RenderPlan: template metadata unavailable")
            validate_required_props(template, render_plan.get("timeline", []))
        except ValueError as exc:
            LogPrint(f"❌ {exc}", file=sys.stderr)
            sys.exit(2)

    from video_dsl.runtime.payload_contract import plan_asset_errors
    reference_errors = plan_asset_errors(render_plan)
    if reference_errors:
        LogPrint("❌ RenderPlan asset references are invalid:", file=sys.stderr)
        for error in reference_errors[:10]:
            LogPrint(f"   - {error}", file=sys.stderr)
        sys.exit(2)

    if not args.skip_asset_resolve:
        # Stub mode: short-circuit gen-image / gen-video before counting pending.
        # CLI flag takes precedence; env vars act as fallback with a visible warning
        # so silent cross-session leakage is always observable.
        stub_image_url = args.stub_image_url
        stub_video_url = args.stub_video_url
        if stub_image_url is None:
            env_v = os.environ.get("STUB_IMAGE_URL", "")
            if env_v:
                LogPrint(f"⚠️  detected env var STUB_IMAGE_URL={env_v}; using it as the image stub URL (pass --stub-image-url explicitly or unset the env var)", file=sys.stderr)
            stub_image_url = env_v
        if stub_video_url is None:
            env_v = os.environ.get("STUB_VIDEO_URL", "")
            if env_v:
                LogPrint(f"⚠️  detected env var STUB_VIDEO_URL={env_v}; using it as the video stub URL (pass --stub-video-url explicitly or unset the env var)", file=sys.stderr)
            stub_video_url = env_v

        if stub_image_url or stub_video_url:
            stubbed = apply_stub_urls(
                render_plan,
                stub_image_url=stub_image_url,
                stub_video_url=stub_video_url,
            )
            if stubbed:
                LogPrint(f"🧪 Stub mode: {stubbed} asset(s) using a stub URL; generation API skipped", file=sys.stderr)
            else:
                LogPrint(f"🧪 Stub mode is on but no matching assets were found (no pending gen-image / gen-video resources)", file=sys.stderr)
        pending_count = sum(1 for a in render_plan["resolvedAssets"] if a["status"] == "pending")
        if pending_count > 0:
            # ── Token pre-check ───────────────────────────────────────────
            # Fail fast with a clear message instead of letting child
            # processes silently hang waiting for interactive token input.
            if not private_token:
                LogPrint("❌ asset generation needs an auth token, but PRIV_TOKEN env var is not set and --priv-token was not passed", file=sys.stderr)
                LogPrint("   configure it one of these ways:", file=sys.stderr)
                LogPrint("   1. export PRIV_TOKEN=<your-token>", file=sys.stderr)
                LogPrint("   2. python3 render_video.py --priv-token <your-token> ...", file=sys.stderr)
                sys.exit(1)
            # ──────────────────────────────────────────────────────────────
            LogPrint(f"🔍 Resolving {pending_count} missing asset(s)...", file=sys.stderr)
            render_plan = resolve_assets(
                render_plan,
                private_token=private_token,
                max_retries=args.max_asset_retries,
                timeout=args.asset_timeout,
            )
        else:
            LogPrint(f"✅ all assets ready", file=sys.stderr)
            render_plan["status"] = "assets-ready"

    render_plan.pop("_dsl_assets", None)

    if not args.skip_asset_resolve:
        render_plan = adjust_timeline_to_audio(render_plan)

    output_path = os.path.join(OUTPUT_DIR, "video.mp4")
    rp_output = args.render_plan_output or os.path.join(OUTPUT_DIR, "render-plan.json")

    # --save-job 模式下 RenderPlan 只存数据库，不写磁盘（避免多用户并发时互相覆盖同名文件）
    if args.save_render_plan or not args.save_job:
        validate_and_fix_render_plan(render_plan)
        save_json(render_plan, rp_output)
        LogPrint(f"💾 RenderPlan saved: {rp_output}", file=sys.stderr)

    # ── 保存 RenderPlan 到数据库 ──────────────────────────────────────────
    # 默认 args.save_job=True，所以无论 --resolve-only 还是直接渲染，都会落库；
    # 没有 PRIV_TOKEN 时已在 main 入口降级为 False。
    if args.save_job:
        # 入库前校验 + 自动修复
        validate_and_fix_render_plan(render_plan)

        from render_job_client import create_job as rjc_create_job, save_plan as rjc_save_plan
        render_plan_json_str = json.dumps(render_plan, ensure_ascii=False)
        # 若已有 job_id 则复用，否则新建
        _job_id = args.job_id
        if not _job_id:
            # 构造 DB 列 dsl_meta（轻量摘要供列表查询使用）。
            # 单一来源是 dsl.meta — render_plan 里只放渲染必需的最小集，不再缓存整段 meta。
            _locals = locals()
            _dsl_obj = _locals.get("dsl")
            _binding_obj = _locals.get("binding")
            _meta_summary: dict = {}
            # 直接渲染（--render-plan / --job-id）这种路径下没有 dsl 对象，
            # 退而用 render_plan 里的字段拼一份摘要，保证列表页有可读信息。
            if _dsl_obj:
                _src_meta = (_dsl_obj.get("meta") or {})
                for _k in ("title", "topic", "platform", "templateId", "targetDuration"):
                    _v = _src_meta.get(_k)
                    if _v:
                        _meta_summary[_k] = _v
                _ratio = (_dsl_obj.get("global") or {}).get("aspectRatio")
                if _ratio:
                    _meta_summary["aspectRatio"] = _ratio
            else:
                for _k in ("title", "templateId", "targetDuration"):
                    _v = render_plan.get(_k)
                    if _v:
                        _meta_summary[_k] = _v
                _rc = render_plan.get("renderConfig") or {}
                if _rc.get("width") and _rc.get("height"):
                    _meta_summary["resolution"] = f"{_rc['width']}x{_rc['height']}"
            if _binding_obj and isinstance(_binding_obj, dict):
                _tid = _binding_obj.get("templateId")
                if _tid and "templateId" not in _meta_summary:
                    _meta_summary["templateId"] = _tid
            _dsl_str = json.dumps(_dsl_obj, ensure_ascii=False) if _dsl_obj else ""
            try:
                _job_id = rjc_create_job(
                    private_token,
                    dsl=_dsl_str,
                    dsl_meta=_meta_summary or None,
                )
                LogPrint(f"✅ render job created: jobId={_job_id}", file=sys.stderr)
            except RuntimeError as e:
                LogPrint(f"⚠️  failed to create render job (--save-job auto-disabled, rendering continues): {e}", file=sys.stderr)
                args.save_job = False
                _job_id = None
        if args.save_job and _job_id:
            try:
                rjc_save_plan(_job_id, render_plan_json_str, private_token)
                LogPrint(f"✅ RenderPlan saved to database: jobId={_job_id}", file=sys.stderr)
                # 把回填的 job_id 暴露给后续 saveManifest 使用
                args.job_id = _job_id
                print(f"\n📦 render job jobId: {_job_id}")
            except RuntimeError as e:
                LogPrint(f"⚠️  failed to save RenderPlan to database (--save-job auto-disabled, rendering continues): {e}", file=sys.stderr)
                args.save_job = False

    if args.resolve_only:
        generated = sum(1 for a in render_plan["resolvedAssets"] if a["status"] == "generated")
        failed = sum(1 for a in render_plan["resolvedAssets"] if a["status"] == "failed")
        total = len(render_plan["resolvedAssets"])
        LogPrint(f"\n📊 Asset resolution finished (resolve-only mode): total {total}, generated {generated}, failed {failed}", file=sys.stderr)

        fps = render_plan.get("renderConfig", {}).get("fps", 30)
        target_dur = render_plan.get("targetDuration")
        actual_dur = render_plan.get("renderConfig", {}).get("totalDuration", 0)

        # ── 摘要输出到 stdout，供 Agent 读取资产 URL（不要改为 LogPrint/stderr）──
        print(f"\n📊 Asset resolution finished: total {total}, generated {generated}, failed {failed}")
        if target_dur:
            diff = actual_dur - target_dur
            sign = "+" if diff >= 0 else ""
            print(f"⏱  Actual duration: {actual_dur}s (target {target_dur}s, {sign}{diff:.1f}s)")
        else:
            print(f"⏱  Actual duration: {actual_dur}s")

        print(f"\n🎬 Scene timeline:")
        for entry in render_plan.get("timeline", []):
            scene_dur = round(entry.get("durationFrames", 0) / fps, 1)
            scene_id = entry.get("sceneId", "?")
            print(f"   {scene_id:<25} {scene_dur:>6.1f}s")

        audio_assets = [a for a in render_plan.get("resolvedAssets", [])
                        if a.get("type") == "audio" and a.get("status") == "generated"]
        if audio_assets:
            print(f"\n🔊 TTS audio:")
            for asset in audio_assets:
                dur_s = round((asset.get("duration") or 0) / 1000, 1)
                url = asset.get("url", "")
                print(f"   {asset['assetId']:<25} {dur_s:>6.1f}s  {url}")

        image_assets = [a for a in render_plan.get("resolvedAssets", [])
                        if a.get("type") == "image" and a.get("status") == "generated"]
        if image_assets:
            print(f"\n🖼  Image assets:")
            for asset in image_assets:
                url = asset.get("url", "")
                print(f"   {asset['assetId']:<25} {url}")
        # ─────────────────────────────────────────────────────────────────────

        if failed > 0:
            sys.exit(1)
        return

    if render_plan["status"] == "failed":
        LogPrint("❌ asset resolution had failures; cannot continue rendering", file=sys.stderr)
        if not args.save_job:
            save_json(render_plan, rp_output)
        sys.exit(1)

    render_mode = args.renderer
    upload_title_hint = (
        args.upload_title
        or render_plan.get("title")
        or os.path.splitext(os.path.basename(output_path))[0]
    )
    # 阿里云 VOD 与 ab-api DTO 都按"字符数"限制 title 长度（max=128）。
    # DSL 的 meta.title 经常是一段长描述（gen-script 把 topic 直接当 title），
    # 这里截断到 128，避免远程渲染跑完之后栽在 CreateUploadVideoToken 校验上。
    if upload_title_hint and len(upload_title_hint) > 128:
        upload_title_hint = upload_title_hint[:128]
    conversation_id = os.environ.get("CONVERSATION_ID") or None
    if render_mode == "remote":
        LogPrint(f"🌐 Using remote render mode (MM_API_BASE_URL)", file=sys.stderr)
        if not private_token:
            LogPrint("❌ remote render requires PRIV_TOKEN or --priv-token", file=sys.stderr)
            sys.exit(1)
        success = render_with_remote_api(
            render_plan,
            output_path,
            private_token=private_token,
            poll_timeout=args.remote_poll_timeout,
            poll_interval=args.remote_poll_interval,
            upload_title=upload_title_hint,
            conversation_id=conversation_id,
            job_id=args.job_id if args.save_job else None,
        )
    else:
        LogPrint(f"💻 Using local render mode", file=sys.stderr)
        render_plan["renderMode"] = "local"
        success = render_with_local_cli(render_plan, output_path)

    # save-job 模式下 RenderPlan 只入库；非 save-job 才写共享磁盘
    if not args.save_job:
        save_json(render_plan, rp_output)

    manifest = {
        "outputPath": output_path,
        "renderPlanPath": "" if args.save_job else rp_output,
        "renderConfig": render_plan["renderConfig"],
        "status": render_plan["status"],
        "renderMode": render_mode,
        "createdAt": render_plan["createdAt"],
        "completedAt": now_iso(),
        "assetCount": len(render_plan["resolvedAssets"]),
        "sceneCount": len(render_plan["timeline"]),
        "errorCount": len(render_plan["errors"]),
    }
    # --save-job 模式下 Manifest 只存数据库，不写磁盘（避免多用户并发时互相覆盖同名文件）
    manifest_path = None if args.save_job else os.path.join(OUTPUT_DIR, "render-manifest.json")
    if manifest_path:
        save_json(manifest, manifest_path)

    if success:
        LogPrint(f"\n🎉 Video render finished!", file=sys.stderr)
        if render_mode == "remote":
            upload_info = render_plan.get("upload") or {}
            remote_url = upload_info.get("fileUrl", "")
            playback_url = upload_info.get("playbackUrl", "")
            LogPrint(f"   Remote video: {remote_url}", file=sys.stderr)
            if playback_url:
                LogPrint(f"   Playback URL: {playback_url}", file=sys.stderr)
            # Print the playable video URL to stdout; prefer the transcoded playbackUrl.
            effective_playback = playback_url or remote_url
            if effective_playback and not effective_playback.startswith("vod://"):
                print(f"\n🎬 Video playback URL: {effective_playback}")

                # 结构化资产标记 —— 宿主（ab-agent）据此把权威 URL 作为 attachment
                # 下发前端渲染播放器，绕开模型正文。
                #
                # 为什么必须绕开：模型复述 32 位不透明 hex 会出错。2026-08-30 实测一次
                # 成片链接被吞掉一个字符（…c1c20102 → …c1c2012），用户拿到 404，
                # 而文件本身好好地在 CDN 上。约定与 web-screenshot/scripts/record.py
                # 的 __web_record_asset__ 一致。
                #
                # 只在拿到**非 vod:// 的真实播放地址**时才打标记（上面的 if 已保证）：
                # playbackUrl 来自 VOD 轮询，可能超时未就绪。宁可前端没有内联播放器
                # （用户仍可从成片面板看），也不要再给出一条不可用的地址。
                asset = {"url": effective_playback}
                cover = upload_info.get("coverUrl")
                if cover:
                    asset["coverUrl"] = cover
                total_duration = (render_plan.get("renderConfig") or {}).get("totalDuration")
                if isinstance(total_duration, (int, float)) and total_duration > 0:
                    asset["durationSec"] = total_duration
                print("__render_video_asset__ " + json.dumps(asset, ensure_ascii=False))
        else:
            LogPrint(f"   Video: {output_path}", file=sys.stderr)
        if args.save_job:
            LogPrint(f"   RenderPlan: persisted to database (jobId={args.job_id})", file=sys.stderr)
        else:
            LogPrint(f"   RenderPlan: {rp_output}", file=sys.stderr)
        if manifest_path:
            LogPrint(f"   Manifest: {manifest_path}", file=sys.stderr)

        if render_mode == "remote":
            # In remote mode the fileUrl came from the server; reuse it into the manifest.
            manifest["upload"] = render_plan.get("upload", {})
            if manifest_path:
                save_json(manifest, manifest_path)
                LogPrint(f"   Manifest updated (remote fileUrl): {manifest_path}", file=sys.stderr)
        elif args.upload and not args.no_upload:
            try:
                from upload_video import upload_media_file

                upload_title = upload_title_hint
                LogPrint(f"\n📤 Uploading video to OSS...", file=sys.stderr)
                upload_result = upload_media_file(file_path=output_path, title=upload_title)
                manifest["upload"] = {
                    "fileUrl": upload_result["fileUrl"],
                    "title": upload_title,
                    "uploadedAt": now_iso(),
                }
                if manifest_path:
                    save_json(manifest, manifest_path)
                    LogPrint(f"   Manifest updated: {manifest_path}", file=sys.stderr)
            except Exception as e:
                LogPrint(f"\n⚠️  upload failed (the video stays local): {e}", file=sys.stderr)

        # ── Save Manifest to the database ────────────────────────────────
        if args.save_job and args.job_id:
            from render_job_client import save_manifest as rjc_save_manifest
            try:
                rjc_save_manifest(args.job_id, manifest, private_token)
                LogPrint(f"✅ Manifest saved to database: jobId={args.job_id}", file=sys.stderr)
            except RuntimeError as e:
                LogPrint(f"⚠️  failed to save Manifest to database (local result unaffected): {e}", file=sys.stderr)
    else:
        LogPrint(f"\n❌ render failed, see RenderPlan for details: {rp_output}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()