xiaoxia-saas/tests/unit/test_unified_render_service.py

"""UnifiedRenderService 单元测试.

测试图层分组算法、filter_complex 构建、以及渲染流程。
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, patch

import pytest
from video_processing.render_audio import (
    RenderContext,
    clip_has_audio,
    merge_audio_video,
    mix_audio,
)
from video_processing.render_subtitles import (
    _hex_to_ass_color,
    _position_to_ass_alignment,
    generate_ass_subtitles,
)
from video_processing.unified_render_service import (
    RenderResult,
    ResolvedClip,
    UnifiedRenderService,
    _resolve_layer_role,
)

# ── Fixtures ──────────────────────────────────────────────────────────────────


@dataclass
class FakeClip:
    """模拟 EditPlanClip。"""

    id: str
    plan_id: str = "plan_001"
    clip_type: str = "main"
    order: int = 0
    asset_id: str = ""
    text_content: str = ""
    start_time: float = 0.0
    duration: float = 0.0
    transition_effect: str = "cut"
    transition_duration: float = 0.0
    status: str = "ready"
    playback_speed: float = 1.0
    config: dict[str, Any] = field(default_factory=dict)


@dataclass
class FakePlan:
    """模拟 EditPlan。"""

    id: str = "plan_001"
    name: str = "测试计划"
    config: dict[str, Any] = field(default_factory=dict)


@pytest.fixture(autouse=True)
def _assume_source_clips_have_audio():
    """默认假设测试中的视频素材都带音频流。

    新行为：保留视频素材原声（不再无条件丢弃）。需要模拟无音频流的用例
    自行 patch probe_has_audio=False（如 test_mix_audio_main_no_audio_stream_returns_none）。
    """
    with patch("video_processing.render_audio.probe_has_audio", return_value=True):
        yield


def _make_clip(
    clip_id: str,
    clip_type: str = "main",
    order: int = 0,
    asset_id: str = "",
    duration: float = 0.0,
    transition_effect: str = "cut",
    transition_duration: float = 0.0,
    config: dict[str, Any] | None = None,
    playback_speed: float = 1.0,
) -> FakeClip:
    return FakeClip(
        id=clip_id,
        clip_type=clip_type,
        order=order,
        asset_id=asset_id or f"asset_{clip_id}.mp4",
        duration=duration,
        transition_effect=transition_effect,
        transition_duration=transition_duration,
        config=config or {},
        playback_speed=playback_speed,
    )


def _make_service(
    clips: list[FakeClip] | None = None,
    asset_paths: dict[str, Path] | None = None,
    work_dir: Path | None = None,
    output_fps: int = 25,
    output_width: int = 1280,
    output_height: int = 720,
) -> UnifiedRenderService:
    """创建测试用的 UnifiedRenderService 实例。

    如果未提供 asset_paths，自动从 clips 生成默认映射
    （asset_id → /tmp/asset_{clip_id}.mp4）。
    """
    plan = FakePlan()
    clips = clips or []
    work_dir = work_dir or Path("/tmp/test_render")
    if asset_paths is None:
        asset_paths = {}
        for c in clips:
            if c.asset_id:
                asset_paths[c.asset_id] = Path(f"/tmp/{c.asset_id}")
    return UnifiedRenderService(
        plan=plan,
        clips=clips,
        asset_path_map=asset_paths,
        work_dir=work_dir,
        output_width=output_width,
        output_height=output_height,
        output_fps=output_fps,
    )


def _patch_path_exists():
    """Patch Path.exists() 让测试路径返回 True。"""
    return patch("pathlib.Path.exists", return_value=True)


def _make_ctx() -> RenderContext:
    """创建测试用 RenderContext。"""
    return RenderContext(work_dir=Path("/tmp/test_render"), plan_id="plan_001")


# ── 测试 _resolve_layer_role ─────────────────────────────────────────────────


class TestResolveLayerRole:
    """测试 clip_type → layer role 映射。"""

    def test_main_default(self):
        assert _resolve_layer_role("main", {}) == "main"

    def test_main_with_b_roll_role(self):
        assert _resolve_layer_role("main", {"role": "b_roll"}) == "broll"

    def test_overlay(self):
        assert _resolve_layer_role("overlay", {}) == "overlay"

    def test_background(self):
        assert _resolve_layer_role("background", {}) == "background"

    def test_corner_voice(self):
        assert _resolve_layer_role("corner_voice", {}) == "corner_voice"

    def test_b_roll(self):
        assert _resolve_layer_role("b_roll", {}) == "broll"

    def test_intro(self):
        assert _resolve_layer_role("intro", {}) == "main"

    def test_outro(self):
        assert _resolve_layer_role("outro", {}) == "main"


# ── 测试图层分组 ──────────────────────────────────────────────────────────────


class TestGroupClipsIntoLayers:
    """测试 _group_clips_into_layers 方法。"""

    def test_group_clips_one_take(self):
        """4 个 main clips → 1 个 main_layer。"""
        clips = [
            _make_clip("c1", "main", order=0),
            _make_clip("c2", "main", order=1),
            _make_clip("c3", "main", order=2),
            _make_clip("c4", "main", order=3),
        ]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)

        assert len(layers) == 1
        assert layers[0].role == "main"
        assert len(layers[0].clips) == 4
        assert layers[0].z_index == 0

    def test_group_clips_pip(self):
        """1 main + 2 overlay → main_layer + overlay_layer。"""
        clips = [
            _make_clip("c1", "main", order=0),
            _make_clip("c2", "overlay", order=1),
            _make_clip("c3", "overlay", order=2),
        ]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)

        roles = {lyr.role for lyr in layers}
        assert "main" in roles
        assert "overlay" in roles

        main_layer = next(lyr for lyr in layers if lyr.role == "main")
        overlay_layer = next(lyr for lyr in layers if lyr.role == "overlay")
        assert len(main_layer.clips) == 1
        assert len(overlay_layer.clips) == 2
        assert overlay_layer.z_index > main_layer.z_index

    def test_group_clips_voice_over(self):
        """3 个 main(b_roll) clips → 1 个 broll_layer。"""
        clips = [
            _make_clip("c1", "main", order=0, config={"role": "b_roll"}),
            _make_clip("c2", "main", order=1, config={"role": "b_roll"}),
            _make_clip("c3", "main", order=2, config={"role": "b_roll"}),
        ]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)

        assert len(layers) == 1
        assert layers[0].role == "broll"
        assert len(layers[0].clips) == 3

    def test_group_clips_voice_pip(self):
        """1 background + 1 corner_voice + 2 b_roll → 3 layers。"""
        clips = [
            _make_clip("c1", "background", order=0),
            _make_clip("c2", "corner_voice", order=1),
            _make_clip("c3", "b_roll", order=2),
            _make_clip("c4", "b_roll", order=3),
        ]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)

        roles = {lyr.role for lyr in layers}
        assert roles == {"background", "corner_voice", "broll"}
        assert len(layers) == 3

        # z_index 排序
        assert layers[0].z_index <= layers[1].z_index <= layers[2].z_index

    def test_group_clips_intro_outro(self):
        """intro + 2 main + outro → 1 main_layer（4 clips，按 order 排序）。"""
        clips = [
            _make_clip("intro", "intro", order=0),
            _make_clip("c1", "main", order=1),
            _make_clip("c2", "main", order=2),
            _make_clip("outro", "outro", order=3),
        ]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)

        assert len(layers) == 1
        assert layers[0].role == "main"
        assert len(layers[0].clips) == 4
        # 按 order 排序
        orders = [c.order for c in layers[0].clips]
        assert orders == [0, 1, 2, 3]


# ── 测试 _resolve_clips ──────────────────────────────────────────────────────


class TestResolveClips:
    """测试 _resolve_clips 方法。"""

    def test_skip_missing_asset(self):
        """跳过 asset_id 在 asset_path_map 中找不到的 clip。"""
        clips = [
            _make_clip("c1", "main", order=0, asset_id="asset_1.mp4"),
            _make_clip("c2", "main", order=1, asset_id="missing.mp4"),
        ]
        # 只有 asset_1.mp4 存在
        asset_paths = {"asset_1.mp4": Path("/tmp/asset_1.mp4")}

        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()

        assert len(resolved) == 1
        assert resolved[0].clip_id == "c1"

    def test_skip_empty_asset_id(self):
        """跳过 asset_id 为空的 clip。"""
        clips = [
            _make_clip("c1", "main", order=0, asset_id=""),
            _make_clip("c2", "main", order=1, asset_id="asset_2.mp4"),
        ]
        asset_paths = {"asset_2.mp4": Path("/tmp/asset_2.mp4")}

        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()

        assert len(resolved) == 1
        assert resolved[0].clip_id == "c2"

    def test_sort_by_order(self):
        """解析后的 clips 按 order 排序。"""
        clips = [
            _make_clip("c3", "main", order=3, asset_id="a3.mp4"),
            _make_clip("c1", "main", order=1, asset_id="a1.mp4"),
            _make_clip("c2", "main", order=2, asset_id="a2.mp4"),
        ]
        asset_paths = {
            "a1.mp4": Path("/tmp/a1.mp4"),
            "a2.mp4": Path("/tmp/a2.mp4"),
            "a3.mp4": Path("/tmp/a3.mp4"),
        }

        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()

        orders = [c.order for c in resolved]
        assert orders == [1, 2, 3]


# ── 测试 _build_filter_complex ───────────────────────────────────────────────


class TestBuildFilterComplex:
    """测试 _build_filter_complex 方法。"""

    def test_single_layer_single_clip(self):
        """只有 1 个 main clip → 简单 scale + setpts。"""
        clips = [_make_clip("c1", "main", order=0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, input_args = svc._build_filter_complex(layers)

        assert "-i" in input_args
        assert "/tmp/asset_c1.mp4" in input_args
        assert "scale=" in fc
        assert "[final_video]" in fc

    def test_single_layer_multi_clips(self):
        """多个 main clips（默认硬切）→ concat 串联。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=3.0),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, input_args = svc._build_filter_complex(layers)

        assert input_args.count("-i") == 2
        # 全硬切场景走 concat filter（性能远优于 xfade）
        assert "concat=n=2:v=1:a=0" in fc
        assert "[final_video]" in fc

    def test_single_layer_multi_clips_with_transition(self):
        """多个 main clips 带转场效果 → xfade 串联。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip(
                "c2",
                "main",
                order=1,
                duration=3.0,
                transition_effect="fade",
                transition_duration=0.5,
            ),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, input_args = svc._build_filter_complex(layers)

        assert input_args.count("-i") == 2
        assert "xfade=" in fc
        assert "[final_video]" in fc

    def test_with_overlay(self):
        """main + overlay → overlay 滤镜。"""
        clips = [
            _make_clip("c1", "main", order=0),
            _make_clip("c2", "overlay", order=1),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, input_args = svc._build_filter_complex(layers)

        assert "overlay=" in fc
        assert "[final_video]" in fc

    def test_setpts_before_fps_in_xfade_inputs(self):
        """多视频 xfade 模式：setpts=PTS-STARTPTS 必须在 fps 之前，确保 xfade 时各片段 PTS 一致。

        构造两个不同时长的视频片段，验证生成的 filter_complex 中每个片段的
        预处理滤镜链里 setpts 都在 fps 前面。
        """
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=5.0, transition_effect="fade", transition_duration=0.5),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        # 确保 xfade 存在
        assert "xfade=" in fc

        # 提取每个 clip 的预处理滤镜链（[i:v]...[vi] 部分）
        # 验证：每个 clip 滤镜链中，setpts=PTS-STARTPTS 的最后一次出现
        # 必须在 fps= 的前面（PTS 归一化后再统一帧率）
        import re

        clip_pattern = re.compile(r"\[(\d+):v\](.+?)\[v\d+\]")
        matches = clip_pattern.findall(fc)
        assert len(matches) == 2, f"Expected 2 clip preprocessing chains, got {len(matches)}"

        for idx, chain_str in matches:
            # 找到所有 setpts 和 fps 的位置
            setpts_positions = [m.start() for m in re.finditer(r"setpts=PTS-STARTPTS", chain_str)]
            fps_positions = [m.start() for m in re.finditer(r"fps=\d+", chain_str)]

            assert setpts_positions, f"clip {idx}: 未找到 setpts=PTS-STARTPTS"
            assert fps_positions, f"clip {idx}: 未找到 fps="

            # 最后一个 setpts 必须在第一个 fps 之前
            last_setpts = max(setpts_positions)
            first_fps = min(fps_positions)
            assert last_setpts < first_fps, (
                f"clip {idx}: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。"
                f"滤镜链: {chain_str}"
            )

    def test_setpts_before_fps_single_clip(self):
        """单视频模式（一镜到底）：setpts 也必须在 fps 之前。

        单视频虽然没有 xfade，但滤镜链顺序应保持一致，确保 PTS 处理逻辑统一。
        """
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        import re

        clip_pattern = re.compile(r"\[(\d+):v\](.+?)\[v\d+\]")
        matches = clip_pattern.findall(fc)
        assert len(matches) == 1

        chain_str = matches[0][1]
        setpts_positions = [m.start() for m in re.finditer(r"setpts=PTS-STARTPTS", chain_str)]
        fps_positions = [m.start() for m in re.finditer(r"fps=\d+", chain_str)]

        assert setpts_positions, "单视频: 未找到 setpts=PTS-STARTPTS"
        assert fps_positions, "单视频: 未找到 fps="

        last_setpts = max(setpts_positions)
        first_fps = min(fps_positions)
        assert last_setpts < first_fps, (
            f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" f"滤镜链: {chain_str}"
        )

    def test_main_clip_uses_pad_strategy(self):
        """main clip 使用 scale+pad 保持比例留黑边（不裁剪内容）。

        渲染引擎不能挑素材，任何素材进来都统一规格后出片；pad留黑边保证内容完整。
        """
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        # 验证：scale 使用 force_original_aspect_ratio=decrease（等比缩小）
        assert "force_original_aspect_ratio=decrease" in fc
        # 验证：有 pad（留黑边）
        assert "pad=" in fc
        assert "black" in fc
        # 验证：没有 crop（不是裁剪模式）
        assert "crop=" not in fc

    def test_broll_clip_uses_pad_strategy(self):
        """broll clip 同样使用 scale+pad 留黑边策略。"""
        clips = [_make_clip("c1", "b_roll", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        assert "force_original_aspect_ratio=decrease" in fc
        assert "pad=" in fc
        assert "crop=" not in fc

    def test_background_uses_fill_crop_strategy(self):
        """background 层也使用铺满裁剪（已有的行为，保持一致）。"""
        clips = [_make_clip("c1", "background", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        assert "force_original_aspect_ratio=increase" in fc
        assert "crop=1280:720" in fc

    def test_empty_layers_raises(self):
        """空图层列表抛出 ValueError。"""
        svc = _make_service()
        with pytest.raises(ValueError, match="没有可渲染的图层"):
            svc._build_filter_complex([])


class TestPassThrough:
    """测试单图层单 clip 直通优化路径。"""

    def test_can_use_pass_through_single_main_clip(self):
        """1个main图层 + 1个clip → 可以直通。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            assert svc._can_use_pass_through(layers) is True

    def test_can_use_pass_through_single_broll_clip(self):
        """1个broll图层 + 1个clip → 可以直通。"""
        clips = [_make_clip("c1", "b_roll", order=0, duration=5.0)]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            assert svc._can_use_pass_through(layers) is True

    def test_can_use_pass_through_single_background_clip(self):
        """1个background图层 + 1个clip → 可以直通。"""
        clips = [_make_clip("c1", "background", order=0, duration=5.0)]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            assert svc._can_use_pass_through(layers) is True

    def test_cannot_pass_through_multi_clips(self):
        """1个图层 + 多个clips → 不能直通（需要xfade）。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=3.0),
        ]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            assert svc._can_use_pass_through(layers) is False

    def test_cannot_pass_through_multi_layers(self):
        """多个图层 → 不能直通。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip("c2", "overlay", order=1, duration=5.0),
        ]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            assert svc._can_use_pass_through(layers) is False

    def test_cannot_pass_through_overlay_only(self):
        """只有overlay图层 → 不能直通（需要叠加到主层）。"""
        clips = [_make_clip("c1", "overlay", order=0, duration=5.0)]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            assert svc._can_use_pass_through(layers) is False

    def test_render_uses_pass_through_for_single_clip(self):
        """单clip渲染时走直通路径（调用_render_pass_through而非_execute_ffmpeg）。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch.object(svc, "_render_pass_through") as mock_pass,
            patch.object(svc, "_execute_ffmpeg") as mock_exec,
            patch("video_processing.unified_render_service.mix_audio", return_value=None),
            patch("shutil.copy2"),
            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
        ):
            result = svc.render()

        mock_pass.assert_called_once()
        mock_exec.assert_not_called()
        assert result.duration == 5.0

    def test_render_uses_filter_complex_for_multi_clips(self):
        """多clip渲染时走完整filter_complex路径。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=3.0),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch.object(svc, "_render_pass_through") as mock_pass,
            patch.object(svc, "_execute_ffmpeg") as mock_exec,
            patch("video_processing.unified_render_service.mix_audio", return_value=None),
            patch("shutil.copy2"),
            patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)),
        ):
            result = svc.render()

        mock_pass.assert_not_called()
        mock_exec.assert_called_once()
        assert result.duration == 5.5


