apps/api/app/api/routes/feature_flags.py

@@ -32,9 +32,7 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/internal/feature-flags", tags=["Internal"])
 
 # 允许管理的 flag 白名单（防止误操作其他系统 flag）
-ALLOWED_FLAGS = {
-    "render_engine",
-}
+ALLOWED_FLAGS: set[str] = set()
 
 
 def _get_feature_flag_store() -> RedisFeatureFlagStore:

apps/worker/video_processing/render_engine_resolver.py

@@ -1,204 +0,0 @@
-"""渲染引擎 Feature Flag 解析器。
-
-封装渲染引擎选择逻辑，支持：
-- 环境变量作为默认值（RENDER_ENGINE=legacy/unified）
-- Redis Feature Flag 运行时覆盖（白名单 + 百分比 + 全局开关）
-- 定时刷新，支持热更新不重启 worker
-
-使用方式：
-    resolver = RenderEngineResolver(redis_url="redis://...", default_engine="legacy")
-    engine = resolver.get_engine(user_id="user123")
-    # engine: "legacy" 或 "unified"
-"""
-
-from __future__ import annotations
-
-import logging
-import threading
-from typing import Optional
-
-from packages.adapters.redis.feature_flag_store import (
-    FeatureFlagConfig,
-    FeatureFlagStore,
-    InMemoryFeatureFlagStore,
-    RedisFeatureFlagStore,
-)
-
-logger = logging.getLogger(__name__)
-
-# Feature Flag 名称常量
-FLAG_RENDER_ENGINE = "render_engine"
-
-# 引擎常量
-ENGINE_LEGACY = "legacy"
-ENGINE_UNIFIED = "unified"
-VALID_ENGINES = {ENGINE_LEGACY, ENGINE_UNIFIED}
-
-
-class RenderEngineResolver:
-    """渲染引擎选择器。
-
-    判定逻辑（从高到低）：
-    1. Redis flag 白名单匹配 → unified
-    2. Redis flag 百分比命中 → unified
-    3. Redis flag 全局开启（100%）→ unified
-    4. 环境变量默认值 → legacy / unified
-
-    当 Redis 不可用时，自动降级到环境变量默认值，不影响业务。
-    """
-
-    def __init__(
-        self,
-        default_engine: str = ENGINE_LEGACY,
-        redis_url: Optional[str] = None,
-        refresh_interval: float = 30.0,
-        store: Optional[FeatureFlagStore] = None,
-    ) -> None:
-        """
-        Args:
-            default_engine: 环境变量默认的引擎名（legacy / unified）
-            redis_url: Redis 连接 URL，传 None 时使用内存实现（测试用）
-            refresh_interval: Redis flag 配置刷新间隔（秒）
-            store: 直接传入 store 实例（测试用，优先级高于 redis_url）
-        """
-        self._default_engine = default_engine.lower() if default_engine else ENGINE_LEGACY
-        if self._default_engine not in VALID_ENGINES:
-            logger.warning(
-                "Invalid default engine '%s', fallback to '%s'",
-                self._default_engine,
-                ENGINE_LEGACY,
-            )
-            self._default_engine = ENGINE_LEGACY
-
-        if store is not None:
-            self._store = store
-        elif redis_url:
-            self._store = RedisFeatureFlagStore(redis_url=redis_url)
-        else:
-            self._store = InMemoryFeatureFlagStore()
-            logger.info("No Redis configured, using in-memory feature flag store")
-
-        self._refresh_interval = refresh_interval
-        self._lock = threading.Lock()
-        self._cached_config: Optional[FeatureFlagConfig] = None
-        self._last_refresh: float = 0.0
-
-    def _maybe_refresh(self) -> None:
-        """惰性刷新配置，超过刷新间隔时从存储重新读取。"""
-        import time
-
-        now = time.time()
-        if now - self._last_refresh < self._refresh_interval:
-            return
-
-        try:
-            config = self._store.get(FLAG_RENDER_ENGINE)
-            with self._lock:
-                self._cached_config = config
-                self._last_refresh = now
-        except Exception as exc:
-            logger.warning("Failed to refresh render engine flag: %s", exc)
-            # 刷新失败时保留旧缓存，不中断业务
-            if self._cached_config is None:
-                # 首次就读失败，设一个默认值
-                with self._lock:
-                    self._cached_config = FeatureFlagConfig(name=FLAG_RENDER_ENGINE)
-                    self._last_refresh = now
-
-    def _get_config(self) -> FeatureFlagConfig:
-        """获取当前 flag 配置（带缓存）。"""
-        if self._cached_config is None:
-            self._maybe_refresh()
-        else:
-            self._maybe_refresh()
-        return self._cached_config or FeatureFlagConfig(name=FLAG_RENDER_ENGINE)
-
-    def get_engine(self, user_id: Optional[str] = None) -> str:
-        """获取当前应该使用的渲染引擎。
-
-        Args:
-            user_id: 用户ID，用于白名单匹配和百分比哈希。
-                    传 None 时只看全局开关。
-
-        Returns:
-            "legacy" 或 "unified"
-        """
-        config = self._get_config()
-
-        # 全局关闭 → 用默认值
-        if not config.enabled:
-            return self._default_engine
-
-        # 白名单匹配 / 百分比命中 → unified
-        if config.is_active(user_id):
-            return ENGINE_UNIFIED
-
-        # 未命中灰度 → 用默认值
-        return self._default_engine
-
-    def should_use_unified(self, user_id: Optional[str] = None) -> bool:
-        """便捷方法：是否应该使用统一渲染引擎。"""
-        return self.get_engine(user_id) == ENGINE_UNIFIED
-
-    def force_refresh(self) -> None:
-        """强制立即刷新配置（用于管理接口修改后立即生效）。"""
-        self._last_refresh = 0.0
-        if isinstance(self._store, RedisFeatureFlagStore):
-            self._store.invalidate_cache(FLAG_RENDER_ENGINE)
-        self._maybe_refresh()
-
-    def get_config_snapshot(self) -> dict:
-        """获取当前配置快照（用于管理接口展示）。"""
-        config = self._get_config()
-        return {
-            "flag_name": FLAG_RENDER_ENGINE,
-            "default_engine": self._default_engine,
-            "enabled": config.enabled,
-            "percentage": config.percentage,
-            "whitelist": sorted(config.whitelist),
-            "refresh_interval": self._refresh_interval,
-            "last_refresh": self._last_refresh,
-        }
-
-    def set_flag(self, config: FeatureFlagConfig) -> None:
-        """设置 flag 配置（管理接口用）。"""
-        config.name = FLAG_RENDER_ENGINE
-        self._store.set(config)
-        self.force_refresh()
-
-
-# 全局单例
-_resolver: Optional[RenderEngineResolver] = None
-_resolver_lock = threading.Lock()
-
-
-def get_render_engine_resolver() -> RenderEngineResolver:
-    """获取全局单例（基于 worker 配置）。"""
-    global _resolver
-    if _resolver is not None:
-        return _resolver
-
-    with _resolver_lock:
-        if _resolver is not None:
-            return _resolver
-
-        try:
-            from worker_app.core.config import get_settings
-
-            settings = get_settings()
-            redis_url = getattr(settings, "redis_url", None) or getattr(settings, "broker_url", None)
-            default = getattr(settings, "render_engine", ENGINE_LEGACY)
-            _resolver = RenderEngineResolver(
-                default_engine=default,
-                redis_url=redis_url,
-            )
-            logger.info(
-                "RenderEngineResolver initialized: default=%s, redis=%s",
-                default,
-                bool(redis_url),
-            )
-        except Exception as exc:
-            logger.warning("Failed to init RenderEngineResolver from settings: %s", exc)
-            _resolver = RenderEngineResolver(default_engine=ENGINE_LEGACY)
-
-    return _resolver

