apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -1,13 +1,18 @@
-"""剪辑计划渲染任务 — Phase 8 任务 2.05.
+"""剪辑计划渲染任务 — 支持 Feature Flag 灰度.
 
 Celery 任务 worker.render_edit_plan:
   1. 加载 EditPlan + EditPlanClips
-  2. 下载各片段素材
-  3. 使用 UnifiedRenderService 按时间线+图层渲染
+  2. 根据 Feature Flag 选择渲染引擎（legacy / unified）
+  3. 下载各片段素材 + 渲染
   4. 上传渲染结果到 OSS
   5. 创建 GeneratedVideo 记录 + 查重
   6. 更新 EditPlan / EditPlanClip 状态
   7. 更新 GenerationTask 进度
+
+渲染引擎灰度：
+  - 走 Feature Flag (render_engine) 控制
+  - legacy: VideoComposeService + FFmpeg filter_complex
+  - unified: UnifiedRenderService 图层架构
 """
 
 from __future__ import annotations
@@ -63,14 +68,268 @@ def _get_repos():
 # ── Celery Task ───────────────────────────────────────────────────────────────
 
 
+def _resolve_render_engine(user_id: str) -> str:
+    """根据 Feature Flag 决定使用哪个渲染引擎。
+
+    Returns:
+        "legacy" 或 "unified"
+    """
+    try:
+        from video_processing.render_engine_resolver import get_render_engine_resolver
+
+        resolver = get_render_engine_resolver()
+        return resolver.get_engine(user_id=user_id)
+    except Exception as exc:
+        logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc)
+        return "legacy"
+
+
+def _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg: str):
+    """统一的计划失败标记工具。"""
+    plan = plan_repo.get(plan_id)
+    if plan and plan.status.value == "rendering":
+        plan.mark_failed()
+        plan_repo.update(plan)
+    if generation_task_id:
+        gen_task = gen_task_repo.get(generation_task_id)
+        if gen_task and gen_task.status.value != "failed":
+            gen_task.status = "failed"
+            gen_task.error_message = error_msg
+            gen_task.completed_at = datetime.now(timezone.utc)
+            gen_task_repo.update(gen_task)
+
+
+def _finalize_render_success(
+    plan,
+    plan_repo,
+    clip_repo,
+    gen_task_repo,
+    db,
+    plan_id: str,
+    output_url: str,
+    storage_key: str,
+    duration: float,
+    file_size: int,
+    width: int,
+    height: int,
+    rendered_clip_ids: list[str],
+    failed_clip_ids: list[str],
+    generation_task_id: str,
+    output_path: Path,
+    engine: str,
+) -> dict:
+    """渲染成功后的统一收尾：查重 + 更新状态 + 返回结果。"""
+    # 创建 GeneratedVideo 记录 + 查重
+    project_id = plan.project_id or ""
+    batch_id = plan.config.get("batch_id", "")
+    mode = plan.config.get("mode", "edit_plan")
+    if generation_task_id and project_id:
+        try:
+            create_video_record_and_dedup(
+                generation_task_id=generation_task_id,
+                project_id=project_id,
+                batch_id=batch_id,
+                file_url=output_url or "",
+                file_size=file_size,
+                duration=duration,
+                video_path=str(output_path),
+                mode=mode,
+                session=db,
+                width=width,
+                height=height,
+                fps=OUTPUT_FPS,
+            )
+        except Exception as dedup_err:
+            logger.warning("查重失败（不影响渲染结果）: %s", dedup_err)
+
+    # 更新片段状态为 rendered
+    for clip_id in rendered_clip_ids:
+        clip = clip_repo.get(clip_id)
+        if clip and clip.status.value == "ready":
+            clip.mark_rendered()
+            clip_repo.update(clip)
+
+    # 更新 EditPlan 状态为 completed
+    plan.config["rendered_url"] = output_url or ""
+    plan.config["rendered_storage_key"] = storage_key
+    plan.mark_completed()
+    plan_repo.update(plan)
+
+    # 更新 GenerationTask 状态为 completed
+    if generation_task_id:
+        gen_task = gen_task_repo.get(generation_task_id)
+        if gen_task:
+            gen_task.status = "completed"
+            gen_task.progress = 100.0
+            gen_task.result_count = len(rendered_clip_ids)
+            gen_task.completed_at = datetime.now(timezone.utc)
+            gen_task_repo.update(gen_task)
+
+    logger.info(
+        "剪辑计划渲染完成: plan_id=%s engine=%s rendered=%d failed=%d duration=%.1fs",
+        plan_id,
+        engine,
+        len(rendered_clip_ids),
+        len(failed_clip_ids),
+        duration,
+    )
+
+    return {
+        "status": "completed",
+        "plan_id": plan_id,
+        "rendered_count": len(rendered_clip_ids),
+        "failed_count": len(failed_clip_ids),
+        "output_url": output_url,
+        "duration": duration,
+    }
+
+
+def _render_with_unified(
+    plan,
+    clips,
+    asset_path_map: dict[str, Path],
+    tmpdir_path: Path,
+    rendered_clip_ids: list[str],
+    plan_id: str,
+    generation_task_id: str,
+    plan_repo,
+    clip_repo,
+    gen_task_repo,
+    db,
+) -> dict:
+    """统一渲染引擎路径（UnifiedRenderService 图层架构）。"""
+    render_service = UnifiedRenderService(
+        plan=plan,
+        clips=clips,
+        asset_path_map=asset_path_map,
+        work_dir=tmpdir_path,
+        output_width=OUTPUT_WIDTH,
+        output_height=OUTPUT_HEIGHT,
+        output_fps=int(OUTPUT_FPS),
+    )
+
+    try:
+        render_result = render_service.render()
+    except Exception as render_err:
+        logger.error("渲染失败(unified): %s — %s", plan_id, render_err)
+        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"渲染失败: {render_err}")
+        return {"status": "error", "message": f"渲染失败: {render_err}"}
+
+    output_path = render_result.output_path
+
+    # 上传到 OSS
+    storage_key = f"rendered/{plan_id}/output.mp4"
+    output_url = upload_to_oss(output_path, storage_key)
+
+    failed_clip_ids: list[str] = []
+    return _finalize_render_success(
+        plan=plan,
+        plan_repo=plan_repo,
+        clip_repo=clip_repo,
+        gen_task_repo=gen_task_repo,
+        db=db,
+        plan_id=plan_id,
+        output_url=output_url or "",
+        storage_key=storage_key,
+        duration=render_result.duration,
+        file_size=render_result.file_size,
+        width=render_result.width,
+        height=render_result.height,
+        rendered_clip_ids=rendered_clip_ids,
+        failed_clip_ids=failed_clip_ids,
+        generation_task_id=generation_task_id,
+        output_path=output_path,
+        engine="unified",
+    )
+
+
+def _render_with_legacy(
+    plan,
+    clips,
+    rendered_clip_ids: list[str],
+    failed_clip_ids: list[str],
+    tmpdir_path: Path,
+    plan_id: str,
+    generation_task_id: str,
+    plan_repo,
+    clip_repo,
+    gen_task_repo,
+    db,
+) -> dict:
+    """旧引擎路径（VideoComposeService + FFmpeg filter_complex）。"""
+    import os
+    import subprocess
+
+    from apps.api.app.services.video_compose_service import VideoComposeService
+
+    compose_svc = VideoComposeService(db)
+
+    # 校验合成条件
+    validation = compose_svc.validate_compose(plan_id)
+    if not validation.valid:
+        error_msg = "; ".join(validation.errors)
+        logger.error("合成校验失败(legacy): %s — %s", plan_id, error_msg)
+        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"合成校验失败: {error_msg}")
+        return {"status": "error", "message": error_msg}
+
+    # 构建 FFmpeg 命令
+    output_dir = os.environ.get("VIDEO_OUTPUT_DIR", str(tmpdir_path))
+    output_path = Path(output_dir) / f"{plan_id}.mp4"
+    compose_cmd = compose_svc.build_compose_command(plan_id, str(output_path))
+
+    logger.info("执行 FFmpeg (legacy): plan_id=%s", plan_id)
+    try:
+        subprocess.run(
+            compose_cmd.command,
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            timeout=3600,
+        )
+    except subprocess.CalledProcessError as e:
+        error_msg = f"FFmpeg 执行失败: {e.stderr[:500]}"
+        logger.error("FFmpeg 执行失败(legacy): %s — %s", plan_id, error_msg)
+        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg)
+        return {"status": "error", "message": error_msg}
+
+    # 获取文件大小
+    file_size = output_path.stat().st_size if output_path.exists() else 0
+    duration = compose_cmd.estimated_duration or 0.0
+
+    # 上传到 OSS
+    storage_key = f"rendered/{plan_id}/output.mp4"
+    output_url = upload_to_oss(output_path, storage_key)
+
+    return _finalize_render_success(
+        plan=plan,
+        plan_repo=plan_repo,
+        clip_repo=clip_repo,
+        gen_task_repo=gen_task_repo,
+        db=db,
+        plan_id=plan_id,
+        output_url=output_url or "",
+        storage_key=storage_key,
+        duration=duration,
+        file_size=file_size,
+        width=OUTPUT_WIDTH,
+        height=OUTPUT_HEIGHT,
+        rendered_clip_ids=rendered_clip_ids,
+        failed_clip_ids=failed_clip_ids,
+        generation_task_id=generation_task_id,
+        output_path=output_path,
+        engine="legacy",
+    )
+
+
 @celery_app.task(name="worker.render_edit_plan", bind=True, max_retries=2)
 def render_edit_plan(self, plan_id: str) -> dict:
     """渲染剪辑计划
 
     流程：
     1. 加载 EditPlan + EditPlanClips
