xiaoxia-saas/apps/worker/video_processing/unified_render_service.py

"""统一渲染引擎 — 输入 EditPlan + EditPlanClips，按时间线+图层渲染视频.

核心原则（灵应）：渲染引擎是统一的，不判断模式，只按 clip_type/config.role
分组为图层再合成。

图层分组：
  main (无 config.role)     → main (z=0)
  main + config.role=b_roll → broll (z=0，与 main 同层替换)
  overlay                   → overlay (z=1，画中画叠加)
  background                → background (z=0，全屏底图)
  corner_voice              → corner_voice (z=1，右上角小窗)
  b_roll                    → broll (z=0)
  intro / outro             → main (z=0，按 order 排在首/尾)

合成流程：
  1. 每个 clip 先 trim + scale + setpts 预处理
  2. 同层 clips 按 order 用 xfade 串联
  3. overlay/corner_voice 层 overlay 到主层
  4. 如有独立音频轨，amix 混入
"""

from __future__ import annotations

import logging
import subprocess
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from video_processing.color_grade_engine import ColorGradeConfig, ColorGradeEngine
from video_processing.ffmpeg_utils import (
    DEFAULT_FPS,
    DEFAULT_OUTPUT_HEIGHT,
    DEFAULT_OUTPUT_WIDTH,
    DEFAULT_TRANSITION_DURATION,
    FFMPEG_BIN,
    FFMPEG_ENCODE_CRF,
    FFMPEG_ENCODE_PRESET,
    FFMPEG_ENCODE_THREADS,
    probe_duration,
    probe_has_audio,
    probe_video_info,
    run_ffmpeg,
)
from video_processing.intro_outro_engine import IntroOutroConfig, IntroOutroEngine
from video_processing.pip_engine import PiPConfig, PiPEngine, PiPLayerConfig
from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio
from video_processing.render_subtitles import generate_ass_subtitles
from video_processing.reverse_engine import ReverseConfig, ReverseEngine
from video_processing.speed_engine import SpeedEngine
from video_processing.sticker_engine import StickerEngine
from video_processing.subtitle_generator import generate_ass_from_timeline
from video_processing.transition_engine import TransitionEngine
from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config
from video_processing.tts_engine import TtsEngine
from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine

from packages.domain.render_layer_utils import LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX
from packages.domain.render_layer_utils import clip_adjusted_duration as _clip_adjusted_duration_pure
from packages.domain.render_layer_utils import clip_effective_duration as _clip_effective_duration_pure
from packages.domain.render_layer_utils import clip_playback_speed as _clip_playback_speed_pure
from packages.domain.render_layer_utils import estimate_total_duration as _estimate_total_duration_pure
from packages.domain.render_layer_utils import resolve_layer_role as _resolve_layer_role_pure
from packages.domain.tts_config import TtsConfig
from packages.shared.gpu_encoder import GpuEncodeError, get_gpu_encoder

logger = logging.getLogger(__name__)


# ── 数据结构 ──────────────────────────────────────────────────────────────────


@dataclass
class ResolvedClip:
    """已解析到本地路径的片段。"""

    clip_id: str
    asset_id: str
    local_path: Path
    clip_type: str
    order: int
    start_time: float = 0.0
    duration: float = 0.0  # 0 表示使用素材完整时长
    transition_effect: str = "cut"
    transition_duration: float = 0.0  # 0 表示使用全局默认值
    playback_speed: float = 1.0  # 0 或 1.0 表示原速
    config: dict[str, Any] = field(default_factory=dict)

    # 运行时填充
    actual_duration: float = 0.0  # 素材实际时长（probe 后填充）
    trim_config: TrimConfig | None = None  # 解析后的裁剪配置（运行时填充）


@dataclass
class RenderLayer:
    """渲染图层。"""

    role: str  # "main" | "overlay" | "pip" | "background" | "corner_voice" | "broll" | "audio"
    clips: list[ResolvedClip] = field(default_factory=list)
    z_index: int = 0
    opacity: float = 1.0
    position: tuple[int, int] | None = None  # (x, y) 偏移，None 表示全屏


@dataclass
class RenderResult:
    """渲染结果。"""

    output_path: Path
    duration: float
    file_size: int
    width: int
    height: int


# ── clip_type → layer role 映射 ──────────────────────────────────────────────


def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str:
    """根据 clip_type 和 config.role 确定图层角色（向后兼容别名）。

    实际实现移至 packages.domain.render_layer_utils.resolve_layer_role。
    """
    return _resolve_layer_role_pure(clip_type, config)


# ── 图层默认 z_index ─────────────────────────────────────────────────────────

_LAYER_Z_INDEX: dict[str, int] = _IMPORTED_LAYER_Z_INDEX

# 图层默认 PiP 位置（相对输出画布的偏移）
_PIP_SCALE = 0.25  # PiP 占主画面的比例


# ── 统一渲染引擎 ─────────────────────────────────────────────────────────────