apps/worker/worker_app/core/config.py

@@ -19,9 +19,6 @@ class WorkerSettings(BaseSettings):
     auto_create_schema: bool = False
     redis_url: str = "redis://redis:6379/0"
 
-    # 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService
-    render_engine: str = "legacy"
-
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

apps/worker/worker_app/tasks/compose_video.py

@@ -1,6 +1,6 @@
 """视频合成 Celery 任务 — Phase 8 任务 2.10.
 
-使用 JobService 管理任务生命周期，集成 VideoComposeService 执行合成。
+使用 JobService 管理任务生命周期，通过 RenderAdapter 调用 UnifiedRenderService 执行合成。
 """
 
 from __future__ import annotations
@@ -35,9 +35,7 @@ def _get_job_service():
 def compose_video(self, job_id: str, **kwargs):
     """视频合成任务。
 
-    根据 RENDER_ENGINE 配置选择渲染引擎：
-      - legacy: 旧 VideoComposeService（filter_complex 模式）
-      - unified: 新 UnifiedRenderService（图层架构）
+    使用 UnifiedRenderService（图层架构）进行渲染。
 
     Args:
         job_id: JobService 中的任务 ID
@@ -56,30 +54,8 @@ def compose_video(self, job_id: str, **kwargs):
             job_service.fail_job(job_id, "Missing plan_id in job payload")
             return {"status": "error", "message": "Missing plan_id"}
 
-        # 判断使用哪个渲染引擎
-        # 优先级：Redis Feature Flag（白名单 > 百分比） > 环境变量默认
-        from video_processing.render_engine_resolver import get_render_engine_resolver
-
-        resolver = get_render_engine_resolver()
-        user_id = job.created_by_user_id or None
-        engine = resolver.get_engine(user_id=user_id)
-        # 灰度期间打印详细 flag 配置，便于排查
-        config = resolver.get_config_snapshot()
-        logger.info(
-            "compose_video 引擎选择: job_id=%s engine=%s user_id=%s enabled=%s percentage=%s whitelist=%d default=%s",
-            job_id,
-            engine,
-            user_id,
-            config.get("enabled"),
-            config.get("percentage"),
-            len(config.get("whitelist", [])),
-            config.get("default_engine"),
-        )
-
-        if engine == "unified":
-            return _compose_with_unified_engine(self, job_service, job, plan_id, db)
-        else:
-            return _compose_with_legacy_engine(self, job_service, job, plan_id, db)
+        # 使用 unified 渲染引擎
+        return _compose_with_unified_engine(self, job_service, job, plan_id, db)
 
     except self.retry_exc as exc:
         logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc)
@@ -95,69 +71,6 @@ def compose_video(self, job_id: str, **kwargs):
         db.close()
 
 
-def _compose_with_legacy_engine(task, job_service, job, plan_id: str, db) -> dict:
-    """旧引擎渲染路径（VideoComposeService）。"""
-    job_id = job.id
-
-    # 标记为 running
-    job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境")
-
-    # 延迟导入 VideoComposeService
-    from apps.api.app.services.video_compose_service import VideoComposeService
-
-    compose_svc = VideoComposeService(db)
-
-    # 校验合成条件
-    job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件")
-    validation = compose_svc.validate_compose(plan_id)
-    if not validation.valid:
-        error_msg = "; ".join(validation.errors)
-        job_service.fail_job(job_id, f"合成校验失败: {error_msg}")
-        return {"status": "error", "message": error_msg}
-
-    # 构建合成命令
-    job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令")
-    _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output"))
-    output_path = os.path.join(_output_dir, f"{job_id}.mp4")
-    compose_cmd = compose_svc.build_compose_command(plan_id, output_path)
-
-    # 执行 FFmpeg
-    job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成")
-    logger.info("Executing FFmpeg for job %s, plan %s", job_id, plan_id)
-
-    try:
-        from video_processing.ffmpeg_utils import run_ffmpeg
-
-        run_ffmpeg(compose_cmd.command, timeout=3600)
-    except Exception as e:
-        error_msg = f"FFmpeg 执行失败: {str(e)[:500]}"
-        job_service.fail_job(job_id, error_msg)
-        raise
-
-    # 上传结果
-    job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果")
-    storage_key = f"rendered/{plan_id}/{job_id}.mp4"
-
-    from worker_app.tasks.edit_plan_generation import _upload_to_oss
-
-    output_url = _upload_to_oss(Path(output_path), storage_key)
-
-    # 更新 Job 状态为完成
-    result_data = {
-        "plan_id": plan_id,
-        "output_path": output_path,
-        "storage_key": storage_key,
-        "output_url": output_url or "",
-        "estimated_duration": compose_cmd.estimated_duration,
-        "clip_count": len(compose_cmd.clip_chains),
-        "engine": "legacy",
-    }
-    job_service.complete_job(job_id, result=result_data)
-
-    logger.info("视频合成完成(legacy): job_id=%s, plan_id=%s", job_id, plan_id)
-    return {"status": "completed", "job_id": job_id, "result": result_data}
-
-
 def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> dict:
     """新引擎渲染路径（UnifiedRenderService + RenderAdapter）。"""
     job_id = job.id

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -1,24 +1,18 @@
-"""剪辑计划渲染任务 — 支持 Feature Flag 灰度.
+"""剪辑计划渲染任务 — 使用 UnifiedRenderService 统一渲染引擎.
 
 Celery 任务 worker.render_edit_plan:
   1. 加载 EditPlan + EditPlanClips
-  2. 根据 Feature Flag 选择渲染引擎（legacy / unified）
+  2. 通过 RenderAdapter 调用 UnifiedRenderService 渲染
   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
 
 import logging
-import tempfile
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -35,10 +29,6 @@ OUTPUT_FPS = 25.0
 # ── 共享工具模块导入 ──────────────────────────────────────────────────────────
 
 from video_processing.dedup_helpers import create_video_record_and_dedup
-from video_processing.oss_helpers import (
-    download_asset,
-    upload_to_oss,
-)
 
 # ── Repository imports (延迟导入避免循环依赖) ─────────────────────────────────
 
@@ -66,34 +56,6 @@ 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()
-        engine = resolver.get_engine(user_id=user_id)
-        # 灰度期间打印详细 flag 配置，便于排查
-        config = resolver.get_config_snapshot()
-        logger.info(
-            "edit_plan 引擎选择: user_id=%s engine=%s enabled=%s percentage=%s whitelist=%d default=%s",
-            user_id,
-            engine,
-            config.get("enabled"),
-            config.get("percentage"),
-            len(config.get("whitelist", [])),
-            config.get("default_engine"),
-        )
-        return engine
-    except Exception as exc:
-        logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc, exc_info=True)
-        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)
@@ -311,331 +273,13 @@ def _render_with_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
-
-    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"
-
-    # 从 plan.config.export 读取输出分辨率，兼容 plan 自定义配置
-    plan_config = plan.config or {}
-    export_config = plan_config.get("export", {}) or {}
-    output_width = OUTPUT_WIDTH
-    output_height = OUTPUT_HEIGHT
-    resolution = export_config.get("resolution", "")
-    if resolution and "x" in resolution:
-        try:
-            w_str, h_str = resolution.lower().split("x", 1)
-            output_width = int(w_str)
-            output_height = int(h_str)
-        except (ValueError, TypeError):
-            pass
-
-    fps = export_config.get("fps", 25)
-    try:
-        fps = int(fps)
-    except (ValueError, TypeError):
-        fps = 25
-
-    compose_cmd = compose_svc.build_compose_command(
-        plan_id,
-        str(output_path),
-        output_width=output_width,
-        output_height=output_height,
-        fps=fps,
-    )
-
-    logger.info("执行 FFmpeg (legacy): plan_id=%s cmd=%s", plan_id, " ".join(compose_cmd.command)[:500])
-
-    # 开始渲染，更新进度
-    if generation_task_id:
-        try:
-            gen_task = gen_task_repo.get(generation_task_id)
-            if gen_task and gen_task.progress < 40.0:
-                gen_task.progress = 40.0
-                gen_task.append_log(
-                    stage="render_start",
-                    message="开始FFmpeg渲染（legacy）",
-                    level="INFO",
-                    progress=40.0,
-                )
-                gen_task_repo.update(gen_task)
-        except Exception:
-            pass
-
-    try:
-        from video_processing.ffmpeg_utils import run_ffmpeg
-
-        run_ffmpeg(compose_cmd.command, timeout=3600)
-    except Exception as e:
-        # 提取完整 stderr（如果是 CalledProcessError）
-        stderr_text = ""
-        if hasattr(e, "stderr"):
-            stderr_raw = e.stderr
-            if isinstance(stderr_raw, bytes):
-                stderr_text = stderr_raw.decode("utf-8", errors="replace")
-            elif isinstance(stderr_raw, str):
-                stderr_text = stderr_raw
-
-        # 完整命令（截断前2000字符，避免日志过大）
-        full_cmd = " ".join(compose_cmd.command)
-        cmd_preview = full_cmd[:2000] + ("..." if len(full_cmd) > 2000 else "")
-
-        # 拼接完整错误信息：命令 + 异常 + stderr最后1500字符
-        error_parts = [f"FFmpeg渲染失败(exit={getattr(e, 'returncode', 'unknown')})"]
-        error_parts.append("--- cmd ---")
-        error_parts.append(cmd_preview)
-        if stderr_text:
-            # 取最后1500字符，通常错误信息在末尾
-            stderr_preview = stderr_text[-1500:] if len(stderr_text) > 1500 else stderr_text
-            error_parts.append("--- stderr (last 1500 chars) ---")
-            error_parts.append(stderr_preview)
-        error_msg = "\n".join(error_parts)
-
-        logger.error("FFmpeg 执行失败(legacy): plan_id=%s\n%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
-    try:
-        from video_processing.ffmpeg_utils import probe_duration
-
-        actual_duration = probe_duration(str(output_path))
-        if actual_duration > 0:
-            duration = actual_duration
-    except Exception:
-        pass
-
-    # ── 标题/字幕叠加（legacy 引擎补齐） ────────────────────────────────
-    plan_config = plan.config or {}
-    title_cfg = plan_config.get("title", {}) or {}
-    subtitle_cfg = plan_config.get("subtitle", {}) or {}
-    title_text = title_cfg.get("text", "") or ""
-    subtitle_text = subtitle_cfg.get("text", "") or ""
-    title_enabled = title_cfg.get("enabled", True) and bool(title_text.strip())
-    subtitle_enabled = subtitle_cfg.get("enabled", True) and bool(subtitle_text.strip())
-    # ASR 自动字幕 legacy 暂不支持（需要额外 ASR 服务，统一用 unified 引擎）
-    has_subtitle_overlay = title_enabled or subtitle_enabled
-
-    if has_subtitle_overlay and output_path.exists() and duration > 0:
-        try:
-            from video_processing.ffmpeg_utils import run_ffmpeg
-            from video_processing.render_subtitles import generate_ass_subtitles
-
-            ass_path = tmpdir_path / f"subtitles_{plan_id}.ass"
-            generate_ass_subtitles(
-                ass_path,
-                video_width=output_width,
-                video_height=output_height,
-                video_duration=duration,
-                title_text=title_text,
-                title_config=title_cfg,
-                subtitle_text=subtitle_text,
-                subtitle_config=subtitle_cfg,
-            )
-            # 用 subtitles 滤镜叠加 ASS 字幕，音频直接 copy
-            subtitled_path = tmpdir_path / f"{plan_id}_subtitled.mp4"
-            # 处理 Windows 路径下的 ass 滤镜转义问题
-            ass_filter_path = str(ass_path).replace("\\", "/").replace(":", r"\:")
-            run_ffmpeg(
-                [
-                    "ffmpeg",
-                    "-y",
-                    "-i",
-                    str(output_path),
-                    "-vf",
-                    f"subtitles='{ass_filter_path}'",
-                    "-c:a",
-                    "copy",
-                    str(subtitled_path),
-                ],
-                timeout=1800,
-            )
-            if subtitled_path.exists() and subtitled_path.stat().st_size > 0:
-                output_path = subtitled_path
-                file_size = subtitled_path.stat().st_size
-                logger.info(
-                    "legacy 标题/字幕叠加完成: plan_id=%s title=%s subtitle=%s",
-                    plan_id,
-                    title_enabled,
-                    subtitle_enabled,
-                )
-        except Exception as sub_err:
-            logger.warning("legacy 标题/字幕叠加失败（不影响主流程）: plan_id=%s err=%s", plan_id, sub_err)
-
-    # ── TTS 配音混音（legacy 引擎补齐） ────────────────────────────────
-    tts_cfg = plan_config.get("tts", {}) or {}
-    tts_enabled = tts_cfg.get("enabled", False) and bool(tts_cfg.get("text", "").strip())
-
-    if tts_enabled and output_path.exists() and duration > 0:
-        try:
-            from packages.domain.tts_config import TtsConfig
-
-            tts_config = TtsConfig.parse(tts_cfg)
-            if tts_config.enabled and tts_config.text.strip():
-                from apps.worker.services.tts_service_factory import get_tts_service
-
-                tts_service = get_tts_service()
-                voiceover_path = tmpdir_path / f"voiceover_{plan_id}.wav"
-
-                # 生成配音音频
-                audio_path = tts_service.synthesize(
-                    text=tts_config.text,
-                    voice_id=tts_config.voice_id,
-                    speed=tts_config.speed,
-                    pitch=tts_config.pitch,
-                    output_path=voiceover_path,
-                )
-
-                if audio_path and audio_path.exists() and audio_path.stat().st_size > 0:
-                    from video_processing.ffmpeg_utils import run_ffmpeg
-
-                    mixed_path = tmpdir_path / f"{plan_id}_with_voiceover.mp4"
-
-                    # 混音：配音音量按配置调整
-                    voice_volume = max(0.0, min(1.0, tts_config.volume))
-
-                    if tts_config.overlap_mode == "mix":
-                        # 混音模式：原音 + 配音混合
-                        filter_complex = (
-                            f"[0:a]volume=1.0[a0];"
-                            f"[1:a]volume={voice_volume:.2f}[a1];"
-                            f"[a0][a1]amix=inputs=2:duration=first:dropout_transition=0[aout]"
-                        )
-                    else:
-                        # replace 模式：配音替换原音
-                        filter_complex = f"[1:a]volume={voice_volume:.2f}[aout]"
-
-                    run_ffmpeg(
-                        [
-                            "ffmpeg",
-                            "-y",
-                            "-i",
-                            str(output_path),
-                            "-i",
-                            str(audio_path),
-                            "-filter_complex",
-                            filter_complex,
-                            "-map",
-                            "0:v",
-                            "-map",
-                            "[aout]",
-                            "-c:v",
-                            "copy",
-                            "-c:a",
-                            "aac",
-                            "-b:a",
-                            "128k",
-                            "-shortest",
-                            str(mixed_path),
-                        ],
-                        timeout=1800,
-                    )
-
-                    if mixed_path.exists() and mixed_path.stat().st_size > 0:
-                        output_path = mixed_path
-                        file_size = mixed_path.stat().st_size
-                        logger.info(
-                            "legacy TTS 配音混音完成: plan_id=%s voice_id=%s mode=%s",
-                            plan_id,
-                            tts_config.voice_id,
-                            tts_config.overlap_mode,
-                        )
-        except Exception as tts_err:
-            logger.warning("legacy TTS 配音混音失败（不影响主流程）: plan_id=%s err=%s", plan_id, tts_err)
-
-    # 渲染完成，更新进度
-    if generation_task_id:
-        try:
-            gen_task = gen_task_repo.get(generation_task_id)
-            if gen_task and gen_task.progress < 80.0:
-                gen_task.progress = 80.0
-                gen_task.append_log(
-                    stage="render_done",
-                    message="FFmpeg渲染完成（legacy）",
-                    level="INFO",
-                    progress=80.0,
-                )
-                gen_task_repo.update(gen_task)
-        except Exception:
-            pass
-
-    # 上传到 OSS
-    storage_key = f"rendered/{plan_id}/output.mp4"
-    output_url = upload_to_oss(output_path, storage_key)
-
-    # 上传完成，更新进度
-    if generation_task_id:
-        try:
-            gen_task = gen_task_repo.get(generation_task_id)
-            if gen_task and gen_task.progress < 95.0:
-                gen_task.progress = 95.0
-                gen_task.append_log(
-                    stage="upload_done",
-                    message="OSS上传完成（legacy）",
-                    level="INFO",
-                    progress=95.0,
-                )
-                gen_task_repo.update(gen_task)
-        except Exception:
-            pass
-
-    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. 根据 Feature Flag 选择渲染引擎（legacy / unified）
+    2. 通过 RenderAdapter 调用 UnifiedRenderService 渲染
     3. 下载素材 + 渲染
     4. 上传渲染结果到 OSS
     5. 创建 GeneratedVideo 记录 + 查重
@@ -645,7 +289,6 @@ 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
@@ -660,10 +303,7 @@ def render_edit_plan(self, plan_id: str) -> dict:
             # 获取 generation_task_id（提前读取，确保 except 块可用）
             generation_task_id = plan.config.get("generation_task_id", "")
 
-            # 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)
+            # 2. 准备渲染（使用 unified 渲染引擎）
 
             # 3. 加载片段列表（按 order 排序）
             clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
@@ -681,9 +321,9 @@ def render_edit_plan(self, plan_id: str) -> dict:
                     gen_task.started_at = datetime.now(timezone.utc)
                     gen_task.append_log(
                         stage="render_start",
-                        message=f"开始渲染，引擎 {engine}，片段数 {len(clips)}",
+                        message=f"开始渲染，片段数 {len(clips)}",
                         level="INFO",
-                        engine=engine,
+                        engine="unified",
                         clip_count=len(clips),
                     )
                     gen_task_repo.update(gen_task)
@@ -706,122 +346,19 @@ def render_edit_plan(self, plan_id: str) -> dict:
                                 pass
                         return {"status": "cancelled", "plan_id": plan_id, "message": "任务已取消"}
 
-            # 4. 根据引擎选择渲染方式
-            if engine == "unified":
-                # ── unified 路径：RenderAdapter 统一处理（下载 + BGM + ASR + 渲染 + 上传）
-                result = _render_with_unified(
-                    plan=plan,
-                    clips=clips,
-                    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:
-                # ── legacy 路径：原有的素材下载 + VideoComposeService
-                with tempfile.TemporaryDirectory(prefix="edit_plan_") as tmpdir:
-                    tmpdir_path = Path(tmpdir)
-                    asset_path_map: dict[str, Path] = {}
-                    rendered_clip_ids: list[str] = []
-                    failed_clip_ids: list[str] = []
+            # 4. 渲染（unified 引擎：RenderAdapter 统一处理下载 + BGM + ASR + 渲染 + 上传）
+            result = _render_with_unified(
+                plan=plan,
+                clips=clips,
+                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,
+            )
 
-                    # 预先批量查询所有素材的 storage_key
-                    # 兼容存量数据：storage_key 为空时 fallback 到 file_url
-                    from packages.adapters.sqlalchemy_impl.models import AssetModel
-
-                    clip_asset_ids = [c.asset_id for c in clips if c.asset_id]
-                    asset_storage_map: dict[str, str] = {}
-                    if clip_asset_ids:
-                        assets = db.query(AssetModel).filter(AssetModel.id.in_(clip_asset_ids)).all()
-                        asset_storage_map = {
-                            a.id: (a.storage_key or a.file_url or "") for a in assets if a.storage_key or a.file_url
-                        }
-
-                    for clip in clips:
-                        if not clip.asset_id:
-                            # 没有素材的片段跳过，标记为失败
-                            clip.mark_failed()
-                            clip_repo.update(clip)
-                            failed_clip_ids.append(clip.id)
-                            continue
-
-                        if clip.asset_id in asset_path_map:
-                            # 同一素材已下载（多个 clip 共享同一素材）
-                            rendered_clip_ids.append(clip.id)
-                            continue
-
-                        storage_key = asset_storage_map.get(clip.asset_id)
-                        if not storage_key:
-                            logger.warning(
-                                "片段素材无 storage_key，跳过: clip_id=%s asset_id=%s",
-                                clip.id,
-                                clip.asset_id,
-                            )
-                            clip.mark_failed()
-                            clip_repo.update(clip)
-                            failed_clip_ids.append(clip.id)
-                            continue
-
-                        # 下载素材
-                        ext = Path(storage_key).suffix or ".mp4"
-                        local_path = tmpdir_path / f"clip_{clip.order:04d}{ext}"
-                        if download_asset(storage_key, local_path):
-                            asset_path_map[clip.asset_id] = local_path
-                            rendered_clip_ids.append(clip.id)
-                        else:
-                            clip.mark_failed()
-                            clip_repo.update(clip)
-                            failed_clip_ids.append(clip.id)
-
-                    if not asset_path_map:
-                        logger.error("所有片段素材下载失败: %s", plan_id)
-                        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 = "所有片段素材下载失败"
-                                gen_task.completed_at = datetime.now(timezone.utc)
-                                gen_task.append_log(
-                                    stage="download_failed",
-                                    message="所有片段素材下载失败",
-                                    level="ERROR",
-                                )
-                                gen_task_repo.update(gen_task)
-                        return {"status": "error", "message": "所有片段素材下载失败"}
-
-                    # 素材下载完成，记录日志
-                    if generation_task_id:
-                        gen_task = gen_task_repo.get(generation_task_id)
-                        if gen_task:
-                            gen_task.append_log(
-                                stage="download_done",
-                                message=f"素材下载完成，成功 {len(asset_path_map)} 个，失败 {len(failed_clip_ids)} 个",
-                                level="INFO",
-                                success_count=len(asset_path_map),
-                                failed_count=len(failed_clip_ids),
-                            )
-                            gen_task.progress = 30.0
-                            gen_task_repo.update(gen_task)
-
-                    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,
-                    )
-
-            result["engine"] = engine
+            result["engine"] = "unified"
             return result
 
         except Exception as exc:

apps/worker/worker_app/tasks/generation.py

@@ -146,7 +146,6 @@ from video_processing.oss_helpers import (
     get_signed_download_url,
     upload_to_oss,
 )
-from video_processing.render_engine_resolver import ENGINE_LEGACY, ENGINE_UNIFIED
 from video_processing.unified_render_service import UnifiedRenderService
 
 # ── 虚拟 Plan / Clip（内存中构建，不写数据库） ────────────────────────────────
@@ -1018,161 +1017,6 @@ def _load_template_plan_config(template_id: str) -> dict:
         return {}
 
 
-# ── 渲染引擎选择 ─────────────────────────────────────────────────────────────
-
-
-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()
-        engine = resolver.get_engine(user_id=user_id)
-        # 灰度期间打印详细 flag 配置，便于排查
-        config = resolver.get_config_snapshot()
-        logger.info(
-            "[渲染引擎] flag 解析: user_id=%s engine=%s enabled=%s percentage=%s whitelist=%d default=%s",
-            user_id,
-            engine,
-            config.get("enabled"),
-            config.get("percentage"),
-            len(config.get("whitelist", [])),
-            config.get("default_engine"),
-        )
-        return engine
-    except Exception as exc:
-        # 异常时 fallback 到 legacy（保守策略，与 edit_plan_generation 一致）
-        logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc, exc_info=True)
-        return ENGINE_LEGACY
-
-
-# ── 旧引擎渲染（FFmpeg filter_complex） ────────────────────────────────────────
-
-
-def _render_with_legacy_engine(
-    task_id: str,
-    virtual_clips: list[_VirtualClip],
-    asset_path_map: dict[str, Path],
-    work_dir: Path,
-    output_path: Path,
-) -> tuple[float, int]:
-    """旧引擎渲染路径：手动构建 FFmpeg filter_complex 命令。
-
-    说明：generate_video 任务使用虚拟 clips（无 EditPlan 数据库记录），
-    因此无法直接复用 VideoComposeService。这里手动构建等价的 filter_complex
-    命令，与旧引擎行为一致（scale → crop → setpts → trim → setpts，
-    无 fps 归一化，保持原帧率）。
-
-    支持模式：one_take / pip / voice_over / voice_pip
-    - 所有模式统一走 concat 滤镜（与旧引擎多片段逻辑一致）
-
-    Returns:
-        (duration_seconds, file_size_bytes)
-    """
-    import subprocess
-
-    main_clips = [
-        c
-        for c in virtual_clips
-        if c.clip_type in ("main", "b_roll", "background")
-        or (c.clip_type == "main" and c.config.get("role") == "b_roll")
-    ]
-    if not main_clips:
-        main_clips = virtual_clips[:1]
-
-    input_args: list[str] = []
-    video_filters: list[str] = []
-    audio_filters: list[str] = []
-
-    for i, clip in enumerate(main_clips):
-        local_path = asset_path_map.get(clip.asset_id)
-        if not local_path:
-            continue
-        input_args.extend(["-i", str(local_path)])
-
-        duration = clip.duration or 0.0
-
-        # 视频滤镜：scale → crop → setpts → trim → setpts（与旧引擎一致）
-        vf = (
-            f"[{i}:v]"
-            f"scale={OUTPUT_WIDTH}:{OUTPUT_HEIGHT}:force_original_aspect_ratio=increase,"
-            f"crop={OUTPUT_WIDTH}:{OUTPUT_HEIGHT},"
-            f"setpts=PTS-STARTPTS,"
-            f"trim=0:{duration:.3f},"
-            f"setpts=PTS-STARTPTS"
-            f"[v{i}]"
-        )
-        video_filters.append(vf)
-
-        # 音频滤镜：atrim → asetpts
-        af = f"[{i}:a]atrim=0:{duration:.3f},asetpts=PTS-STARTPTS[a{i}]"
-        audio_filters.append(af)
-
-    n = len(main_clips)
-
-    if n == 1:
-        video_label = "[v0]"
-        audio_label = "[a0]"
-    else:
-        # concat 视频
-        v_inputs = "".join(f"[v{i}]" for i in range(n))
-        video_filters.append(f"{v_inputs}concat=n={n}:v=1:a=0[outv]")
-        # concat 音频
-        a_inputs = "".join(f"[a{i}]" for i in range(n))
-        audio_filters.append(f"{a_inputs}concat=n={n}:v=0:a=1[outa]")
-        video_label = "[outv]"
-        audio_label = "[outa]"
-
-    # 组装 filter_complex
-    fc_parts = video_filters + audio_filters
-    filter_complex = ";".join(fc_parts)
-
-    command = [
-        FFMPEG_BIN,
-        "-y",
-        *input_args,
-        "-filter_complex",
-        filter_complex,
-        "-map",
-        video_label,
-        "-map",
-        audio_label,
-        "-c:v",
-        "libx264",
-        "-crf",
-        "23",
-        "-preset",
-        "medium",
-        "-c:a",
-        "aac",
-        "-b:a",
-        "192k",
-        "-movflags",
-        "+faststart",
-        str(output_path),
-    ]
-
-    logger.info("[task_id=%s] [渲染] legacy 引擎 FFmpeg 开始: clips=%d", task_id, n)
-    try:
-        run_ffmpeg(command)
-    except subprocess.CalledProcessError as e:
-        logger.error(
-            "[task_id=%s] [渲染] legacy 引擎 FFmpeg 失败: %s\nfilter_complex: %s",
-            task_id,
-            e,
-            filter_complex[:500],
-        )
-        raise
-
-    file_size = output_path.stat().st_size if output_path.exists() else 0
-    duration = probe_duration(output_path)
-    return duration, file_size
-
-
 # ── generate_video 阶段子函数 ─────────────────────────────────────────────────
 
 
@@ -1299,59 +1143,45 @@ def _render_video(
         total_duration,
     )
 
-    # 选择渲染引擎
-    engine = _resolve_render_engine(user_id) if user_id else ENGINE_UNIFIED
-    logger.info("[task_id=%s] [渲染] 引擎选择: %s (user_id=%s)", task_id, engine, user_id)
-
     render_start = time.monotonic()
     render_output_path = temp_path / f"rendered-{task_id}.mp4"
 
-    if engine == ENGINE_LEGACY:
-        render_duration, _ = _render_with_legacy_engine(
-            task_id=task_id,
-            virtual_clips=virtual_clips,
-            asset_path_map=asset_path_map,
-            work_dir=temp_path,
-            output_path=render_output_path,
-        )
-    else:
-        logger.info("[task_id=%s] [渲染] unified 引擎 FFmpeg 渲染开始", task_id)
+    logger.info("[task_id=%s] [渲染] unified 引擎 FFmpeg 渲染开始", task_id)
 
-        # ── 准备 BGM 音频 ──
-        bgm_path: str | None = None
-        plan_config = virtual_plan.config or {}
-        bgm_config = plan_config.get("bgm", {}) or {}
-        if bgm_config.get("enabled", False):
-            try:
-                bgm_path = _prepare_bgm_track(
-                    bgm_config=bgm_config,
-                    temp_path=temp_path,
-                    task_id=task_id,
-                )
-            except Exception as bgm_err:
-                logger.warning("[task_id=%s] [BGM] 准备失败，跳过BGM: %s", task_id, bgm_err)
-                bgm_path = None
+    # ── 准备 BGM 音频 ──
+    bgm_path: str | None = None
+    plan_config = virtual_plan.config or {}
+    bgm_config = plan_config.get("bgm", {}) or {}
+    if bgm_config.get("enabled", False):
+        try:
+            bgm_path = _prepare_bgm_track(
+                bgm_config=bgm_config,
+                temp_path=temp_path,
+                task_id=task_id,
+            )
+        except Exception as bgm_err:
+            logger.warning("[task_id=%s] [BGM] 准备失败，跳过BGM: %s", task_id, bgm_err)
+            bgm_path = None
 
-        render_service = UnifiedRenderService(
-            plan=virtual_plan,
-            clips=virtual_clips,
-            asset_path_map=asset_path_map,
-            work_dir=temp_path,
-            output_width=OUTPUT_WIDTH,
-            output_height=OUTPUT_HEIGHT,
-            output_fps=int(OUTPUT_FPS),
-            asr_service=get_asr_service(),
-            bgm_path=bgm_path,
-        )
-        render_result = render_service.render()
-        render_output_path = render_result.output_path
-        render_duration = render_result.duration
+    render_service = UnifiedRenderService(
+        plan=virtual_plan,
+        clips=virtual_clips,
+        asset_path_map=asset_path_map,
+        work_dir=temp_path,
+        output_width=OUTPUT_WIDTH,
+        output_height=OUTPUT_HEIGHT,
+        output_fps=int(OUTPUT_FPS),
+        asr_service=get_asr_service(),
+        bgm_path=bgm_path,
+    )
+    render_result = render_service.render()
+    render_output_path = render_result.output_path
+    render_duration = render_result.duration
 
     render_elapsed = time.monotonic() - render_start
     logger.info(
-        "[task_id=%s] [渲染] %s 引擎完成: 耗时=%.1fs, 时长=%.2fs",
+        "[task_id=%s] [渲染] unified 引擎完成: 耗时=%.1fs, 时长=%.2fs",
         task_id,
-        engine,
         render_elapsed,
         render_duration,
     )

scripts/ci/run_validate.sh

@@ -95,6 +95,16 @@ else
 fi
 
 if [ "$SCAN_MODE" = "incremental" ]; then
+  # 防御性过滤：磁盘上不存在的文件（已删除文件）不参与检查，
+  # 避免 black/isort/ruff 报 "Path does not exist" 错误。
+  EXISTING_PY_FILES=""
+  for f in $CHANGED_PY_FILES; do
+    if [ -f "$f" ]; then
+      EXISTING_PY_FILES="$EXISTING_PY_FILES $f"
+    fi
+  done
+  CHANGED_PY_FILES="$EXISTING_PY_FILES"
+
   python3 -m compileall -q $CHANGED_PY_FILES
   python3 -m black --check --fast $CHANGED_PY_FILES
   python3 -m isort --check-only $CHANGED_PY_FILES

tests/unit/test_feature_flag.py

@@ -1,6 +1,6 @@
 """Feature Flag 单元测试。
 