# ── 测试 ASS 字幕生成 ────────────────────────────────────────────────────────


class TestAssSubtitles:
    """测试 ASS 字幕生成功能。"""

    def test_hex_to_ass_color_white(self):
        """#ffffff → &HFFFFFF（ASS 是 BGR 顺序）。"""
        assert _hex_to_ass_color("#ffffff") == "&HFFFFFF"

    def test_hex_to_ass_color_black(self):
        """#000000 → &H000000。"""
        assert _hex_to_ass_color("#000000") == "&H000000"

    def test_hex_to_ass_color_red(self):
        """#ff0000 红 → &H0000FF（B=00, G=00, R=FF）。"""
        assert _hex_to_ass_color("#ff0000") == "&H0000FF"

    def test_hex_to_ass_color_blue(self):
        """#0000ff 蓝 → &HFF0000（B=FF, G=00, R=00）。"""
        assert _hex_to_ass_color("#0000ff") == "&HFF0000"

    def test_hex_to_ass_color_no_hash(self):
        """不带 # 的颜色值也能解析。"""
        assert _hex_to_ass_color("ff0000") == "&H0000FF"

    def test_position_to_ass_alignment_top(self):
        """top → 8（顶部居中）。"""
        assert _position_to_ass_alignment("top") == 8

    def test_position_to_ass_alignment_center(self):
        """center → 5（居中）。"""
        assert _position_to_ass_alignment("center") == 5

    def test_position_to_ass_alignment_bottom(self):
        """bottom → 2（底部居中）。"""
        assert _position_to_ass_alignment("bottom") == 2

    def test_generate_ass_with_title_only(self, tmp_path):
        """只有标题时生成 ASS 文件。"""
        ass_path = tmp_path / "test.ass"
        result = generate_ass_subtitles(
            ass_path,
            video_width=1280,
            video_height=720,
            video_duration=10.0,
            title_text="测试标题",
            title_config={
                "enabled": True,
                "font": "思源黑体",
                "size": 48,
                "color": "#ffffff",
                "bold": True,
                "position": "top",
                "stroke": {"enabled": True, "color": "#000000", "width": 2},
                "shadow": {"enabled": True, "blur": 4, "offset_x": 2, "offset_y": 2},
            },
        )

        assert result == ass_path
        content = ass_path.read_text(encoding="utf-8")
        assert "[Script Info]" in content
        assert "PlayResX: 1280" in content
        assert "PlayResY: 720" in content
        assert "[V4+ Styles]" in content
        assert "TitleStyle" in content
        assert "测试标题" in content
        assert "Dialogue:" in content

    def test_generate_ass_with_subtitle_only(self, tmp_path):
        """只有字幕时生成 ASS 文件。"""
        ass_path = tmp_path / "test.ass"
        generate_ass_subtitles(
            ass_path,
            video_width=1280,
            video_height=720,
            video_duration=10.0,
            subtitle_text="测试字幕内容",
            subtitle_config={
                "enabled": True,
                "font": "思源黑体",
                "size": 24,
                "color": "#ffffff",
                "position": "bottom",
            },
        )

        content = ass_path.read_text(encoding="utf-8")
        assert "SubtitleStyle" in content
        assert "测试字幕内容" in content
        assert "Dialogue:" in content

    def test_generate_ass_with_both_title_and_subtitle(self, tmp_path):
        """同时有标题和字幕。"""
        ass_path = tmp_path / "test.ass"
        generate_ass_subtitles(
            ass_path,
            video_width=1280,
            video_height=720,
            video_duration=10.0,
            title_text="大标题",
            title_config={"enabled": True, "position": "top"},
            subtitle_text="底部字幕",
            subtitle_config={"enabled": True, "position": "bottom"},
        )

        content = ass_path.read_text(encoding="utf-8")
        assert "TitleStyle" in content
        assert "SubtitleStyle" in content
        assert "大标题" in content
        assert "底部字幕" in content
        # 两条 Dialogue 行
        assert content.count("Dialogue:") == 2

    def test_generate_ass_disabled_returns_empty(self, tmp_path):
        """标题和字幕都禁用时返回空文件。"""
        ass_path = tmp_path / "test.ass"
        generate_ass_subtitles(
            ass_path,
            video_width=1280,
            video_height=720,
            video_duration=10.0,
            title_text="不显示",
            title_config={"enabled": False},
            subtitle_text="也不显示",
            subtitle_config={"enabled": False},
        )

        content = ass_path.read_text(encoding="utf-8")
        assert content == ""

    def test_generate_ass_empty_text_returns_empty(self, tmp_path):
        """文本为空时不生成字幕。"""
        ass_path = tmp_path / "test.ass"
        generate_ass_subtitles(
            ass_path,
            video_width=1280,
            video_height=720,
            video_duration=10.0,
            title_text="",
            title_config={"enabled": True},
            subtitle_text="   ",
            subtitle_config={"enabled": True},
        )

        content = ass_path.read_text(encoding="utf-8")
        assert content == ""

    def test_generate_ass_time_format(self, tmp_path):
        """验证 ASS 时间格式正确（H:MM:SS.cc）。"""
        ass_path = tmp_path / "test.ass"
        generate_ass_subtitles(
            ass_path,
            video_width=1280,
            video_height=720,
            video_duration=125.5,  # 2分5.5秒
            title_text="测试",
            title_config={"enabled": True},
        )

        content = ass_path.read_text(encoding="utf-8")
        # 结束时间应该是 0:02:05.50
        assert "0:02:05.50" in content

    def test_ass_text_escape_newlines(self, tmp_path):
        """换行符转义为 ASS 的 \\N。"""
        ass_path = tmp_path / "test.ass"
        generate_ass_subtitles(
            ass_path,
            video_width=1280,
            video_height=720,
            video_duration=10.0,
            title_text="第一行\n第二行",
            title_config={"enabled": True},
        )

        content = ass_path.read_text(encoding="utf-8")
        assert "第一行\\N第二行" in content