-    2. 下载各片段素材到临时目录，构建 asset_path_map
-    3. 使用 UnifiedRenderService 按时间线+图层渲染
+    2. 根据 Feature Flag 选择渲染引擎（legacy / unified）
+    3. 下载素材 + 渲染
     4. 上传渲染结果到 OSS
     5. 创建 GeneratedVideo 记录 + 查重
     6. 更新 EditPlan → completed, EditPlanClips → rendered
@@ -79,6 +338,7 @@ def render_edit_plan(self, plan_id: str) -> dict:
     logger.info("开始渲染剪辑计划: plan_id=%s", plan_id)
 
     generation_task_id = ""
+    engine = "legacy"
 
     for repos in _get_repos():
         plan_repo, clip_repo, gen_task_repo, db = repos
@@ -93,7 +353,12 @@ def render_edit_plan(self, plan_id: str) -> dict:
             # 获取 generation_task_id（提前读取，确保 except 块可用）
             generation_task_id = plan.config.get("generation_task_id", "")
 
-            # 2. 加载片段列表（按 order 排序）
+            # 2. 选择渲染引擎（Feature Flag 灰度控制）
+            user_id = plan.created_by_user_id or ""
+            engine = _resolve_render_engine(user_id)
+            logger.info("剪辑计划渲染引擎: plan_id=%s engine=%s user_id=%s", plan_id, engine, user_id)
+
+            # 3. 加载片段列表（按 order 排序）
             clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
             if not clips:
                 logger.warning("剪辑计划没有片段: %s", plan_id)