-测试 FeatureFlagConfig、InMemoryFeatureFlagStore、RenderEngineResolver 的核心逻辑。
+测试 FeatureFlagConfig、InMemoryFeatureFlagStore、RedisFeatureFlagStore 的核心逻辑。
 """
 
 from __future__ import annotations
@@ -184,148 +184,6 @@ class TestInMemoryFeatureFlagStore:
         assert store.is_active("nonexistent") is False
 
 
-# ── RenderEngineResolver 测试 ───────────────────────────────────────────────
-
-
-class TestRenderEngineResolver:
-    """渲染引擎选择器测试。"""
-
-    def test_default_legacy_when_flag_disabled(self):
-        """flag 关闭时使用默认引擎（legacy）。"""
-        store = InMemoryFeatureFlagStore()
-        resolver = self._make_resolver(store=store, default="legacy")
-        assert resolver.get_engine() == "legacy"
-        assert resolver.get_engine("user1") == "legacy"
-
-    def test_default_unified_when_flag_disabled(self):
-        """flag 关闭但默认值是 unified 时返回 unified。"""
-        store = InMemoryFeatureFlagStore()
-        resolver = self._make_resolver(store=store, default="unified")
-        assert resolver.get_engine() == "unified"
-
-    def test_whitelist_user_uses_unified(self):
-        """白名单用户走新引擎。"""
-        store = InMemoryFeatureFlagStore()
-        store.set(
-            FeatureFlagConfig(
-                name="render_engine",
-                enabled=True,
-                percentage=0,
-                whitelist={"beta_tester"},
-            )
-        )
-        resolver = self._make_resolver(store=store, default="legacy")
-        assert resolver.get_engine("beta_tester") == "unified"
-        assert resolver.get_engine("normal_user") == "legacy"
-
-    def test_100_percent_all_unified(self):
-        """100% 时所有用户走新引擎。"""
-        store = InMemoryFeatureFlagStore()
-        store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
-        resolver = self._make_resolver(store=store, default="legacy")
-        for i in range(50):
-            assert resolver.get_engine(f"user_{i}") == "unified"
-
-    def test_invalid_default_engine_fallback(self):
-        """无效默认值回退到 legacy。"""
-        store = InMemoryFeatureFlagStore()
-        resolver = self._make_resolver(store=store, default="invalid_value")
-        assert resolver.get_engine() == "legacy"
-
-    def test_should_use_unified_helper(self):
-        """should_use_unified 便捷方法。"""
-        store = InMemoryFeatureFlagStore()
-        store.set(
-            FeatureFlagConfig(
-                name="render_engine",
-                enabled=True,
-                percentage=0,
-                whitelist={"user_a"},
-            )
-        )
-        resolver = self._make_resolver(store=store)
-        assert resolver.should_use_unified("user_a") is True
-        assert resolver.should_use_unified("user_b") is False
-
-    def test_config_snapshot(self):
-        """配置快照。"""
-        store = InMemoryFeatureFlagStore()
-        store.set(
-            FeatureFlagConfig(
-                name="render_engine",
-                enabled=True,
-                percentage=30,
-                whitelist={"u1", "u2"},
-            )
-        )
-        resolver = self._make_resolver(store=store)
-        snapshot = resolver.get_config_snapshot()
-        assert snapshot["flag_name"] == "render_engine"
-        assert snapshot["enabled"] is True
-        assert snapshot["percentage"] == 30
-        assert snapshot["whitelist"] == ["u1", "u2"]
-
-    def test_set_flag_updates_config(self):
-        """通过 set_flag 修改后立即生效。"""
-        store = InMemoryFeatureFlagStore()
-        resolver = self._make_resolver(store=store, default="legacy")
-
-        # 初始：关闭
-        assert resolver.get_engine("user1") == "legacy"
-
-        # 开启 100%
-        resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
-        assert resolver.get_engine("user1") == "unified"
-
-        # 关闭
-        resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=False))
-        assert resolver.get_engine("user1") == "legacy"
-
-    def test_force_refresh(self):
-        """强制刷新不报错。"""
-        store = InMemoryFeatureFlagStore()
-        resolver = self._make_resolver(store=store)
-        resolver.force_refresh()  # 不抛异常即可
-
-    def test_does_not_affect_in_flight_tasks(self):
-        """
-        热更新不影响在途任务验证：
-        任务开始时确定引擎，中途配置变更不改变当前任务的引擎选择。
-        （这是通过"每次调用 get_engine 时读取当前配置"来保证的，
-        任务开始时调用一次拿到结果，之后不再变化）
-        """
-        store = InMemoryFeatureFlagStore()
-        store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
-        resolver = self._make_resolver(store=store, default="legacy")
-
-        # 模拟任务开始时获取引擎
-        engine_at_start = resolver.get_engine("user1")
-        assert engine_at_start == "unified"
-
-        # 任务进行中关闭 flag
-        store.set(FeatureFlagConfig(name="render_engine", enabled=False))
-        resolver.force_refresh()
-
-        # 在途任务持有的 engine_at_start 仍然是 unified（不随配置变化）
-        assert engine_at_start == "unified"
-        # 新任务会拿到 legacy
-        assert resolver.get_engine("user1") == "legacy"
-
-    # ── 辅助方法 ──
-
-    @staticmethod
-    def _make_resolver(store=None, default="legacy"):
-        from apps.worker.video_processing.render_engine_resolver import (
-            RenderEngineResolver,
-        )
-
-        return RenderEngineResolver(
-            default_engine=default,
-            store=store or InMemoryFeatureFlagStore(),
-            refresh_interval=9999,  # 测试时禁用自动刷新
-        )
-
-
 # ── RedisFeatureFlagStore 降级测试（无 Redis 环境） ───────────────────────
 
 

tests/unit/test_generation_feature_flag.py

@@ -1,332 +0,0 @@
-"""generate_video 任务 Feature Flag 灰度引擎选择单元测试.
-
-覆盖：
-- _resolve_render_engine 正常返回 unified / legacy
-- Feature Flag 不可用时 fallback 到 unified
-- 白名单 / 百分比 / 全局开关各场景
-- _render_with_legacy_engine 命令构建与输出验证
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from types import ModuleType
-from unittest.mock import MagicMock
-
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-from pathlib import Path
-
-# ── Mock worker 模块以避免数据库连接 ──────────────────────────────────────────
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-
-_mock_db_mod = ModuleType("worker_app.db")
-_mock_db_mod.SessionLocal = MagicMock()
-sys.modules.setdefault("worker_app.db", _mock_db_mod)
-
-_mock_celery_mod = ModuleType("worker_app.celery_app")
-_mock_celery_app = MagicMock()
-_mock_celery_app.task = lambda **kwargs: lambda fn: fn
-_mock_celery_mod.celery_app = _mock_celery_app
-sys.modules.setdefault("worker_app.celery_app", _mock_celery_mod)
-
-# Mock worker_app.core.config 避免 settings 加载
-_mock_config_mod = ModuleType("worker_app.core.config")
-_mock_settings = MagicMock()
-_mock_settings.redis_url = None
-_mock_settings.render_engine = "unified"
-_mock_config_mod.get_settings = lambda: _mock_settings
-sys.modules.setdefault("worker_app.core", ModuleType("worker_app.core"))
-sys.modules.setdefault("worker_app.core.config", _mock_config_mod)
-
-
-# ── 测试用数据类 ──────────────────────────────────────────────────────────────
-
-
-class _TestClip:
-    def __init__(self, asset_id, duration=30.0, clip_type="main", config=None, order=0):
-        self.id = f"clip_{asset_id}"
-        self.plan_id = "test-plan"
-        self.clip_type = clip_type
-        self.order = order
-        self.asset_id = asset_id
-        self.duration = duration
-        self.config = config or {}
-        self.start_time = 0.0
-        self.transition_effect = "cut"
-
-
-# ── RenderEngineResolver 基础行为测试 ───────────────────────────────────────
-
-
-def test_resolver_unified_when_enabled_100_percent():
-    """flag 全局开启（percentage=100）时，返回 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "unified"
-
-
-def test_resolver_legacy_when_flag_disabled():
-    """flag 全局关闭时，返回默认引擎 legacy。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=False, percentage=100))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "legacy"
-
-
-def test_resolver_whitelist_overrides_percentage_0():
-    """白名单用户即使 percentage=0 也走 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(
-        FeatureFlagConfig(
-            name="render_engine",
-            enabled=True,
-            percentage=0,
-            whitelist={"user-vip"},
-        )
-    )
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-vip") == "unified"
-    assert resolver.get_engine(user_id="user-other") == "legacy"
-
-
-def test_resolver_percentage_0_all_legacy():
-    """percentage=0 且无白名单时，全部走 legacy。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=0))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    for i in range(50):
-        assert resolver.get_engine(user_id=f"user-{i}") == "legacy"
-
-
-def test_resolver_default_unified_when_flag_off():
-    """默认引擎设为 unified 且 flag 关闭时，返回 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=False, percentage=0))
-    resolver = RenderEngineResolver(default_engine="unified", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "unified"
-
-
-# ── _render_with_legacy_engine 集成测试 ──────────────────────────────────────
-
-
-def test_legacy_engine_single_clip_keeps_original_fps():
-    """单 clip 场景：输出保持原帧率（不做 fps 归一化），分辨率缩放正确。"""
-    import subprocess
-    import tempfile
-
-    from video_processing.ffmpeg_utils import probe_video_info
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input_path = tmp_path / "input.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        # 生成 1 秒 30fps 测试视频（带音频）
-        subprocess.run(
-            [
-                "ffmpeg",
-                "-y",
-                "-f",
-                "lavfi",
-                "-i",
-                "color=c=red:s=640x360:d=1:r=30",
-                "-f",
-                "lavfi",
-                "-i",
-                "anullsrc=r=44100:cl=stereo:d=1",
-                "-c:v",
-                "libx264",
-                "-pix_fmt",
-                "yuv420p",
-                "-c:a",
-                "aac",
-                "-shortest",
-                str(input_path),
-            ],
-            check=True,
-            capture_output=True,
-        )
-
-        clip = _TestClip(asset_id="asset-1", duration=1.0)
-        asset_path_map = {"asset-1": input_path}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert duration > 0
-
-        # 旧引擎保持原帧率（30fps），不做 fps 归一化
-        info = probe_video_info(str(output_path))
-        assert abs(info.get("fps", 0) - 30.0) < 0.5
-        assert info.get("width") == 1280
-        assert info.get("height") == 720
-
-
-def test_legacy_engine_two_clips_concat_duration():
-    """多 clip 场景：concat 后时长为两片段之和。"""
-    import subprocess
-    import tempfile
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input1 = tmp_path / "input1.mp4"
-        input2 = tmp_path / "input2.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        for idx, inp in enumerate([input1, input2]):
-            color = "red" if idx == 0 else "blue"
-            subprocess.run(
-                [
-                    "ffmpeg",
-                    "-y",
-                    "-f",
-                    "lavfi",
-                    "-i",
-                    f"color=c={color}:s=640x360:d=1:r=30",
-                    "-f",
-                    "lavfi",
-                    "-i",
-                    "anullsrc=r=44100:cl=stereo:d=1",
-                    "-c:v",
-                    "libx264",
-                    "-pix_fmt",
-                    "yuv420p",
-                    "-c:a",
-                    "aac",
-                    "-shortest",
-                    str(inp),
-                ],
-                check=True,
-                capture_output=True,
-            )
-
-        clip1 = _TestClip(asset_id="asset-1", duration=1.0, clip_type="main", order=0)
-        clip2 = _TestClip(asset_id="asset-2", duration=1.0, clip_type="main", order=1)
-        asset_path_map = {"asset-1": input1, "asset-2": input2}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip1, clip2],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert abs(duration - 2.0) < 0.2
-
-
-def test_legacy_engine_broll_mode_supported():
-    """b_roll 类型的 clip 也被正确识别为主图层并渲染。"""
-    import subprocess
-    import tempfile
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input_path = tmp_path / "input.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        subprocess.run(
-            [
-                "ffmpeg",
-                "-y",
-                "-f",
-                "lavfi",
-                "-i",
-                "color=c=green:s=640x360:d=1:r=30",
-                "-f",
-                "lavfi",
-                "-i",
-                "anullsrc=r=44100:cl=stereo:d=1",
-                "-c:v",
-                "libx264",
-                "-pix_fmt",
-                "yuv420p",
-                "-c:a",
-                "aac",
-                "-shortest",
-                str(input_path),
-            ],
-            check=True,
-            capture_output=True,
-        )
-
-        clip = _TestClip(
-            asset_id="asset-1",
-            duration=1.0,
-            clip_type="main",
-            config={"role": "b_roll"},
-        )
-        asset_path_map = {"asset-1": input_path}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert duration > 0

tests/unit/test_generation_logging.py

@@ -17,11 +17,18 @@ from __future__ import annotations
 import json
 import sys
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
 
+# ── 预注入 mock 模块，防止 worker_app.db 触发真实数据库连接 ──
+_mock_db_module = MagicMock()
+_mock_db_module.SessionLocal = MagicMock()
+sys.modules.setdefault("worker_app.db", _mock_db_module)
+if "worker_app" in sys.modules:
+    sys.modules["worker_app"].db = _mock_db_module
+
 from app.schemas.generation_task import GenerationTaskResponse
 
 from packages.domain.generation_task import GenerationTask, GenerationTaskStatus

tests/unit/test_generation_p3_optimizations.py

@@ -17,6 +17,15 @@ import pytest
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
 
+# ── 预注入 mock 模块，防止 worker_app.db 触发真实数据库连接 ──
+# worker_app.db 在模块级别调用 ensure_database_exists() 尝试连接 PostgreSQL，
+# 增量测试单独跑这些文件时会失败。与 test_voice_clone_task.py 同理。
+_mock_db_module = MagicMock()
+_mock_db_module.SessionLocal = MagicMock()
+sys.modules.setdefault("worker_app.db", _mock_db_module)
+if "worker_app" in sys.modules:
+    sys.modules["worker_app"].db = _mock_db_module
+
 
 # ── P3-3: _verify_url_accessible 重试 ───────────────────────────────────────