# ── 测试 render 方法 ─────────────────────────────────────────────────────────


class TestRender:
    """测试 render 方法。"""

    def test_render_empty_clips_raises(self):
        """没有 clips 时抛出 ValueError。"""
        svc = _make_service(clips=[], asset_paths={})
        with pytest.raises(ValueError, match="没有可渲染的片段"):
            svc.render()

    def test_render_with_missing_assets_raises(self):
        """所有 clips 素材缺失时抛出 ValueError。"""
        clips = [_make_clip("c1", "main", order=0, asset_id="missing.mp4")]
        svc = _make_service(clips, asset_paths={})
        with pytest.raises(ValueError, match="没有可渲染的片段"):
            svc.render()

    def test_render_success_single_clip(self):
        """单clip正常渲染（走直通路径）。"""
        clips = [_make_clip("c1", "main", order=0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch.object(svc, "_render_pass_through") as mock_pass,
            patch("video_processing.unified_render_service.mix_audio", return_value=None),
            patch("shutil.copy2"),
            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
        ):
            result = svc.render()

        assert isinstance(result, RenderResult)
        assert result.duration == 5.0
        assert result.file_size == 1024
        assert result.width == 1280
        assert result.height == 720
        mock_pass.assert_called_once()

    def test_render_success_multi_clips(self):
        """多clip正常渲染（走完整filter_complex路径）。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=3.0),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch.object(svc, "_execute_ffmpeg") as mock_exec,
            patch("video_processing.unified_render_service.mix_audio", return_value=None),
            patch("shutil.copy2"),
            patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)),
        ):
            result = svc.render()

        assert isinstance(result, RenderResult)
        assert result.duration == 5.5
        assert result.file_size == 2048
        assert result.width == 1280
        assert result.height == 720
        mock_exec.assert_called_once()


# ── 测试音频后处理 ──────────────────────────────────────────────────────────


class TestAudioMixing:
    """测试音频后处理混音功能。"""

    def test_clip_effective_duration_with_both(self):
        """指定时长和实际时长都有时取较小值。"""
        clip = ResolvedClip(
            clip_id="c1",
            asset_id="a1",
            local_path=Path("/tmp/c1.mp4"),
            clip_type="main",
            order=0,
            duration=3.0,
            actual_duration=5.0,
        )
        assert UnifiedRenderService._clip_effective_duration(clip) == 3.0

    def test_clip_effective_duration_only_actual(self):
        """只有实际时长时用实际时长。"""
        clip = ResolvedClip(
            clip_id="c1",
            asset_id="a1",
            local_path=Path("/tmp/c1.mp4"),
            clip_type="main",
            order=0,
            duration=0.0,
            actual_duration=5.0,
        )
        assert UnifiedRenderService._clip_effective_duration(clip) == 5.0

    def test_clip_effective_duration_only_specified(self):
        """只有指定时长时用指定时长。"""
        clip = ResolvedClip(
            clip_id="c1",
            asset_id="a1",
            local_path=Path("/tmp/c1.mp4"),
            clip_type="main",
            order=0,
            duration=3.0,
            actual_duration=0.0,
        )
        assert UnifiedRenderService._clip_effective_duration(clip) == 3.0

    def test_clip_effective_duration_zero(self):
        """都没有时返回0。"""
        clip = ResolvedClip(
            clip_id="c1",
            asset_id="a1",
            local_path=Path("/tmp/c1.mp4"),
            clip_type="main",
            order=0,
            duration=0.0,
            actual_duration=0.0,
        )
        assert UnifiedRenderService._clip_effective_duration(clip) == 0.0

    def test_mix_audio_single_main_clip(self):
        """只有 main clip（无独立音频轨）→ 保留源视频原声，走单轨 concat。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        # 保留源视频原声
        assert result is not None
        mock_run.assert_called_once()
        assert "asset_c1.mp4" in " ".join(mock_run.call_args[0][0])

    def test_mix_audio_multi_main_clips(self):
        """main 原声与独立音频轨通过 amix 混音。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("tts1", "main", order=0, duration=3.0, config={"role": "audio"}),
            _make_clip("tts2", "main", order=1, duration=2.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
            "asset_tts2.mp4": Path("/tmp/asset_tts2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 4.5)

        assert result is not None
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)
        # 原声 + 2 个独立音频轨 → amix 混音（3 路输入）
        assert "amix=inputs=3" in cmd_str
        # main 原声 c1 参与混音
        assert "asset_c1.mp4" in cmd_str
        assert "asset_tts1.mp4" in cmd_str
        assert "asset_tts2.mp4" in cmd_str

    def test_mix_audio_with_independent_audio_track(self):
        """main 原声与独立音频轨通过 amix 混音。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip(
                "bgm1",
                "main",
                order=0,
                duration=5.0,
                config={"role": "audio", "volume": 0.5},
            ),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        assert result is not None
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)
        # main 原声 c1 与独立音频轨 bgm1 都参与 amix
        assert "amix=inputs=2" in cmd_str
        assert "asset_c1.mp4" in cmd_str
        assert "asset_bgm1.mp4" in cmd_str

    def test_mix_with_independent_audio_amix(self):
        """直接验证 mix_with_independent_audio：main 音频 + 独立音频轨走 amix 混音。"""
        from video_processing.render_audio import mix_with_independent_audio
        from video_processing.unified_render_service import ResolvedClip

        main = ResolvedClip(
            clip_id="c1",
            asset_id="asset_c1",
            clip_type="main",
            order=0,
            local_path=Path("/tmp/asset_c1.mp4"),
            duration=5.0,
            actual_duration=5.0,
            config={},
        )
        aud = ResolvedClip(
            clip_id="bgm1",
            asset_id="asset_bgm1",
            clip_type="main",
            order=1,
            local_path=Path("/tmp/asset_bgm1.mp4"),
            duration=5.0,
            actual_duration=5.0,
            config={"volume": 0.5},
        )
        ctx = _make_ctx()
        output_path = ctx.work_dir / "audio_plan_001.aac"

        with patch("video_processing.render_audio.run_ffmpeg") as mock_run:
            mix_with_independent_audio(ctx, [main], [aud], output_path, 5.0)

        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)
        assert "amix" in cmd_str
        assert "volume=0.5" in cmd_str
        assert "asset_c1.mp4" in cmd_str
        assert "asset_bgm1.mp4" in cmd_str

    def test_mix_audio_no_audio_returns_none(self):
        """没有音频素材时返回None。"""
        # 构造一个没有音频的场景（比如纯文字）
        clips = [_make_clip("t1", "title", order=0, duration=3.0)]
        clips[0].asset_id = ""  # 无素材
        asset_paths: dict[str, Path] = {}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=3.0),
        ):
            # 没有素材的clip会被跳过，layers为空
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 3.0)

        assert result is None

    def test_mix_audio_background_not_used_as_main(self):
        """background 图层不参与主音频；main 原声与独立音频轨混音。"""
        clips = [
            _make_clip("bg1", "background", order=0, duration=5.0),
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip("tts1", "main", order=0, duration=5.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_bg1.mp4": Path("/tmp/asset_bg1.mp4"),
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        assert result is not None
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)
        # background 图层不参与主音频
        assert "asset_bg1.mp4" not in cmd_str
        # main 原声 + 独立音频轨都参与
        assert "amix=inputs=2" in cmd_str
        assert "asset_c1.mp4" in cmd_str
        assert "asset_tts1.mp4" in cmd_str

    def test_mix_audio_main_priority_over_broll(self):
        """main 图层优先作为主音频，broll 不参与；与独立音频轨 amix。"""
        clips = [
            _make_clip("b1", "b_roll", order=0, duration=5.0),
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip("tts1", "main", order=0, duration=5.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_b1.mp4": Path("/tmp/asset_b1.mp4"),
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        assert result is not None
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)
        # main 原声 c1 优先参与；broll b1 不参与主音频
        assert "amix=inputs=2" in cmd_str
        assert "asset_c1.mp4" in cmd_str
        assert "asset_b1.mp4" not in cmd_str
        assert "asset_tts1.mp4" in cmd_str

    def test_mix_audio_broll_used_when_no_main(self):
        """无 main 图层时 broll 原声作为主音频。"""
        clips = [_make_clip("b1", "b_roll", order=0, duration=5.0)]
        asset_paths = {"asset_b1.mp4": Path("/tmp/asset_b1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        # 保留 broll 原声
        assert result is not None
        mock_run.assert_called_once()
        assert "asset_b1.mp4" in " ".join(mock_run.call_args[0][0])

    def test_mix_audio_single_clip_truncated_to_video_duration(self):
        """主音频截断到 video_duration（video_duration < clip有效时长）。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=10.0),
            _make_clip("tts1", "main", order=0, duration=10.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            # video_duration 只有 3.0，小于独立音频轨的 10.0
            result = mix_audio(ctx, layers, 3.0)

        assert result is not None
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        # 验证截断到 3.0（-t 3.0 或 atrim=0:3.000）
        cmd_str = " ".join(cmd)
        assert "3.000" in cmd_str or "3.0" in cmd_str
        # main 原声参与
        assert "asset_c1.mp4" in cmd_str

    def test_merge_audio_video(self):
        """合并音视频命令正确。"""
        video_path = Path("/tmp/video.mp4")
        audio_path = Path("/tmp/audio.aac")
        output_path = Path("/tmp/output.mp4")
        ctx = _make_ctx()

        with patch("video_processing.render_audio.run_ffmpeg") as mock_run:
            merge_audio_video(ctx, video_path, audio_path, output_path)

        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        assert "-c:v" in cmd
        assert "copy" in cmd
        assert "-map" in cmd
        assert "-shortest" in cmd

    def test_render_calls_audio_mixing(self):
        """多clip完整render流程会调用音频混音（非直通路径）。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=2.0),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch.object(svc, "_execute_ffmpeg"),
            patch("video_processing.unified_render_service.mix_audio", return_value=Path("/tmp/audio.aac")) as mock_mix,
            patch("video_processing.unified_render_service.merge_audio_video") as mock_merge,
            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
        ):
            result = svc.render()

        mock_mix.assert_called_once()
        mock_merge.assert_called_once()
        assert result.duration == 5.0

    def test_render_without_audio_copies_video(self):
        """多clip无音频时走copy路径（非直通路径）。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=2.0),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch.object(svc, "_execute_ffmpeg"),
            patch("video_processing.unified_render_service.mix_audio", return_value=None),
            patch("shutil.copy2") as mock_copy,
            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
        ):
            result = svc.render()

        mock_copy.assert_called_once()
        assert result.duration == 5.0

    def test_render_pass_through_skips_audio_mix(self):
        """直通场景下视频+音频一次完成，跳过 _mix_audio 和 _merge_audio_video。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch.object(svc, "_render_pass_through", return_value=True) as mock_pt,
            patch("video_processing.unified_render_service.mix_audio") as mock_mix,
            patch("video_processing.unified_render_service.merge_audio_video") as mock_merge,
            patch("shutil.copy2") as mock_copy,
            patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
        ):
            result = svc.render()

        # 直通场景调用了 _render_pass_through，跳过了 _mix_audio / _merge / copy
        mock_pt.assert_called_once()
        mock_mix.assert_not_called()
        mock_merge.assert_not_called()
        mock_copy.assert_not_called()
        assert result.duration == 5.0

    def test_pass_through_main_has_aac_audio(self):
        """直通main/broll场景输出带aac音频。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            result = svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=5.0)

        assert result is True  # main 类型返回有音频
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        assert "-an" not in cmd  # 不再是无声
        assert "aac" in cmd  # 有aac音频编码
        assert "-b:a" in cmd

    def test_pass_through_background_no_audio(self):
        """直通background场景不带音频（图片素材）。"""
        clips = [_make_clip("bg1", "background", order=0, duration=5.0)]
        asset_paths = {"asset_bg1.mp4": Path("/tmp/asset_bg1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            result = svc._render_pass_through(layers, Path("/tmp/out.mp4"))

        assert result is False  # background 返回无音频
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        assert "aac" not in cmd  # 没有音频编码参数

    # ── 无音轨视频防御测试 ──

    def test_mix_audio_main_no_audio_stream_returns_none(self):
        """主图层clip无音频流且无独立音频轨时，返回None（不报错）。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.probe_has_audio", return_value=False),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        assert result is None
        # 没有音频流时不应调用 FFmpeg
        mock_run.assert_not_called()

    def test_mix_audio_partial_clips_no_audio_filtered(self):
        """main 原声正常保留；无音频流的 clip 被过滤。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=2.0),
            _make_clip("tts1", "main", order=0, duration=2.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        assert result is not None
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)
        # main 原声 c1/c2 + 独立音频轨 tts1 全部参与
        assert "asset_c1.mp4" in cmd_str
        assert "asset_c2.mp4" in cmd_str
        assert "asset_tts1.mp4" in cmd_str

    def test_mix_audio_main_plus_independent_amix(self):
        """main 原声与独立音频轨 amix 混音。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip(
                "bgm1",
                "main",
                order=0,
                duration=5.0,
                config={"role": "audio", "volume": 0.5},
            ),  # 独立音频轨
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        assert result is not None
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)
        # main 原声 c1 与独立音频轨 bgm1 都参与 amix
        assert "amix=inputs=2" in cmd_str
        assert "asset_c1.mp4" in cmd_str
        assert "asset_bgm1.mp4" in cmd_str

    def test_mix_audio_both_no_audio_returns_none(self):
        """主图层和独立音频轨都无音频时，返回None。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip(
                "bgm1",
                "main",
                order=0,
                duration=5.0,
                config={"role": "audio"},
            ),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.probe_has_audio", return_value=False),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            result = mix_audio(ctx, layers, 5.0)

        assert result is None
        mock_run.assert_not_called()

    def test_clip_has_audio_cache(self):
        """_clip_has_audio 带缓存，同一clip只探测一次。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
        ):
            resolved = svc._resolve_clips()
            clip = resolved[0]

        with patch("video_processing.render_audio.probe_has_audio", return_value=True) as mock_probe:
            # 调用 3 次
            ctx = _make_ctx()
            r1 = clip_has_audio(ctx, clip)
            r2 = clip_has_audio(ctx, clip)
            r3 = clip_has_audio(ctx, clip)

        assert r1 is True and r2 is True and r3 is True
        # 实际只探测了 1 次
        assert mock_probe.call_count == 1


