refactor: 统一渲染链路，预览即所得（-3370行） (#1479)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

tests/integration/test_four_mode_rendering.py

@@ -1,234 +0,0 @@
-"""剪辑模式渲染集成测试.
-
-验证剪辑模式（ONE_TAKE / VOICE_OVER）通过
-_build_plan_and_clips_from_task + UnifiedRenderService 的完整渲染流程。
-注：PIP / VOICE_PIP 已下线，统一映射为 ONE_TAKE。
-
-需要 ffmpeg 可用；CI 无 ffmpeg 时自动跳过。
-"""
-
-from __future__ import annotations
-
-import shutil
-import subprocess
-import tempfile
-from pathlib import Path
-
-import pytest
-from video_processing.unified_render_service import (
-    RenderResult,
-    UnifiedRenderService,
-    _resolve_layer_role,
-)
-from worker_app.tasks.generation import _build_plan_and_clips_from_task
-
-pytestmark = pytest.mark.skipif(
-    not shutil.which("ffmpeg"),
-    reason="ffmpeg not available",
-)
-
-
-# ── 辅助函数 ──────────────────────────────────────────────────────────────────
-
-
-def _generate_test_video(path: Path, duration: float = 3.0, color: str = "red") -> None:
-    """生成一个纯色测试视频。"""
-    cmd = [
-        "ffmpeg",
-        "-y",
-        "-f",
-        "lavfi",
-        "-i",
-        f"color=c={color}:s=640x360:d={duration}:r=25",
-        "-c:v",
-        "libx264",
-        "-pix_fmt",
-        "yuv420p",
-        "-movflags",
-        "+faststart",
-        str(path),
-    ]
-    subprocess.run(cmd, check=True, capture_output=True, timeout=30)
-
-
-def _render_with_mode(
-    mode: str,
-    num_clips: int = 3,
-    duration: float = 2.0,
-) -> tuple[RenderResult, Path]:
-    """用指定模式生成测试视频并渲染，返回 (result, work_dir)。
-
-    调用方负责清理 work_dir。
-    """
-    work_dir = Path(tempfile.mkdtemp(prefix="test_4mode_"))
-
-    # 生成测试视频素材
-    colors = ["red", "green", "blue", "yellow", "purple"]
-    downloaded_paths: list[Path] = []
-    for i in range(num_clips):
-        p = work_dir / f"test_{i:03d}.mp4"
-        _generate_test_video(p, duration=duration, color=colors[i % len(colors)])
-        downloaded_paths.append(p)
-
-    # 构建虚拟 plan + clips
-    task_id = f"test_task_{mode}"
-    plan, clips, asset_path_map = _build_plan_and_clips_from_task(
-        task_id=task_id,
-        downloaded_paths=downloaded_paths,
-        mode=mode,
-    )
-
-    # 渲染
-    service = UnifiedRenderService(
-        plan=plan,
-        clips=clips,
-        asset_path_map=asset_path_map,
-        work_dir=work_dir,
-        output_width=640,
-        output_height=360,
-        output_fps=25,
-    )
-    result = service.render()
-    return result, work_dir
-
-
-# ── 测试 _build_plan_and_clips_from_task ──────────────────────────────────────
-
-
-class TestBuildPlanAndClips:
-    """测试 4 种模式的虚拟 plan 构建。"""
-
-    def _make_paths(self, n: int) -> list[Path]:
-        return [Path(f"/tmp/test_{i}.mp4") for i in range(n)]
-
-    def test_one_take_mode(self):
-        paths = self._make_paths(3)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t1", paths, "one_take")
-
-        assert plan.id == "t1"
-        assert len(clips) == 3
-        assert all(c.clip_type == "main" for c in clips)
-        assert len(asset_map) == 3
-
-    def test_pip_mode_maps_to_one_take(self):
-        """PIP 已下线，映射为 one_take → 全部 main clips。"""
-        paths = self._make_paths(3)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t2", paths, "pip")
-
-        assert len(clips) == 3
-        assert all(c.clip_type == "main" for c in clips)
-
-    def test_voice_over_mode(self):
-        paths = self._make_paths(3)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t3", paths, "voice_over")
-
-        assert len(clips) == 3
-        assert all(c.clip_type == "main" for c in clips)
-        assert all(c.config.get("role") == "b_roll" for c in clips)
-
-    def test_voice_pip_mode_maps_to_one_take(self):
-        """VOICE_PIP 已下线，映射为 one_take → 全部 main clips。"""
-        paths = self._make_paths(4)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t4", paths, "voice_pip")
-
-        assert len(clips) == 4
-        assert all(c.clip_type == "main" for c in clips)
-
-    def test_unknown_mode_defaults_to_one_take(self):
-        paths = self._make_paths(2)
-        plan, clips, asset_map = _build_plan_and_clips_from_task("t5", paths, "unknown_mode")
-
-        assert len(clips) == 2
-        assert all(c.clip_type == "main" for c in clips)
-
-    def test_asset_path_map_keys_match_clip_asset_ids(self):
-        paths = self._make_paths(3)
-        _, clips, asset_map = _build_plan_and_clips_from_task("t6", paths, "one_take")
-
-        clip_asset_ids = {c.asset_id for c in clips}
-        map_keys = set(asset_map.keys())
-        assert clip_asset_ids == map_keys
-
-
-# ── 测试图层分组（4 模式） ────────────────────────────────────────────────────
-
-
-class TestFourModeLayerGrouping:
-    """验证 4 种模式的 clip_type 分布经 _resolve_layer_role 后产生正确的图层。"""
-
-    def test_one_take_layers(self):
-        """ONE_TAKE: 3 main → 1 main layer。"""
-        paths = [Path(f"/tmp/ot_{i}.mp4") for i in range(3)]
-        _, clips, _ = _build_plan_and_clips_from_task("ot", paths, "one_take")
-
-        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
-        assert roles == {"main"}
-
-    def test_pip_layers_now_one_take(self):
-        """PIP 已下线 → one_take: 3 main → 1 main layer。"""
-        paths = [Path(f"/tmp/pip_{i}.mp4") for i in range(3)]
-        _, clips, _ = _build_plan_and_clips_from_task("pip", paths, "pip")
-
-        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
-        assert roles == {"main"}
-
-    def test_voice_over_layers(self):
-        """VOICE_OVER: 3 main(b_roll) → broll。"""
-        paths = [Path(f"/tmp/vo_{i}.mp4") for i in range(3)]
-        _, clips, _ = _build_plan_and_clips_from_task("vo", paths, "voice_over")
-
-        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
-        assert roles == {"broll"}
-
-    def test_voice_pip_layers_now_one_take(self):
-        """VOICE_PIP 已下线 → one_take: 4 main → main layer。"""
-        paths = [Path(f"/tmp/vpip_{i}.mp4") for i in range(4)]
-        _, clips, _ = _build_plan_and_clips_from_task("vpip", paths, "voice_pip")
-
-        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
-        assert roles == {"main"}
-
-
-# ── 端到端渲染测试（需要 ffmpeg） ─────────────────────────────────────────────
-
-
-class TestEndToEndRendering:
-    """4 种模式的完整渲染测试，验证输出文件存在且时长合理。"""
-
-    def test_one_take_render(self):
-        result, work_dir = _render_with_mode("one_take", num_clips=2, duration=2.0)
-        try:
-            assert result.output_path.exists()
-            assert result.file_size > 0
-            assert result.duration > 0
-            assert result.width == 640
-            assert result.height == 360
-        finally:
-            shutil.rmtree(work_dir, ignore_errors=True)
-
-    def test_pip_render(self):
-        result, work_dir = _render_with_mode("pip", num_clips=2, duration=2.0)
-        try:
-            assert result.output_path.exists()
-            assert result.file_size > 0
-            assert result.duration > 0
-        finally:
-            shutil.rmtree(work_dir, ignore_errors=True)
-
-    def test_voice_over_render(self):
-        result, work_dir = _render_with_mode("voice_over", num_clips=2, duration=2.0)
-        try:
-            assert result.output_path.exists()
-            assert result.file_size > 0
-            assert result.duration > 0
-        finally:
-            shutil.rmtree(work_dir, ignore_errors=True)
-
-    def test_voice_pip_render(self):
-        result, work_dir = _render_with_mode("voice_pip", num_clips=3, duration=2.0)
-        try:
-            assert result.output_path.exists()
-            assert result.file_size > 0
-            assert result.duration > 0
-        finally:
-            shutil.rmtree(work_dir, ignore_errors=True)

tests/integration/test_full_pipeline.py

@@ -1,189 +0,0 @@
-"""全链路集成测试.
-
-验证 PlanGeneratorService → UnifiedRenderService → 查重 的端到端流程。
-"""
-
-from __future__ import annotations
-
-import shutil
-import subprocess
-import tempfile
-from pathlib import Path
-
-import pytest
-from video_processing.unified_render_service import (
-    UnifiedRenderService,
-)
-from worker_app.tasks.generation import _build_plan_and_clips_from_task
-
-pytestmark = pytest.mark.skipif(
-    not shutil.which("ffmpeg"),
-    reason="ffmpeg not available",
-)
-
-
-def _generate_test_video(path: Path, duration: float = 3.0) -> None:
-    """生成一个测试视频。"""
-    cmd = [
-        "ffmpeg",
-        "-y",
-        "-f",
-        "lavfi",
-        "-i",
-        f"color=c=blue:s=640x360:d={duration}:r=25",
-        "-c:v",
-        "libx264",
-        "-pix_fmt",
-        "yuv420p",
-        str(path),
-    ]
-    subprocess.run(cmd, check=True, capture_output=True, timeout=30)
-
-
-def _generate_test_audio(path: Path, duration: float = 5.0) -> None:
-    """生成一个测试音频文件。"""
-    cmd = [
-        "ffmpeg",
-        "-y",
-        "-f",
-        "lavfi",
-        "-i",
-        f"sine=frequency=440:duration={duration}",
-        "-c:a",
-        "aac",
-        "-b:a",
-        "128k",
-        str(path),
-    ]
-    subprocess.run(cmd, check=True, capture_output=True, timeout=30)
-
-
-class TestFullPipeline:
-    """验证从虚拟 plan 构建到渲染输出的完整流程。"""
-
-    def test_one_take_pipeline(self):
-        """ONE_TAKE 模式完整流程。"""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            work_dir = Path(tmpdir)
-
-            # 生成测试素材
-            paths = []
-            for i in range(3):
-                p = work_dir / f"clip_{i}.mp4"
-                _generate_test_video(p, duration=2.0)
-                paths.append(p)
-
-            # 构建虚拟 plan
-            plan, clips, asset_map = _build_plan_and_clips_from_task("pipeline_test", paths, "one_take")
-
-            # 渲染
-            service = UnifiedRenderService(
-                plan=plan,
-                clips=clips,
-                asset_path_map=asset_map,
-                work_dir=work_dir,
-                output_width=640,
-                output_height=360,
-            )
-            result = service.render()
-
-            assert result.output_path.exists()
-            assert result.duration > 0
-            assert result.file_size > 0
-            assert result.width == 640
-            assert result.height == 360
-
-    def test_pipeline_with_audio_mux(self):
-        """渲染 + 混音后处理。"""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            work_dir = Path(tmpdir)
-
-            # 生成测试素材
-            video_path = work_dir / "clip_0.mp4"
-            _generate_test_video(video_path, duration=3.0)
-
-            # 构建虚拟 plan
-            plan, clips, asset_map = _build_plan_and_clips_from_task("audio_test", [video_path], "one_take")
-
-            # 渲染
-            service = UnifiedRenderService(
-                plan=plan,
-                clips=clips,
-                asset_path_map=asset_map,
-                work_dir=work_dir,
-                output_width=640,
-                output_height=360,
-            )
-            render_result = service.render()
-
-            # 混音 - 直接用 ffmpeg（_mux_audio_track 已被清理）
-            audio_path = work_dir / "voice.aac"
-            _generate_test_audio(audio_path, duration=5.0)
-
-            final_path = work_dir / "final.mp4"
-            mux_cmd = [
-                "ffmpeg",
-                "-y",
-                "-i",
-                str(render_result.output_path),
-                "-i",
-                str(audio_path),
-                "-c:v",
-                "copy",
-                "-c:a",
-                "aac",
-                "-shortest",
-                str(final_path),
-            ]
-            subprocess.run(mux_cmd, check=True, capture_output=True, timeout=30)
-
-            assert final_path.exists()
-            assert final_path.stat().st_size > 0
-
-    def test_single_clip_pipeline(self):
-        """单 clip 渲染（无转场）。"""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            work_dir = Path(tmpdir)
-
-            video_path = work_dir / "single.mp4"
-            _generate_test_video(video_path, duration=5.0)
-
-            plan, clips, asset_map = _build_plan_and_clips_from_task("single_test", [video_path], "one_take")
-
-            service = UnifiedRenderService(
-                plan=plan,
-                clips=clips,
-                asset_path_map=asset_map,
-                work_dir=work_dir,
-                output_width=640,
-                output_height=360,
-            )
-            result = service.render()
-
-            assert result.output_path.exists()
-            assert result.duration > 0
-
-    def test_dedup_helper_integration(self):
-        """验证 dedup_helpers.create_video_record_and_dedup 的导入和签名。"""
-        # 只验证函数存在且签名正确（不实际调用，需要数据库）
-        import inspect
-
-        from video_processing.dedup_helpers import create_video_record_and_dedup
-
-        sig = inspect.signature(create_video_record_and_dedup)
-        params = set(sig.parameters.keys())
-        expected = {
-            "generation_task_id",
-            "project_id",
-            "batch_id",
-            "file_url",
-            "file_size",
-            "duration",
-            "video_path",
-            "mode",
-            "session",
-            "width",
-            "height",
-            "fps",
-        }
-        assert expected.issubset(params), f"Missing params: {expected - params}"