class UnifiedRenderService:
    """统一渲染引擎。

    输入 EditPlan + EditPlanClips + 素材路径映射，按时间线+图层执行渲染。
    """

    def __init__(
        self,
        plan: Any,  # EditPlan
        clips: list[Any],  # list[EditPlanClip]
        asset_path_map: dict[str, Path],  # asset_id → local_path
        work_dir: Path,
        *,
        output_width: int = DEFAULT_OUTPUT_WIDTH,
        output_height: int = DEFAULT_OUTPUT_HEIGHT,
        output_fps: int = DEFAULT_FPS,
        transition_duration: float = DEFAULT_TRANSITION_DURATION,
        asr_service: Any = None,  # ASRService 实例，用于自动生成字幕
        bgm_path: str | None = None,  # BGM 本地文件路径
        voiceover_audio_path: str | None = None,  # 配音素材库音频本地路径
        clip_has_text: list[bool] | None = None,  # 源视频片段是否有文字（来自 atom_clip.ai_tags.has_text）
    ):
        self.plan = plan
        self.clips = clips
        self.asset_path_map = asset_path_map
        self.work_dir = work_dir
        self.output_width = output_width
        self.output_height = output_height
        self.output_fps = output_fps
        self.transition_duration = transition_duration
        self.asr_service = asr_service
        self.bgm_path = bgm_path
        self.voiceover_audio_path = voiceover_audio_path
        # #1970：片段级文字检测（顺序与非 audio 的源视频片段一致）；None 表示无可靠检测，保守不翻转
        self._clip_has_text = clip_has_text
        self._transition_engine = TransitionEngine(default_duration=transition_duration)
        self._speed_engine = SpeedEngine()
        self._asr_timeline_cache: Any = None  # ASR 字幕结果缓存，避免重复调用
        self._asr_timeline_cached = False
        # #1970 PR2：片段级微变换计划缓存（懒构建，dedup_enabled=False 时为 None）
        self._micro_plan_cache: Any = None
        self._micro_plan_loaded = False

    # ── #1970 PR2 智能降重：片段级微变换 ───────────────────────────────────
    def _dedup_enabled(self) -> bool:
        """读取 plan.config.dedup_enabled，缺省视为 True（向后兼容）。"""
        cfg = self.plan.config or {}
        return bool(cfg.get("dedup_enabled", True))

    def _get_micro_transform_plan(self, clip_count: int) -> Any:
        """按 task_id+视频序号构建可复现的片段级微变换计划。

        种子 hash(generation_task_id + video_index)%10000，同一任务重渲结果一致。
        dedup_enabled=False 时返回 None，调用方不注入任何微变换。
        hflip 放开（#1970）：clip_has_text 来自 atom_clip.ai_tags.has_text，
        仅 AI 明确判定无文字的片段可参与 50% 翻转；未打标签 / has_text 为
        true/null 或缺位时一律视为有文字，保持保守不翻转。
        """
        if self._micro_plan_loaded:
            return self._micro_plan_cache
        self._micro_plan_loaded = True
        if not self._dedup_enabled() or clip_count <= 0:
            self._micro_plan_cache = None
            return None
        try:
            from video_processing.micro_transform_pure import build_micro_transform_plan

            cfg = self.plan.config or {}
            task_id = str(cfg.get("generation_task_id", "") or "")
            video_index = int(cfg.get("video_index", 0) or 0)
            # self._clip_has_text 顺序与非 audio 源片段一致；
            # None（未提供检测，如内存直渲/旧任务）→ 纯函数层按全有文字保守处理；
            # 列表短于片段数时缺位片段同样按有文字处理
            self._micro_plan_cache = build_micro_transform_plan(
                task_id,
                video_index,
                clip_count,
                clip_has_text=self._clip_has_text,
                enable_bgm_offset=bool(cfg.get("bgm")),
            )
        except Exception as e:
            logger.warning("[unified-render] 微变换计划构建失败，本次不注入: %s", e)
            self._micro_plan_cache = None
        return self._micro_plan_cache

    @staticmethod
    def _apply_micro_transform_video(filters: list[str], mt: Any) -> None:
        """把片段视频微变换就地追加到 filter 链（post-scale 阶段调用）。

        顺序：hflip 在 pre-scale 阶段由 _apply_micro_hflip 处理，这里只加
        eq 亮度/对比度/饱和度。速度 setpts 与既有 clip speed 相乘（见调用点），
        避免出现两条 setpts 互相覆盖。
        """
        if mt is None:
            return
        if abs(mt.brightness) > 1e-4 or abs(mt.contrast - 1.0) > 1e-4 or abs(mt.saturation - 1.0) > 1e-4:
            filters.append(
                f"eq=brightness={mt.brightness:+.4f}:contrast={mt.contrast:.4f}:saturation={mt.saturation:.4f}"
            )

    @staticmethod
    def _apply_micro_hflip(filters: list[str], mt: Any) -> None:
        """片段级水平翻转（pre-scale 阶段）。P1 有文字/无法判定时 mt.hflip=False。"""
        if mt is not None and mt.hflip and not mt.has_text:
            filters.append("hflip")

    @staticmethod
    def _micro_speed_factor(mt: Any) -> float:
        """片段微变换速度因子（0.97~1.03），无计划返回 1.0。"""
        if mt is None:
            return 1.0
        return float(getattr(mt, "speed", 1.0) or 1.0)

    def _get_micro_bgm_offset(self) -> float:
        """#1970 PR2：读取本视频 BGM 起始偏移（秒），无 BGM/禁用时为 0。"""
        if not self.plan.config:
            return 0.0
        try:
            count = len([c for c in (self.plan.clips or []) if getattr(c, "clip_type", "main") != "audio"])
            plan = self._get_micro_transform_plan(count)
            if plan:
                return round(float(plan.bgm_start_offset or 0.0), 3)
        except Exception:
            logger.debug("微变换 BGM 偏移读取失败，按 0 处理: plan_id=%s", getattr(self.plan, "id", "?"))
        return 0.0

    def render(self) -> RenderResult:
        """执行渲染，返回 RenderResult.

        优化路径：
        - 单图层单 clip → 直通模式（-vf），性能最优
        - 其他情况 → 完整 filter_complex 渲染

        字幕渲染流程：
        1. 视频主渲染（直通或完整链路）
        2. 如有 title/subtitle，叠加 ASS 字幕

        音频后处理：
        1. 主图层音频 concat 拼接
        2. 独立音频轨 amix 混入
        3. 合并到输出视频

        Raises:
            ValueError: 没有可渲染的片段时抛出
        """
        t_start = time.time()

        # 1. 解析 clips → ResolvedClips（跳过无素材的 clip）
        resolved = self._resolve_clips()
        if not resolved:
            raise ValueError("没有可渲染的片段（所有片段素材缺失或下载失败）")

        # 2. 分组为 RenderLayers
        layers = self._group_clips_into_layers(resolved)

        # 2.5 配音时长对齐：如果有配音素材，调整片段时长以匹配配音时长
        voice_duration = self._get_voice_audio_duration()
        if voice_duration > 0:
            self._align_clips_to_voice_duration(layers, voice_duration)

        # 3. 计算视频总时长（用于字幕显示时长）
        video_duration_final = self._estimate_total_duration(layers)
        # Debug: 输出各图层时长明细
        for layer in layers:
            layer_total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in layer.clips)
            clip_details = [
                f"{c.clip_id}(dur={c.duration:.3f},actual={c.actual_duration:.3f},speed={getattr(c, 'playback_speed', 1.0):.4f})"
                for c in layer.clips
            ]
            logger.info(
                "[debug] layer=%s clips=%d total=%.3f transition_duration=%.3f details=%s",
                layer.role,
                len(layer.clips),
                layer_total,
                self.transition_duration,
                ", ".join(clip_details),
            )
        logger.info("[debug] estimated video_duration=%.3f", video_duration_final)

        # 3.5 TTS 配音生成（如果配置了）
        self._maybe_add_voiceover_layer(layers, video_duration=video_duration_final)

        # 3.6 配音素材库音频（如果传入了本地路径）
        self._maybe_add_voice_library_layer(layers, video_duration=video_duration_final)

        # 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置）
        ass_path = self._maybe_generate_ass(video_duration_final)

        # 4.5 解析画中画配置
        pip_config = PiPConfig.from_dict((self.plan.config or {}).get("pip_config"))
        pip_sources = self._resolve_pip_sources(pip_config) if pip_config.enabled else []
        has_pip = len(pip_sources) > 0

        # 灰度埋点：开始渲染
        layer_roles = [layer.role for layer in layers]
        clip_counts = {layer.role: len(layer.clips) for layer in layers}
        logger.info(
            "[unified-render] start render: plan_id=%s clip_count=%d layers=%s clip_counts=%s pip_layers=%d",
            self.plan.id,
            len(resolved),
            layer_roles,
            clip_counts,
            len(pip_sources),
        )

        # 4.8 全 GPU 直连管线（P1）：命中主流场景则跳过 mezzanine/边缘裁剪 CPU 重编码
        output_path = self.work_dir / f"rendered_{self.plan.id}.mp4"
        direct_result = self._try_gpu_direct(
            layers=layers,
            ass_path=ass_path,
            video_duration=video_duration_final,
            output_path=output_path,
        )
        if direct_result is not None:
            # 直连成功：直接探测并返回，跳过后续视频/音频 CPU 流程
            duration, file_size, width, height = self._probe_output(output_path)
            logger.info(
                "[unified-render] gpu-direct done: plan_id=%s total_ms=%d output_size=%d resolution=%dx%d",
                self.plan.id,
                int((time.time() - t_start) * 1000),
                file_size,
                width,
                height,
            )
            return RenderResult(
                output_path=output_path,
                duration=duration,
                file_size=file_size,
                width=width,
                height=height,
            )

        # 5. 视频主渲染
        t_video_start = time.time()
        video_only_path = self.work_dir / f"rendered_{self.plan.id}_video.mp4"
        output_path = self.work_dir / f"rendered_{self.plan.id}.mp4"

        # 有画中画时不走直通（需要额外图层叠加）
        is_pass_through = self._can_use_pass_through(layers) and not has_pip
        pass_through_has_audio = False
        used_stream_copy = False

        if is_pass_through:
            # 先尝试 stream copy 优化（无重编码，性能提升 10 倍+）
            # 条件不满足或失败时回退到带滤镜的直通渲染
            stream_copy_ok = self._try_render_stream_copy(
                layers, output_path, ass_path=ass_path, video_duration=video_duration_final
            )
            if stream_copy_ok:
                used_stream_copy = True
                # stream copy 模式下，直接探测输出是否有音频
                clip = layers[0].clips[0]
                info = probe_video_info(str(clip.local_path))
                pass_through_has_audio = info.get("has_audio", True)
            else:
                # 回退到带滤镜的直通渲染
                pass_through_has_audio = self._render_pass_through(
                    layers,
                    output_path,
                    ass_path=ass_path,
                    video_duration=video_duration_final,
                )
        else:
            filter_complex, input_args = self._build_filter_complex(layers, ass_path=ass_path)

            # 追加画中画滤镜
            if has_pip:
                filter_complex, input_args = self._append_pip_filters(filter_complex, input_args, pip_sources)

            self._execute_ffmpeg(filter_complex, input_args, video_only_path)

        t_video_end = time.time()
        video_render_ms = int((t_video_end - t_video_start) * 1000)
        logger.info(
            "[unified-render] video render done: plan_id=%s duration_ms=%d pass_through=%s stream_copy=%s",
            self.plan.id,
            video_render_ms,
            is_pass_through,
            used_stream_copy,
        )

        # 6. 音频后处理混音（直通场景已合并处理，跳过）
        t_audio_start = time.time()
        audio_mix_ms = 0
        has_audio = False

        if is_pass_through:
            # 直通场景已在一次调用中完成视频+音频
            has_audio = pass_through_has_audio
            # 直通模式下也支持 BGM 混音：提取音频 → 混 BGM → 合并回视频
            if self.bgm_path and pass_through_has_audio:
                config = self.plan.config or {}
                bgm_config = config.get("bgm", {}) or {}
                if bgm_config.get("enabled", False):
                    ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id)
                    from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main

                    _bgm_off = self._get_micro_bgm_offset()
                    if _bgm_off and not (bgm_config or {}).get("audio_offset"):
                        bgm_config = {**bgm_config, "audio_offset": _bgm_off}
                    bgm_cfg = BGMConfig.from_config_dict(self.bgm_path, bgm_config)
                    # 从直通输出中提取音频
                    main_audio_path = self.work_dir / f"pass_through_audio_{self.plan.id}.aac"
                    extract_cmd = [
                        FFMPEG_BIN,
                        "-y",
                        "-i",
                        str(output_path),
                        "-vn",
                        "-acodec",
                        "aac",
                        "-b:a",
                        "128k",
                        str(main_audio_path),
                    ]
                    try:
                        from video_processing.ffmpeg_utils import run_ffmpeg

                        run_ffmpeg(extract_cmd)
                        final_audio = mix_bgm_with_main(ctx, main_audio_path, bgm_cfg, video_duration_final)
                        # 合并回视频

                        bgm_output = self.work_dir / f"rendered_{self.plan.id}_bgm.mp4"
                        merge_audio_video(ctx, output_path, final_audio, bgm_output)
                        output_path = bgm_output
                        logger.info("[unified-render] pass-through BGM mix done: plan_id=%s", self.plan.id)
                    except Exception:
                        logger.exception(
                            "[unified-render] pass-through BGM mix failed, skipping: plan_id=%s", self.plan.id
                        )
        else:
            config = self.plan.config or {}
            bgm_config = config.get("bgm", {}) or {}
            if not isinstance(bgm_config, dict):
                bgm_config = {}
            audio_tracks_config = config.get("audio_tracks") or {}
            noise_reduction_config = config.get("audio_noise_reduction")
            ctx = RenderContext(
                work_dir=self.work_dir,
                plan_id=self.plan.id,
                noise_reduction_config=noise_reduction_config,
            )
            audio_path = mix_audio(
                ctx,
                layers,
                video_duration_final,
                bgm_path=self.bgm_path,
                bgm_config=bgm_config,
                audio_tracks_config=audio_tracks_config,
                bgm_start_offset=self._get_micro_bgm_offset(),
            )
            t_audio_end = time.time()
            audio_mix_ms = int((t_audio_end - t_audio_start) * 1000)
            has_audio = audio_path is not None
            if has_audio:
                logger.info(
                    "[unified-render] audio mix done: plan_id=%s duration_ms=%d",
                    self.plan.id,
                    audio_mix_ms,
                )
                # 7. 合并音视频
                merge_audio_video(ctx, video_only_path, audio_path, output_path)
            else:
                # 无音频，直接用无声视频
                import shutil

                shutil.copy2(video_only_path, output_path)

        # 8. 探测输出
        duration, file_size, width, height = self._probe_output(output_path)

        # 9. 片头片尾拼接（后处理）
        intro_outro_config = IntroOutroConfig.from_dict((self.plan.config or {}).get("intro_outro"))
        if intro_outro_config.has_intro or intro_outro_config.has_outro:
            io_valid, io_err = intro_outro_config.validate()
            if io_valid:
                final_with_io = self.work_dir / f"rendered_{self.plan.id}_with_io.mp4"
                intro_path = None
                outro_path = None

                # 生成片头
                if intro_outro_config.has_intro:
                    intro_path = self.work_dir / f"intro_{self.plan.id}.mp4"
                    intro_ok = False
                    if intro_outro_config.intro_type == "video":
                        import shutil

                        src = Path(intro_outro_config.intro_video_path)
                        if src.exists():
                            shutil.copy2(src, intro_path)
                            intro_ok = True
                        else:
                            logger.warning("片头视频不存在，跳过片头: %s", src)
                    elif intro_outro_config.intro_type == "text":
                        intro_ok = IntroOutroEngine.generate_text_intro(
                            intro_path,
                            intro_outro_config,
                            self.output_width,
                            self.output_height,
                            self.output_fps,
                        )

                    if not intro_ok:
                        intro_path = None

                # 生成片尾
                if intro_outro_config.has_outro:
                    outro_path = self.work_dir / f"outro_{self.plan.id}.mp4"
                    outro_ok = False
                    if intro_outro_config.outro_type == "video":
                        import shutil

                        src = Path(intro_outro_config.outro_video_path)
                        if src.exists():
                            shutil.copy2(src, outro_path)
                            outro_ok = True
                        else:
                            logger.warning("片尾视频不存在，跳过片尾: %s", src)
                    elif intro_outro_config.outro_type in ("text", "follow"):
                        outro_ok = IntroOutroEngine.generate_text_outro(
                            outro_path,
                            intro_outro_config,
                            self.output_width,
                            self.output_height,
                            self.output_fps,
                        )

                    if not outro_ok:
                        outro_path = None

                # 拼接
                if intro_path or outro_path:
                    concat_ok = IntroOutroEngine.concat_with_intro_outro(
                        output_path,
                        intro_path,
                        outro_path,
                        final_with_io,
                        transition_duration=intro_outro_config.transition_duration,
                        transition_effect=intro_outro_config.transition_effect,
                    )
                    if concat_ok and final_with_io.exists():
                        output_path = final_with_io
                        # 重新探测
                        duration, file_size, width, height = self._probe_output(output_path)
                        logger.info("[unified-render] 片头片尾拼接完成: plan_id=%s", self.plan.id)
                    else:
                        logger.warning("[unified-render] 片头片尾拼接失败，使用原视频: plan_id=%s", self.plan.id)
            else:
                logger.warning("[unified-render] 片头片尾配置无效，跳过: %s", io_err)

        t_total = int((time.time() - t_start) * 1000)
        logger.info(
            "[unified-render] render done: plan_id=%s total_ms=%d video_ms=%d audio_ms=%d "
            "output_duration=%.2fs output_size=%d resolution=%dx%d has_audio=%s",
            self.plan.id,
            t_total,
            video_render_ms,
            audio_mix_ms if has_audio else 0,
            duration,
            file_size,
            width,
            height,
            has_audio,
        )

        return RenderResult(
            output_path=output_path,
            duration=duration,
            file_size=file_size,
            width=width,
            height=height,
        )

    def _estimate_total_duration(self, layers: list[RenderLayer]) -> float:
        """估算视频总时长（用于字幕等需要）。

        实际实现移至 packages.domain.render_layer_utils.estimate_total_duration。
        """
        return _estimate_total_duration_pure(layers, self.transition_duration)

    def _get_voice_audio_duration(self) -> float:
        """获取配音音频文件的时长（秒）。

        Returns:
            配音音频时长，如果无配音或探测失败则返回 0.0
        """
        if not self.voiceover_audio_path:
            return 0.0

        audio_path = Path(self.voiceover_audio_path)
        if not audio_path.exists() or audio_path.stat().st_size == 0:
            return 0.0

        try:
            duration = probe_duration(audio_path)
            logger.info("[voice-align] 配音音频时长: %.3fs path=%s", duration, self.voiceover_audio_path)
            return duration
        except Exception as e:
            logger.warning("[voice-align] 探测配音音频时长失败: %s", e)
            return 0.0

    def _align_clips_to_voice_duration(
        self,
        layers: list[RenderLayer],
        voice_duration: float,
    ) -> None:
        """#1749：配音对齐仅保留 ±5% 守卫日志，不再做全局裁剪/慢放。

        段长已在选片阶段由 voice_duration_planner 按配音时长精确分配
        （成片总时长=配音，素材不足走末帧冻结 tpad/apad）。渲染期任何全局
        裁剪会截断配音、全局慢放会导致失真——两者均已删除。此处仅在偏差
        超阈值时记录 warning 供排查，不修改任何 clip。
        """
        if voice_duration <= 0:
            return

        video_layers = [layer for layer in layers if layer.role in ("main", "broll", "background")]
        if not video_layers:
            return

        total_clips_duration = 0.0
        for layer in video_layers:
            for clip in layer.clips:
                total_clips_duration += self._clip_adjusted_duration(clip)
        if total_clips_duration <= 0:
            return

        estimated = self._estimate_total_duration(video_layers)
        ratio = voice_duration / total_clips_duration if total_clips_duration > 0 else 1.0
        if abs(ratio - 1.0) > 0.05:
            logger.warning(
                "[voice-align] 段长合计与配音偏差>5%%（仅守卫，不裁剪/不慢放）: "
                "ratio=%.4f voice=%.3f clips_sum=%.3f estimated_final=%.3f",
                ratio,
                voice_duration,
                total_clips_duration,
                estimated,
            )
        else:
            logger.info(
                "[voice-align] 段长与配音偏差≤5%%，无需处理: ratio=%.4f voice=%.3f estimated=%.3f",
                ratio,
                voice_duration,
                estimated,
            )

    def _maybe_generate_ass(self, video_duration: float) -> Path | None:
        """根据 plan.config 生成 ASS 字幕文件。

        支持两种字幕模式：
        1. 静态字幕 — title/subtitle 配置了 text 时，生成整段静态字幕
        2. ASR 自动字幕 — subtitle.auto_generated=true 时，从音频自动识别生成时间轴字幕

        Returns:
            ASS 文件路径，没有字幕时返回 None
        """
        config = self.plan.config or {}
        # #1901 统一读 "title"，兼容老数据 "title_config"
        title_cfg = config.get("title", {}) or {}
        if not isinstance(title_cfg, dict) or not (title_cfg.get("text") or "").strip():
            _alt = config.get("title_config") or {}
            if isinstance(_alt, dict):
                title_cfg = _alt
        if not isinstance(title_cfg, dict):
            title_cfg = {}
        subtitle_cfg = config.get("subtitle", {}) or {}
        if not isinstance(subtitle_cfg, dict):
            subtitle_cfg = {}

        title_enabled = title_cfg.get("enabled", True)
        subtitle_enabled = subtitle_cfg.get("enabled", True)
        title_text = title_cfg.get("text", "") or ""
        subtitle_text = subtitle_cfg.get("text", "") or ""
        auto_generated = subtitle_cfg.get("auto_generated", False)

        has_title = title_enabled and bool(title_text.strip())
        has_static_subtitle = subtitle_enabled and bool(subtitle_text.strip())
        has_auto_subtitle = subtitle_enabled and auto_generated and self.asr_service is not None

        if not has_title and not has_static_subtitle and not has_auto_subtitle:
            return None

        ass_path = self.work_dir / f"subtitles_{self.plan.id}.ass"

        # ASR 自动字幕模式
        if has_auto_subtitle:
            try:
                timeline = self._generate_asr_subtitles(video_duration, subtitle_cfg)
                if timeline and timeline.segments:
                    generate_ass_from_timeline(
                        ass_path,
                        timeline,
                        video_width=self.output_width,
                        video_height=self.output_height,
                        video_duration=video_duration,
                        subtitle_config=subtitle_cfg,
                        title_text=title_text,
                        title_config=title_cfg,
                    )
                    logger.info(
                        "ASR自动字幕生成完成: plan_id=%s segments=%d duration=%.1fs",
                        self.plan.id,
                        timeline.segment_count,
                        video_duration,
                    )
                    return ass_path
                else:
                    # ASR 无结果，不生成字幕
                    logger.info("ASR自动字幕无识别结果，跳过字幕: plan_id=%s", self.plan.id)
                    return None
            except Exception:
                # ASR 失败降级：不生成字幕，不阻断主流程
                logger.warning("ASR自动字幕生成失败，跳过字幕", exc_info=True)
                return None

        # 静态字幕模式（原有逻辑）
        generate_ass_subtitles(
            ass_path,
            video_width=self.output_width,
            video_height=self.output_height,
            video_duration=video_duration,
            title_text=title_text,
            title_config=title_cfg,
            subtitle_text=subtitle_text,
            subtitle_config=subtitle_cfg,
        )

        logger.info(
            "生成字幕: plan_id=%s title=%s subtitle=%s ass=%s",
            self.plan.id,
            has_title,
            has_static_subtitle,
            ass_path,
        )
        return ass_path

    def _generate_asr_subtitles(self, video_duration: float, subtitle_cfg: dict) -> Any:  # SubtitleTimeline
        """从视频素材音频中自动识别生成字幕时间轴。

        MVP 版本：使用第一个有音频的素材做ASR，然后按比例映射到整个视频时长。
        后续优化：支持多片段拼接后的完整音频ASR。

        带缓存：同一 plan 只做一次 ASR，TTS 配音和字幕共用结果。
        """
        # 检查缓存
        if self._asr_timeline_cached:
            return self._asr_timeline_cache

        from packages.domain.subtitle import SubtitleTimeline

        # 找第一个有本地路径的素材
        first_asset_path = None
        for clip in self.clips:
            asset_id = getattr(clip, "asset_id", None)
            if asset_id and asset_id in self.asset_path_map:
                first_asset_path = self.asset_path_map[asset_id]
                break

        if first_asset_path is None:
            logger.warning("ASR字幕生成失败：找不到可用素材音频")
            result = SubtitleTimeline(segments=[], total_duration=video_duration)
            self._asr_timeline_cache = result
            self._asr_timeline_cached = True
            return result

        # 提取素材音频为 wav（16kHz单声道，ASR友好格式）
        audio_path = self.work_dir / f"asr_audio_{self.plan.id}.wav"
        try:
            self._extract_audio(first_asset_path, audio_path)
        except Exception:
            logger.warning("ASR音频提取失败", exc_info=True)
            return SubtitleTimeline(segments=[], total_duration=video_duration)

        if not audio_path.exists():
            return SubtitleTimeline(segments=[], total_duration=video_duration)

        # 调用 ASR 服务
        language = subtitle_cfg.get("language", "") or None
        timeline = self.asr_service.transcribe(
            audio_path,
            language=language,
            with_word_timestamps=True,
        )

        # 字幕后处理：合并短片段 + 拆分长片段
        min_chars = int(subtitle_cfg.get("min_chars_per_segment", 8))
        max_chars = int(subtitle_cfg.get("max_chars_per_line", 20))

        if timeline.segments:
            timeline = timeline.merge_short_segments(min_chars=min_chars)
            timeline = timeline.split_long_segments(max_chars=max_chars)

        # 清理临时音频文件
        try:
            audio_path.unlink(missing_ok=True)
        except Exception:
            pass

        # 存入缓存
        self._asr_timeline_cache = timeline
        self._asr_timeline_cached = True
        return timeline

    def _extract_audio(self, video_path: Path, output_path: Path) -> None:
        """从视频中提取音频为16kHz单声道wav（ASR友好格式）。"""
        cmd = [
            FFMPEG_BIN,
            "-y",
            "-i",
            str(video_path),
            "-vn",
            "-acodec",
            "pcm_s16le",
            "-ar",
            "16000",
            "-ac",
            "1",
            str(output_path),
        ]

        try:
            run_ffmpeg(cmd, timeout=120)
        except Exception as e:
            raise RuntimeError(f"音频提取失败: {str(e)[:200]}") from e

    def _maybe_add_voiceover_layer(
        self,
        layers: list[RenderLayer],
        *,
        video_duration: float,
    ) -> bool:
        """根据 plan.config 生成 TTS 配音，加到 audio 图层.

        支持三种触发方式：
        1. config.tts.enabled = true → 标准 TTS 配置
        2. 顶层 voice_id + custom_text → 桥接模式（自定义文案配音）
        3. 顶层 voice_id + subtitle.auto_generated=true → ASR 字幕对齐配音（预设配音）

        Returns:
            是否成功添加了配音音轨
        """
        config = self.plan.config or {}
        tts_cfg = config.get("tts", {}) or {}
        if not isinstance(tts_cfg, dict):
            tts_cfg = {}
        subtitle_cfg = config.get("subtitle", {}) or {}
        if not isinstance(subtitle_cfg, dict):
            subtitle_cfg = {}
        use_subtitle_align = False  # 是否使用字幕对齐模式

        # 兼容前端顶层字段：voice_id / custom_text / voice_clone_profile_id
        if not tts_cfg.get("enabled"):
            top_voice_id = config.get("voice_id", "") or ""
            top_text = config.get("custom_text", "") or ""

            # 方式A：voice_id + custom_text → 整段配音
            if top_voice_id and top_text:
                tts_cfg = {
                    "enabled": True,
                    "voice_id": top_voice_id,
                    "text": top_text,
                    "align_mode": "full",
                    "overlap_mode": "replace",
                }
                logger.info(
                    "[unified-render] 检测到顶层 voice_id+custom_text，桥接到 tts 配置（整段）: plan_id=%s voice_id=%s text_len=%d",
                    self.plan.id,
                    top_voice_id,
                    len(top_text),
                )
            # 方式B：voice_id + 自动字幕 → 字幕对齐配音（预设配音模式）
            # ASR 可用时走字幕对齐模式，不可用时降级为整段配音
            elif top_voice_id and subtitle_cfg.get("auto_generated", False):
                if self.asr_service is not None:
                    # 字幕对齐模式
                    tts_cfg = {
                        "enabled": True,
                        "voice_id": top_voice_id,
                        "text": "",
                        "align_mode": "subtitle",
                        "overlap_mode": "replace",
                    }
                    use_subtitle_align = True
                    logger.info(
                        "[unified-render] 检测到预设配音+自动字幕，使用字幕对齐模式: plan_id=%s voice_id=%s",
                        self.plan.id,
                        top_voice_id,
                    )
                else:
                    # 降级：整段配音（预览模式 ASR 不可用时）
                    # 拼接字幕文本作为配音内容
                    subtitle_text_content = subtitle_cfg.get("text", "") or ""
                    tts_cfg = {
                        "enabled": True,
                        "voice_id": top_voice_id,
                        "text": subtitle_text_content,
                        "align_mode": "full",
                        "overlap_mode": "replace",
                    }
                    logger.info(
                        "[unified-render] ASR 不可用，降级为整段配音模式: plan_id=%s voice_id=%s text_len=%d",
                        self.plan.id,
                        top_voice_id,
                        len(subtitle_text_content),
                    )

        # 兼容前端顶层字段：voice_id / custom_text / voice_clone_profile_id
        # 前端一键生成页面传 config.voice_id + config.custom_text，
        # 统一渲染引擎从 config.tts 读，这里做桥接映射。
        if not tts_cfg.get("enabled"):
            top_voice_id = config.get("voice_id", "") or ""
            top_text = config.get("custom_text", "") or ""
            if top_voice_id and top_text:
                tts_cfg = {
                    "enabled": True,
                    "voice_id": top_voice_id,
                    "text": top_text,
                    "align_mode": "full",
                    "overlap_mode": "replace",
                }
                logger.info(
                    "[unified-render] 检测到顶层 voice_id+custom_text，桥接到 tts 配置: plan_id=%s voice_id=%s text_len=%d",
                    self.plan.id,
                    top_voice_id,
                    len(top_text),
                )

        tts_config = TtsConfig.parse(tts_cfg)
        if not tts_config.enabled:
            return False

        try:
            from apps.worker.services.tts_service_factory import get_tts_service

            tts_service = get_tts_service()
            tts_engine = TtsEngine(tts_service, self.work_dir / "tts")

            # 根据对齐模式选择生成方式
            if use_subtitle_align or tts_config.align_mode == "subtitle":
                # 字幕对齐模式：先做 ASR，再按字幕生成配音
                if not self._asr_timeline_cached:
                    self._generate_asr_subtitles(video_duration, subtitle_cfg)
                timeline = self._asr_timeline_cache
                if timeline is None or not timeline.segments:
                    logger.warning("TTS 字幕对齐配音：ASR 无识别结果，跳过配音")
                    return False

                # 转换为 TtsEngine 需要的字幕格式
                subtitles = [
                    {
                        "text": seg.text,
                        "start_time": seg.start,
                        "end_time": seg.end,
                    }
                    for seg in timeline.segments
                    if getattr(seg, "text", "").strip()
                ]
                if not subtitles:
                    logger.warning("TTS 字幕对齐配音：字幕文本为空，跳过配音")
                    return False

                result = tts_engine.generate_subtitle_voiceover(tts_config, subtitles)
            else:
                # 整段配音模式
                result = tts_engine.generate_full_voiceover(tts_config, total_duration=video_duration)

            if not result.success or not result.segments:
                logger.warning("TTS 配音生成失败，跳过: %s", result.error_message)
                return False

            # 获取主音轨图层（用于判断 replace 模式下是否静音原音）
            # 这里只处理混音添加，replace 模式在外部处理

            # 找到或创建 audio 图层
            audio_layer = None
            for layer in layers:
                if layer.role == "audio":
                    audio_layer = layer
                    break

            if audio_layer is None:
                from video_processing.unified_render_service import _LAYER_Z_INDEX  # type: ignore

                z_index = _LAYER_Z_INDEX.get("audio", 2)
                audio_layer = RenderLayer(role="audio", z_index=z_index)
                layers.append(audio_layer)

            # 把配音片段作为 audio clip 加入
            for seg in result.segments:
                if seg.audio_path is None:
                    continue
                vo_clip = ResolvedClip(
                    clip_id=f"tts_{seg.start_time:.3f}",
                    asset_id="tts_voiceover",
                    local_path=seg.audio_path,
                    clip_type="audio",
                    order=len(audio_layer.clips),
                    start_time=seg.start_time,
                    duration=seg.duration,
                    config={"volume": tts_config.volume, "tts": True},
                    actual_duration=seg.duration,
                )
                audio_layer.clips.append(vo_clip)

            # replace 模式：静音原视频音轨（main + broll 图层）
            if tts_config.overlap_mode == "replace":
                for layer in layers:
                    if layer.role in ("main", "broll"):
                        for clip in layer.clips:
                            clip.config["volume"] = 0
                logger.info(
                    "TTS replace 模式：已静音原视频音轨: plan_id=%s",
                    self.plan.id,
                )

            logger.info(
                "TTS 配音已添加: plan_id=%s voice_id=%s segments=%d total_%.2fs",
                self.plan.id,
                tts_config.voice_id,
                len(result.segments),
                result.total_duration,
            )
            return True

        except Exception as e:
            logger.warning("TTS 配音异常，跳过: %s", e)
            return False

    def _maybe_add_voice_library_layer(
        self,
        layers: list[RenderLayer],
        *,
        video_duration: float,
    ) -> bool:
        """将配音素材库音频作为整段配音加到 audio 图层.

        与 TTS 配音共享同一套 audio 图层混音架构，
        支持与 BGM、TTS 的音量平衡，不再走独立的后处理 mux 链路。

        Returns:
            是否成功添加了配音音轨
        """
        if not self.voiceover_audio_path:
            return False

        audio_path = Path(self.voiceover_audio_path)
        if not audio_path.exists() or audio_path.stat().st_size == 0:
            logger.warning("配音素材库音频文件不存在或为空，跳过: %s", self.voiceover_audio_path)
            return False

        try:
            # 找到或创建 audio 图层
            audio_layer = None
            for layer in layers:
                if layer.role == "audio":
                    audio_layer = layer
                    break

            if audio_layer is None:
                from video_processing.unified_render_service import _LAYER_Z_INDEX  # type: ignore

                z_index = _LAYER_Z_INDEX.get("audio", 2)
                audio_layer = RenderLayer(role="audio", z_index=z_index)
                layers.append(audio_layer)

            # 配音素材作为整段配音：从 0 开始，覆盖整个视频时长
            # 音频不足视频时长时，混音层会按实际长度处理（amix 不自动循环）
            vo_clip = ResolvedClip(
                clip_id="voice_library_main",
                asset_id="voice_library",
                local_path=audio_path,
                clip_type="audio",
                order=len(audio_layer.clips),
                start_time=0.0,
                duration=video_duration,
                config={"volume": 1.0, "voice_library": True},
                actual_duration=video_duration,
            )
            audio_layer.clips.append(vo_clip)

            # 配音素材库默认替换原音：静音原视频音轨（main + broll 图层）
            for layer in layers:
                if layer.role in ("main", "broll"):
                    for clip in layer.clips:
                        clip.config["volume"] = 0
            logger.info(
                "配音素材库：已静音原视频音轨: plan_id=%s",
                self.plan.id,
            )

            logger.info(
                "配音素材库音频已添加到 audio 图层: plan_id=%s duration=%.2fs",
                self.plan.id,
                video_duration,
            )
            return True

        except Exception as e:
            logger.warning("配音素材库音频添加失败，跳过: %s", e)
            return False

    @staticmethod
    def _resolve_watermark_config(plan_config: dict[str, Any] | None) -> WatermarkConfig | None:
        """从 plan config 中解析水印配置，兼容两种存储格式.

        支持格式：
        1. 嵌套格式：config.watermark = {enabled, mode, text, image_path, ...}
        2. 扁平格式（导出配置）：config.export.watermark_enabled + config.export.watermark_text

        Returns:
            WatermarkConfig 或 None（未启用水印时）
        """
        if not plan_config or not isinstance(plan_config, dict):
            return None

        # 格式1: 嵌套 watermark 对象（优先）
        wm_data = plan_config.get("watermark")
        if isinstance(wm_data, dict) and wm_data:
            config = WatermarkConfig.from_dict(wm_data)
            if config is not None:
                return config

        # 格式2: 扁平 export.watermark_enabled + export.watermark_text
        export_cfg = plan_config.get("export")
        if isinstance(export_cfg, dict) and export_cfg:
            enabled = export_cfg.get("watermark_enabled", False)
            text = export_cfg.get("watermark_text", "") or ""
            if enabled and text:
                return WatermarkConfig(
                    mode="text",
                    text=str(text),
                    position=export_cfg.get("watermark_position", "bottom_right"),
                    opacity=float(export_cfg.get("watermark_opacity", 0.6)),
                    font_size=int(export_cfg.get("watermark_font_size", 24)),
                    font_color=str(export_cfg.get("watermark_font_color", "white")),
                )

        return None

    def _can_use_pass_through(self, layers: list[RenderLayer]) -> bool:
        """判断是否可以走直通优化路径。

        条件：
        1. 只有 1 个图层
        2. 该图层是视频图层（main/broll/background），不是 overlay/corner_voice/audio
        3. 该图层只有 1 个 clip（无转场需求）
        4. 没有贴纸（贴纸需要 filter_complex 或额外输入）
        """
        if len(layers) != 1:
            return False
        layer = layers[0]
        if layer.role not in ("main", "broll", "background"):
            return False
        if len(layer.clips) != 1:
            return False
        # 有贴纸时禁用直通（图片贴纸需要额外输入，统一走 filter_complex）
        plan_config = getattr(self.plan, "config", None) or {}
        if isinstance(plan_config, dict) and plan_config.get("stickers"):
            return False

        # 有水印时禁用直通（图片水印需要额外输入，统一走 filter_complex；
        # 文字水印虽然可以 -vf 叠加，但为了保持路径统一也走 filter_complex）
        wm_config = UnifiedRenderService._resolve_watermark_config(plan_config)
        if wm_config is not None and wm_config.validate()[0]:
            return False

        # 有调速时仍然可以走直通（视频调速通过 setpts 实现，单输入即可）

        return True

    def _can_use_stream_copy(
        self,
        clip: ResolvedClip,
        *,
        ass_path: Path | None = None,
        video_duration: float = 0.0,
    ) -> tuple[bool, str]:
        """判断是否可以走 stream copy（流拷贝，不重编码）。

        性能提升：10 倍以上（典型场景从 20s → 1-2s）。

        条件：
        1. 视频编码为 h264（输出目标也是 h264）
        2. 像素格式为 yuv420p
        3. 分辨率与输出一致（不需要 scale/crop）
        4. 帧率与输出一致（误差 < 0.1fps）
        5. 无字幕叠加（字幕需要滤镜）
        6. 无 trim 需求（或 trim 后恰好等于原时长）
        7. 无转场、无特效（单 clip 直通已保证）

        Returns:
            (是否可以 copy, 原因说明)
        """
        # 有字幕 → 需要滤镜 → 不能 copy
        if ass_path is not None:
            return False, "有字幕叠加"

        # #1970 PR2：片段级微变换（变速/hflip/亮度/对比度/饱和度）需要重编码
        try:
            _video_sources = [c for c in (self.clips or []) if getattr(c, "clip_type", "main") != "audio"]
            _ordinal = -1
            for _i, _c in enumerate(_video_sources):
                if getattr(_c, "id", None) == getattr(clip, "clip_id", None):
                    _ordinal = _i
                    break
            _mt_plan = self._get_micro_transform_plan(len(_video_sources))
            if _mt_plan and 0 <= _ordinal < len(_mt_plan.clips):
                _mt = _mt_plan.clips[_ordinal]
                if (
                    abs(UnifiedRenderService._micro_speed_factor(_mt) - 1.0) >= 1e-6
                    or (_mt.hflip and not _mt.has_text)
                    or abs(_mt.brightness) > 1e-4
                    or abs(_mt.contrast - 1.0) > 1e-4
                    or abs(_mt.saturation - 1.0) > 1e-4
                ):
                    return False, "启用了片段级微变换"
        except Exception:
            logger.debug("stream copy 微变换门控检查异常，按可 copy 处理", exc_info=True)

        # 有调速 → 需要重编码 → 不能 copy
        speed = UnifiedRenderService._clip_speed(clip)
        if abs(speed - 1.0) >= 1e-6:
            return False, f"有调速: speed={speed:.2f}x"

        # 音量非默认（静音/放大）→ 需要音频滤镜重编码 → 不能 copy
        vol = UnifiedRenderService._clip_volume(clip)
        if abs(vol - 1.0) >= 1e-6:
            return False, f"音量非默认: volume={vol:.2f}"

        # 有倒放 → 需要重编码 → 不能 copy
        reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
        if reverse_config.enabled and (reverse_config.reverse_video or reverse_config.reverse_audio):
            return False, "有倒放效果"

        # 探测输入视频参数
        info = probe_video_info(str(clip.local_path))

        # 编码必须是 h264
        if info.get("video_codec", "") != "h264":
            return False, f"视频编码不是h264: {info.get('video_codec', 'unknown')}"

        # 像素格式必须是 yuv420p
        if info.get("pix_fmt", "") != "yuv420p":
            return False, f"像素格式不是yuv420p: {info.get('pix_fmt', 'unknown')}"

        # 分辨率必须一致
        if info.get("width", 0) != self.output_width or info.get("height", 0) != self.output_height:
            return False, (
                f"分辨率不匹配: "
                f"{info.get('width', 0)}x{info.get('height', 0)} "
                f"vs {self.output_width}x{self.output_height}"
            )

        # 帧率必须一致（误差 < 0.1fps）
        fps_diff = abs(info.get("fps", 0) - self.output_fps)
        if fps_diff > 0.1:
            return False, f"帧率不匹配: {info.get('fps', 0)} vs {self.output_fps}"

        # 检查是否需要 trim
        effective_duration = UnifiedRenderService._clip_effective_duration(clip)
        if effective_duration > 0:
            # 有 trim 需求但视频时长足够，可用 -ss/-t 实现 copy trim
            input_duration = info.get("duration", 0)
            if input_duration <= 0:
                return False, "无法探测输入时长"
            # trim 起始点 + 目标时长 <= 输入时长
            start_time = getattr(clip, "start_time", 0) or 0
            if start_time + effective_duration > input_duration + 0.1:
                return False, "trim 超出输入时长"

        # video_duration 截断
        if video_duration > 0 and effective_duration > 0:
            final_duration = min(effective_duration, video_duration)
            if final_duration != effective_duration:
                # 也需要截断，但 -t 可以 copy 模式下用
                pass

        return True, "所有条件满足"

    def _try_render_stream_copy(
        self,
        layers: list[RenderLayer],
        output_path: Path,
        *,
        ass_path: Path | None = None,
        video_duration: float = 0.0,
    ) -> bool:
        """尝试 stream copy 渲染，成功返回 True，失败返回 False（调用方回退到重编码）。

        stream copy 模式：不重编码，直接拷贝视频/音频流，性能提升 10 倍+。
        仅用于单 clip 直通场景且满足 copy 条件。
        """
        clip = layers[0].clips[0]
        role = layers[0].role

        # 判断是否满足 copy 条件
        can_copy, reason = self._can_use_stream_copy(clip, ass_path=ass_path, video_duration=video_duration)
        if not can_copy:
            logger.info(
                "[unified-render] stream_copy 跳过: plan_id=%s reason=%s",
                self.plan.id,
                reason,
            )
            return False

        # 构建 copy 命令
        command = [
            FFMPEG_BIN,
            "-y",
        ]

        # trim 支持（-ss 放在 -i 前 = input seeking，速度更快但精度稍差；
        # 放在 -i 后 = output seeking，精度高但慢）
        # 这里用 output seeking 保证精度，反正 copy 模式已经很快了
        start_time = getattr(clip, "start_time", 0) or 0
        effective_duration = UnifiedRenderService._clip_effective_duration(clip)

        command.extend(["-i", str(clip.local_path)])

        if start_time > 0:
            command.extend(["-ss", f"{start_time:.3f}"])

        # 计算最终时长
        final_duration = effective_duration
        if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
            final_duration = video_duration
        if final_duration > 0:
            command.extend(["-t", f"{final_duration:.3f}"])

        # 流拷贝
        command.extend(
            [
                "-c:v",
                "copy",
                "-c:a",
                "copy",
                "-movflags",
                "+faststart",
                str(output_path),
            ]
        )

        logger.info(
            "[unified-render] stream_copy 渲染: plan_id=%s clip=%s role=%s duration=%.2fs",
            self.plan.id,
            clip.clip_id,
            role,
            final_duration,
        )

        try:
            run_ffmpeg(command)
            # 验证输出文件存在且有大小
            if output_path.exists() and output_path.stat().st_size > 0:
                logger.info(
                    "[unified-render] stream_copy 成功: plan_id=%s size=%d",
                    self.plan.id,
                    output_path.stat().st_size,
                )
                return True
            else:
                logger.warning("[unified-render] stream_copy 输出为空: plan_id=%s", self.plan.id)
                return False
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
            logger.warning(
                "[unified-render] stream_copy 失败，回退到重编码: plan_id=%s error=%s",
                self.plan.id,
                str(e)[:200],
            )
            # 清理可能的损坏输出文件
            if output_path.exists():
                try:
                    output_path.unlink()
                except OSError as unlink_err:
                    logger.warning(
                        "[unified-render] 损坏输出文件清理失败: path=%s error=%s",
                        output_path,
                        unlink_err,
                    )
            return False

    def _render_pass_through(
        self,
        layers: list[RenderLayer],
        output_path: Path,
        *,
        ass_path: Path | None = None,
        video_duration: float = 0.0,
    ) -> bool:
        """单图层单 clip 直通渲染（使用 -vf 而非 -filter_complex），一次性输出带音频的最终视频。

        性能优化：
        - 避免 filter_complex 的解析和调度开销，单clip场景性能提升 ~30%
        - 视频+音频一次FFmpeg调用完成，省去后续音频提取+音视频合并两次调用

        Args:
            layers: 图层列表（只有1个图层1个clip）
            output_path: 输出文件路径
            ass_path: ASS 字幕文件路径，有则叠加字幕
            video_duration: 视频总时长（用于截断音频，0表示不额外截断）

        Returns:
            True 表示输出包含音频（近似判断，实际以输出文件为准）
        """
        clip = layers[0].clips[0]
        role = layers[0].role

        # 构建视频滤镜链（与 _build_filter_complex 中预处理逻辑一致）
        filters: list[str] = []

        # trim
        effective_duration = UnifiedRenderService._clip_effective_duration(clip)
        freeze_seconds = float((clip.config or {}).get("_freeze_seconds", 0.0) or 0.0)
        # #1749 冻结时素材内截取时长 = 目标段长 − freeze
        trim_dur = max(0.0, effective_duration - freeze_seconds) if freeze_seconds > 0 else effective_duration

        if trim_dur > 0:
            filters.append(f"trim=duration={trim_dur}")
            filters.append("setpts=PTS-STARTPTS")
        # #1749 末帧冻结（禁慢放）
        if freeze_seconds > 0:
            filters.append(f"tpad=stop_mode=clone:stop_duration={freeze_seconds:.3f}")

        # 视觉扰动（plan 级别，直通模式同样适用）
        vp = self._get_visual_perturbation()
        # #1970 PR2：单片段直通；计划按源视频片段数构建，序号取 config._micro_index
        _src_video_count = len([c for c in (self.clips or []) if getattr(c, "clip_type", "main") != "audio"])
        mt_plan = self._get_micro_transform_plan(max(1, _src_video_count))
        _mi = int(clip.config.get("_micro_index", 0)) if isinstance(clip.config, dict) else 0
        mt = mt_plan.clips[_mi] if mt_plan and 0 <= _mi < len(mt_plan.clips) else None

        # 调速 — 与 filter_complex 路径一致（叠加视觉扰动 speed_factor 与 #1970 微变换速度）
        speed = UnifiedRenderService._clip_speed(clip)
        vp_speed = vp.get("speed_factor", 1.0) if vp else 1.0
        effective_speed = speed * vp_speed  # 微变换速度已烘焙进 playback_speed
        if abs(effective_speed - 1.0) >= 1e-6:
            filters.append(f"setpts=PTS/{effective_speed:.4f}")

        # 倒放滤镜
        reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
        if reverse_config.enabled and reverse_config.reverse_video:
            reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration)
            if reverse_filter:
                filters.append(reverse_filter)

        # 视觉扰动：hflip（在 scale 之前）
        if vp:
            self._apply_visual_perturbation_pre_scale(filters, vp)
        # #1970 PR2：片段级 hflip（P1 保守：有文字/无法判定时不翻转）
        UnifiedRenderService._apply_micro_hflip(filters, mt)

        # scale + pad（等比缩放+留黑边）
        if role in ("overlay", "corner_voice"):
            pip_w = int(self.output_width * _PIP_SCALE)
            pip_h = int(self.output_height * _PIP_SCALE)
            filters.append(f"scale={pip_w}:{pip_h}")
        elif role == "background":
            # background: 铺满裁剪（作为底图，覆盖全屏）
            filters.append(f"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=increase")
            filters.append(f"crop={self.output_width}:{self.output_height}")
        else:
            # main / broll: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容）
            filters.append(f"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=decrease")
            filters.append(f"pad={self.output_width}:{self.output_height}:trunc((ow-iw)/2):trunc((oh-ih)/2):black")

        # 视觉扰动：zoom + brightness（在 scale+pad 之后、调色之前）
        if vp:
            self._apply_visual_perturbation_post_scale(filters, vp)
        # #1970 PR2：片段级亮度/对比度/饱和度微调
        UnifiedRenderService._apply_micro_transform_video(filters, mt)

        # 调色滤镜
        color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade"))
        if color_grade.enabled and color_grade.has_effect():
            grade_filter = ColorGradeEngine.build_filter(color_grade)
            if grade_filter:
                filters.append(grade_filter)
        # chroma key 绿幕抠像
        try:
            from video_processing.chroma_key_engine import ChromaKeyConfig, ChromaKeyEngine

            ck_config = ChromaKeyConfig.from_dict(clip.config.get("chroma_key"))
            if ck_config.has_effect():
                ck_engine = ChromaKeyEngine(ck_config)
                ck_full = ck_engine.build_filter("[in]", "[out]")
                ck_filter_part = ck_full[len("[in]") : -len("[out]")]
                filters.append(ck_filter_part)
        except Exception as e:
            logger.warning("[unified-render] chroma key 直通模式应用失败，跳过: %s", e)

        filters.append("setpts=PTS-STARTPTS")
        filters.append(f"fps={self.output_fps}")
        # #1749 末帧冻结：素材短于目标段长时克隆末帧铺满（禁慢放）
        freeze_seconds = float((clip.config or {}).get("_freeze_seconds", 0.0) or 0.0)
        if freeze_seconds > 0:
            filters.append(f"tpad=stop_mode=clone:stop_duration={freeze_seconds:.3f}")
        filters.append("format=yuv420p")

        # 字幕叠加
        if ass_path is not None:
            ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:")
            filters.append(f"subtitles='{ass_filter_path}'")

        vf_str = ",".join(filters)

        # 最终输出时长：取 clip 调速后有效时长和 video_duration 的较小值
        # 注意：必须用调速后的时长，否则减速场景（speed<1）会被 -t 截断
        adjusted_duration = UnifiedRenderService._clip_adjusted_duration(clip)
        final_duration = adjusted_duration
        if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
            final_duration = video_duration

        command = [
            FFMPEG_BIN,
            "-y",
            "-i",
            str(clip.local_path),
            "-vf",
            vf_str,
            "-c:v",
            "libx264",
            "-crf",
            FFMPEG_ENCODE_CRF,
            "-preset",
            FFMPEG_ENCODE_PRESET,
            "-threads",
            FFMPEG_ENCODE_THREADS,
            "-pix_fmt",
            "yuv420p",
            "-movflags",
            "+faststart",
        ]

        # 音频处理：background 通常是图片无音频，跳过；其他角色先探测是否真有音频流。
        # volume=0 不丢弃音频流，而是保留后通过 volume=0 滤镜静音，保持时间轴对齐。
        clip_volume = UnifiedRenderService._clip_volume(clip)
        if role != "background":
            try:
                has_audio = probe_has_audio(clip.local_path)
            except Exception as e:
                # probe_has_audio 内部已保守返回 True；只有极端错误才会到这里。
                # 此时不静默丢音频，记录 error 并向上抛出，让任务失败而不是产出无声视频。
                logger.error("[unified-render] 探测音频流发生致命错误，终止渲染: %s: %s", clip.local_path, e)
                raise
        else:
            has_audio = False
        if has_audio:
            af_parts: list[str] = []

            # 音频降噪（最先处理：在原始信号上降噪效果最好）
            try:
                from video_processing.noise_reduction_engine import NoiseReductionConfig, NoiseReductionEngine

                plan_config = getattr(self.plan, "config", {}) or {}
                nr_config = NoiseReductionConfig.from_dict(plan_config.get("audio_noise_reduction"))
                if nr_config.has_effect():
                    nr_engine = NoiseReductionEngine(nr_config)
                    nr_full = nr_engine.build_filter("[in]", "[out]")
                    nr_filter_part = nr_full[len("[in]") : -len("[out]")]
                    af_parts.append(nr_filter_part)
            except Exception as e:
                logger.warning("[unified-render] 直通模式音频降噪应用失败，跳过: %s", e)

            # 音频调速（在降噪之后、音量之前，与 render_audio.py concat 路径保持一致）
            # SpeedEngine.build_audio_filter 内部已实现多级 atempo 串联，
            # 自动处理超出 [0.5, 2.0] 范围的速度（如 0.25x → atempo=0.5,atempo=0.5）。
            # #1970 PR2：叠加片段微变换速度因子，保持音画同步。
            speed = UnifiedRenderService._clip_speed(clip)  # 微变换速度已烘焙进 playback_speed
            if abs(speed - 1.0) >= 1e-6:
                try:
                    from video_processing.speed_engine import SpeedConfig, SpeedEngine

                    speed_cfg = SpeedConfig(speed=speed)
                    speed_cfg.clamp()
                    speed_engine = SpeedEngine()
                    af_parts.append(speed_engine.build_audio_filter(speed_cfg))
                except Exception as e:
                    logger.warning("[unified-render] 直通模式音频调速应用失败，跳过: %s", e)

            # 音频倒放
            reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
            if reverse_config.enabled and reverse_config.reverse_audio:
                af_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
                if af_filter:
                    af_parts.append(af_filter)

            # 片段音量（最后应用：确保调速/倒放后的最终输出音量准确，与 concat 路径一致）
            if abs(clip_volume - 1.0) >= 1e-6:
                af_parts.append(f"volume={clip_volume:.4f}")

            # #1749 音频冻结：视频 tpad 延长后音频需等长补静音，否则时间轴错位
            if freeze_seconds > 0:
                af_parts.append(f"apad=whole_dur={final_duration:.3f}")

            if af_parts:
                command.extend(["-af", ",".join(af_parts)])

            command.extend(["-c:a", "aac", "-b:a", "128k"])

        # 统一截断时长（同时作用于视频和音频）
        if final_duration > 0:
            command.extend(["-t", f"{final_duration:.3f}"])

        command.append(str(output_path))

        logger.info(
            "直通渲染: plan_id=%s clip=%s role=%s duration=%.2fs has_audio=%s",
            self.plan.id,
            clip.clip_id,
            role,
            effective_duration,
            has_audio,
        )
        # 尝试 GPU NVENC 加速
        gpu_ok = False
        if self._gpu_encode_available():
            mezz_path = output_path.parent / f".{output_path.stem}.mezz{output_path.suffix}"
            gpu_ok = self._ffmpeg_output_to_mezzanine(command, mezz_path, output_path)

        if not gpu_ok:
            try:
                run_ffmpeg(command)
            except subprocess.CalledProcessError as e:
                stderr_text = (e.stderr or "").strip()
                stderr_tail = stderr_text[-1500:] if len(stderr_text) > 1500 else stderr_text
                logger.error(
                    "直通渲染失败: plan_id=%s clip=%s exit_code=%d\nvf=%s\nstderr(last 1500):\n%s",
                    self.plan.id,
                    clip.clip_id,
                    e.returncode,
                    vf_str[:2000],
                    stderr_tail,
                )
                raise

        return has_audio

    # ── 内部方法 ──────────────────────────────────────────────────────────────

    def _resolve_clips(self) -> list[ResolvedClip]:
        """将 EditPlanClip 列表解析为 ResolvedClip 列表。

        跳过 asset_id 为空或在 asset_path_map 中找不到的片段。
        支持多段裁剪：一个 clip 配置了 trim_segments 时会展开为多个 ResolvedClip。
        """
        resolved: list[ResolvedClip] = []
        # #1970 PR2：预建片段级微变换计划，按源视频片段序号取速度因子，
        # 烘焙进 playback_speed，保证视频 setpts 与音频 atempo 一致。
        video_source_clips = [c for c in self.clips if getattr(c, "clip_type", "main") != "audio"]
        mt_plan = self._get_micro_transform_plan(len(video_source_clips))
        _video_ordinal = {id(c): i for i, c in enumerate(video_source_clips)}

        for clip in self.clips:
            asset_id = clip.asset_id
            if not asset_id:
                logger.warning("片段无素材: clip_id=%s", clip.id)
                continue
            _mt_idx = _video_ordinal.get(id(clip), -1)
            _mt = mt_plan.clips[_mt_idx] if mt_plan and 0 <= _mt_idx < len(mt_plan.clips) else None
            _micro_speed = UnifiedRenderService._micro_speed_factor(_mt)

            local_path = self.asset_path_map.get(asset_id)
            if local_path is None or not local_path.exists():
                logger.warning("素材不存在: clip_id=%s asset_id=%s", clip.id, asset_id)
                continue

            # 探测实际时长
            try:
                actual_duration = probe_duration(local_path)
            except Exception:
                actual_duration = clip.duration or 5.0

            # 检查是否有多段裁剪配置
            clip_config = clip.config or {}
            trim_segments = TrimEngine.parse_segments_from_config(clip_config)

            if trim_segments and len(trim_segments) > 1:
                # 多段裁剪：展开为多个 clip
                resolved_segments = TrimEngine.resolve_segments(trim_segments, actual_duration)
                configured_speed = getattr(clip, "playback_speed", 1.0) or 1.0
                for i, seg in enumerate(resolved_segments):
                    # 每个段生成一个独立的 ResolvedClip
                    seg_clip_id = f"{clip.id}_seg_{seg.segment_id}"
                    seg_order = clip.order + seg.order * 0.001 + i * 0.0001  # 保持排序
                    seg_start = seg.trim.start_time
                    seg_duration = seg.trim.duration

                    # 多段裁剪：如果段的时长超过素材实际时长，减速补偿
                    seg_speed = configured_speed * _micro_speed
                    if actual_duration > 0 and seg_duration > actual_duration + 0.05:
                        seg_speed = max(0.25, round(configured_speed * actual_duration / seg_duration, 4))
                        logger.info(
                            "[debug] multi-seg clip=%s seg=%s duration=%.3f actual=%.3f → speed=%.4f",
                            clip.id,
                            seg.segment_id,
                            seg_duration,
                            actual_duration,
                            seg_speed,
                        )

                    rc = ResolvedClip(
                        clip_id=seg_clip_id,
                        asset_id=asset_id,
                        local_path=local_path,
                        clip_type=clip.clip_type,
                        order=seg_order,
                        start_time=seg_start,
                        duration=seg_duration,
                        transition_effect=clip.transition_effect or "cut",
                        transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
                        playback_speed=seg_speed,
                        config={**clip_config, "_segment_id": seg.segment_id, "_micro_index": _mt_idx},
                        actual_duration=actual_duration,
                        trim_config=seg.trim,
                    )
                    resolved.append(rc)
                continue

            # 单段裁剪（或无裁剪）
            # #1749 定稿：目标段长（clip.duration，配音分配结果）始终为准。
            # trim 只决定素材内截取区间；素材短于段长 → 末帧冻结（tpad）/音频补静音（apad）
            # 铺满，禁止慢放、禁止截断配音。
            trim_config = extract_trim_from_clip_config(clip_config)
            if trim_config is None and (clip.start_time > 0 or clip.duration > 0):
                # 用旧字段构造
                trim_config = TrimConfig(
                    start_time=clip.start_time,
                    duration=clip.duration,
                )

            effective_trim: TrimConfig | None = None
            # 目标段长：配音分配后的 clip.duration 始终为准（不被素材时长钳制）
            target_duration = float(clip.duration or 0.0)
            final_start = clip.start_time
            final_duration = target_duration
            configured_speed = getattr(clip, "playback_speed", 1.0) or 1.0
            freeze_seconds = 0.0

            if trim_config is not None and actual_duration > 0:
                resolved_trim = trim_config.validate_and_resolve(actual_duration)
                if resolved_trim.is_valid:
                    # trim 仅决定素材内截取区间 [start, min(start+dur, actual)]
                    final_start = resolved_trim.start_time
                    avail_in_asset = max(0.0, actual_duration - final_start)
                    effective_trim = TrimConfig(
                        start_time=final_start, duration=min(resolved_trim.duration, avail_in_asset)
                    )
                else:
                    # 裁剪无效 → 从素材头取
                    logger.warning("裁剪配置无效，使用素材起点: clip_id=%s", clip.id)
                    final_start = 0.0
                    avail_in_asset = actual_duration
            else:
                final_start = 0.0
                avail_in_asset = actual_duration if actual_duration > 0 else target_duration

            # 素材内可用时长不足目标段长 → 末帧冻结（禁慢放）
            if target_duration > 0 and avail_in_asset < target_duration - 0.05:
                freeze_seconds = round(target_duration - avail_in_asset, 3)
                logger.info(
                    "[freeze] clip=%s target=%.3f avail=%.3f → 末帧冻结 %.3fs（禁慢放）",
                    clip.id,
                    target_duration,
                    avail_in_asset,
                    freeze_seconds,
                )
            final_speed = configured_speed * _micro_speed

            # freeze 标记写入 config，供视频 tpad / 音频 apad 读取
            resolved_config = dict(clip_config)
            if freeze_seconds > 0:
                resolved_config["_freeze_seconds"] = freeze_seconds
            resolved_config["_micro_index"] = _mt_idx

            rc = ResolvedClip(
                clip_id=clip.id,
                asset_id=asset_id,
                local_path=local_path,
                clip_type=clip.clip_type,
                order=clip.order,
                start_time=final_start,
                duration=final_duration,
                transition_effect=clip.transition_effect or "cut",
                transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
                playback_speed=final_speed,
                config=resolved_config,
                actual_duration=actual_duration,
                trim_config=effective_trim,
            )
            resolved.append(rc)

            # Debug日志：记录每个clip的时长信息
            eff_dur = _clip_effective_duration_pure(final_duration, actual_duration)
            logger.info(
                "[debug] resolved clip=%s duration=%.3f actual=%.3f effective=%.3f speed=%.4f start=%.3f freeze=%.3f",
                clip.id,
                final_duration,
                actual_duration,
                eff_dur,
                final_speed,
                final_start,
                freeze_seconds,
            )

        # 按 order 排序
        resolved.sort(key=lambda c: c.order)
        return resolved

    def _group_clips_into_layers(self, resolved_clips: list[ResolvedClip]) -> list[RenderLayer]:
        """将 ResolvedClips 分组为 RenderLayers。

        分组规则见 _resolve_layer_role 函数文档。
        """
        layer_map: dict[str, RenderLayer] = {}

        for clip in resolved_clips:
            role = _resolve_layer_role(clip.clip_type, clip.config)
            if role not in layer_map:
                z = _LAYER_Z_INDEX.get(role, 0)
                layer_map[role] = RenderLayer(role=role, z_index=z)
            layer_map[role].clips.append(clip)

        # 每个 layer 内的 clips 按 order 排序
        for layer in layer_map.values():
            layer.clips.sort(key=lambda c: c.order)

        # 计算 PiP 位置
        pip_width = int(self.output_width * _PIP_SCALE)
        margin = 20  # 边距

        if "overlay" in layer_map:
            layer_map["overlay"].position = (
                self.output_width - pip_width - margin,
                margin,
            )
        if "corner_voice" in layer_map:
            layer_map["corner_voice"].position = (
                self.output_width - pip_width - margin,
                margin,
            )

        # 按 z_index 排序返回
        layers = sorted(layer_map.values(), key=lambda lyr: lyr.z_index)
        return layers

    def _build_filter_complex(
        self, layers: list[RenderLayer], *, ass_path: Path | None = None
    ) -> tuple[str, list[str]]:
        """构建 FFmpeg filter_complex 字符串和输入参数列表。

        Args:
            layers: 图层列表
            ass_path: ASS 字幕文件路径，有则在最后叠加字幕

        Returns:
            (filter_complex_str, input_args_list)
            input_args_list 是 ["-i", path1, "-i", path2, ...] 格式
        """
        if not layers:
            raise ValueError("没有可渲染的图层")

        # 收集所有 clips（按图层顺序，同层按 order）
        # 排除纯音频 clips — 它们由 mix_audio() 独立处理，不应出现在视频 filter_complex 中
        # 例如：voice.mp3 没有视频流，如果加入 all_clips 会生成 [N:v] 引用导致 FFmpeg 报错
        all_clips: list[ResolvedClip] = []
        for layer in layers:
            for clip in layer.clips:
                if clip.clip_type == "audio":
                    continue
                all_clips.append(clip)

        if not all_clips:
            raise ValueError("没有可渲染的视频片段（所有片段均为纯音频）")

        # 构建输入参数
        input_args: list[str] = []
        clip_to_input_idx: dict[str, int] = {}
        for i, clip in enumerate(all_clips):
            input_args.extend(["-i", str(clip.local_path)])
            clip_to_input_idx[clip.clip_id] = i

        filter_parts: list[str] = []

        # Step 1: 预处理每个 clip — trim + scale + setpts
        # 为每个 clip 生成预处理后的标签 [v0], [v1], ...
        preprocessed_labels: list[str] = []
        # 视觉扰动（plan 级别，所有 clip 共享同一套扰动参数）
        vp = self._get_visual_perturbation()
        # #1970 PR2：片段级微变换（每片段独立参数，dedup_enabled=False 时为 None）
        # 计划按源视频片段数构建，trim 多段展开时各段通过 config._micro_index 找参数
        _src_video_count = len([c for c in (self.clips or []) if getattr(c, "clip_type", "main") != "audio"])
        mt_plan = self._get_micro_transform_plan(_src_video_count)
        for i, clip in enumerate(all_clips):
            label = f"v{i}"
            role = _resolve_layer_role(clip.clip_type, clip.config)
            _mi = int(clip.config.get("_micro_index", i)) if isinstance(clip.config, dict) else i
            mt = mt_plan.clips[_mi] if mt_plan and 0 <= _mi < len(mt_plan.clips) else None

            filters: list[str] = []

            # trim — 裁剪到指定区间，精确到帧
            effective_duration = UnifiedRenderService._clip_effective_duration(clip)
            trim_start = getattr(clip, "start_time", 0) or 0
            # #1749 冻结时素材内截取时长 = 目标段长 − freeze（末帧冻结由 tpad 铺满，与直通路径同口径）
            freeze_seconds = float((clip.config or {}).get("_freeze_seconds", 0.0) or 0.0)
            trim_dur = max(0.0, effective_duration - freeze_seconds) if freeze_seconds > 0 else effective_duration

            if trim_dur > 0:
                if trim_start > 0:
                    filters.append(f"trim=start={trim_start:.3f}:duration={trim_dur:.3f}")
                else:
                    filters.append(f"trim=duration={trim_dur:.3f}")
                filters.append("setpts=PTS-STARTPTS")

            # 调速 — 基于 setpts 改变播放速度（叠加视觉扰动 speed_factor 与 #1970 微变换速度）
            speed = UnifiedRenderService._clip_speed(clip)
            vp_speed = vp.get("speed_factor", 1.0) if vp else 1.0
            effective_speed = speed * vp_speed  # 微变换速度已烘焙进 playback_speed
            if abs(effective_speed - 1.0) >= 1e-6:
                filters.append(f"setpts=PTS/{effective_speed:.4f}")

            # 倒放滤镜（在 trim 之后、scale 之前应用）
            reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
            if reverse_config.enabled and reverse_config.reverse_video:
                reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration)
                if reverse_filter:
                    filters.append(reverse_filter)

            # 视觉扰动：hflip（在 scale 之前，翻转原始画面）
            if vp:
                self._apply_visual_perturbation_pre_scale(filters, vp)
            # #1970 PR2：片段级 hflip（P1 保守：有文字/无法判定时不翻转）
            UnifiedRenderService._apply_micro_hflip(filters, mt)

            # scale
            if role in ("overlay", "corner_voice"):
                pip_w = int(self.output_width * _PIP_SCALE)
                pip_h = int(self.output_height * _PIP_SCALE)
                filters.append(f"scale={pip_w}:{pip_h}")
            elif role == "background":
                filters.append(f"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=increase")
                filters.append(f"crop={self.output_width}:{self.output_height}")
            else:
                # main / broll: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容）
                # concat 要求所有输入分辨率完全一致，pad 模式确保不同宽高比的素材都能正常拼接
                filters.append(f"scale={self.output_width}:{self.output_height}:force_original_aspect_ratio=decrease")
                filters.append(f"pad={self.output_width}:{self.output_height}:trunc((ow-iw)/2):trunc((oh-ih)/2):black")

            # 视觉扰动：zoom + brightness（在 scale+pad 之后、调色之前）
            if vp:
                self._apply_visual_perturbation_post_scale(filters, vp)
            # #1970 PR2：片段级亮度/对比度/饱和度微调
            UnifiedRenderService._apply_micro_transform_video(filters, mt)

            # 调色滤镜（每个 clip 独立的 color grade 配置）
            color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade"))
            if color_grade.enabled and color_grade.has_effect():
                grade_filter = ColorGradeEngine.build_filter(color_grade)
                if grade_filter:
                    filters.append(grade_filter)
            # chroma key 绿幕抠像（在 scale 之后，fps 之前）
            try:
                from video_processing.chroma_key_engine import ChromaKeyConfig, ChromaKeyEngine

                ck_config = ChromaKeyConfig.from_dict(clip.config.get("chroma_key"))
                if ck_config.has_effect():
                    ck_engine = ChromaKeyEngine(ck_config)
                    # 提取滤镜部分（不带输入输出标签）
                    ck_full = ck_engine.build_filter("[in]", "[out]")
                    ck_filter_part = ck_full[len("[in]") : -len("[out]")]
                    filters.append(ck_filter_part)
            except Exception as e:
                logger.warning("[unified-render] chroma key 应用失败，跳过 clip=%s: %s", clip.clip_id, e)

            filters.append("setpts=PTS-STARTPTS")
            filters.append(f"fps={self.output_fps}")
            # #1749 末帧冻结：素材短于目标段长时克隆末帧铺满（禁慢放）
            if freeze_seconds > 0:
                filters.append(f"tpad=stop_mode=clone:stop_duration={freeze_seconds:.3f}")

            filter_str = f"[{i}:v]{','.join(filters)}[{label}]"
            filter_parts.append(filter_str)
            preprocessed_labels.append(label)

        # Step 2: 同层 clips 用 xfade 串联（跳过纯音频层，由 mix_audio() 独立处理）
        layer_output_labels: dict[str, str] = {}
        for layer in layers:
            # 音频层不参与视频 filter_complex，跳过
            video_clips_in_layer = [c for c in layer.clips if c.clip_type != "audio"]
            if not video_clips_in_layer:
                continue
            layer_clip_indices = [all_clips.index(c) for c in video_clips_in_layer]
            layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
            # 使用调速后的实际时长，与 Step 1 的调速处理保持一致
            layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[i]) for i in layer_clip_indices]
            layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
            layer_transition_durations = [all_clips[i].transition_duration for i in layer_clip_indices]

            if len(layer_labels) == 1:
                # 单 clip 层，直接使用预处理标签
                layer_output_labels[layer.role] = layer_labels[0]
            else:
                out_label = f"{layer.role}_merged"
                # 判断是否全部为硬切：是则用 concat filter，否则用 xfade 转场链
                all_cut = all(
                    t is None or t == "" or str(t).lower() == "cut"
                    for t in layer_transitions[1:]  # 第一个 clip 的转场忽略
                )
                if all_cut:
                    # 全硬切：用 concat filter，性能远优于 xfade
                    concat_inputs = "".join(f"[{label}]" for label in layer_labels)
                    filter_parts.append(f"{concat_inputs}concat=n={len(layer_labels)}:v=1:a=0[{out_label}]")
                    logger.info(
                        "[unified-render] layer=%s clips=%d using concat (all hard-cut)",
                        layer.role,
                        len(layer_labels),
                    )
                else:
                    # 有转场效果：用 TransitionEngine 构建 xfade 链
                    # #1766: 提取逐转场时长（每个 clip 的 transition_duration，跳过第一个）
                    # layer_transition_durations[i] 对应 clip i 的转场，第 0 个忽略
                    per_transition_durations = [d for idx, d in enumerate(layer_transition_durations) if idx > 0]
                    # #1766: 提取每个 clip 的 jitter（存在 config 中），跳过第一个
                    layer_jitters = [
                        (
                            all_clips[layer_clip_indices[idx]].config.get("transition_jitter", 0.0)
                            if isinstance(all_clips[layer_clip_indices[idx]].config, dict)
                            else 0.0
                        )
                        for idx in range(len(layer_clip_indices))
                    ]
                    per_transition_jitters = [j for idx, j in enumerate(layer_jitters) if idx > 0]
                    xfade_filter, xfade_estimated_dur = self._transition_engine.build_xfade_chain(
                        clip_durations=layer_durations,
                        clip_video_labels=layer_labels,
                        transitions=layer_transitions,
                        transition_durations=(
                            per_transition_durations if any(d > 0 for d in per_transition_durations) else None
                        ),
                        jitters=per_transition_jitters if any(j != 0.0 for j in per_transition_jitters) else None,
                        output_label=out_label,
                    )
                    if xfade_filter:
                        filter_parts.append(xfade_filter)
                    logger.info(
                        "[unified-render] layer=%s xfade: clips=%d durations=%s estimated_dur=%.3f",
                        layer.role,
                        len(layer_labels),
                        [round(d, 3) for d in layer_durations],
                        xfade_estimated_dur,
                    )
                layer_output_labels[layer.role] = out_label

        # Step 3: 合成各层
        # 找到主层 — background 优先作为底图，其次 broll / main
        final_video_label = None

        if "background" in layer_output_labels:
            final_video_label = layer_output_labels["background"]
            # b_roll / main 叠加到 background 上
            for role in ("broll", "main"):
                if role in layer_output_labels:
                    base_label = layer_output_labels[role]
                    combined_label = f"combined_{role}"
                    filter_parts.append(f"[{final_video_label}][{base_label}]overlay=(W-w)/2:(H-h)/2[{combined_label}]")
                    final_video_label = combined_label
        else:
            # 无 background 时，取 broll 或 main 作为基础
            for role in ("broll", "main"):
                if role in layer_output_labels:
                    final_video_label = layer_output_labels[role]
                    break

        if final_video_label is None:
            # 没有任何主层，使用第一个层
            final_video_label = layer_output_labels[layers[0].role]

        # 叠加 overlay 层
        for layer in layers:
            if layer.role in ("overlay", "corner_voice"):
                if layer.role not in layer_output_labels:
                    continue
                overlay_label = layer_output_labels[layer.role]
                x, y = layer.position or (
                    self.output_width - int(self.output_width * _PIP_SCALE) - 20,
                    20,
                )
                combined_label = f"combined_{layer.role}"
                filter_parts.append(f"[{final_video_label}][{overlay_label}]overlay={x}:{y}[{combined_label}]")
                final_video_label = combined_label

        # 叠加水印（在字幕之前）
        watermark_config = UnifiedRenderService._resolve_watermark_config(self.plan.config)
        if watermark_config is not None:
            wm_valid, wm_err = watermark_config.validate()
            if wm_valid:
                wm_label = "watermarked"
                if watermark_config.mode == "image":
                    # 图片水印：检查图片是否存在
                    wm_path = Path(watermark_config.image_path)
                    if wm_path.exists():
                        # 图片水印需要额外输入，放在 filter 开头
                        wm_idx = len(all_clips)  # 水印图是最后一个输入
                        wm_scale = int(self.output_width * watermark_config.scale)

                        # 透明度
                        wm_filters = f"scale={wm_scale}:-1"
                        if watermark_config.opacity < 1.0:
                            wm_filters += f",format=rgba,colorchannelmixer=aa={watermark_config.opacity}"

                        filter_parts.insert(0, f"[{wm_idx}:v]{wm_filters}[wm_scaled]")
                        input_args.extend(["-i", str(wm_path)])

                        # 位置计算（水印高度用 scale 后的宽度近似）
                        wm_h = wm_scale  # 近似（正方形假设）
                        x, y = WatermarkEngine.calc_position(
                            watermark_config.position,
                            self.output_width,
                            self.output_height,
                            wm_scale,
                            wm_h,
                            watermark_config.margin_x,
                            watermark_config.margin_y,
                        )

                        # 滚动水印
                        if watermark_config.scroll:
                            x_expr = f"W-mod({watermark_config.scroll_speed}*t\\,W+w)"
                            overlay = f"[{final_video_label}][wm_scaled]overlay=x={x_expr}:y={y}[{wm_label}]"
                        else:
                            overlay = f"[{final_video_label}][wm_scaled]overlay=x={x}:y={y}[{wm_label}]"

                        filter_parts.append(overlay)
                        final_video_label = wm_label
                    else:
                        logger.warning("水印图片不存在，跳过水印: %s", wm_path)
                elif watermark_config.mode == "text":
                    # 文字水印
                    try:
                        text_wm = WatermarkEngine.build_text_watermark_filter(
                            f"[{final_video_label}]",
                            f"[{wm_label}]",
                            watermark_config,
                            self.output_width,
                            self.output_height,
                        )
                        filter_parts.append(text_wm)
                        final_video_label = wm_label
                    except Exception as e:
                        logger.warning("文字水印构建失败，跳过: %s", e)
        # 贴纸叠加（图片贴纸 + 文字贴纸）
        sticker_filter, sticker_extra_inputs = self._build_sticker_filters(final_video_label, "after_stickers")
        if sticker_filter:
            filter_parts.append(sticker_filter)
            # 图片贴纸需要额外输入
            for img_path in sticker_extra_inputs:
                input_args.extend(["-i", img_path])
            final_video_label = "after_stickers"

        # 叠加字幕（如有）+ 最终像素格式
        if ass_path is not None:
            ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:")
            filter_parts.append(f"[{final_video_label}]subtitles='{ass_filter_path}',format=yuv420p[final_video]")
        else:
            filter_parts.append(f"[{final_video_label}]format=yuv420p[final_video]")

        filter_complex = ";".join(filter_parts)
        return filter_complex, input_args

    # ── GPU NVENC 加速 ────────────────────────────────────────────────────

    # ── 全 GPU 直连渲染（P1）─────────────────────────────────────────────

    def _can_use_gpu_direct(self, layers: list[RenderLayer]) -> bool:
        """判断是否命中直连支持的场景：单一主视频轨、全硬切、无复杂合成。"""
        try:
            cfg = self.plan.config or {}
            # 特性开关（默认开启；可经 env/plan config 关闭灰度回退）
            if not bool(cfg.get("gpu_direct_enabled", True)):
                return False

            video_layers = [_lyr for _lyr in layers if l.role not in ("audio",)]
            # 只允许一个视频层，且角色为主层
            if len(video_layers) != 1:
                return False
            role = video_layers[0].role
            if role not in ("main", "broll"):
                return False

            clips_v = [c for c in video_layers[0].clips if c.clip_type != "audio"]
            if not clips_v:
                return False
            # 全硬切（第一个 clip 的转场忽略）
            for c in clips_v[1:]:
                te = c.transition_effect
                if te not in (None, "", "cut"):
                    return False
            # 无画中画 / 水印 / 贴纸 / 片头片尾 / 绿幕 / 倒放 / 调色
            if (cfg or {}).get("pip_config"):
                return False
            if (cfg or {}).get("intro_outro"):
                return False
            for c in clips_v:
                cc = c.config or {}
                if cc.get("watermark") or cc.get("stickers") or cc.get("chroma_key"):
                    return False
                if ReverseConfig.from_dict(cc.get("reverse")).enabled:
                    return False
                cg = ColorGradeConfig.from_dict(cc.get("color_grade"))
                if cg.enabled and cg.has_effect():
                    return False
                if not (c.config or {}).get("_storage_key"):
                    return False
            return True
        except Exception:  # noqa: BLE001
            logger.warning("[gpu-direct] eligibility check failed (fallback)", exc_info=True)
            return False

    def _try_gpu_direct(
        self,
        *,
        layers: list[RenderLayer],
        ass_path: Path | None,
        video_duration: float,
        output_path: Path,
    ) -> bool | None:
        """尝试全 GPU 直连渲染。成功返回 True，不支持/失败返回 None（调用方走旧链路）。"""
        if not self._can_use_gpu_direct(layers):
            return None
        if not self._gpu_encode_available():
            return None

        try:
            from video_processing import gpu_direct_pipeline as gdp

            cfg = self.plan.config or {}
            video_layer = next(_lyr for _lyr in layers if l.role not in ("audio",))
            video_clips = [c for c in video_layer.clips if c.clip_type != "audio"]

            # TTS：把 audio 层的 TTS 片段合并成一个文件给 P4000
            tts_merged: Path | None = None
            audio_layer = next((_lyr for _lyr in layers if l.role == "audio"), None)
            if audio_layer:
                tts_clips = [c for c in audio_layer.clips if (c.config or {}).get("tts") and c.local_path.exists()]
                if tts_clips:
                    tts_merged = self._concat_audio_clips(tts_clips, tag="tts_direct")

            # BGM 本地文件
            bgm_path = Path(self.bgm_path) if self.bgm_path else None
            if bgm_path is not None and not bgm_path.exists():
                bgm_path = None

            # 字幕：标题 + ASR 时间轴
            title_cfg = cfg.get("title", {}) or cfg.get("title_config", {}) or {}
            title_text = ""
            if isinstance(title_cfg, dict) and title_cfg.get("enabled", True):
                title_text = title_cfg.get("text", "") or ""

            subtitle_segments: list[Any] = []
            sub_cfg = cfg.get("subtitle", {}) or {}
            if isinstance(sub_cfg, dict) and sub_cfg.get("enabled", True):
                if sub_cfg.get("auto_generated") and self._asr_timeline_cache is not None:
                    subtitle_segments = list(self._asr_timeline_cache.segments)

            # 边缘裁剪比例（与 random_edge_crop 默认 2~5% 同口径，取固定 3%）
            dedup = self._dedup_enabled()
            edge_pct = 0.03 if dedup else 0.0

            plan = gdp.build_direct_render(
                resolved_clips=video_clips,
                output_width=self.output_width,
                output_height=self.output_height,
                output_fps=self.output_fps,
                tts_audio=tts_merged,
                bgm_audio=bgm_path,
                title_text=title_text,
                subtitle_segments=subtitle_segments,
                edge_crop_pct=edge_pct,
                total_duration=video_duration,
            )

            client = get_gpu_encoder()
            client.render_inputs_to_output(plan.inputs, plan.ffmpeg_args, output_path)

            # 清理本次上传的临时音频
            for key in plan.oss_keys:
                try:
                    from video_processing.oss_helpers import _storage

                    _storage().delete_file(key) if hasattr(_storage(), "delete_file") else None
                except Exception:  # noqa: BLE001
                    pass

            logger.info("[gpu-direct] success: plan_id=%s clips=%d", self.plan.id, len(video_clips))
            return True

        except GpuEncodeError as e:
            logger.warning("[gpu-direct] failed (fallback to legacy): %s", e)
            try:
                if output_path.exists():
                    output_path.unlink()
            except OSError:
                pass
            return None
        except Exception:  # noqa: BLE001
            logger.warning("[gpu-direct] unexpected error (fallback)", exc_info=True)
            return None

    def _concat_audio_clips(self, clips: list[Any], *, tag: str) -> Path:
        """把多个本地音频片段无间隙 concat 成一个 m4a（TTS 分段→单文件）。"""
        out = self.work_dir / f"{tag}_{self.plan.id}.m4a"
        listfile = self.work_dir / f"{tag}_{self.plan.id}.txt"
        lines = []
        for c in clips:
            ap = str(c.local_path).replace("'", "'\\''")
            lines.append(f"file '{ap}'")
        listfile.write_text("\n".join(lines), encoding="utf-8")
        cmd = [
            FFMPEG_BIN,
            "-y",
            "-f",
            "concat",
            "-safe",
            "0",
            "-i",
            str(listfile),
            "-c:a",
            "aac",
            "-b:a",
            "128k",
            str(out),
        ]
        run_ffmpeg(cmd)
        return out

    def _gpu_encode_available(self) -> bool:
        """GPU 编码客户端是否已配置且健康（缓存健康状态，单任务内只探测一次）。"""
        if not getattr(self, "_gpu_health_ok", None):
            client = get_gpu_encoder()
            if client is None:
                self._gpu_health_ok = False
                return False
            try:
                health = client.check_health()
                if health.ready:
                    logger.info(
                        "[gpu-encoder] healthy endpoint=%s gpu=%s",
                        client.endpoint,
                        health.gpu_name,
                    )
                    self._gpu_health_ok = True
                else:
                    logger.warning(
                        "[gpu-encoder] not ready: %s (endpoint=%s)",
                        health.error,
                        client.endpoint,
                    )
                    self._gpu_health_ok = False
            except Exception as e:  # noqa: BLE001
                logger.warning("[gpu-encoder] health probe error (CPU fallback): %s", e)
                self._gpu_health_ok = False
        return self._gpu_health_ok

    def _ffmpeg_output_to_mezzanine(
        self,
        base_command: list[str],
        mezzanine_path: Path,
        output_path: Path,
    ) -> bool:
        """用 CPU ultrafast 把滤镜链输出到 mezzanine_path，然后调 GPU 做最终编码。

        base_command: 原本要执行的完整 ffmpeg 命令（含 -c:v libx264 -crf X -preset Y ... output_path）
        我们把最后一个参数（output_path）替换成 mezzanine_path，并把编码参数改成 ultrafast，
        成功后调用 gpu_encoder 做 nvenc 编码到 output_path。

        任何失败返回 False，调用方走原始 CPU 路径。
        """
        client = get_gpu_encoder()
        if client is None:
            return False

        # 构造 mezzanine 命令：替换编码参数和输出路径
        mezz_cmd = list(base_command)
        # 找到编码参数位置并替换
        try:
            i_crf = mezz_cmd.index("-crf")
            mezz_cmd[i_crf + 1] = "20"
            i_preset = mezz_cmd.index("-preset")
            mezz_cmd[i_preset + 1] = "ultrafast"
        except ValueError:
            logger.warning("[gpu-encoder] could not find -crf/-preset in command, skip gpu")
            return False

        # 如果命令有音频编码 -c:a aac，我们保留音频让 GPU 侧不用单独处理
        # （P4000 的 ffmpeg_args 可以直接 copy 音频？这里简单起见：把音频编码留在 mezzanine，
        # 然后 GPU 侧直接 -c:a copy，避免重编码损失）
        has_audio = "-c:a" in mezz_cmd

        # 替换输出路径（最后一个参数）
        mezz_cmd[-1] = str(mezzanine_path)

        # 1) 跑 mezzanine
        mezzanine_path.parent.mkdir(parents=True, exist_ok=True)
        t0 = time.time()
        try:
            run_ffmpeg(mezz_cmd)
        except subprocess.CalledProcessError as e:
            logger.warning("[gpu-encoder] mezzanine encode failed (CPU fallback): %s", e)
            return False
        logger.info(
            "[gpu-encoder] mezzanine ready: %s (%.1fs, %d bytes), dispatching to P4000 nvenc...",
            mezzanine_path.name,
            time.time() - t0,
            mezzanine_path.stat().st_size if mezzanine_path.exists() else 0,
        )

        # 2) GPU nvenc encode（含上传 mezzanine → OSS → P4000 下载+编码 → relay 回传）
        try:
            # GPU 侧：-i in.mp4 -c:v h264_nvenc ... 音频 copy（mezzanine 里音频已是 aac）
            audio_args = ["-c:a", "copy"] if has_audio else None
            client.encode_mezzanine_to_output(
                mezzanine_path,
                output_path,
                audio_args=audio_args,
            )
            logger.info(
                "[gpu-encoder] GPU nvenc encode done: %s (total %.1fs)",
                output_path.name,
                time.time() - t0,
            )
            return True
        except GpuEncodeError as e:
            logger.warning("[gpu-encoder] GPU encode failed (CPU fallback): %s", e)
            # 删除可能残留的不完整 output
            try:
                if output_path.exists():
                    output_path.unlink()
            except OSError:
                pass
            return False
        except Exception as e:  # noqa: BLE001
            logger.warning("[gpu-encoder] GPU encode unexpected error (CPU fallback): %s", e)
            return False
        finally:
            # 清理 mezzanine
            try:
                if mezzanine_path.exists():
                    mezzanine_path.unlink()
            except OSError:
                pass

    def _execute_ffmpeg(
        self,
        filter_complex: str,
        input_args: list[str],
        output_path: Path,
    ) -> None:
        """执行 FFmpeg 渲染命令。

        失败时记录完整 filter_complex 以便排查（如 exit code 183）。
        """
        command = [
            FFMPEG_BIN,
            "-y",
            *input_args,
            "-filter_complex",
            filter_complex,
            "-map",
            "[final_video]",
            "-c:v",
            "libx264",
            "-crf",
            FFMPEG_ENCODE_CRF,
            "-preset",
            FFMPEG_ENCODE_PRESET,
            "-threads",
            FFMPEG_ENCODE_THREADS,
            "-pix_fmt",
            "yuv420p",
            "-movflags",
            "+faststart",
            str(output_path),
        ]

        logger.info(
            "执行渲染: plan_id=%s inputs=%d output=%s",
            self.plan.id,
            input_args.count("-i"),
            output_path,
        )

        # 尝试 GPU NVENC 加速：先出 ultrafast mezzanine，再交给 P4000 做最终编码
        gpu_ok = False
        if self._gpu_encode_available():
            mezz_path = output_path.parent / f".{output_path.stem}.mezz{output_path.suffix}"
            gpu_ok = self._ffmpeg_output_to_mezzanine(command, mezz_path, output_path)

        if not gpu_ok:
            try:
                run_ffmpeg(command)
            except subprocess.CalledProcessError as e:
                # 额外记录 filter_complex + stderr，方便排查滤镜链构建问题
                stderr_text = (e.stderr or "").strip()
                stderr_tail = stderr_text[-1500:] if len(stderr_text) > 1500 else stderr_text
                logger.error(
                    "渲染失败: plan_id=%s exit_code=%d\nfilter_complex:\n%s\nstderr(last 1500):\n%s",
                    self.plan.id,
                    e.returncode,
                    filter_complex[:5000],
                    stderr_tail,
                )
                raise

    def _build_sticker_filters(self, input_label: str, output_label: str) -> tuple[str, list[str]]:
        """构建贴纸叠加滤镜链.

        Args:
            input_label: 输入视频标签
            output_label: 输出视频标签

        Returns:
            (filter_str, extra_input_paths)
            filter_str: 贴纸滤镜字符串（空表示无贴纸）
            extra_input_paths: 额外需要的输入文件路径（图片贴纸）
        """
        plan_config = getattr(self.plan, "config", None) or {}
        if isinstance(plan_config, dict):
            stickers_data = plan_config.get("stickers", [])
        else:
            stickers_data = []

        if not stickers_data:
            return "", []

        try:
            result = StickerEngine.build_sticker_chain(
                stickers=stickers_data,
                input_label=f"[{input_label}]",
                output_label=f"[{output_label}]",
                canvas_w=self.output_width,
                canvas_h=self.output_height,
            )
            return result.filter_str, result.extra_inputs
        except Exception as e:
            logger.warning("贴纸滤镜构建失败，跳过贴纸: %s", e)
            return "", []

    def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]:
        """探测输出文件的时长、大小、宽高.

        Returns:
            (duration, file_size, width, height)
        """
        info = probe_video_info(str(output_path))
        file_size = output_path.stat().st_size if output_path.exists() else 0
        return (
            info["duration"],
            file_size,
            info["width"],
            info["height"],
        )

    @staticmethod
    def _clip_effective_duration(clip: ResolvedClip) -> float:
        """计算 clip 的有效时长（#1749：目标段长为准，素材不足走末帧冻结）。

        实际实现移至 packages.domain.render_layer_utils.clip_effective_duration
        （duration>0 直接返回；慢放补偿已删除）。
        """
        return _clip_effective_duration_pure(clip.duration, clip.actual_duration)

    # ── 画中画（PiP）相关方法 ──────────────────────────────────────────────────

    def _resolve_pip_sources(self, pip_config: PiPConfig) -> list[tuple[str, PiPLayerConfig, Path]]:
        """解析画中画图层的素材源，返回可用的图层列表.

        降级策略：素材不存在或无效的图层自动跳过，不阻断渲染。

        Returns:
            [(input_label_placeholder, layer_config, local_path), ...]
            input_label 在 build_pip_filters 中会用实际的输入索引替换
        """
        if not pip_config.enabled:
            return []

        engine = PiPEngine(
            output_width=self.output_width,
            output_height=self.output_height,
            output_fps=self.output_fps,
        )

        result = []
        for i, layer in enumerate(pip_config.layers):
            path = engine.validate_layer_source(layer, self.asset_path_map)
            if path is None:
                logger.warning("PiP图层素材不可用，跳过: layer_index=%d source=%s", i, layer.source)
                continue
            # 标签占位，实际输入索引由 build_pip_filters 内部管理
            result.append((f"pip_src_{i}", layer, path))

        return result

    def _append_pip_filters(
        self,
        filter_complex: str,
        input_args: list[str],
        pip_sources: list[tuple[str, Any, Path]],
    ) -> tuple[str, list[str]]:
        """将画中画滤镜追加到 filter_complex 末尾.

        处理逻辑：
        1. 将原 final_video 标签重命名为 pip_base（作为PiP的底层视频）
        2. 追加 PiP 预处理和 overlay 滤镜
        3. PiP 最终输出命名为 final_video

        Args:
            filter_complex: 原 filter_complex 字符串
            input_args: 原输入参数列表
            pip_sources: PiP 素材列表 [(label, layer_config, path), ...]

        Returns:
            (new_filter_complex, new_input_args)
        """
        if not pip_sources:
            return filter_complex, input_args

        pip_engine = PiPEngine(
            output_width=self.output_width,
            output_height=self.output_height,
            output_fps=self.output_fps,
        )

        # 1. 将原 final_video 改为 pip_base
        new_filter = filter_complex.replace("[final_video]", "[pip_base]")

        # 2. 构建 PiP 滤镜链
        # 主输入数量 = len(input_args) // 2（每个输入占 "-i path" 两个参数）
        base_input_idx = len(input_args) // 2
        pip_filter_parts, pip_input_args, final_label = pip_engine.build_pip_filters(
            base_label="pip_base",
            pip_sources=pip_sources,
            base_input_idx=base_input_idx,
        )

        if not pip_filter_parts:
            # 没有有效PiP滤镜，恢复原标签
            return filter_complex, input_args

        # 3. 追加 PiP 滤镜 + 最终格式转换（输出为 final_video）
        pip_filter_str = ";".join(pip_filter_parts)
        final_format = f"[{final_label}]format=yuv420p[final_video]"
        new_filter = f"{new_filter};{pip_filter_str};{final_format}"

        # 4. 追加输入参数
        new_input_args = list(input_args) + pip_input_args

        logger.info(
            "[unified-render] appended PiP filters: layers=%d new_inputs=%d",
            len(pip_sources),
            len(pip_input_args) // 2,
        )

        return new_filter, new_input_args

    @staticmethod
    def _clip_speed(clip: ResolvedClip) -> float:
        """获取 clip 的播放速度，无效值回退到 1.0。

        实际实现移至 packages.domain.render_layer_utils.clip_playback_speed。
        """
        return _clip_playback_speed_pure(getattr(clip, "playback_speed", 1.0))

    def _get_visual_perturbation(self) -> dict:
        # #2034：dedup_enabled=False 时跳过视觉/像素扰动（与 edge_crop、micro_transform 一致）
        if not self._dedup_enabled():
            return {}
        # 读取当前 plan 的视觉扰动参数（plan.config.visual_perturbation）
        perturbation = (self.plan.config or {}).get("visual_perturbation") or {}
        if not perturbation:
            return {}
        result = {
            "hflip": bool(perturbation.get("hflip", False)),
            "zoom_ratio": max(1.0, min(1.2, float(perturbation.get("zoom_ratio", 1.0) or 1.0))),
            "speed_factor": max(0.8, min(1.2, float(perturbation.get("speed_factor", 1.0) or 1.0))),
            "brightness_shift": max(-30, min(30, int(perturbation.get("brightness_shift", 0) or 0))),
        }
        # #1765：同时读取像素级扰动滤镜
        pixel_pert = (self.plan.config or {}).get("pixel_perturbation") or {}
        if pixel_pert:
            result["pixel_perturbation"] = pixel_pert
        return result

    def _apply_visual_perturbation_pre_scale(self, filters: list[str], perturbation: dict) -> None:
        # scale+pad 之前的扰动（hflip），就地修改 filters
        if perturbation.get("hflip"):
            filters.append("hflip")

    def _apply_visual_perturbation_post_scale(self, filters: list[str], perturbation: dict) -> None:
        # scale+pad 之后的扰动（zoom/brightness），就地修改 filters
        zoom = perturbation.get("zoom_ratio", 1.0)
        if zoom > 1.0 + 1e-6:
            zoomed_w = int(self.output_width * zoom)
            zoomed_h = int(self.output_height * zoom)
            filters.append(f"scale={zoomed_w}:{zoomed_h}")
            filters.append(f"crop={self.output_width}:{self.output_height}")
        brightness = perturbation.get("brightness_shift", 0)
        if brightness != 0:
            filters.append(f"eq=brightness={brightness / 100.0:.3f}")

        # #1765：追加像素级扰动滤镜
        pixel_pert = perturbation.get("pixel_perturbation") or {}
        if pixel_pert:
            self._apply_pixel_perturbation(filters, pixel_pert)

    def _apply_pixel_perturbation(self, filters: list[str], pixel_pert: dict) -> None:
        """应用像素级扰动滤镜（Issue #1765）。

        滤镜参数幅度确保肉眼不可见（SSIM > 0.95），但能让同素材不同变体
        在帧级产生 > 3% 的差异，降低平台查重风险。
        """
        filter_list = pixel_pert.get("filters") or []

        for filt in filter_list:
            if filt == "noise":
                # 轻微噪声：noise=alls=0.015:allf=t+u
                strength = pixel_pert.get("noise_strength", 0.015)
                filters.append(f"noise=alls={strength}:allf=t+u")

            elif filt == "unsharp":
                # 锐化/柔化：unsharp=3:3:amount
                # amount > 0 锐化，< 0 柔化
                amount = pixel_pert.get("unsharp_amount", 0.0)
                if abs(amount) > 0.01:
                    filters.append(f"unsharp=3:3:{amount:.2f}")

            elif filt == "curves":
                # 对比度微调：curves 用 preset 或手动定义
                # 简单方案：用 eq=contrast 代替（curves 语法复杂）
                contrast = pixel_pert.get("curves_contrast", 1.0)
                if abs(contrast - 1.0) > 0.01:
                    filters.append(f"eq=contrast={contrast:.3f}")

            elif filt == "color_balance":
                # RGB 通道偏移：colorbalance=rs=...:gs=...:bs=...
                r = pixel_pert.get("color_r", 0)
                g = pixel_pert.get("color_g", 0)
                b = pixel_pert.get("color_b", 0)
                if r != 0 or g != 0 or b != 0:
                    # color_balance 参数范围 -1.0 ~ 1.0，这里用 /100 转换
                    filters.append(f"colorbalance=rs={r / 100:.3f}:gs={g / 100:.3f}:bs={b / 100:.3f}")

    @staticmethod
    def _clip_volume(clip: ResolvedClip) -> float:
        """获取 clip 的音量（config.volume）。缺省 1.0 原声，0.0 静音。"""
        cfg = getattr(clip, "config", None) or {}
        try:
            vol = float(cfg.get("volume", 1.0))
        except (TypeError, ValueError):
            return 1.0
        return max(0.0, vol)

    @staticmethod
    def _clip_adjusted_duration(clip: ResolvedClip) -> float:
        """计算调速后的 clip 实际时长（用于拼接计算）。

        实际实现移至 packages.domain.render_layer_utils.clip_adjusted_duration。
        """
        return _clip_adjusted_duration_pure(
            clip.duration,
            clip.actual_duration,
            getattr(clip, "playback_speed", 1.0),
        )