# ── 测试 stream copy 流拷贝优化 ───────────────────────────────────────────────


class TestStreamCopy:
    """stream copy 流拷贝优化测试。"""

    def _make_single_clip_service(self):
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        svc = _make_service(clips)
        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
        return svc, resolved[0], layers

    def test_can_use_stream_copy_all_conditions_met(self):
        """所有条件满足 → 可以 stream copy。"""
        svc, clip, layers = self._make_single_clip_service()
        probe_result = {
            "width": 1280,
            "height": 720,
            "fps": 25.0,
            "video_codec": "h264",
            "pix_fmt": "yuv420p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }
        with patch(
            "video_processing.unified_render_service.probe_video_info",
            return_value=probe_result,
        ):
            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
        assert can_copy is True
        assert "所有条件满足" in reason

    def test_cannot_copy_with_subtitles(self):
        """有字幕 → 不能 stream copy。"""
        svc, clip, layers = self._make_single_clip_service()
        can_copy, reason = svc._can_use_stream_copy(clip, ass_path=Path("/tmp/sub.ass"), video_duration=0)
        assert can_copy is False
        assert "字幕" in reason

    def test_cannot_copy_wrong_codec(self):
        """编码不是 h264 → 不能 stream copy。"""
        svc, clip, layers = self._make_single_clip_service()
        probe_result = {
            "width": 1280,
            "height": 720,
            "fps": 25.0,
            "video_codec": "hevc",
            "pix_fmt": "yuv420p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }
        with patch(
            "video_processing.unified_render_service.probe_video_info",
            return_value=probe_result,
        ):
            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
        assert can_copy is False
        assert "编码" in reason

    def test_cannot_copy_wrong_resolution(self):
        """分辨率不匹配 → 不能 stream copy。"""
        svc, clip, layers = self._make_single_clip_service()
        probe_result = {
            "width": 1920,
            "height": 1080,
            "fps": 25.0,
            "video_codec": "h264",
            "pix_fmt": "yuv420p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }
        with patch(
            "video_processing.unified_render_service.probe_video_info",
            return_value=probe_result,
        ):
            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
        assert can_copy is False
        assert "分辨率" in reason

    def test_cannot_copy_wrong_fps(self):
        """帧率不匹配 → 不能 stream copy。"""
        svc, clip, layers = self._make_single_clip_service()
        probe_result = {
            "width": 1280,
            "height": 720,
            "fps": 30.0,
            "video_codec": "h264",
            "pix_fmt": "yuv420p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }
        with patch(
            "video_processing.unified_render_service.probe_video_info",
            return_value=probe_result,
        ):
            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
        assert can_copy is False
        assert "帧率" in reason

    def test_cannot_copy_wrong_pix_fmt(self):
        """像素格式不匹配 → 不能 stream copy。"""
        svc, clip, layers = self._make_single_clip_service()
        probe_result = {
            "width": 1280,
            "height": 720,
            "fps": 25.0,
            "video_codec": "h264",
            "pix_fmt": "yuv422p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }
        with patch(
            "video_processing.unified_render_service.probe_video_info",
            return_value=probe_result,
        ):
            can_copy, reason = svc._can_use_stream_copy(clip, ass_path=None, video_duration=0)
        assert can_copy is False
        assert "像素格式" in reason

    def test_try_render_stream_copy_success(self):
        """stream copy 渲染成功 → 返回 True。"""
        svc, clip, layers = self._make_single_clip_service()
        probe_result = {
            "width": 1280,
            "height": 720,
            "fps": 25.0,
            "video_codec": "h264",
            "pix_fmt": "yuv420p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }
        output_path = Path("/tmp/test_output.mp4")

        def fake_stat():
            m = MagicMock()
            m.st_size = 1024000
            return m

        with (
            patch(
                "video_processing.unified_render_service.probe_video_info",
                return_value=probe_result,
            ),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
            patch("pathlib.Path.exists", return_value=True),
            patch("pathlib.Path.stat", side_effect=fake_stat),
        ):
            result = svc._try_render_stream_copy(layers, output_path, ass_path=None, video_duration=0)

        assert result is True
        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]
        assert "-c:v" in cmd
        assert "copy" in cmd
        assert "-c:a" in cmd

    def test_try_render_stream_copy_fallback_on_ffmpeg_error(self):
        """stream copy FFmpeg 失败 → 返回 False（调用方回退到重编码）。"""
        svc, clip, layers = self._make_single_clip_service()
        probe_result = {
            "width": 1280,
            "height": 720,
            "fps": 25.0,
            "video_codec": "h264",
            "pix_fmt": "yuv420p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }
        output_path = Path("/tmp/test_output.mp4")

        import subprocess as sp

        with (
            patch(
                "video_processing.unified_render_service.probe_video_info",
                return_value=probe_result,
            ),
            patch(
                "video_processing.unified_render_service.run_ffmpeg",
                side_effect=sp.CalledProcessError(1, ["ffmpeg"], stderr="copy failed"),
            ),
            patch("pathlib.Path.exists", return_value=False),
        ):
            result = svc._try_render_stream_copy(layers, output_path, ass_path=None, video_duration=0)

        assert result is False

    def test_render_uses_stream_copy_when_eligible(self):
        """完整渲染流程：满足条件时走 stream copy。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        svc = _make_service(clips)

        probe_result = {
            "width": 1280,
            "height": 720,
            "fps": 25.0,
            "video_codec": "h264",
            "pix_fmt": "yuv420p",
            "duration": 5.0,
            "has_audio": True,
            "audio_codec": "aac",
        }

        def fake_stat():
            m = MagicMock()
            m.st_size = 1024000
            return m

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch(
                "video_processing.unified_render_service.probe_video_info",
                return_value=probe_result,
            ),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
            patch("pathlib.Path.stat", side_effect=fake_stat),
            patch("shutil.copy2"),
        ):
            result = svc.render()

        assert mock_run.call_count == 1
        cmd = mock_run.call_args[0][0]
        assert "copy" in cmd
        assert isinstance(result.output_path, Path)


# ══════════════════════════════════════════════════════════════════════════════
# _extract_audio 安全下沉测试（P1 技术债务）
# ══════════════════════════════════════════════════════════════════════════════


class TestExtractAudioUsesRunFfmpeg:
    """_extract_audio 必须使用 run_ffmpeg 统一管理，不能用裸 subprocess."""

    def test_extract_audio_calls_run_ffmpeg(self, tmp_path):
        """_extract_audio 内部应调用 ffmpeg_utils.run_ffmpeg 而非裸 subprocess."""
        from video_processing.unified_render_service import UnifiedRenderService

        plan = MagicMock()
        plan.config = {}
        plan.id = "test-plan"
        plan.canvas_config = MagicMock()
        plan.canvas_config.width = 1080
        plan.canvas_config.height = 1920
        plan.canvas_config.fps = 30
        plan.canvas_config.output_width = 1080
        plan.canvas_config.output_height = 1920

        svc = UnifiedRenderService(plan, [], {}, tmp_path)

        video_path = tmp_path / "input.mp4"
        output_path = tmp_path / "output.wav"
        video_path.write_bytes(b"fake")

        with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run:
            svc._extract_audio(video_path, output_path)

        # 验证调用了 run_ffmpeg
        assert mock_run.called, "_extract_audio 必须通过 run_ffmpeg 执行 FFmpeg"
        cmd = mock_run.call_args[0][0]

        # 验证命令参数正确
        assert cmd[0].endswith("ffmpeg") or "ffmpeg" in cmd[0]
        assert "-i" in cmd
        assert str(video_path) in cmd
        assert "-vn" in cmd  # 无视频流
        assert "pcm_s16le" in cmd  # 16bit PCM
        assert "16000" in cmd  # 16kHz
        assert str(output_path) in cmd
        assert mock_run.call_args[1].get("timeout") == 120

    def test_extract_audio_failure_raises_runtime_error(self, tmp_path):
        """_extract_audio 失败时应抛出 RuntimeError."""
        from video_processing.unified_render_service import UnifiedRenderService

        plan = MagicMock()
        plan.config = {}
        plan.id = "test-plan"
        plan.canvas_config = MagicMock()
        plan.canvas_config.width = 1080
        plan.canvas_config.height = 1920
        plan.canvas_config.fps = 30
        plan.canvas_config.output_width = 1080
        plan.canvas_config.output_height = 1920

        svc = UnifiedRenderService(plan, [], {}, tmp_path)

        video_path = tmp_path / "input.mp4"
        output_path = tmp_path / "output.wav"
        video_path.write_bytes(b"fake")

        import subprocess

        with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run:
            mock_run.side_effect = subprocess.CalledProcessError(1, "ffmpeg", stderr="error")
            with pytest.raises(RuntimeError, match="音频提取失败"):
                svc._extract_audio(video_path, output_path)


# ══════════════════════════════════════════════════════════════════════════════
# concat 前 4 项归一化测试（修复 exit=234）
# ══════════════════════════════════════════════════════════════════════════════


class TestConcatNormalizeVideoResolution:
    """视频分辨率归一化：scale + pad 保持比例留黑边，确保 concat 前分辨率一致。

    对应 4 项 normalize 之：视频分辨率（scale + pad 到 canvas_width × canvas_height）
    修复前：main/broll 用 scale+crop（铺满裁剪），横屏素材内容被裁掉
    修复后：main/broll 用 scale+pad（留黑边），保持原始比例不裁剪
    """

    def test_main_layer_uses_pad_not_crop_in_filter_complex(self):
        """_build_filter_complex 中 main 层用 scale+pad（letterbox），不用 crop。"""
        clips = [_make_clip("c1", "main", order=0, duration=3.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        # 必须用 pad（留黑边），不能用 crop（裁剪）
        assert "force_original_aspect_ratio=decrease" in fc, "main 层应使用 decrease 模式（等比缩放到画布内）"
        assert "pad=" in fc, "main 层应有 pad 滤镜（留黑边到目标分辨率）"
        assert "trunc((ow-iw)/2):trunc((oh-ih)/2):black" in fc, "pad 应该居中取整+黑边"
        # 注意：background 层也用 crop，但 main/broll 应该用 pad
        # 检查第一个 [0:v] 处理链（第一个 clip 是 main）
        first_v_chain = fc.split("[v0]")[0]
        assert "crop=" not in first_v_chain, "main 层不应该有 crop 滤镜"

    def test_broll_layer_uses_pad_not_crop_in_filter_complex(self):
        """_build_filter_complex 中 b_roll 层用 scale+pad，不用 crop。"""
        clips = [
            _make_clip("c1", "b_roll", order=0, duration=3.0),
        ]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        assert "force_original_aspect_ratio=decrease" in fc
        assert "pad=" in fc
        first_v_chain = fc.split("[v0]")[0]
        assert "crop=" not in first_v_chain, "broll 层不应该有 crop 滤镜"

    def test_background_layer_still_uses_crop_in_filter_complex(self):
        """_build_filter_complex 中 background 层保持 scale+crop（作为底图铺满）。"""
        clips = [
            _make_clip("bg1", "background", order=0, duration=3.0),
            _make_clip("c1", "main", order=0, duration=3.0),
        ]
        asset_paths = {
            "asset_bg1.mp4": Path("/tmp/asset_bg1.mp4"),
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        # background 应该用 increase + crop（cover 模式）
        # 统计 crop 出现次数：background 1次 + 其他位置（如果有）
        # 关键是 background 走 increase 模式
        assert "force_original_aspect_ratio=increase" in fc, "background 层应使用 increase 模式（铺满裁剪）"

    def test_multi_clip_concat_all_same_resolution(self):
        """多 clip concat 时，所有 clip 预处理后分辨率一致（pad 到同一尺寸）。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=2.0),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        # 两个 clip 都应该有 pad
        pad_count = fc.count("pad=")
        assert pad_count >= 2, f"至少应有 2 个 pad（每个 clip 一个），实际 {pad_count}"
        # 都用 decrease 模式
        decrease_count = fc.count("force_original_aspect_ratio=decrease")
        assert decrease_count >= 2, f"至少应有 2 个 decrease，实际 {decrease_count}"
        # 有 concat
        assert "concat=n=2:v=1:a=0" in fc

    def test_pass_through_main_uses_pad_not_crop(self):
        """_render_pass_through 中 main 层用 scale+pad，不用 crop。"""
        clips = [_make_clip("c1", "main", order=0, duration=3.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=3.0)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)

        # 直通模式 main 层应该用 pad
        assert "force_original_aspect_ratio=decrease" in cmd_str, "直通模式 main 层应使用 decrease 模式"
        assert "pad=" in cmd_str, "直通模式应有 pad 滤镜"
        # 检查 vf 中没有 crop
        vf_idx = cmd.index("-vf")
        vf_value = cmd[vf_idx + 1]
        assert "crop=" not in vf_value, "直通模式 main 层 vf 中不应该有 crop"

    def test_pass_through_background_uses_crop(self):
        """_render_pass_through 中 background 层保持 scale+crop。"""
        clips = [_make_clip("bg1", "background", order=0, duration=3.0)]
        asset_paths = {"asset_bg1.mp4": Path("/tmp/asset_bg1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=3.0)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)

        # background 层应该用 increase + crop
        assert "force_original_aspect_ratio=increase" in cmd_str
        vf_idx = cmd.index("-vf")
        vf_value = cmd[vf_idx + 1]
        assert "crop=" in vf_value, "background 层应有 crop 滤镜"

    def test_pass_through_speed_up_correct_duration(self):
        """直通模式加速（speed=2x）：视频+音频均调速，-t 时长为原始的 1/2。"""
        clips = [_make_clip("c1", "main", order=0, duration=10.0, playback_speed=2.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            svc._render_pass_through(layers, Path("/tmp/out.mp4"))

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)

        # 视频调速：setpts=PTS/2.0
        assert "setpts=PTS/2.0" in cmd_str, "加速场景应有 setpts=PTS/speed 滤镜"

        # 音频调速：atempo
        assert "atempo" in cmd_str, "加速场景应有 atempo 音频调速滤镜"

        # 输出时长应为原始 / speed = 10 / 2 = 5 秒
        t_idx = cmd.index("-t")
        t_value = float(cmd[t_idx + 1])
        assert abs(t_value - 5.0) < 0.01, f"加速后 -t 时长应为 5.0s，实际 {t_value}s"

    def test_pass_through_slow_down_correct_duration(self):
        """直通模式减速（speed=0.5x）：视频+音频均调速，-t 时长为原始的 2 倍（不被截断）。"""
        clips = [_make_clip("c1", "main", order=0, duration=10.0, playback_speed=0.5)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            svc._render_pass_through(layers, Path("/tmp/out.mp4"))

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)

        # 视频调速：setpts=PTS/0.5
        assert "setpts=PTS/0.5" in cmd_str, "减速场景应有 setpts=PTS/speed 滤镜"

        # 音频调速：atempo
        assert "atempo" in cmd_str, "减速场景应有 atempo 音频调速滤镜"

        # 输出时长应为原始 / speed = 10 / 0.5 = 20 秒（减速后视频变长，不应被截断）
        t_idx = cmd.index("-t")
        t_value = float(cmd[t_idx + 1])
        assert abs(t_value - 20.0) < 0.01, f"减速后 -t 时长应为 20.0s，实际 {t_value}s"

    def test_pass_through_speed_with_video_duration_cap(self):
        """直通模式调速 + video_duration 截断：取调速后时长与 video_duration 的较小值。"""
        clips = [_make_clip("c1", "main", order=0, duration=10.0, playback_speed=2.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            # video_duration=3.0 < 调速后时长 5.0，应取 3.0
            svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=3.0)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        t_idx = cmd.index("-t")
        t_value = float(cmd[t_idx + 1])
        assert abs(t_value - 3.0) < 0.01, f"video_duration 更小时应取 video_duration，实际 {t_value}s"


class TestConcatNormalizeVideoFps:
    """视频帧率归一化：fps 滤镜统一到目标 fps。

    对应 4 项 normalize 之：视频帧率（fps 滤镜统一到目标 fps）
    """

    def test_filter_complex_has_fps_filter(self):
        """_build_filter_complex 中每个 clip 都有 fps 滤镜。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("c2", "main", order=1, duration=2.0),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
        }
        svc = _make_service(clips, asset_paths, output_fps=30)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        fps_count = fc.count("fps=30")
        assert fps_count >= 2, f"至少应有 2 个 fps=30（每个 clip 一个），实际 {fps_count}"

    def test_fps_after_pad_before_final_setpts(self):
        """fps 滤镜在 pad 之后、setpts 之前（确保分辨率和帧率都统一后再归一化 PTS）。"""
        import re

        clips = [_make_clip("c1", "main", order=0, duration=3.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths, output_fps=30)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        # 提取第一个 clip 的视频处理链
        chain_str = fc.split("[v0]")[0]

        pad_pos = chain_str.find("pad=")
        fps_pos = chain_str.find("fps=")
        assert pad_pos >= 0, "应找到 pad 滤镜"
        assert fps_pos >= 0, "应找到 fps 滤镜"
        assert fps_pos > pad_pos, "fps 应在 pad 之后"


class TestConcatNormalizeAudioFormat:
    """音频格式归一化：aformat 统一为 stereo + 48000Hz + fltp。

    对应 4 项 normalize 之：音频声道（aformat 统一为 stereo + 48000Hz + fltp）
    修复前：concat 前音频无归一化，不同采样率/声道导致 exit=234
    修复后：concat/amix 前所有音频统一为 48000Hz + stereo + fltp
    """

    def test_multi_clip_concat_has_aformat(self):
        """main 原声 + 独立音频轨在 concat/amix 前都有 aformat 归一化。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=3.0),
            _make_clip("tts1", "main", order=0, duration=3.0, config={"role": "audio"}),
            _make_clip("tts2", "main", order=1, duration=2.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
            "asset_tts2.mp4": Path("/tmp/asset_tts2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            mix_audio(ctx, layers, 4.5)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)

        # 必须有 aformat 归一化
        assert "aformat=" in cmd_str, "concat 前应有 aformat 滤镜"
        assert "sample_rates=48000" in cmd_str, "采样率应统一为 48000Hz"
        assert "channel_layouts=stereo" in cmd_str, "声道应统一为 stereo"
        assert "sample_fmts=fltp" in cmd_str, "采样格式应统一为 fltp"

        # main 原声 + 2 个独立音频轨都应有 aformat
        aformat_count = cmd_str.count("aformat=")
        assert aformat_count >= 3, f"3 路音频都应有 aformat，实际 {aformat_count} 个"

        # amix 3 路输入
        assert "amix=inputs=3" in cmd_str
        # main 源视频原声参与
        assert "asset_c1.mp4" in cmd_str

    def test_aformat_before_concat(self):
        """aformat 应在 concat 之前（每个独立音频轨处理链中 aformat 在 concat 之前）。"""
        clips = [
            _make_clip("tts1", "main", order=0, duration=3.0, config={"role": "audio"}),
            _make_clip("tts2", "main", order=1, duration=2.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
            "asset_tts2.mp4": Path("/tmp/asset_tts2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            mix_audio(ctx, layers, 4.5)

        cmd = mock_run.call_args[0][0]
        fc_idx = cmd.index("-filter_complex")
        fc_str = cmd[fc_idx + 1]

        concat_pos = fc_str.find("concat=")
        assert concat_pos > 0, "应找到 concat filter"

        before_concat = fc_str[:concat_pos]
        aformat_before_count = before_concat.count("aformat=")
        assert aformat_before_count >= 2, f"concat 之前每个独立音频轨都应有 aformat，实际 {aformat_before_count} 个"

    def test_single_clip_audio_has_normalized_output(self):
        """原声+独立音频轨输出统一格式（一致性保障）。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip("tts1", "main", order=0, duration=5.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            mix_audio(ctx, layers, 5.0)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)

        # 原声 + 独立音频轨走 amix：两路都有 aformat 归一化
        assert "amix=inputs=2" in cmd_str
        assert cmd_str.count("aformat=") >= 2
        assert "sample_rates=48000" in cmd_str
        assert "channel_layouts=stereo" in cmd_str
        assert "asset_c1.mp4" in cmd_str
        assert "asset_tts1.mp4" in cmd_str

    def test_independent_audio_track_has_aformat(self):
        """原声+独立音频轨输出统一格式（48000Hz + stereo + aac）。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip(
                "audio1",
                "main",
                order=0,
                duration=5.0,
                config={"role": "audio", "volume": 0.5},
            ),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_audio1.mp4": Path("/tmp/asset_audio1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            mix_audio(ctx, layers, 5.0)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)

        # 原声 + 独立音频轨走 amix：两路都有 aformat 归一化
        assert "amix=inputs=2" in cmd_str
        assert cmd_str.count("aformat=") >= 2
        assert "sample_rates=48000" in cmd_str
        assert "channel_layouts=stereo" in cmd_str
        # main 原声 c1 与独立音频轨 audio1 都参与
        assert "asset_c1.mp4" in cmd_str
        assert "asset_audio1.mp4" in cmd_str


class TestConcatNormalizeAudioCodec:
    """音频编码归一化：统一为 aac。

    对应 4 项 normalize 之：音频编码（统一为 aac）
    """

    def test_multi_clip_output_is_aac(self):
        """多独立音频轨 concat 后输出编码为 aac。"""
        clips = [
            _make_clip("tts1", "main", order=0, duration=3.0, config={"role": "audio"}),
            _make_clip("tts2", "main", order=1, duration=2.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
            "asset_tts2.mp4": Path("/tmp/asset_tts2.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            mix_audio(ctx, layers, 4.5)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        assert "aac" in cmd, "音频输出编码应为 aac"
        assert "-b:a" in cmd, "应指定音频码率"

    def test_single_clip_output_is_aac(self):
        """原声+独立音频轨输出编码为 aac。"""
        clips = [
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip("tts1", "main", order=0, duration=5.0, config={"role": "audio"}),
        ]
        asset_paths = {
            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
            "asset_tts1.mp4": Path("/tmp/asset_tts1.mp4"),
        }
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            mix_audio(ctx, layers, 5.0)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        assert "aac" in cmd
        # main 原声参与
        assert "asset_c1.mp4" in " ".join(cmd)


class TestConcatNormalizeFourItemsComplete:
    """4 项 normalize 完整性验证：one_take 多素材 concat 场景全链路检查。

    模拟 one_take 模式（多个 main clip）的完整渲染路径，
    验证 concat 前所有 4 项 normalize 都已到位。
    """

    def _make_one_take_clips(self):
        """构造 one_take 模式的典型素材：3 个 main clip（模拟横屏+竖屏混合）。"""
        return [
            _make_clip("c1", "main", order=0, duration=5.0),
            _make_clip("c2", "main", order=1, duration=4.0),
            _make_clip("c3", "main", order=2, duration=3.0),
        ]

    def test_video_resolution_normalized(self):
        """[1/4] 视频分辨率：所有 clip 都有 pad 到统一分辨率。"""
        clips = self._make_one_take_clips()
        asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)}
        svc = _make_service(clips, asset_paths)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=10.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        # 3 个 clip 都应该有 pad（decrease + black padding）
        pad_count = fc.count("pad=")
        decrease_count = fc.count("force_original_aspect_ratio=decrease")
        assert pad_count >= 3, f"3个 clip 都应有 pad，实际 {pad_count} 个"
        assert decrease_count >= 3, f"3个 clip 都应用 decrease 模式，实际 {decrease_count} 个"

    def test_video_fps_normalized(self):
        """[2/4] 视频帧率：所有 clip 都有 fps 滤镜。"""
        clips = self._make_one_take_clips()
        asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)}
        svc = _make_service(clips, asset_paths, output_fps=30)

        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=10.0):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            fc, _ = svc._build_filter_complex(layers)

        fps_count = fc.count("fps=30")
        assert fps_count >= 3, f"3个 clip 都应有 fps=30，实际 {fps_count} 个"

    def test_audio_format_normalized(self):
        """[3/4] 音频格式：3 个 main 原声 + 3 个独立音频轨都有 aformat 归一化。"""
        clips = self._make_one_take_clips() + [
            _make_clip("tts1", "main", order=10, duration=5.0, config={"role": "audio"}),
            _make_clip("tts2", "main", order=11, duration=4.0, config={"role": "audio"}),
            _make_clip("tts3", "main", order=12, duration=3.0, config={"role": "audio"}),
        ]
        asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)}
        asset_paths.update({f"asset_tts{i}.mp4": Path(f"/tmp/asset_tts{i}.mp4") for i in range(1, 4)})
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            mix_audio(ctx, layers, 12.0)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        cmd_str = " ".join(cmd)

        aformat_count = cmd_str.count("aformat=")
        assert aformat_count >= 6, f"6 路音频（3原声+3独立轨）都应有 aformat，实际 {aformat_count} 个"
        assert "sample_rates=48000" in cmd_str
        assert "channel_layouts=stereo" in cmd_str
        assert "sample_fmts=fltp" in cmd_str
        # main 图层原声参与
        for i in range(1, 4):
            assert f"asset_c{i}.mp4" in cmd_str

    def test_audio_codec_aac(self):
        """[4/4] 音频编码：输出为 aac（仅独立音频轨参与）。"""
        clips = self._make_one_take_clips() + [
            _make_clip("tts1", "main", order=10, duration=5.0, config={"role": "audio"}),
            _make_clip("tts2", "main", order=11, duration=4.0, config={"role": "audio"}),
            _make_clip("tts3", "main", order=12, duration=3.0, config={"role": "audio"}),
        ]
        asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)}
        asset_paths.update({f"asset_tts{i}.mp4": Path(f"/tmp/asset_tts{i}.mp4") for i in range(1, 4)})
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(svc._resolve_clips())
            ctx = _make_ctx()
            mix_audio(ctx, layers, 12.0)

        assert mock_run.called
        cmd = mock_run.call_args[0][0]
        assert "aac" in cmd
        # 3 个 main 原声 concat 后再与独立轨 amix
        assert "concat=n=3:v=0:a=1" in " ".join(cmd)


class TestVoiceoverTopLevelConfigBridge:
    """顶层 voice_id + custom_text 桥接到 tts 配置的兼容性测试.

    前端一键生成页面传 config.voice_id + config.custom_text（顶层字段），
    统一渲染引擎从 config.tts 读取。桥接逻辑确保两条路径都能工作。
    """

    def test_top_level_voice_id_with_text_triggers_tts(self):
        """顶层 voice_id + custom_text 能触发 TTS 配音（桥接生效）。"""
        from unittest.mock import MagicMock

        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        plan = FakePlan(
            id="plan_tts_001",
            config={
                "voice_id": "longxiaoxia_v3",
                "custom_text": "大家好，欢迎来到我的频道",
            },
        )
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts"),
        )

        mock_seg = MagicMock()
        mock_seg.audio_path = Path("/tmp/test_tts/tts/voiceover_full.wav")
        mock_seg.start_time = 0.0
        mock_seg.duration = 3.0
        mock_result = MagicMock()
        mock_result.success = True
        mock_result.segments = [mock_seg]
        mock_result.total_duration = 3.0

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch(
                "video_processing.tts_engine.TtsEngine.generate_full_voiceover",
                return_value=mock_result,
            ),
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            result = svc._maybe_add_voiceover_layer(layers, video_duration=5.0)

        assert result is True, "顶层 voice_id + custom_text 应触发 TTS 配音"
        # 应有 audio 图层
        audio_layer = next((layer for layer in layers if layer.role == "audio"), None)
        assert audio_layer is not None, "应添加 audio 图层"
        assert len(audio_layer.clips) == 1, "应有 1 个配音片段"

    def test_tts_config_takes_priority(self):
        """config.tts.enabled 已配置时，以 tts 配置为准，不触发桥接。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        # tts.enabled=True 但 text 为空（应失败），顶层有 text
        plan = FakePlan(
            id="plan_tts_002",
            config={
                "voice_id": "longxiaoxia_v3",
                "custom_text": "顶层文本不生效",
                "tts": {
                    "enabled": True,
                    "voice_id": "longxiaochun_v3",
                    "text": "",  # tts 配置里 text 为空
                },
            },
        )
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts"),
        )

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            result = svc._maybe_add_voiceover_layer(layers, video_duration=5.0)

        # tts.enabled=True 但 text 为空 → 生成失败 → 返回 False
        # 关键是不触发桥接（不会用顶层的 custom_text）
        assert result is False

    def test_top_level_voice_id_without_text_no_trigger(self):
        """只有 voice_id 没有 custom_text 不触发 TTS 配音。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        plan = FakePlan(
            id="plan_tts_003",
            config={"voice_id": "longxiaoxia_v3", "custom_text": ""},
        )
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts"),
        )

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            result = svc._maybe_add_voiceover_layer(layers, video_duration=5.0)

        assert result is False

    def test_no_voice_config_no_trigger(self):
        """没有 voice_id 也没有 tts 配置时，不触发配音。"""
        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        svc = _make_service(clips, asset_paths)

        with (
            _patch_path_exists(),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
        ):
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            result = svc._maybe_add_voiceover_layer(layers, video_duration=5.0)

        assert result is False
        # 没有 audio 图层
        assert not any(layer.role == "audio" for layer in layers)


class TestVoiceoverSubtitleAlign:
    """预设配音 + 自动字幕 → 字幕对齐 TTS 配音.

    前端预设配音模式只传 voice_id，不传 custom_text。
    配合自动字幕时，用 ASR 识别结果生成逐字幕配音。
    """

    def _make_mock_timeline(self, segments_data):
        """构造模拟的 SubtitleTimeline."""
        from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline

        segments = [SubtitleSegment(text=s["text"], start=s["start"], end=s["end"]) for s in segments_data]
        return SubtitleTimeline(segments=segments, total_duration=10.0)

    def test_preset_voice_with_auto_subtitle_triggers_tts(self):
        """预设配音 + 自动字幕 → 触发字幕对齐 TTS 配音。"""
        from unittest.mock import MagicMock, PropertyMock

        clips = [_make_clip("c1", "main", order=0, duration=10.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        plan = FakePlan(
            id="plan_sub_001",
            config={
                "voice_id": "longxiaoxia_v3",
                "subtitle": {
                    "enabled": True,
                    "auto_generated": True,
                },
            },
        )

        mock_asr = MagicMock()
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts_sub"),
            asr_service=mock_asr,
        )

        # 模拟 ASR 结果（通过缓存注入）
        svc._asr_timeline_cache = self._make_mock_timeline(
            [
                {"text": "大家好欢迎来到我的频道", "start": 0.0, "end": 2.5},
                {"text": "今天给大家分享一个小技巧", "start": 2.5, "end": 5.0},
                {"text": "记得点赞关注哦", "start": 5.0, "end": 7.0},
            ]
        )
        svc._asr_timeline_cached = True

        with _patch_path_exists():
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            with patch(
                "video_processing.tts_engine.TtsEngine.generate_subtitle_voiceover",
            ) as mock_sub_vo:
                # 构造模拟返回
                mock_seg1 = MagicMock()
                mock_seg1.audio_path = Path("/tmp/tts/seg_000.wav")
                mock_seg1.start_time = 0.0
                mock_seg1.duration = 2.5
                mock_seg2 = MagicMock()
                mock_seg2.audio_path = Path("/tmp/tts/seg_001.wav")
                mock_seg2.start_time = 2.5
                mock_seg2.duration = 2.5
                mock_result = MagicMock()
                mock_result.success = True
                mock_result.segments = [mock_seg1, mock_seg2]
                mock_result.total_duration = 5.0
                mock_sub_vo.return_value = mock_result

                result = svc._maybe_add_voiceover_layer(layers, video_duration=10.0)

        assert result is True, "预设配音+自动字幕应触发 TTS 配音"
        # 应调用字幕对齐模式
        mock_sub_vo.assert_called_once()
        # 应有 audio 图层
        audio_layer = next((layer for layer in layers if layer.role == "audio"), None)
        assert audio_layer is not None
        assert len(audio_layer.clips) == 2  # 2 个字幕对应 2 段配音

    def test_preset_voice_without_auto_subtitle_no_trigger(self):
        """只有 voice_id 没有自动字幕 → 不触发配音。"""
        clips = [_make_clip("c1", "main", order=0, duration=10.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        plan = FakePlan(
            id="plan_sub_002",
            config={
                "voice_id": "longxiaoxia_v3",
                "subtitle": {"enabled": True, "auto_generated": False},
            },
        )
        mock_asr = MagicMock()
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts_sub"),
            asr_service=mock_asr,
        )

        with _patch_path_exists():
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            result = svc._maybe_add_voiceover_layer(layers, video_duration=10.0)

        assert result is False
        assert not any(layer.role == "audio" for layer in layers)

    def test_preset_voice_no_asr_service_no_trigger(self):
        """有 voice_id + 自动字幕但没有 ASR 服务 → 不触发。"""
        clips = [_make_clip("c1", "main", order=0, duration=10.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        plan = FakePlan(
            id="plan_sub_003",
            config={
                "voice_id": "longxiaoxia_v3",
                "subtitle": {"enabled": True, "auto_generated": True},
            },
        )
        # 不传 asr_service
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts_sub"),
        )

        with _patch_path_exists():
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            result = svc._maybe_add_voiceover_layer(layers, video_duration=10.0)

        assert result is False

    def test_preset_voice_asr_empty_segments_skip(self):
        """ASR 无识别结果 → 跳过配音，不报错。"""
        clips = [_make_clip("c1", "main", order=0, duration=10.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        plan = FakePlan(
            id="plan_sub_004",
            config={
                "voice_id": "longxiaoxia_v3",
                "subtitle": {"enabled": True, "auto_generated": True},
            },
        )
        mock_asr = MagicMock()
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts_sub"),
            asr_service=mock_asr,
        )

        # ASR 返回空结果
        svc._asr_timeline_cache = self._make_mock_timeline([])
        svc._asr_timeline_cached = True

        with _patch_path_exists():
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            # 不抛异常，返回 False 即可
            result = svc._maybe_add_voiceover_layer(layers, video_duration=10.0)

        assert result is False

    def test_asr_cache_reuse_between_subtitle_and_voiceover(self):
        """ASR 结果缓存：字幕和配音共用一次 ASR 调用。"""
        clips = [_make_clip("c1", "main", order=0, duration=10.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        plan = FakePlan(
            id="plan_sub_005",
            config={
                "voice_id": "longxiaoxia_v3",
                "subtitle": {"enabled": True, "auto_generated": True},
            },
        )
        mock_asr = MagicMock()
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts_sub"),
            asr_service=mock_asr,
        )

        # 先模拟调用过一次 ASR（比如字幕模块先调用）
        svc._asr_timeline_cache = self._make_mock_timeline(
            [
                {"text": "测试字幕", "start": 0.0, "end": 2.0},
            ]
        )
        svc._asr_timeline_cached = True

        with _patch_path_exists():
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            with patch(
                "video_processing.tts_engine.TtsEngine.generate_subtitle_voiceover",
            ) as mock_sub_vo:
                mock_seg = MagicMock()
                mock_seg.audio_path = Path("/tmp/tts/seg_000.wav")
                mock_seg.start_time = 0.0
                mock_seg.duration = 2.0
                mock_result = MagicMock()
                mock_result.success = True
                mock_result.segments = [mock_seg]
                mock_result.total_duration = 2.0
                mock_sub_vo.return_value = mock_result

                result = svc._maybe_add_voiceover_layer(layers, video_duration=10.0)

        assert result is True
        # ASR 服务不应被再次调用（使用缓存）
        mock_asr.transcribe.assert_not_called()

    def test_tts_config_align_mode_subtitle_also_works(self):
        """标准 tts 配置 + align_mode=subtitle 也走字幕对齐模式。"""
        clips = [_make_clip("c1", "main", order=0, duration=10.0)]
        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
        plan = FakePlan(
            id="plan_sub_006",
            config={
                "tts": {
                    "enabled": True,
                    "voice_id": "longxiaoxia_v3",
                    "text": "",
                    "align_mode": "subtitle",
                },
                "subtitle": {"enabled": True, "auto_generated": True},
            },
        )
        mock_asr = MagicMock()
        svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_paths,
            work_dir=Path("/tmp/test_tts_sub"),
            asr_service=mock_asr,
        )

        svc._asr_timeline_cache = self._make_mock_timeline(
            [
                {"text": "字幕1", "start": 0.0, "end": 3.0},
                {"text": "字幕2", "start": 3.0, "end": 6.0},
            ]
        )
        svc._asr_timeline_cached = True

        with _patch_path_exists():
            resolved = svc._resolve_clips()
            layers = svc._group_clips_into_layers(resolved)
            with patch(
                "video_processing.tts_engine.TtsEngine.generate_subtitle_voiceover",
            ) as mock_sub_vo:
                mock_seg = MagicMock()
                mock_seg.audio_path = Path("/tmp/tts/seg_000.wav")
                mock_seg.start_time = 0.0
                mock_seg.duration = 6.0
                mock_result = MagicMock()
                mock_result.success = True
                mock_result.segments = [mock_seg, mock_seg]
                mock_result.total_duration = 6.0
                mock_sub_vo.return_value = mock_result

                result = svc._maybe_add_voiceover_layer(layers, video_duration=10.0)

        assert result is True
        mock_sub_vo.assert_called_once()