@@ -174,100 +439,38 @@ def render_edit_plan(self, plan_id: str) -> dict:
                             gen_task_repo.update(gen_task)
                     return {"status": "error", "message": "所有片段素材下载失败"}
 
-                # 4. 使用 UnifiedRenderService 渲染
-                render_service = UnifiedRenderService(
-                    plan=plan,
-                    clips=clips,
-                    asset_path_map=asset_path_map,
-                    work_dir=tmpdir_path,
-                    output_width=OUTPUT_WIDTH,
-                    output_height=OUTPUT_HEIGHT,
-                    output_fps=int(OUTPUT_FPS),
-                )
+                # 4. 根据引擎选择渲染方式
+                if engine == "unified":
+                    result = _render_with_unified(
+                        plan=plan,
+                        clips=clips,
+                        asset_path_map=asset_path_map,
+                        tmpdir_path=tmpdir_path,
+                        rendered_clip_ids=rendered_clip_ids,
+                        plan_id=plan_id,
+                        generation_task_id=generation_task_id,
+                        plan_repo=plan_repo,
+                        clip_repo=clip_repo,
+                        gen_task_repo=gen_task_repo,
+                        db=db,
+                    )
+                else:
+                    result = _render_with_legacy(
+                        plan=plan,
+                        clips=clips,
+                        rendered_clip_ids=rendered_clip_ids,
+                        failed_clip_ids=failed_clip_ids,
+                        tmpdir_path=tmpdir_path,
+                        plan_id=plan_id,
+                        generation_task_id=generation_task_id,
+                        plan_repo=plan_repo,
+                        clip_repo=clip_repo,
+                        gen_task_repo=gen_task_repo,
+                        db=db,
+                    )
 
