diff --git a/apps/api/app/api/routes/feature_flags.py b/apps/api/app/api/routes/feature_flags.py index c5834a5a6..f8b7ff378 100755 --- a/apps/api/app/api/routes/feature_flags.py +++ b/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: diff --git a/apps/worker/video_processing/render_engine_resolver.py b/apps/worker/video_processing/render_engine_resolver.py deleted file mode 100755 index b39b63049..000000000 --- a/apps/worker/video_processing/render_engine_resolver.py +++ /dev/null @@ -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 diff --git a/apps/worker/worker_app/core/config.py b/apps/worker/worker_app/core/config.py index a47fcd2bc..67ffdcfe7 100755 --- a/apps/worker/worker_app/core/config.py +++ b/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", diff --git a/apps/worker/worker_app/tasks/compose_video.py b/apps/worker/worker_app/tasks/compose_video.py index c9307d2da..f4b706395 100755 --- a/apps/worker/worker_app/tasks/compose_video.py +++ b/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 diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py index 80b26b9c2..7ecbe3e8c 100755 --- a/apps/worker/worker_app/tasks/edit_plan_generation.py +++ b/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: diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index e638bf1ef..144ce3af2 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/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, ) diff --git a/scripts/ci/run_validate.sh b/scripts/ci/run_validate.sh index ade0dd049..eba19661d 100755 --- a/scripts/ci/run_validate.sh +++ b/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 diff --git a/tests/unit/test_feature_flag.py b/tests/unit/test_feature_flag.py index 8bab8c04e..99cd76850 100644 --- a/tests/unit/test_feature_flag.py +++ b/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 环境） ─────────────────────── diff --git a/tests/unit/test_generation_feature_flag.py b/tests/unit/test_generation_feature_flag.py deleted file mode 100644 index 5c743a598..000000000 --- a/tests/unit/test_generation_feature_flag.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_generation_logging.py b/tests/unit/test_generation_logging.py index 7d4d1b373..6b10b0688 100644 --- a/tests/unit/test_generation_logging.py +++ b/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 diff --git a/tests/unit/test_generation_p3_optimizations.py b/tests/unit/test_generation_p3_optimizations.py index ff4229083..f8dceb68c 100644 --- a/tests/unit/test_generation_p3_optimizations.py +++ b/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 重试 ───────────────────────────────────────