-                try:
-                    render_result = render_service.render()
-                except Exception as render_err:
-                    logger.error("渲染失败: %s — %s", plan_id, render_err)
-                    plan.mark_failed()
-                    plan_repo.update(plan)
-                    if generation_task_id:
-                        gen_task = gen_task_repo.get(generation_task_id)
-                        if gen_task:
-                            gen_task.status = "failed"
-                            gen_task.error_message = f"渲染失败: {render_err}"
-                            gen_task.completed_at = datetime.now(timezone.utc)
-                            gen_task_repo.update(gen_task)
-                    return {"status": "error", "message": f"渲染失败: {render_err}"}
-
-                output_path = render_result.output_path
-
-                # 5. 上传到 OSS
-                storage_key = f"rendered/{plan_id}/output.mp4"
-                output_url = upload_to_oss(output_path, storage_key)
-
-                # 6. 创建 GeneratedVideo 记录 + 查重
-                project_id = plan.project_id or ""
-                batch_id = plan.config.get("batch_id", "")
-                mode = plan.config.get("mode", "edit_plan")
-                if generation_task_id and project_id:
-                    try:
-                        create_video_record_and_dedup(
-                            generation_task_id=generation_task_id,
-                            project_id=project_id,
-                            batch_id=batch_id,
-                            file_url=output_url or "",
-                            file_size=render_result.file_size,
-                            duration=render_result.duration,
-                            video_path=str(output_path),
-                            mode=mode,
-                            session=db,
-                            width=render_result.width,
-                            height=render_result.height,
-                            fps=OUTPUT_FPS,
-                        )
-                    except Exception as dedup_err:
-                        logger.warning("查重失败（不影响渲染结果）: %s", dedup_err)
-
-                # 7. 更新片段状态为 rendered
-                for clip_id in rendered_clip_ids:
-                    clip = clip_repo.get(clip_id)
-                    if clip and clip.status.value == "ready":
-                        clip.mark_rendered()
-                        clip_repo.update(clip)
-
-                # 8. 更新 EditPlan 状态为 completed
-                plan.config["rendered_url"] = output_url or ""
-                plan.config["rendered_storage_key"] = storage_key
-                plan.mark_completed()
-                plan_repo.update(plan)
-
-                # 9. 更新 GenerationTask 状态为 completed
-                if generation_task_id:
-                    gen_task = gen_task_repo.get(generation_task_id)
-                    if gen_task:
-                        gen_task.status = "completed"
-                        gen_task.progress = 100.0
-                        gen_task.result_count = len(rendered_clip_ids)
-                        gen_task.completed_at = datetime.now(timezone.utc)
-                        gen_task_repo.update(gen_task)
-
-                logger.info(
-                    "剪辑计划渲染完成: plan_id=%s rendered=%d failed=%d duration=%.1fs",
-                    plan_id,
-                    len(rendered_clip_ids),
-                    len(failed_clip_ids),
-                    render_result.duration,
-                )
-
-                return {
-                    "status": "completed",
-                    "plan_id": plan_id,
-                    "rendered_count": len(rendered_clip_ids),
-                    "failed_count": len(failed_clip_ids),
-                    "output_url": output_url,
-                    "duration": render_result.duration,
-                }
+                result["engine"] = engine
+                return result
 
         except Exception as exc:
             logger.exception("渲染剪辑计划异常: %s", plan_id)

tests/unit/test_edit_plan_worker_failure.py

@@ -63,6 +63,8 @@ class StubEditPlan:
     template_id: str = "tmpl-001"
     status: Any = None
     config: dict = field(default_factory=dict)
+    project_id: str = ""
+    created_by_user_id: str = "user-001"
 
     def mark_failed(self):
         self.status = _StubStatus("failed")