From b37e812b1b79c6a1dea91936bf1bc2a5d40b2217 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 24 Aug 2026 13:42:38 +0800 Subject: [PATCH 1/4] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E6=B8=B2?= =?UTF-8?q?=E6=9F=93=E9=93=BE=E8=B7=AF=EF=BC=8C=E9=A2=84=E8=A7=88=E5=8D=B3?= =?UTF-8?q?=E6=89=80=E5=BE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动: - 删除旧内存渲染路径(render_from_memory)及所有关联函数 - 预览和正式生成都走 _render_from_edit_plan(基于 EditPlan 数据库) - source_edit_plan_id 为空时直接报错,不再兜底旧路径 - 删除 custom_title 字段全链路(domain/application/schema/route/worker) - 统一使用 title_config 结构化字段传递标题配置 删除的函数(generation.py -1031行): - _build_plan_and_clips_from_task:内存构建虚拟 plan/clips - _load_template_plan_config:模板配置加载(旧路径辅助) - _load_template_clip_configs / _load_template_segment_durations - _download_library_assets / _download_all_assets:旧路径素材下载 - _render_video:旧路径渲染入口(854x480 低质量预览) - generate_video 中 DEPRECATED 旧路径代码块 删除的测试文件(测试已删除函数的旧测试): - test_full_pipeline.py / test_four_mode_rendering.py(集成测试) - test_preview_voice_ids_fallback.py / test_preview_title_render.py - test_generation_p3_optimizations.py / test_generation_worker_fixes.py - test_bool_config_defense.py / test_preview_title_config.py 清理的测试(删除引用已删函数的测试类): - test_oneclick_gen_p0_fixes.py:移除 TestDownloadLibraryAssets / TestTemplatePlanConfigLoading - test_1294_preview_voice_injection.py:移除 TestRenderVideoVoiceInjection / TestGenerateVideoPassesVoiceIds - test_1280_preview_speedup.py:移除 test_parallel_download_still_works - test_worker_cover_meta_and_status.py:更新 _meta_session 断言 - test_confirm_generation.py / test_three_bugs_fix.py:移除 custom_title 断言 单测验证:13647 passed, 0 failed --- apps/api/app/api/routes/generation_preview.py | 14 +- apps/api/app/api/routes/generation_tasks.py | 5 - apps/api/app/schemas/generation_task.py | 5 +- apps/worker/worker_app/tasks/generation.py | 1047 +---------------- .../generation_task_repository.py | 3 - packages/application/generation_tasks.py | 2 - packages/domain/generation_task.py | 6 - tests/integration/test_four_mode_rendering.py | 234 ---- tests/integration/test_full_pipeline.py | 189 --- tests/unit/test_1280_preview_speedup.py | 8 - .../unit/test_1294_preview_voice_injection.py | 181 --- tests/unit/test_bool_config_defense.py | 130 -- tests/unit/test_confirm_generation.py | 11 +- .../unit/test_generation_p3_optimizations.py | 389 ------ tests/unit/test_generation_worker_fixes.py | 228 ---- tests/unit/test_oneclick_gen_p0_fixes.py | 247 ---- tests/unit/test_preview_title_config.py | 190 --- tests/unit/test_preview_title_render.py | 256 ---- tests/unit/test_preview_voice_ids_fallback.py | 151 --- tests/unit/test_render_subtitles_pure.py | 2 +- tests/unit/test_three_bugs_fix.py | 3 - .../unit/test_worker_cover_meta_and_status.py | 9 +- 22 files changed, 16 insertions(+), 3294 deletions(-) delete mode 100644 tests/integration/test_four_mode_rendering.py delete mode 100644 tests/integration/test_full_pipeline.py delete mode 100644 tests/unit/test_bool_config_defense.py delete mode 100644 tests/unit/test_generation_p3_optimizations.py delete mode 100644 tests/unit/test_generation_worker_fixes.py delete mode 100644 tests/unit/test_preview_title_config.py delete mode 100644 tests/unit/test_preview_title_render.py delete mode 100644 tests/unit/test_preview_voice_ids_fallback.py diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index a7eda77a0..6ae78cb6d 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -268,19 +268,7 @@ def create_preview_generation_task( # 从模板读取 editing_mode / mode 作为 strategy_id(渲染 pipeline 的 mode 参数) strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id) - # 处理标题配置:如果有标题文本,序列化到 custom_title 字段传递给 worker title_config = request.title_config or {} - title_text = (title_config.get("text") or "").strip() - custom_title_value = "" - if title_text: - # 将标题文本和样式配置序列化为 JSON 存入 custom_title - # Worker 端会解析 JSON 获取完整标题配置 - custom_title_value = json.dumps(title_config, ensure_ascii=False) - logger.info( - "[预览生成] 标题配置: text=%s, config_keys=%s", - title_text[:30], - list(title_config.keys()), - ) use_case = CreateGenerationTaskUseCase(generation_task_repository) @@ -305,7 +293,7 @@ def create_preview_generation_task( auto_retry_enabled=False, auto_retry_max=0, is_preview=True, - custom_title=custom_title_value, + title_config=title_config, ) ) except ValueError as e: diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 84019f2e3..9af9d0ecb 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -70,7 +70,6 @@ def _to_generation_task_response(task) -> GenerationTaskResponse: output_width=getattr(task, "output_width", 1280), output_height=getattr(task, "output_height", 720), cover_url=getattr(task, "cover_url", ""), - custom_title=getattr(task, "custom_title", ""), title_config=getattr(task, "title_config", {}) or {}, logs=getattr(task, "logs", "[]"), status=task.status, @@ -355,7 +354,6 @@ def create_generation_task( output_width=request.output_width, output_height=request.output_height, cover_url=request.cover_url, - custom_title=request.custom_title, title_config=request.title_config or {}, ) ) @@ -473,7 +471,6 @@ def confirm_generation( if resolution_match: source_task.mark_confirmed( cover_url=request.cover_url, - custom_title=request.custom_title, output_width=request.output_width, output_height=request.output_height, ) @@ -519,7 +516,6 @@ def confirm_generation( output_width=request.output_width, output_height=request.output_height, cover_url=request.cover_url, - custom_title=request.custom_title, ) ) @@ -652,7 +648,6 @@ def retry_generation_task( output_width=getattr(task, "output_width", 1280), output_height=getattr(task, "output_height", 720), cover_url=getattr(task, "cover_url", ""), - custom_title=getattr(task, "custom_title", ""), ) ) try: diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py index 8d27abd86..9c61684d2 100755 --- a/apps/api/app/schemas/generation_task.py +++ b/apps/api/app/schemas/generation_task.py @@ -10,7 +10,6 @@ class ConfirmGenerationRequest(BaseModel): output_width: int = Field(default=1080, ge=100, description="输出视频宽度") output_height: int = Field(default=1920, ge=100, description="输出视频高度") cover_url: str = Field(default="", description="自定义封面图片 URL") - custom_title: str = Field(default="", description="自定义视频标题") class CreateGenerationTaskRequest(BaseModel): @@ -33,7 +32,7 @@ class CreateGenerationTaskRequest(BaseModel): voice_ids: list[str] = Field(default_factory=list) # ── 来源剪辑计划 ── source_edit_plan_id: str = "" - # ── 标题配置(结构化,优先于 custom_title 纯文本)── + # ── 标题配置(结构化)── title_config: dict | None = Field( default=None, description="标题样式对象,包含 text/font/font_size/font_color/position/bold/stroke/shadow 等。为空时不影响现有行为。", @@ -77,7 +76,6 @@ class CreateGenerationTaskRequest(BaseModel): output_width: int = Field(default=1280, description="输出视频宽度") output_height: int = Field(default=720, description="输出视频高度") cover_url: str = Field(default="", description="封面图片 URL") - custom_title: str = Field(default="", description="自定义视频标题") @model_validator(mode="after") def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest": @@ -113,7 +111,6 @@ class GenerationTaskResponse(BaseModel): output_width: int = 1280 output_height: int = 720 cover_url: str = "" - custom_title: str = "" title_config: dict = Field(default_factory=dict) status: str progress: float diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 1ca355a1f..1ba1bf399 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -158,175 +158,6 @@ from video_processing.oss_helpers import ( upload_to_oss, ) - -def _load_template_clip_configs(template_id: str) -> list: - """从数据库读取模板的片段配置列表。 - - 失败返回空列表,不阻断主流程。 - """ - if not template_id: - return [] - try: - from worker_app.db import SessionLocal - - from packages.adapters.sqlalchemy_impl.template_clip_config_repository import ( - SQLAlchemyTemplateClipConfigRepository, - ) - - session = SessionLocal() - try: - repo = SQLAlchemyTemplateClipConfigRepository(session) - configs = repo.list_by_template(template_id, limit=200) - logger.info("读取模板片段配置: template_id=%s count=%d", template_id, len(configs)) - return configs - finally: - session.close() - except Exception as e: - logger.warning("读取模板片段配置失败,跳过效果层映射: template_id=%s error=%s", template_id, e) - return [] - - -def _load_template_segment_durations(template_id: str) -> list[float]: - """从数据库读取模板各 segment 的 duration_max 列表(按 segment_order 排序)。 - - 用于限制每个 clip 的最大时长,防止素材完整时长超过模板约束。 - 失败返回空列表,不阻断主流程。 - """ - if not template_id: - return [] - try: - from worker_app.db import SessionLocal - - from packages.adapters.sqlalchemy_impl.models import TemplateSegmentModel - - session = SessionLocal() - try: - segments = ( - session.query(TemplateSegmentModel) - .filter(TemplateSegmentModel.template_id == template_id) - .order_by(TemplateSegmentModel.segment_order) - .all() - ) - durations = [s.duration_max for s in segments if s.duration_max and s.duration_max > 0] - if durations: - logger.info( - "读取模板segment时长约束: template_id=%s segments=%d durations=%s", - template_id, - len(durations), - durations, - ) - return durations - finally: - session.close() - except Exception as e: - logger.warning("读取模板segment时长约束失败: template_id=%s error=%s", template_id, e) - return [] - - -# DEPRECATED: 仅兼容无 source_edit_plan_id 的旧调用,后续移除 -def _build_plan_and_clips_from_task( - task_id: str, - downloaded_paths: list[Path], - mode: str, - template_id: str = "", -) -> tuple[_VirtualPlan, list[_VirtualClip], dict[str, Path]]: - """根据模式和下载的素材路径,构建虚拟 plan + clips + asset_path_map。 - - 模式 → clip_type 映射: - ONE_TAKE: N 个 main clips(默认,pip/voice_pip 已统一映射为此模式) - VOICE_OVER: N 个 main(config.role=b_roll) - - Returns: - (virtual_plan, virtual_clips, asset_path_map) - """ - plan = _VirtualPlan(id=task_id, name=f"Generated-{task_id[:8]}") - - # 为每个下载路径生成合成 asset_id,并预探测素材时长 - asset_path_map: dict[str, Path] = {} - path_to_asset_id: dict[Path, str] = {} - path_duration: dict[Path, float] = {} - for i, p in enumerate(downloaded_paths): - asset_id = f"gen_{task_id[:8]}_{i:03d}{p.suffix or '.mp4'}" - asset_path_map[asset_id] = p - path_to_asset_id[p] = asset_id - path_duration[p] = probe_duration(p) - - # 产品已确认全面下线画中画,pip/voice_pip统一走one_take(顺序拼接) - if mode in ("pip", "voice_pip"): - logger.info("画中画模式已下线,%s 强制映射为 one_take", mode) - mode = "one_take" - - clips: list[_VirtualClip] = [] - - if mode == "voice_over": - # N 个 main(config.role=b_roll) - for i, p in enumerate(downloaded_paths): - clips.append( - _VirtualClip( - id=f"vc_{i:03d}", - plan_id=task_id, - clip_type="main", - order=i, - asset_id=path_to_asset_id[p], - duration=path_duration[p], - config={"role": "b_roll"}, - ) - ) - else: - # ONE_TAKE (default): N 个 main clips - for i, p in enumerate(downloaded_paths): - clips.append( - _VirtualClip( - id=f"vc_{i:03d}", - plan_id=task_id, - clip_type="main", - order=i, - asset_id=path_to_asset_id[p], - duration=path_duration[p], - ) - ) - - # ── P1-2: 模板 segment 时长约束 ── - if template_id and clips: - seg_durations = _load_template_segment_durations(template_id) - if seg_durations: - capped_count = 0 - for idx, clip in enumerate(clips): - if idx < len(seg_durations): - max_dur = seg_durations[idx] - if clip.duration > max_dur: - clip.duration = max_dur - capped_count += 1 - if capped_count > 0: - logger.info( - "模板时长约束已应用: template_id=%s capped_clips=%d/%d", - template_id, - capped_count, - len(clips), - ) - - # ── P1: 模板效果层映射 ── - if template_id: - clip_configs = _load_template_clip_configs(template_id) - if clip_configs: - # 1. clip级效果层(转场、滤镜、调速等) - _apply_template_clip_effects(clips, clip_configs, mode) - - # 2. 片头片尾(从 intro/outro 类型 clip 提取 plan 级配置) - intro_outro_config = _extract_intro_outro_from_clip_configs(clip_configs) - if intro_outro_config: - plan_config = plan.config or {} - plan_config["intro_outro"] = intro_outro_config - plan.config = plan_config - logger.info( - "模板片头片尾配置已注入: has_intro=%s has_outro=%s", - intro_outro_config.get("has_intro", False), - intro_outro_config.get("has_outro", False), - ) - - return plan, clips, asset_path_map - - def _download_voice_asset(voice_library_id: str, local_path: Path) -> bool: """下载配音文件。 @@ -473,253 +304,6 @@ def _verify_url_accessible( logger.warning("URL 可访问性校验最终失败: url=%s error=%s", url, last_error) return False - -def _download_library_assets( - temp_path: Path, - *, - asset_library_id: str = "", - project_id: str = "", - asset_ids: list[str] | None = None, - strict: bool = True, - task_id: str = "", - gen_task=None, -) -> list[Path]: - """下载视频素材 — 同时支持素材库模式和项目级模式。 - - 两种查询路径: - - 素材库模式:asset_library_id 非空时,按 asset_library_id + asset_ids 查 - - 项目级模式:project_id 非空时,按 project_id + asset_ids 查 - - 两者都提供时优先素材库模式;两者都为空时抛异常 - - 归属校验与下载在同一 DB session 中完成,避免多次连接开销(P3-2)。 - - Args: - temp_path: 临时目录路径 - asset_library_id: 素材库 ID(可选,与 project_id 二选一) - project_id: 项目 ID(可选,与 asset_library_id 二选一) - asset_ids: 指定素材 ID 列表,为空则下载全部 ready 视频素材 - strict: 严格模式(默认 True)。 - True — 任何素材下载失败立即抛 RuntimeError; - False — 跳过失败素材,返回成功列表(调用方可通过日志感知失败)。 - - Returns: - 下载成功的视频文件 Path 列表 - - Raises: - ValueError: 当 asset_library_id、project_id 和 asset_ids 都为空时 - RuntimeError: strict=True 时任何下载失败;或指定了 asset_ids 但全部下载失败 - """ - if not asset_library_id and not project_id and not asset_ids: - raise ValueError("asset_library_id、project_id 或 asset_ids 至少需要提供一个") - - try: - from packages.adapters.sqlalchemy_impl.models import AssetModel - - session = SessionLocal() - try: - # 构建查询 - query = session.query(AssetModel).filter( - AssetModel.status == "ready", - AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]), - ) - - if asset_ids: - # 明确指定了 asset_ids:直接按 ID 查,不预先按 library/project 过滤 - # 避免项目级素材或跨库素材因为 library_id 不匹配而查不到 - # 归属安全由后面的归属校验保证 - query = query.filter(AssetModel.id.in_(asset_ids)) - logger.info( - "下载指定素材: asset_ids=%d 个, asset_library_id=%s, project_id=%s", - len(asset_ids), - asset_library_id or "none", - project_id or "none", - ) - else: - # 未指定 asset_ids:按 library 或 project 下载全部 ready 视频 - if asset_library_id: - query = query.filter(AssetModel.asset_library_id == asset_library_id) - logger.info( - "下载素材库全部视频: asset_library_id=%s", - asset_library_id, - ) - else: - query = query.filter(AssetModel.project_id == project_id) - logger.info( - "下载项目全部视频: project_id=%s", - project_id, - ) - - assets = query.order_by(AssetModel.created_at).all() - - if not assets: - mode_desc = ( - f"素材库 {asset_library_id}" - if asset_library_id - else (f"项目 {project_id}" if project_id else "指定素材") - ) - msg = f"未找到视频素材: {mode_desc}, asset_ids={asset_ids or 'all'}" - logger.error(msg) - raise RuntimeError(msg) - - # P3-2: 归属校验合并到同一 session - if asset_ids: - found_ids = {a.id for a in assets} - missing_ids = set(asset_ids) - found_ids - if missing_ids: - raise ValueError(f"素材不存在: asset_ids={sorted(missing_ids)}") - for asset in assets: - # 校验素材库归属(只要传了 asset_library_id 就校验) - if asset_library_id and asset.asset_library_id != asset_library_id: - raise ValueError( - f"素材不属于指定素材库: asset_id={asset.id}, " - f"expected_asset_library_id={asset_library_id}, " - f"actual_asset_library_id={asset.asset_library_id}" - ) - # 校验项目归属(只要传了 project_id 就校验) - if project_id and asset.project_id != project_id: - raise ValueError( - f"素材不属于指定项目: asset_id={asset.id}, " - f"expected_project_id={project_id}, " - f"actual_project_id={asset.project_id}" - ) - logger.info( - "素材归属校验通过 (同 session): %d 个 asset_ids", - len(asset_ids), - ) - - # 构建待下载列表 (index, asset, storage_key, local_file) - download_jobs: list[tuple[int, Any, str, Path]] = [] - failed_assets: list[str] = [] - for i, asset in enumerate(assets): - storage_key = asset.file_url if asset.file_url else None - if not storage_key: - failed_assets.append(f"{asset.name}({asset.id})") - logger.warning( - "[task_id=%s] 素材缺少 file_url, 跳过: asset_id=%s name=%s", - task_id, - asset.id, - asset.name, - ) - if gen_task: - gen_task.append_log( - "下载素材", - "素材缺少file_url, 跳过", - level="WARN", - asset_id=asset.id, - asset_name=asset.name, - success=False, - file_size=0, - duration=0.0, - ) - if strict: - raise RuntimeError(f"素材缺少 file_url: asset_id={asset.id}, name={asset.name}") - continue - - ext = Path(storage_key).suffix or ".mp4" - local_file = temp_path / f"asset_{i:03d}_{asset.id}{ext}" - download_jobs.append((i, asset, storage_key, local_file)) - - # 并行下载素材(线程池,IO 密集型) - downloaded: list[Path] = [] - if download_jobs: - from concurrent.futures import ThreadPoolExecutor, as_completed - - max_workers = min(len(download_jobs), 6) - logger.info( - "[task_id=%s] 并行下载素材: count=%d, workers=%d", - task_id, - len(download_jobs), - max_workers, - ) - - def _download_one(item: tuple) -> tuple[int, Any, Path, bool, float]: - idx, asset, skey, lfile = item - t0 = time.monotonic() - ok = download_asset(skey, lfile) - elapsed = time.monotonic() - t0 - return idx, asset, lfile, ok, elapsed - - with ThreadPoolExecutor(max_workers=max_workers) as executor: - futures = {executor.submit(_download_one, job): job for job in download_jobs} - # 按原始顺序收集结果,保证 downloaded 列表顺序稳定 - results_map: dict[int, tuple[Path, bool, float, Any]] = {} - for future in as_completed(futures): - idx, asset, lfile, ok, elapsed = future.result() - results_map[idx] = (lfile, ok, elapsed, asset) - - # 按原始顺序处理结果 - for idx in sorted(results_map.keys()): - lfile, ok, elapsed, asset = results_map[idx] - if ok: - file_size = lfile.stat().st_size if lfile.exists() else 0 - downloaded.append(lfile) - logger.info( - "[task_id=%s] Downloaded asset: %s -> %s (size=%d, time=%.1fs)", - task_id, - asset.name, - lfile, - file_size, - elapsed, - ) - if gen_task: - gen_task.append_log( - "下载素材", - f"下载成功: {asset.name}", - asset_id=asset.id, - asset_name=asset.name, - success=True, - file_size=file_size, - duration=round(elapsed, 2), - ) - else: - failed_assets.append(f"{asset.name}({asset.id})") - logger.warning( - "[task_id=%s] Failed to download asset: %s (id=%s)", - task_id, - asset.name, - asset.id, - ) - if gen_task: - gen_task.append_log( - "下载素材", - f"下载失败: {asset.name}", - level="WARN", - asset_id=asset.id, - asset_name=asset.name, - success=False, - file_size=0, - duration=round(elapsed, 2), - ) - if strict: - raise RuntimeError(f"素材下载失败: asset_id={asset.id}, name={asset.name}") - - # 指定了 asset_ids 但全部下载失败 → 无论 strict 与否都报错 - if asset_ids and not downloaded: - msg = f"指定的 {len(asset_ids)} 个素材全部下载失败, failed={failed_assets}" - logger.error(msg) - raise RuntimeError(msg) - - # 非严格模式有部分失败,记录警告 - if failed_assets and not strict: - logger.warning( - "素材下载部分失败 (非严格模式): failed=%s, succeeded=%d", - failed_assets, - len(downloaded), - ) - - return downloaded - finally: - session.close() - except (ValueError, RuntimeError): - raise - except Exception as e: - logger.error("Error downloading library assets: %s", e, exc_info=True) - raise RuntimeError(f"素材下载异常: {e}") from e - - -# ── P1 校验函数 ────────────────────────────────────────────────────────────── - - def _validate_template_exists(template_id: str) -> None: """校验 template_id 是否存在且可用。 @@ -763,97 +347,6 @@ def _validate_template_exists(template_id: str) -> None: raise ValueError(f"模板不存在或已禁用: template_id={template_id}") finally: session.close() - - -def _load_template_plan_config(template_id: str) -> dict: - """从模板加载 plan 级配置(BGM、字幕、标题等效果层)。 - - 优先读新模板系统(EditTemplate.config + TemplateClipConfig), - 找不到 fallback 到旧模板系统(TemplateModel 独立字段)。 - - 模板不存在时返回空 dict,不阻塞主流程。 - """ - if not template_id: - return {} - try: - from packages.adapters.sqlalchemy_impl import ( - SQLAlchemyEditTemplateRepository, - SQLAlchemyTemplateClipConfigRepository, - ) - - session = SessionLocal() - try: - # 优先读新模板系统 - tpl_repo = SQLAlchemyEditTemplateRepository(session) - clip_repo = SQLAlchemyTemplateClipConfigRepository(session) - template = tpl_repo.get(template_id) - - if template is not None: - # 新系统:config 直接就是 plan.config 格式 - plan_config = dict(template.config or {}) - - # 从片段配置中提取 intro/outro 配置 - clip_configs = clip_repo.list_by_template(template_id, limit=200) - if clip_configs: - intro_outro = _extract_intro_outro_from_clip_configs(clip_configs) - if intro_outro: - plan_config["intro_outro"] = intro_outro - - # 把 editing_mode 也带过去 - if template.editing_mode: - plan_config["editing_mode"] = template.editing_mode - - logger.info( - "模板配置加载成功(新系统): template_id=%s keys=%s", - template_id, - list(plan_config.keys()), - ) - return plan_config - - # fallback: 旧模板系统 - from packages.adapters.sqlalchemy_impl.models import TemplateModel - - template = ( - session.query(TemplateModel) - .filter( - TemplateModel.id == template_id, - TemplateModel.is_active.is_(True), - ) - .first() - ) - if template is None: - logger.warning("模板不存在,跳过配置加载: template_id=%s", template_id) - return {} - - # 从独立字段组装成 plan.config 格式 - plan_config: dict[str, Any] = {} - title_cfg = template.title_config if isinstance(template.title_config, dict) else {} - subtitle_cfg = template.subtitle_config if isinstance(template.subtitle_config, dict) else {} - bgm_cfg = template.bgm_config if isinstance(template.bgm_config, dict) else {} - - if title_cfg: - plan_config["title"] = title_cfg - if subtitle_cfg: - plan_config["subtitle"] = subtitle_cfg - if bgm_cfg: - plan_config["bgm"] = bgm_cfg - - logger.info( - "模板配置加载成功(旧系统): template_id=%s keys=%s", - template_id, - list(plan_config.keys()), - ) - return plan_config - finally: - session.close() - except Exception as e: - logger.warning("加载模板配置失败,跳过: template_id=%s err=%s", template_id, e) - return {} - - -# ── generate_video 阶段子函数 ───────────────────────────────────────────────── - - def _load_task_info(task_id: str) -> dict | None: """从数据库加载 GenerationTask 元数据。 @@ -887,7 +380,6 @@ def _load_task_info(task_id: str) -> dict | None: "output_width": getattr(gen_task, "output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH, "output_height": getattr(gen_task, "output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT, "cover_url": getattr(gen_task, "cover_url", "") or "", - "custom_title": getattr(gen_task, "custom_title", "") or "", "title_config": dict(getattr(gen_task, "title_config", {}) or {}), "voice_ids": list(getattr(gen_task, "voice_ids", []) or []), "source_edit_plan_id": getattr(gen_task, "source_edit_plan_id", "") or "", @@ -895,241 +387,6 @@ def _load_task_info(task_id: str) -> dict | None: finally: session.close() - -def _download_all_assets( - temp_path: Path, - asset_library_id: str, - project_id: str, - task_asset_ids: list[str], - voice_library_id: str, - task_id: str, - voice_ids: list[str] | None = None, -) -> tuple[list[Path], str | None]: - """下载视频素材和配音素材。 - - Returns: - (downloaded_videos, audio_path) - - Note: gen_task 不传入下载函数(session 已关闭), - 主函数在下载前后已有汇总日志。 - - 配音下载逻辑:优先使用 voice_library_id(配音素材库资产); - 若为空则 fallback 到 voice_ids[0](前端选择的音频 asset_id)。 - """ - logger.info("[task_id=%s] [下载素材] 开始下载视频素材", task_id) - download_start = time.monotonic() - downloaded_videos = _download_library_assets( - temp_path, - asset_library_id=asset_library_id, - project_id=project_id, - asset_ids=task_asset_ids or None, - task_id=task_id, - ) - download_elapsed = time.monotonic() - download_start - logger.info( - "[task_id=%s] [下载素材] 完成: 成功=%d个, 耗时=%.1fs", - task_id, - len(downloaded_videos), - download_elapsed, - ) - - audio_path: str | None = None - # 配音下载:优先 voice_library_id,fallback 到 voice_ids[0] - effective_voice_id = voice_library_id - if not effective_voice_id and voice_ids: - effective_voice_id = voice_ids[0] - logger.info( - "[task_id=%s] [下载配音] voice_library_id 为空,fallback 到 voice_ids[0]=%s", - task_id, - effective_voice_id, - ) - if effective_voice_id: - local_audio = temp_path / "voice.mp3" - if _download_voice_asset(effective_voice_id, local_audio): - audio_path = str(local_audio) - logger.info( - "[task_id=%s] [下载配音] 配音下载成功 (source=%s)", - task_id, - "voice_library_id" if voice_library_id else "voice_ids", - ) - - return downloaded_videos, audio_path - - -def _render_video( - task_id: str, - downloaded_videos: list[Path], - voice_path: str | None, - editing_mode, - project_id: str, - template_id: str, - user_id: str, - temp_path: Path, - output_name: str, - resolution: str = "", - bgm_config: dict | None = None, - voice_ids: list[str] | None = None, - custom_title: str = "", - title_config: dict | None = None, -) -> tuple[Path, float, list[dict] | None]: - """渲染视频(含配音混音)。 - - 使用 RenderAdapter 统一渲染入口,复用 BGM/ASR/分辨率/缩略图逻辑。 - - Args: - Returns: - (output_path, render_duration, cover_candidates, voiceover_path) - """ - if not downloaded_videos: - raise RuntimeError(f"素材下载结果为空: task_id={task_id}") - - # 构建虚拟 plan + clips + asset_path_map - virtual_plan, virtual_clips, asset_path_map = _build_plan_and_clips_from_task( - task_id=task_id, - downloaded_paths=downloaded_videos, - mode=editing_mode.value, - template_id=template_id, - ) - - # 注入模板配置(BGM、字幕等效果层) - if template_id: - template_config = _load_template_plan_config(template_id) - if template_config: - base_config = virtual_plan.config or {} - virtual_plan.config = {**template_config, **base_config} - logger.info( - "[task_id=%s] [渲染] 模板配置已注入: keys=%s", - task_id, - list(template_config.keys()), - ) - - # ── 用户自定义标题:title_config 优先,custom_title 兜底 ───────────── - effective_title_cfg: dict | None = None - if title_config and isinstance(title_config, dict) and title_config.get("text", "").strip(): - effective_title_cfg = dict(title_config) - elif custom_title: - try: - parsed = json.loads(custom_title) if isinstance(custom_title, str) else custom_title - if isinstance(parsed, dict) and parsed.get("text", "").strip(): - effective_title_cfg = parsed - except (json.JSONDecodeError, TypeError): - logger.warning("[task_id=%s] custom_title JSON解析失败: %s", task_id, custom_title[:100]) - - if effective_title_cfg: - # 字段名归一化: 前端 font_size/font_color → 后端 size/color - if "font_size" in effective_title_cfg and "size" not in effective_title_cfg: - effective_title_cfg["size"] = effective_title_cfg["font_size"] - if "font_color" in effective_title_cfg and "color" not in effective_title_cfg: - effective_title_cfg["color"] = effective_title_cfg["font_color"] - plan_cfg = dict(virtual_plan.config or {}) - plan_cfg["title"] = effective_title_cfg - virtual_plan.config = plan_cfg - logger.info( - "[task_id=%s] [渲染] 标题配置已注入(source=%s): text=%s", - task_id, - "title_config" if title_config else "custom_title", - effective_title_cfg.get("text", "")[:30], - ) - - # 用户自定义 BGM 覆盖模板 BGM(用户指定优先级最高) - if bgm_config: - plan_cfg = virtual_plan.config or {} - template_bgm = plan_cfg.get("bgm", {}) or {} - if not isinstance(template_bgm, dict): - template_bgm = {} - merged_bgm = merge_bgm_config(template_bgm, bgm_config) - plan_cfg["bgm"] = merged_bgm - virtual_plan.config = plan_cfg - logger.info( - "[task_id=%s] [渲染] 用户自定义BGM已合并: enabled=%s source=%s", - task_id, - merged_bgm.get("enabled", False), - merged_bgm.get("source", ""), - ) - - # 确保输出分辨率配置存在 - # 优先级:用户指定 > 模板配置 > 默认 1280x720 - # 预览模式:强制 854x480 + 低码率 - # 注意:必须拷贝字典,避免预览模式修改污染源对象(模板配置) - plan_cfg = dict(virtual_plan.config or {}) - export_cfg = dict(plan_cfg.get("export", {}) or {}) - if resolution: - # 用户在 API 调用时指定的分辨率优先级最高 - export_cfg["resolution"] = resolution - elif not export_cfg.get("resolution"): - export_cfg["resolution"] = f"{OUTPUT_WIDTH}x{OUTPUT_HEIGHT}" - # 将修改后的配置写回 virtual_plan(拷贝后的副本,不影响原始数据源) - plan_cfg["export"] = export_cfg - virtual_plan.config = plan_cfg - - # 注入用户选择的配音 voice_id(ASR 字幕对齐模式) - if voice_ids: - plan_cfg = dict(virtual_plan.config or {}) - plan_cfg["voice_id"] = voice_ids[0] - subtitle_cfg = plan_cfg.get("subtitle", {}) or {} - if not isinstance(subtitle_cfg, dict): - subtitle_cfg = {} - subtitle_cfg["auto_generated"] = True - plan_cfg["subtitle"] = subtitle_cfg - virtual_plan.config = plan_cfg - logger.info( - "[task_id=%s] [渲染] 预览配音已注入: voice_id=%s", - task_id, - voice_ids[0], - ) - - total_duration = sum(c.duration for c in virtual_clips) - logger.info( - "[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs", - task_id, - len(virtual_clips), - total_duration, - ) - - render_start = time.monotonic() - logger.info("[task_id=%s] [渲染] RenderAdapter 统一渲染开始", task_id) - - # 使用 RenderAdapter 统一渲染入口(复用 BGM/ASR/分辨率/缩略图逻辑) - from video_processing.render_adapter import RenderAdapter - from worker_app.db import SessionLocal - - db = SessionLocal() - try: - adapter = RenderAdapter(db) - render_result = adapter.render_from_memory( - plan=virtual_plan, - clips=virtual_clips, - asset_path_map=asset_path_map, - plan_id=f"gen_{task_id}", - job_id=task_id, - work_dir=temp_path, - voiceover_audio_path=voice_path, - ) - finally: - db.close() - - if not render_result.success: - raise RuntimeError(f"渲染失败: {render_result.error_message}") - - render_output_path = render_result.output_path - render_duration = render_result.duration - - render_elapsed = time.monotonic() - render_start - logger.info( - "[task_id=%s] [渲染] RenderAdapter 完成: 耗时=%.1fs, 时长=%.2fs", - task_id, - render_elapsed, - render_duration, - ) - - # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理 - output_path = render_output_path - # RenderAdapter 在渲染完成后用本地 ffmpeg 抽取的封面候选帧(已上传 OSS) - cover_candidates = getattr(render_result, "cover_candidates", None) - - return output_path, render_duration, cover_candidates - - def _upload_and_record( task_id: str, output_path: Path, @@ -1604,307 +861,19 @@ def generate_video(self, task_id: str) -> dict: except OSError: logger.warning("[task_id=%s] 清理临时配音文件失败: %s", task_id, voiceover_tmp_path) - # DEPRECATED: 以下为旧路径,仅兼容无 source_edit_plan_id 的旧调用,后续移除 - with tempfile.TemporaryDirectory(prefix="xiaoxia-generation-") as temp_dir: - temp_path = Path(temp_dir) - # ── 2. 下载素材 ────────────────────────────────────────────────── - downloaded_videos, audio_path = _download_all_assets( - temp_path, - asset_library_id=asset_library_id, - project_id=project_id, - task_asset_ids=task_asset_ids, - voice_library_id=voice_library_id, - task_id=task_id, - voice_ids=task_info.get("voice_ids", []), - ) - - if gen_task: - gen_task.append_log( - "下载素材", - f"成功下载 {len(downloaded_videos)} 个视频素材", - count=len(downloaded_videos), - ) - _flush_logs(task_id, gen_task) - - _update_task_progress(task_id, 30, "素材下载完成") - - # ── 2.5 MediaKit 视频理解(渲染前分析素材内容)───────────────── - asset_analyses = {} - if task_asset_ids: - try: - from packages.shared.mediakit_client import get_mediakit_client - from packages.shared.storage import get_shared_storage_service - - mk_client = get_mediakit_client() - if mk_client.is_available: - storage_svc = get_shared_storage_service() - asset_urls = [] - valid_asset_ids = [] - - # 获取素材 URL(从 downloaded_videos 获取本地路径或从 asset 表获取 OSS URL) - for i, asset_id in enumerate(task_asset_ids[:5]): - try: - # 优先使用已下载的本地文件 - if i < len(downloaded_videos) and downloaded_videos[i]: - # 本地文件路径,需要上传或直接用 OSS URL - from packages.adapters.sqlalchemy_impl.asset_repository import ( - SQLAlchemyAssetRepository, - ) - - _s = SessionLocal() - try: - _ar = SQLAlchemyAssetRepository(_s) - asset = _ar.get(asset_id) - if asset and getattr(asset, "storage_key", ""): - asset_urls.append(storage_svc.get_url(asset.storage_key)) - valid_asset_ids.append(asset_id) - finally: - _s.close() - except Exception: - continue - - if asset_urls: - _update_task_progress(task_id, 35, "MediaKit 视频理解中...") - analyses = mk_client.analyze_videos( - prompt="分析这个视频的主要内容,描述场景、物体、人物动作和主题", - video_urls=asset_urls, - level="Economy", - poll_interval=2.0, - max_poll_attempts=15, - ) - if analyses: - for i, content in enumerate(analyses): - if i < len(valid_asset_ids) and content: - asset_analyses[valid_asset_ids[i]] = content - logger.info("[task_id=%s] MediaKit 视频理解完成: %d 个素材", task_id, len(asset_urls)) - - # 保存分析结果到 extra_meta - if asset_analyses: - _meta_session = SessionLocal() - try: - from packages.adapters.sqlalchemy_impl.models import ( - GenerationTaskModel, - ) - - _m = ( - _meta_session.query(GenerationTaskModel) - .filter(GenerationTaskModel.id == task_id) - .first() - ) - if _m: - existing = dict(_m.extra_meta or {}) - existing["asset_analyses"] = asset_analyses - _m.extra_meta = existing - _meta_session.commit() - finally: - _meta_session.close() - if gen_task: - _flush_logs(task_id, gen_task) - except Exception: - logger.warning("[task_id=%s] MediaKit 视频理解失败,继续渲染", task_id, exc_info=True) - - # ── 3. 渲染 + 混音 ─────────────────────────────────────────────── - _update_task_progress(task_id, 40, "开始渲染") - # 动态分辨率:优先使用 output_width/output_height,其次 resolution 字符串 - _ow = task_info.get("output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH - _oh = task_info.get("output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT - # 防护:前端可能误传宽高比(如 parseInt("9:16") = 9),宽度 < 100 时忽略 - if _ow < 100 or _oh < 100: - logger.warning( - "[task_id=%s] output_width/output_height 异常 (%dx%d),回退到默认", - task_id, - _ow, - _oh, - ) - _ow = OUTPUT_WIDTH - _oh = OUTPUT_HEIGHT - if _ow != OUTPUT_WIDTH or _oh != OUTPUT_HEIGHT: - _resolved_resolution = f"{_ow}x{_oh}" - else: - _resolved_resolution = task_info.get("resolution", "") - - output_path, render_duration, cover_candidates = _render_video( - task_id=task_id, - downloaded_videos=downloaded_videos, - voice_path=audio_path, - editing_mode=editing_mode, - project_id=project_id, - template_id=template_id, - user_id=user_id, - temp_path=temp_path, - output_name=output_name, - resolution=_resolved_resolution, - bgm_config=task_info.get("bgm_config", {}), - voice_ids=task_info.get("voice_ids", []), - custom_title=task_info.get("custom_title", ""), - title_config=task_info.get("title_config", {}), - ) - - if gen_task: - gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s") - _flush_logs(task_id, gen_task) - - _update_task_progress(task_id, 80, "渲染完成") - - # ── 4. 上传 OSS + 查重记录 ─────────────────────────────────────── - _update_task_progress(task_id, 85, "开始上传") - file_url, duration, file_size, video_count = _upload_and_record( - task_id=task_id, - output_path=output_path, - project_id=project_id, - batch_id=batch_id, - editing_mode=editing_mode, - user_id=user_id, - video_name=task_info.get("video_title", ""), - ) - - if gen_task: - gen_task.append_log( - "OSS上传", - f"上传成功, 大小={file_size}", - file_size=file_size, - file_url=file_url, - ) - _flush_logs(task_id, gen_task) - - _update_task_progress(task_id, 95, "上传完成") - - # ── 4.5 封面帧持久化 ──────────────────────────────────────────── - # RenderAdapter 在渲染完成后已用本地 ffmpeg 从 output_path 抽帧 - # (标题通过 ASS 烧录,帧天然带标题),并上传 OSS 返回 cover_candidates。 - # 这里把第一帧写入 gen_task.cover_url,完整列表写入 metadata, - # 封面路由(generation_cover.py)的 A/B/C/D 步骤即可直接命中。 - try: - if cover_candidates: - first = cover_candidates[0] - # 候选帧字段兼容:RenderAdapter 用 image_url,thumbnail_generator 用 url - cover_frame_url = first.get("image_url") or first.get("url") or "" - if cover_frame_url: - _cover_session = SessionLocal() - try: - from packages.adapters.sqlalchemy_impl.models import ( - GenerationTaskModel, - ) - - _cover_model = ( - _cover_session.query(GenerationTaskModel) - .filter(GenerationTaskModel.id == task_id) - .first() - ) - if _cover_model: - _cover_model.cover_url = cover_frame_url - # 持久化完整候选列表到 extra_meta - meta = dict(_cover_model.extra_meta or {}) - meta["cover_candidates"] = cover_candidates - _cover_model.extra_meta = meta - _cover_session.commit() - logger.info( - "[task_id=%s] 封面帧已持久化(ffmpeg本地抽帧): cover_url=%s candidates=%d", - task_id, - cover_frame_url[:80], - len(cover_candidates), - ) - finally: - _cover_session.close() - else: - logger.warning("[task_id=%s] 渲染未产出 cover_candidates,封面将依赖 API 兜底", task_id) - except Exception: - logger.warning("[task_id=%s] 封面帧持久化失败(不影响主流程)", task_id, exc_info=True) - - # ── 5. 标记完成 ────────────────────────────────────────────────── - _update_task_status(task_id, "mark_completed", result_count=video_count) - - # 5.1 更新标题使用次数 - try: - _title_session = SessionLocal() - try: - from packages.adapters.sqlalchemy_impl.generation_task_repository import ( - SQLAlchemyGenerationTaskRepository, - ) - from packages.adapters.sqlalchemy_impl.title_library_repository import ( - SQLAlchemyTitleLibraryRepository, - ) - - _task_repo = SQLAlchemyGenerationTaskRepository(_title_session) - _gen_task = _task_repo.get(task_id) - if _gen_task and _gen_task.title_ids and _gen_task.created_by_user_id: - _title_repo = SQLAlchemyTitleLibraryRepository(_title_session) - for _tid in _gen_task.title_ids: - try: - _title_repo.increment_usage_count(_tid, _gen_task.created_by_user_id) - except Exception: - logger.warning( - "[task_id=%s] 更新标题使用次数失败: title_id=%s", - task_id, - _tid, - exc_info=True, - ) - finally: - _title_session.close() - except Exception: - logger.warning("[task_id=%s] 更新标题使用次数异常(不影响主流程)", task_id, exc_info=True) - - # 5.2 更新素材使用次数 + 最近使用时间 - try: - from worker_app.core.asset_usage import mark_asset_used_for_generation - - _asset_session = SessionLocal() - try: - from packages.adapters.sqlalchemy_impl.asset_repository import ( - SQLAlchemyAssetRepository, - ) - from packages.adapters.sqlalchemy_impl.generation_task_repository import ( - SQLAlchemyGenerationTaskRepository, - ) - - _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session) - _asset_repo = SQLAlchemyAssetRepository(_asset_session) - _gen_task = _task_repo.get(task_id) - if _gen_task and _gen_task.asset_ids: - for _aid in _gen_task.asset_ids: - try: - _asset = _asset_repo.get(_aid) - if _asset: - mark_asset_used_for_generation(_asset) - _asset_repo.update(_asset) - except Exception: - logger.warning( - "[task_id=%s] 更新素材使用次数失败: asset_id=%s", - task_id, - _aid, - exc_info=True, - ) - finally: - _asset_session.close() - except Exception: - logger.warning("[task_id=%s] 更新素材使用次数异常(不影响主流程)", task_id, exc_info=True) - if gen_task: - gen_task.append_log( - "任务完成", - f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}", - duration=round(duration, 2), - file_size=file_size, - video_count=video_count, - ) - _flush_logs(task_id, gen_task) - - logger.info( - "[task_id=%s] [任务完成] duration=%.2fs file_size=%d", + else: + logger.error( + "[task_id=%s] source_edit_plan_id 为空,无法渲染。所有任务必须通过预览 API 创建并关联 EditPlan。", task_id, - duration, - file_size, ) - + if gen_task: + gen_task.append_log("任务失败", "缺少 source_edit_plan_id", level="ERROR") + _flush_logs(task_id, gen_task) return { - "status": "completed", + "status": "failed", "task_id": task_id, - "output_path": str(output_path), - "file_size": file_size, - "duration": duration, - "width": OUTPUT_WIDTH, - "height": OUTPUT_HEIGHT, - "mode": editing_mode.value, + "error": "source_edit_plan_id is required. Please create a preview task first.", } except Exception as error: logger.error("[task_id=%s] [任务失败] %s", task_id, error, exc_info=True) diff --git a/packages/adapters/sqlalchemy_impl/generation_task_repository.py b/packages/adapters/sqlalchemy_impl/generation_task_repository.py index 5f3d1d489..38aa44f9d 100755 --- a/packages/adapters/sqlalchemy_impl/generation_task_repository.py +++ b/packages/adapters/sqlalchemy_impl/generation_task_repository.py @@ -41,7 +41,6 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask: output_width=getattr(model, "output_width", 1280) or 1280, output_height=getattr(model, "output_height", 720) or 720, cover_url=getattr(model, "cover_url", "") or "", - custom_title=getattr(model, "custom_title", "") or "", title_config=dict(getattr(model, "title_config", {}) or {}), logs=model.logs or "[]", created_at=model.created_at, @@ -86,7 +85,6 @@ class SQLAlchemyGenerationTaskRepository: output_width=task.output_width, output_height=task.output_height, cover_url=task.cover_url or "", - custom_title=task.custom_title or "", title_config=dict(task.title_config) if task.title_config else {}, logs=task.logs, created_at=task.created_at, @@ -274,7 +272,6 @@ class SQLAlchemyGenerationTaskRepository: model.output_width = task.output_width model.output_height = task.output_height model.cover_url = task.cover_url or "" - model.custom_title = task.custom_title or "" model.title_config = dict(task.title_config) if task.title_config else {} model.logs = task.logs self.session.commit() diff --git a/packages/application/generation_tasks.py b/packages/application/generation_tasks.py index d916314ec..025ef6f82 100755 --- a/packages/application/generation_tasks.py +++ b/packages/application/generation_tasks.py @@ -31,7 +31,6 @@ class CreateGenerationTaskCommand: output_width: int = 1280 output_height: int = 720 cover_url: str = "" - custom_title: str = "" title_config: dict = field(default_factory=dict) @@ -68,7 +67,6 @@ class CreateGenerationTaskUseCase: output_width=command.output_width, output_height=command.output_height, cover_url=command.cover_url, - custom_title=command.custom_title, title_config=command.title_config, ) return self.generation_task_repository.create(task) diff --git a/packages/domain/generation_task.py b/packages/domain/generation_task.py index 34e891c18..707890741 100755 --- a/packages/domain/generation_task.py +++ b/packages/domain/generation_task.py @@ -120,7 +120,6 @@ class GenerationTask: output_width: int = 1280 output_height: int = 720 cover_url: str = "" - custom_title: str = "" title_config: dict = field(default_factory=dict) extra_meta: dict = field(default_factory=dict) logs: str = "[]" @@ -153,7 +152,6 @@ class GenerationTask: output_width: int = 1280, output_height: int = 720, cover_url: str = "", - custom_title: str = "", title_config: dict | None = None, extra_meta: dict | None = None, ) -> "GenerationTask": @@ -185,7 +183,6 @@ class GenerationTask: output_width=output_width, output_height=output_height, cover_url=cover_url, - custom_title=custom_title, title_config=dict(title_config) if title_config else {}, extra_meta=dict(extra_meta) if extra_meta else {}, ) @@ -306,7 +303,6 @@ class GenerationTask: self, *, cover_url: str = "", - custom_title: str = "", extra_meta: dict | None = None, output_width: int = 0, output_height: int = 0, @@ -319,8 +315,6 @@ class GenerationTask: self.is_preview = False if cover_url: self.cover_url = cover_url - if custom_title: - self.custom_title = custom_title if output_width > 0: self.output_width = output_width if output_height > 0: diff --git a/tests/integration/test_four_mode_rendering.py b/tests/integration/test_four_mode_rendering.py deleted file mode 100644 index 581aa0729..000000000 --- a/tests/integration/test_four_mode_rendering.py +++ /dev/null @@ -1,234 +0,0 @@ -"""剪辑模式渲染集成测试. - -验证剪辑模式(ONE_TAKE / VOICE_OVER)通过 -_build_plan_and_clips_from_task + UnifiedRenderService 的完整渲染流程。 -注:PIP / VOICE_PIP 已下线,统一映射为 ONE_TAKE。 - -需要 ffmpeg 可用;CI 无 ffmpeg 时自动跳过。 -""" - -from __future__ import annotations - -import shutil -import subprocess -import tempfile -from pathlib import Path - -import pytest -from video_processing.unified_render_service import ( - RenderResult, - UnifiedRenderService, - _resolve_layer_role, -) -from worker_app.tasks.generation import _build_plan_and_clips_from_task - -pytestmark = pytest.mark.skipif( - not shutil.which("ffmpeg"), - reason="ffmpeg not available", -) - - -# ── 辅助函数 ────────────────────────────────────────────────────────────────── - - -def _generate_test_video(path: Path, duration: float = 3.0, color: str = "red") -> None: - """生成一个纯色测试视频。""" - cmd = [ - "ffmpeg", - "-y", - "-f", - "lavfi", - "-i", - f"color=c={color}:s=640x360:d={duration}:r=25", - "-c:v", - "libx264", - "-pix_fmt", - "yuv420p", - "-movflags", - "+faststart", - str(path), - ] - subprocess.run(cmd, check=True, capture_output=True, timeout=30) - - -def _render_with_mode( - mode: str, - num_clips: int = 3, - duration: float = 2.0, -) -> tuple[RenderResult, Path]: - """用指定模式生成测试视频并渲染,返回 (result, work_dir)。 - - 调用方负责清理 work_dir。 - """ - work_dir = Path(tempfile.mkdtemp(prefix="test_4mode_")) - - # 生成测试视频素材 - colors = ["red", "green", "blue", "yellow", "purple"] - downloaded_paths: list[Path] = [] - for i in range(num_clips): - p = work_dir / f"test_{i:03d}.mp4" - _generate_test_video(p, duration=duration, color=colors[i % len(colors)]) - downloaded_paths.append(p) - - # 构建虚拟 plan + clips - task_id = f"test_task_{mode}" - plan, clips, asset_path_map = _build_plan_and_clips_from_task( - task_id=task_id, - downloaded_paths=downloaded_paths, - mode=mode, - ) - - # 渲染 - service = UnifiedRenderService( - plan=plan, - clips=clips, - asset_path_map=asset_path_map, - work_dir=work_dir, - output_width=640, - output_height=360, - output_fps=25, - ) - result = service.render() - return result, work_dir - - -# ── 测试 _build_plan_and_clips_from_task ────────────────────────────────────── - - -class TestBuildPlanAndClips: - """测试 4 种模式的虚拟 plan 构建。""" - - def _make_paths(self, n: int) -> list[Path]: - return [Path(f"/tmp/test_{i}.mp4") for i in range(n)] - - def test_one_take_mode(self): - paths = self._make_paths(3) - plan, clips, asset_map = _build_plan_and_clips_from_task("t1", paths, "one_take") - - assert plan.id == "t1" - assert len(clips) == 3 - assert all(c.clip_type == "main" for c in clips) - assert len(asset_map) == 3 - - def test_pip_mode_maps_to_one_take(self): - """PIP 已下线,映射为 one_take → 全部 main clips。""" - paths = self._make_paths(3) - plan, clips, asset_map = _build_plan_and_clips_from_task("t2", paths, "pip") - - assert len(clips) == 3 - assert all(c.clip_type == "main" for c in clips) - - def test_voice_over_mode(self): - paths = self._make_paths(3) - plan, clips, asset_map = _build_plan_and_clips_from_task("t3", paths, "voice_over") - - assert len(clips) == 3 - assert all(c.clip_type == "main" for c in clips) - assert all(c.config.get("role") == "b_roll" for c in clips) - - def test_voice_pip_mode_maps_to_one_take(self): - """VOICE_PIP 已下线,映射为 one_take → 全部 main clips。""" - paths = self._make_paths(4) - plan, clips, asset_map = _build_plan_and_clips_from_task("t4", paths, "voice_pip") - - assert len(clips) == 4 - assert all(c.clip_type == "main" for c in clips) - - def test_unknown_mode_defaults_to_one_take(self): - paths = self._make_paths(2) - plan, clips, asset_map = _build_plan_and_clips_from_task("t5", paths, "unknown_mode") - - assert len(clips) == 2 - assert all(c.clip_type == "main" for c in clips) - - def test_asset_path_map_keys_match_clip_asset_ids(self): - paths = self._make_paths(3) - _, clips, asset_map = _build_plan_and_clips_from_task("t6", paths, "one_take") - - clip_asset_ids = {c.asset_id for c in clips} - map_keys = set(asset_map.keys()) - assert clip_asset_ids == map_keys - - -# ── 测试图层分组(4 模式) ──────────────────────────────────────────────────── - - -class TestFourModeLayerGrouping: - """验证 4 种模式的 clip_type 分布经 _resolve_layer_role 后产生正确的图层。""" - - def test_one_take_layers(self): - """ONE_TAKE: 3 main → 1 main layer。""" - paths = [Path(f"/tmp/ot_{i}.mp4") for i in range(3)] - _, clips, _ = _build_plan_and_clips_from_task("ot", paths, "one_take") - - roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} - assert roles == {"main"} - - def test_pip_layers_now_one_take(self): - """PIP 已下线 → one_take: 3 main → 1 main layer。""" - paths = [Path(f"/tmp/pip_{i}.mp4") for i in range(3)] - _, clips, _ = _build_plan_and_clips_from_task("pip", paths, "pip") - - roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} - assert roles == {"main"} - - def test_voice_over_layers(self): - """VOICE_OVER: 3 main(b_roll) → broll。""" - paths = [Path(f"/tmp/vo_{i}.mp4") for i in range(3)] - _, clips, _ = _build_plan_and_clips_from_task("vo", paths, "voice_over") - - roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} - assert roles == {"broll"} - - def test_voice_pip_layers_now_one_take(self): - """VOICE_PIP 已下线 → one_take: 4 main → main layer。""" - paths = [Path(f"/tmp/vpip_{i}.mp4") for i in range(4)] - _, clips, _ = _build_plan_and_clips_from_task("vpip", paths, "voice_pip") - - roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips} - assert roles == {"main"} - - -# ── 端到端渲染测试(需要 ffmpeg) ───────────────────────────────────────────── - - -class TestEndToEndRendering: - """4 种模式的完整渲染测试,验证输出文件存在且时长合理。""" - - def test_one_take_render(self): - result, work_dir = _render_with_mode("one_take", num_clips=2, duration=2.0) - try: - assert result.output_path.exists() - assert result.file_size > 0 - assert result.duration > 0 - assert result.width == 640 - assert result.height == 360 - finally: - shutil.rmtree(work_dir, ignore_errors=True) - - def test_pip_render(self): - result, work_dir = _render_with_mode("pip", num_clips=2, duration=2.0) - try: - assert result.output_path.exists() - assert result.file_size > 0 - assert result.duration > 0 - finally: - shutil.rmtree(work_dir, ignore_errors=True) - - def test_voice_over_render(self): - result, work_dir = _render_with_mode("voice_over", num_clips=2, duration=2.0) - try: - assert result.output_path.exists() - assert result.file_size > 0 - assert result.duration > 0 - finally: - shutil.rmtree(work_dir, ignore_errors=True) - - def test_voice_pip_render(self): - result, work_dir = _render_with_mode("voice_pip", num_clips=3, duration=2.0) - try: - assert result.output_path.exists() - assert result.file_size > 0 - assert result.duration > 0 - finally: - shutil.rmtree(work_dir, ignore_errors=True) diff --git a/tests/integration/test_full_pipeline.py b/tests/integration/test_full_pipeline.py deleted file mode 100644 index 70307d4a1..000000000 --- a/tests/integration/test_full_pipeline.py +++ /dev/null @@ -1,189 +0,0 @@ -"""全链路集成测试. - -验证 PlanGeneratorService → UnifiedRenderService → 查重 的端到端流程。 -""" - -from __future__ import annotations - -import shutil -import subprocess -import tempfile -from pathlib import Path - -import pytest -from video_processing.unified_render_service import ( - UnifiedRenderService, -) -from worker_app.tasks.generation import _build_plan_and_clips_from_task - -pytestmark = pytest.mark.skipif( - not shutil.which("ffmpeg"), - reason="ffmpeg not available", -) - - -def _generate_test_video(path: Path, duration: float = 3.0) -> None: - """生成一个测试视频。""" - cmd = [ - "ffmpeg", - "-y", - "-f", - "lavfi", - "-i", - f"color=c=blue:s=640x360:d={duration}:r=25", - "-c:v", - "libx264", - "-pix_fmt", - "yuv420p", - str(path), - ] - subprocess.run(cmd, check=True, capture_output=True, timeout=30) - - -def _generate_test_audio(path: Path, duration: float = 5.0) -> None: - """生成一个测试音频文件。""" - cmd = [ - "ffmpeg", - "-y", - "-f", - "lavfi", - "-i", - f"sine=frequency=440:duration={duration}", - "-c:a", - "aac", - "-b:a", - "128k", - str(path), - ] - subprocess.run(cmd, check=True, capture_output=True, timeout=30) - - -class TestFullPipeline: - """验证从虚拟 plan 构建到渲染输出的完整流程。""" - - def test_one_take_pipeline(self): - """ONE_TAKE 模式完整流程。""" - with tempfile.TemporaryDirectory() as tmpdir: - work_dir = Path(tmpdir) - - # 生成测试素材 - paths = [] - for i in range(3): - p = work_dir / f"clip_{i}.mp4" - _generate_test_video(p, duration=2.0) - paths.append(p) - - # 构建虚拟 plan - plan, clips, asset_map = _build_plan_and_clips_from_task("pipeline_test", paths, "one_take") - - # 渲染 - service = UnifiedRenderService( - plan=plan, - clips=clips, - asset_path_map=asset_map, - work_dir=work_dir, - output_width=640, - output_height=360, - ) - result = service.render() - - assert result.output_path.exists() - assert result.duration > 0 - assert result.file_size > 0 - assert result.width == 640 - assert result.height == 360 - - def test_pipeline_with_audio_mux(self): - """渲染 + 混音后处理。""" - with tempfile.TemporaryDirectory() as tmpdir: - work_dir = Path(tmpdir) - - # 生成测试素材 - video_path = work_dir / "clip_0.mp4" - _generate_test_video(video_path, duration=3.0) - - # 构建虚拟 plan - plan, clips, asset_map = _build_plan_and_clips_from_task("audio_test", [video_path], "one_take") - - # 渲染 - service = UnifiedRenderService( - plan=plan, - clips=clips, - asset_path_map=asset_map, - work_dir=work_dir, - output_width=640, - output_height=360, - ) - render_result = service.render() - - # 混音 - 直接用 ffmpeg(_mux_audio_track 已被清理) - audio_path = work_dir / "voice.aac" - _generate_test_audio(audio_path, duration=5.0) - - final_path = work_dir / "final.mp4" - mux_cmd = [ - "ffmpeg", - "-y", - "-i", - str(render_result.output_path), - "-i", - str(audio_path), - "-c:v", - "copy", - "-c:a", - "aac", - "-shortest", - str(final_path), - ] - subprocess.run(mux_cmd, check=True, capture_output=True, timeout=30) - - assert final_path.exists() - assert final_path.stat().st_size > 0 - - def test_single_clip_pipeline(self): - """单 clip 渲染(无转场)。""" - with tempfile.TemporaryDirectory() as tmpdir: - work_dir = Path(tmpdir) - - video_path = work_dir / "single.mp4" - _generate_test_video(video_path, duration=5.0) - - plan, clips, asset_map = _build_plan_and_clips_from_task("single_test", [video_path], "one_take") - - service = UnifiedRenderService( - plan=plan, - clips=clips, - asset_path_map=asset_map, - work_dir=work_dir, - output_width=640, - output_height=360, - ) - result = service.render() - - assert result.output_path.exists() - assert result.duration > 0 - - def test_dedup_helper_integration(self): - """验证 dedup_helpers.create_video_record_and_dedup 的导入和签名。""" - # 只验证函数存在且签名正确(不实际调用,需要数据库) - import inspect - - from video_processing.dedup_helpers import create_video_record_and_dedup - - sig = inspect.signature(create_video_record_and_dedup) - params = set(sig.parameters.keys()) - expected = { - "generation_task_id", - "project_id", - "batch_id", - "file_url", - "file_size", - "duration", - "video_path", - "mode", - "session", - "width", - "height", - "fps", - } - assert expected.issubset(params), f"Missing params: {expected - params}" diff --git a/tests/unit/test_1280_preview_speedup.py b/tests/unit/test_1280_preview_speedup.py index 1511352fd..2db7a90d7 100644 --- a/tests/unit/test_1280_preview_speedup.py +++ b/tests/unit/test_1280_preview_speedup.py @@ -149,14 +149,6 @@ class TestWorkerGenerationNoPreviewOverride: assert 'resolution = "854x480"' not in source, "Should not override resolution to 480p in preview mode" assert 'bitrate = "1M"' not in source, "Should not override bitrate to 1M in preview mode" - def test_parallel_download_still_works(self): - """并行下载逻辑保留。""" - with open("apps/worker/worker_app/tasks/generation.py") as f: - source = f.read() - - assert "ThreadPoolExecutor" in source, "Should use ThreadPoolExecutor for parallel downloads" - assert "as_completed" in source, "Should use as_completed for result collection" - # ── 5. generation_preview.py 不再有 PREVIEW_RESOLUTION ── diff --git a/tests/unit/test_1294_preview_voice_injection.py b/tests/unit/test_1294_preview_voice_injection.py index 3b5c11363..7401e7d98 100644 --- a/tests/unit/test_1294_preview_voice_injection.py +++ b/tests/unit/test_1294_preview_voice_injection.py @@ -2,7 +2,6 @@ 验证: 1. _load_task_info 正确加载 voice_ids -2. _render_video 接受 voice_ids 参数 3. voice_ids 正确注入到 plan config 中(实际执行代码路径,diff-cover 可达) """ @@ -76,183 +75,3 @@ class TestLoadTaskInfoVoiceIds: assert result["voice_ids"] == [] -class TestRenderVideoVoiceInjection: - """验证 _render_video 正确注入 voice_id 到 plan config(实际执行代码路径)""" - - def test_render_video_accepts_voice_ids(self): - """_render_video 签名包含 voice_ids 参数""" - import inspect - - from worker_app.tasks.generation import _render_video - - sig = inspect.signature(_render_video) - assert "voice_ids" in sig.parameters - - def test_voice_ids_default_none(self): - """voice_ids 参数默认为 None""" - import inspect - - from worker_app.tasks.generation import _render_video - - sig = inspect.signature(_render_video) - param = sig.parameters["voice_ids"] - assert param.default is None - - def test_voice_ids_injected_into_plan_config(self): - """voice_ids 非空时,voice_id 和 subtitle.auto_generated 被注入到 plan config。 - - 此测试实际执行 _render_video 的配音注入代码路径,确保 diff-cover 覆盖新增行。 - """ - from dataclasses import dataclass, field - - @dataclass - class MockClip: - """模拟 VirtualClip,至少需要 duration 属性。""" - - id: str = "clip_1" - duration: float = 5.0 - config: dict = field(default_factory=dict) - - @dataclass - class MockPlan: - """模拟 VirtualPlan,至少需要 config 属性。""" - - id: str = "test_plan" - name: str = "test" - config: dict = field(default_factory=dict) - - mock_plan = MockPlan(config={"some_key": "some_value"}) - mock_clips = [MockClip(duration=5.0), MockClip(duration=3.0)] - mock_asset_path_map = {"asset_1": Path("/tmp/video1.mp4")} - - # Mock RenderAdapter 和 render 结果 - mock_render_result = MagicMock() - mock_render_result.success = True - mock_render_result.output_path = Path("/tmp/output.mp4") - mock_render_result.duration = 8.0 - - mock_adapter_cls = MagicMock(return_value=MagicMock()) - mock_adapter_cls.return_value.render_from_memory.return_value = mock_render_result - - mock_db = MagicMock() - - with ( - patch( - "worker_app.tasks.generation._build_plan_and_clips_from_task", - return_value=(mock_plan, mock_clips, mock_asset_path_map), - ), - patch( - "worker_app.tasks.generation._load_template_plan_config", - return_value=None, - ), - patch( - "video_processing.render_adapter.RenderAdapter", - mock_adapter_cls, - ), - patch( - "worker_app.tasks.generation.SessionLocal", - return_value=mock_db, - ), - ): - from worker_app.tasks.generation import _render_video - - from packages.domain import EditingMode - - output_path, render_duration, cover_candidates = _render_video( - task_id="test_task_123", - downloaded_videos=[Path("/tmp/video1.mp4")], - voice_path=None, - editing_mode=EditingMode.ONE_TAKE, - project_id="proj_1", - template_id="tmpl_1", - user_id="user_1", - temp_path=Path("/tmp"), - output_name="test_output.mp4", - resolution="854x480", - voice_ids=["voice_abc"], - ) - - # 验证 voice_id 被注入到 plan config(覆盖新增代码行) - assert mock_plan.config.get("voice_id") == "voice_abc" - # 验证 subtitle.auto_generated 被设置为 True - assert mock_plan.config.get("subtitle", {}).get("auto_generated") is True - # 验证 RenderAdapter 被调用 - mock_adapter_cls.return_value.render_from_memory.assert_called_once() - # 验证返回值 - assert output_path == Path("/tmp/output.mp4") - assert render_duration == 8.0 - - def test_voice_ids_empty_skips_injection(self): - """voice_ids 为空时,不注入 voice_id 到 plan config""" - from dataclasses import dataclass, field - - @dataclass - class MockClip: - id: str = "clip_1" - duration: float = 5.0 - - @dataclass - class MockPlan: - id: str = "test_plan" - name: str = "test" - config: dict = field(default_factory=dict) - - mock_plan = MockPlan(config={"export": {"resolution": "854x480"}}) - mock_clips = [MockClip(duration=5.0)] - - mock_render_result = MagicMock() - mock_render_result.success = True - mock_render_result.output_path = Path("/tmp/output.mp4") - mock_render_result.duration = 5.0 - - mock_adapter_cls = MagicMock(return_value=MagicMock()) - mock_adapter_cls.return_value.render_from_memory.return_value = mock_render_result - - with ( - patch( - "worker_app.tasks.generation._build_plan_and_clips_from_task", - return_value=(mock_plan, mock_clips, {}), - ), - patch( - "worker_app.tasks.generation._load_template_plan_config", - return_value=None, - ), - patch( - "video_processing.render_adapter.RenderAdapter", - mock_adapter_cls, - ), - patch( - "worker_app.tasks.generation.SessionLocal", - return_value=MagicMock(), - ), - ): - from worker_app.tasks.generation import _render_video - - from packages.domain import EditingMode - - _render_video( - task_id="test_task_456", - downloaded_videos=[Path("/tmp/video1.mp4")], - voice_path=None, - editing_mode=EditingMode.ONE_TAKE, - project_id="proj_1", - template_id="", - user_id="user_1", - temp_path=Path("/tmp"), - output_name="test_output.mp4", - voice_ids=[], - ) - - # 验证 voice_id 没有被注入 - assert "voice_id" not in mock_plan.config - - -class TestGenerateVideoPassesVoiceIds: - """验证 generate_video 调用 _render_video 时传递 voice_ids""" - - def test_generate_video_passes_voice_ids(self): - """generate_video 中 _render_video 调用包含 voice_ids 参数""" - with open("apps/worker/worker_app/tasks/generation.py", "r") as f: - content = f.read() - - assert 'voice_ids=task_info.get("voice_ids", [])' in content diff --git a/tests/unit/test_bool_config_defense.py b/tests/unit/test_bool_config_defense.py deleted file mode 100644 index 331f5214a..000000000 --- a/tests/unit/test_bool_config_defense.py +++ /dev/null @@ -1,130 +0,0 @@ -""" -测试:模板 config 字段存储了非 dict 值(如 True / False / str)时, -渲染链路不会崩溃('bool' object has no attribute 'get')。 - -覆盖两个关键文件: -1. generation.py — _load_template_plan_config 旧系统路径 -2. unified_render_service.py — _maybe_generate_ass -""" - -from __future__ import annotations - -import os -import sys -from pathlib import Path -from unittest.mock import MagicMock, patch - -import pytest - -# Add worker app to path -sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) -os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret") -os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") - - -class TestLoadTemplatePlanConfigBoolDefense: - """_load_template_plan_config 旧系统路径对非 dict 值的防护。""" - - def _call_old_path(self, title_cfg, subtitle_cfg, bgm_cfg): - """通过 mock 新模板系统返回 None,强制走旧模板系统 fallback 路径。""" - from worker_app.tasks.generation import _load_template_plan_config - - mock_old_template = MagicMock() - mock_old_template.title_config = title_cfg - mock_old_template.subtitle_config = subtitle_cfg - mock_old_template.bgm_config = bgm_cfg - - mock_session = MagicMock() - # 旧系统 query 返回 mock template - mock_session.query.return_value.filter.return_value.first.return_value = mock_old_template - - # Mock 新模板系统 repo.get() 返回 None(强制走 fallback) - mock_repo_cls = MagicMock() - mock_repo_cls.return_value.get.return_value = None - - with ( - patch("worker_app.tasks.generation.SessionLocal", return_value=mock_session), - patch("packages.adapters.sqlalchemy_impl.SQLAlchemyEditTemplateRepository", mock_repo_cls), - patch("packages.adapters.sqlalchemy_impl.SQLAlchemyTemplateClipConfigRepository", MagicMock()), - ): - return _load_template_plan_config("fake-id") - - def test_bool_values_return_empty(self): - """title_config=True / subtitle_config=False / bgm_config='str' → 全部过滤掉""" - result = self._call_old_path(True, False, "not_a_dict") - assert isinstance(result, dict) - assert "title" not in result - assert "subtitle" not in result - assert "bgm" not in result - - def test_valid_dict_passes_through(self): - """正常 dict 正常传递""" - result = self._call_old_path( - {"text": "标题", "enabled": True}, - {"text": "副标题"}, - {"enabled": True, "source": "test.mp3"}, - ) - assert result["title"] == {"text": "标题", "enabled": True} - assert result["subtitle"] == {"text": "副标题"} - assert result["bgm"] == {"enabled": True, "source": "test.mp3"} - - def test_none_returns_empty(self): - """None → 空 dict""" - result = self._call_old_path(None, None, None) - assert result == {} - - def test_mixed_valid_and_invalid(self): - """部分有效、部分无效时只保留有效的""" - result = self._call_old_path({"text": "OK"}, True, None) - assert "title" in result - assert "subtitle" not in result - assert "bgm" not in result - - def test_int_and_list_also_filtered(self): - """int / list 类型也被过滤""" - result = self._call_old_path(42, [1, 2, 3], 0) - assert result == {} - - -class TestUnifiedRenderBoolConfigDefense: - """_maybe_generate_ass 对 plan.config 中非 dict title/subtitle 的防护。""" - - def _make_service(self, config): - from video_processing.unified_render_service import UnifiedRenderService - - service = UnifiedRenderService.__new__(UnifiedRenderService) - mock_plan = MagicMock() - mock_plan.config = config - service.plan = mock_plan - service.task_id = "test-task" - return service - - def test_bool_title_does_not_crash(self): - """config['title']=True → 不崩溃,返回 None""" - service = self._make_service({"title": True, "subtitle": {}}) - result = service._maybe_generate_ass(video_duration=10.0) - assert result is None - - def test_bool_subtitle_does_not_crash(self): - """config['subtitle']=False → 不崩溃,返回 None""" - service = self._make_service({"title": {}, "subtitle": False}) - result = service._maybe_generate_ass(video_duration=10.0) - assert result is None - - def test_str_title_does_not_crash(self): - """config['title']='plain string' → 不崩溃""" - service = self._make_service({"title": "plain string", "subtitle": {}}) - result = service._maybe_generate_ass(video_duration=10.0) - assert result is None - - def test_none_config_does_not_crash(self): - """config=None → 不崩溃""" - service = self._make_service(None) - result = service._maybe_generate_ass(video_duration=10.0) - assert result is None - - def test_int_title_does_not_crash(self): - """config['title']=42 → 不崩溃""" - service = self._make_service({"title": 42, "subtitle": 0}) - result = service._maybe_generate_ass(video_duration=10.0) - assert result is None diff --git a/tests/unit/test_confirm_generation.py b/tests/unit/test_confirm_generation.py index 9072fd120..69877544d 100644 --- a/tests/unit/test_confirm_generation.py +++ b/tests/unit/test_confirm_generation.py @@ -5,7 +5,7 @@ - 预览任务未完成 → 创建新任务走渲染流程 - 预览任务不存在 → 404 - 权限不足 → 403 - - cover_url 和 custom_title 正确传递 + - cover_url 正确传递 """ from __future__ import annotations @@ -201,7 +201,6 @@ def _make_preview_task(**kwargs: Any) -> GenerationTask: output_width=1080, output_height=1920, cover_url="", - custom_title="", video_title="", resolution="", bgm_config={}, @@ -233,7 +232,6 @@ class TestConfirmGenerationReuse: "output_width": 1080, "output_height": 1920, "cover_url": "https://example.com/cover.jpg", - "custom_title": "我的视频", }, ) @@ -251,7 +249,6 @@ class TestConfirmGenerationReuse: assert item["output_height"] == 1920 # 封面和标题更新 assert item["cover_url"] == "https://example.com/cover.jpg" - assert item["custom_title"] == "我的视频" # 没有创建新任务 assert len(gen_task_repo._store) == initial_count @@ -277,7 +274,6 @@ class TestConfirmGenerationReuse: assert updated is not None assert updated.is_preview is False assert updated.cover_url == "https://cdn.example.com/cover.png" - assert updated.custom_title == "测试标题" def test_confirm_creates_new_task_when_preview_not_completed( self, @@ -356,7 +352,6 @@ class TestConfirmGenerationErrors: assert item["output_height"] == 1080 # 默认封面和标题为空 assert item["cover_url"] == "" - assert item["custom_title"] == "" # 配置保留 assert item["voice_library_id"] == "voice-001" assert item["template_id"] == "tmpl-001" @@ -368,7 +363,7 @@ class TestConfirmGenerationErrors: client: TestClient, gen_task_repo: StubGenerationTaskRepository, ) -> None: - """cover_url 和 custom_title 正确传递""" + """cover_url 正确传递""" preview = _make_preview_task() gen_task_repo.create(preview) @@ -378,14 +373,12 @@ class TestConfirmGenerationErrors: "output_width": 1080, "output_height": 1920, "cover_url": "https://cdn.example.com/my-cover.png", - "custom_title": "测试视频标题", }, ) assert resp.status_code == 200 item = resp.json()["items"][0] assert item["cover_url"] == "https://cdn.example.com/my-cover.png" - assert item["custom_title"] == "测试视频标题" def test_confirm_default_resolution( self, diff --git a/tests/unit/test_generation_p3_optimizations.py b/tests/unit/test_generation_p3_optimizations.py deleted file mode 100644 index f8dceb68c..000000000 --- a/tests/unit/test_generation_p3_optimizations.py +++ /dev/null @@ -1,389 +0,0 @@ -"""P3 优化单元测试 — generation.py 三项优化. - -覆盖: - P3-1: _download_library_assets strict 模式 - P3-2: 归属校验合并到同一 DB session - P3-3: _verify_url_accessible HEAD 重试 -""" - -from __future__ import annotations - -import sys -from pathlib import Path -from types import SimpleNamespace -from unittest.mock import MagicMock, patch - -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 重试 ─────────────────────────────────────── - - -class TestVerifyUrlAccessibleRetry: - """_verify_url_accessible 重试逻辑.""" - - @patch("time.sleep") - @patch("urllib.request.OpenerDirector.open") - def test_first_attempt_success(self, mock_open, mock_sleep): - """首次成功,不重试.""" - from worker_app.tasks.generation import _verify_url_accessible - - mock_resp = MagicMock() - mock_resp.status = 200 - mock_resp.__enter__ = MagicMock(return_value=mock_resp) - mock_resp.__exit__ = MagicMock(return_value=False) - mock_open.return_value = mock_resp - - assert _verify_url_accessible("https://example.com/file.mp4") is True - assert mock_open.call_count == 1 - mock_sleep.assert_not_called() - - @patch("time.sleep") - @patch("urllib.request.OpenerDirector.open") - def test_retry_then_success(self, mock_open, mock_sleep): - """首次失败,重试后成功.""" - from worker_app.tasks.generation import _verify_url_accessible - - # 第一次失败(网络异常),第二次成功 - mock_resp_ok = MagicMock() - mock_resp_ok.status = 200 - mock_resp_ok.__enter__ = MagicMock(return_value=mock_resp_ok) - mock_resp_ok.__exit__ = MagicMock(return_value=False) - - mock_open.side_effect = [ - OSError("connection reset"), - mock_resp_ok, - ] - - assert _verify_url_accessible("https://example.com/file.mp4") is True - assert mock_open.call_count == 2 - mock_sleep.assert_called_once_with(1) - - @patch("time.sleep") - @patch("urllib.request.OpenerDirector.open") - def test_all_retries_exhausted(self, mock_open, mock_sleep): - """全部重试耗尽,返回 False.""" - from worker_app.tasks.generation import _verify_url_accessible - - mock_open.side_effect = OSError("connection refused") - - assert _verify_url_accessible("https://example.com/file.mp4") is False - # 1 首次 + 2 重试 = 3 次 - assert mock_open.call_count == 3 - assert mock_sleep.call_count == 2 - - @patch("time.sleep") - @patch("urllib.request.OpenerDirector.open") - def test_http_500_then_success(self, mock_open, mock_sleep): - """HTTP 500 后重试成功.""" - from worker_app.tasks.generation import _verify_url_accessible - - mock_resp_500 = MagicMock() - mock_resp_500.status = 500 - mock_resp_500.__enter__ = MagicMock(return_value=mock_resp_500) - mock_resp_500.__exit__ = MagicMock(return_value=False) - - mock_resp_200 = MagicMock() - mock_resp_200.status = 200 - mock_resp_200.__enter__ = MagicMock(return_value=mock_resp_200) - mock_resp_200.__exit__ = MagicMock(return_value=False) - - mock_open.side_effect = [mock_resp_500, mock_resp_200] - - assert _verify_url_accessible("https://example.com/file.mp4") is True - assert mock_open.call_count == 2 - - @patch("time.sleep") - @patch("urllib.request.OpenerDirector.open") - def test_custom_retries_zero(self, mock_open, mock_sleep): - """retries=0 时不重试.""" - from worker_app.tasks.generation import _verify_url_accessible - - mock_open.side_effect = OSError("timeout") - - assert _verify_url_accessible("https://example.com/file.mp4", retries=0) is False - assert mock_open.call_count == 1 - mock_sleep.assert_not_called() - - -# ── P3-1: _download_library_assets strict 模式 ────────────────────────────── - - -def _make_mock_asset( - asset_id: str, - name: str, - file_url: str | None, - asset_library_id: str = "lib-1", - project_id: str = "", -): - """构造 mock AssetModel 实例.""" - return SimpleNamespace( - id=asset_id, - name=name, - file_url=file_url, - asset_library_id=asset_library_id, - project_id=project_id, - status="ready", - file_type="video", - created_at="2026-01-01", - ) - - -def _setup_mock_session(assets): - """构造 mock session,返回 (mock_session, mock_query_chain).""" - mock_session = MagicMock() - mock_query = MagicMock() - - # chain: session.query().filter().filter().order_by().all() - mock_session.query.return_value = mock_query - mock_query.filter.return_value = mock_query - mock_query.order_by.return_value = mock_query - mock_query.all.return_value = assets - - return mock_session - - -class TestDownloadLibraryAssetsStrictMode: - """_download_library_assets strict 模式.""" - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_strict_mode_raises_on_download_failure(self, mock_download, mock_session_factory): - """strict=True 时,单个素材下载失败立即抛 RuntimeError.""" - from worker_app.tasks.generation import _download_library_assets - - assets = [ - _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"), - _make_mock_asset("a2", "video2.mp4", "uploads/video2.mp4"), - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - - # 第一个成功,第二个失败 - mock_download.side_effect = [True, False] - - with pytest.raises(RuntimeError, match="素材下载失败"): - _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - asset_ids=["a1", "a2"], - strict=True, - ) - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_non_strict_mode_returns_partial_results(self, mock_download, mock_session_factory): - """strict=False 时,跳过失败素材,返回成功列表.""" - from worker_app.tasks.generation import _download_library_assets - - assets = [ - _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"), - _make_mock_asset("a2", "video2.mp4", "uploads/video2.mp4"), - _make_mock_asset("a3", "video3.mp4", "uploads/video3.mp4"), - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - - # 第一个成功,第二个失败,第三个成功 - mock_download.side_effect = [True, False, True] - - result = _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - asset_ids=["a1", "a2", "a3"], - strict=False, - ) - - assert len(result) == 2 - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_non_strict_all_fail_raises(self, mock_download, mock_session_factory): - """strict=False 但全部失败时仍抛 RuntimeError.""" - from worker_app.tasks.generation import _download_library_assets - - assets = [ - _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"), - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - mock_download.return_value = False - - with pytest.raises(RuntimeError, match="全部下载失败"): - _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - asset_ids=["a1"], - strict=False, - ) - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_strict_mode_raises_on_missing_file_url(self, mock_download, mock_session_factory): - """strict=True 时,素材缺少 file_url 立即抛异常.""" - from worker_app.tasks.generation import _download_library_assets - - assets = [ - _make_mock_asset("a1", "video1.mp4", None), # file_url 为空 - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - - with pytest.raises(RuntimeError, match="素材缺少 file_url"): - _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - strict=True, - ) - - # download_asset 不应被调用 - mock_download.assert_not_called() - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_default_is_strict(self, mock_download, mock_session_factory): - """默认 strict=True.""" - from worker_app.tasks.generation import _download_library_assets - - assets = [ - _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4"), - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - mock_download.return_value = False - - # 不传 strict 参数,默认严格模式 - with pytest.raises(RuntimeError, match="素材下载失败"): - _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - ) - - -# ── P3-2: 归属校验合并到同一 session ──────────────────────────────────────── - - -class TestDownloadLibraryAssetsOwnershipValidation: - """归属校验合并到 _download_library_assets 同一 session.""" - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_ownership_mismatch_raises_value_error(self, mock_download, mock_session_factory): - """asset_ids 不属于指定素材库时抛 ValueError.""" - from worker_app.tasks.generation import _download_library_assets - - # asset 属于 lib-2,但请求的是 lib-1 - assets = [ - _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-2"), - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - - with pytest.raises(ValueError, match="素材不属于指定素材库"): - _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - asset_ids=["a1"], - ) - - # 不应调用 download_asset(校验在下载前) - mock_download.assert_not_called() - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_missing_asset_ids_raises_value_error(self, mock_download, mock_session_factory): - """指定的 asset_ids 不存在时抛 ValueError.""" - from worker_app.tasks.generation import _download_library_assets - - # DB 返回空(asset_ids 不存在,query 过滤后无结果) - mock_session = _setup_mock_session([]) - mock_session_factory.return_value = mock_session - - with pytest.raises(RuntimeError, match="未找到视频素材"): - _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - asset_ids=["nonexistent-id"], - ) - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_project_ownership_mismatch_raises(self, mock_download, mock_session_factory): - """项目级模式下归属不匹配抛 ValueError.""" - from worker_app.tasks.generation import _download_library_assets - - assets = [ - _make_mock_asset( - "a1", - "video1.mp4", - "uploads/video1.mp4", - asset_library_id="", - project_id="proj-2", - ), - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - - with pytest.raises(ValueError, match="素材不属于指定项目"): - _download_library_assets( - Path("/tmp"), - project_id="proj-1", - asset_ids=["a1"], - ) - - mock_download.assert_not_called() - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_ownership_pass_then_download(self, mock_download, mock_session_factory): - """归属校验通过后正常下载.""" - from worker_app.tasks.generation import _download_library_assets - - assets = [ - _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-1"), - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - mock_download.return_value = True - - result = _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - asset_ids=["a1"], - ) - - assert len(result) == 1 - mock_download.assert_called_once() - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_single_session_used(self, mock_download, mock_session_factory): - """验证只创建了一个 DB session(P3-2 核心).""" - from worker_app.tasks.generation import _download_library_assets - - assets = [ - _make_mock_asset("a1", "video1.mp4", "uploads/video1.mp4", asset_library_id="lib-1"), - ] - mock_session = _setup_mock_session(assets) - mock_session_factory.return_value = mock_session - mock_download.return_value = True - - _download_library_assets( - Path("/tmp"), - asset_library_id="lib-1", - asset_ids=["a1"], - ) - - # SessionLocal 只调用一次(合并前会调用两次:校验 + 下载) - assert mock_session_factory.call_count == 1 diff --git a/tests/unit/test_generation_worker_fixes.py b/tests/unit/test_generation_worker_fixes.py deleted file mode 100644 index a1fd969e1..000000000 --- a/tests/unit/test_generation_worker_fixes.py +++ /dev/null @@ -1,228 +0,0 @@ -"""Tests for generation.py worker-side fixes: segment durations + preview resolution.""" - -from __future__ import annotations - -import os -import sys -from pathlib import Path -from unittest.mock import MagicMock, patch - -# Add worker app to path -sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) -os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret") -os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") - - -def _patch_session_local(mock_session): - """Patch worker_app.db.SessionLocal robustly even when other tests - have pre-registered a MagicMock for worker_app.db in sys.modules. - Uses patch.dict to inject a clean module so that - 'from worker_app.db import SessionLocal' resolves correctly.""" - from types import ModuleType - - _fresh_db = ModuleType("worker_app.db") - _fresh_db.SessionLocal = lambda *a, **kw: mock_session - return patch.dict(sys.modules, {"worker_app.db": _fresh_db}) - - -class TestLoadTemplateSegmentDurations: - """_load_template_segment_durations 单元测试 (covers lines 198-226).""" - - def test_empty_template_id(self): - """空 template_id 直接返回空列表。""" - from worker_app.tasks.generation import _load_template_segment_durations - - result = _load_template_segment_durations("") - assert result == [] - - def test_loads_durations_ordered(self): - """按 segment_order 排序返回 duration_max 列表。""" - from worker_app.tasks.generation import _load_template_segment_durations - - mock_seg1 = MagicMock(duration_max=5.0) - mock_seg2 = MagicMock(duration_max=8.0) - mock_seg3 = MagicMock(duration_max=3.0) - - mock_query = MagicMock() - mock_query.filter.return_value.order_by.return_value.all.return_value = [ - mock_seg1, - mock_seg2, - mock_seg3, - ] - mock_session = MagicMock() - mock_session.query.return_value = mock_query - - with _patch_session_local(mock_session): - result = _load_template_segment_durations("tpl_123") - - assert result == [5.0, 8.0, 3.0] - - def test_filters_zero_and_negative(self): - """duration_max <= 0 的 segment 被过滤。""" - from worker_app.tasks.generation import _load_template_segment_durations - - mock_seg_valid = MagicMock(duration_max=5.0) - mock_seg_zero = MagicMock(duration_max=0.0) - mock_seg_none = MagicMock(duration_max=None) - - mock_query = MagicMock() - mock_query.filter.return_value.order_by.return_value.all.return_value = [ - mock_seg_valid, - mock_seg_zero, - mock_seg_none, - ] - mock_session = MagicMock() - mock_session.query.return_value = mock_query - - with _patch_session_local(mock_session): - result = _load_template_segment_durations("tpl_456") - - assert result == [5.0] - - def test_db_error_returns_empty(self): - """数据库异常返回空列表,不抛出。""" - from types import ModuleType - - from worker_app.tasks.generation import _load_template_segment_durations - - _err_db = ModuleType("worker_app.db") - - def _raise(*a, **kw): - raise Exception("DB down") - - _err_db.SessionLocal = _raise - with patch.dict(sys.modules, {"worker_app.db": _err_db}): - result = _load_template_segment_durations("tpl_789") - - assert result == [] - - def test_empty_segments_returns_empty(self): - """没有 segment 时返回空列表。""" - from worker_app.tasks.generation import _load_template_segment_durations - - mock_query = MagicMock() - mock_query.filter.return_value.order_by.return_value.all.return_value = [] - mock_session = MagicMock() - mock_session.query.return_value = mock_query - - with _patch_session_local(mock_session): - result = _load_template_segment_durations("tpl_empty") - - assert result == [] - - -class TestDurationCappingInBuildPlan: - """_build_plan_and_clips_from_task 时长约束测试 (covers lines 322-333).""" - - def _make_temp_video(self, tmpdir: Path, name: str = "v.mp4") -> Path: - p = tmpdir / name - p.write_bytes(b"\x00" * 100) - return p - - def test_clips_capped_by_segment_max(self): - """clip 时长超过 segment duration_max 时截断。""" - import tempfile - - from worker_app.tasks.generation import _build_plan_and_clips_from_task - - with tempfile.TemporaryDirectory() as tmpdir: - tmpdir = Path(tmpdir) - paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)] - - with patch("worker_app.tasks.generation.probe_duration", return_value=30.0): - with patch( - "worker_app.tasks.generation._load_template_segment_durations", - return_value=[5.0, 4.0, 3.0], - ): - with patch( - "worker_app.tasks.generation._load_template_clip_configs", - return_value=[], - ): - _, clips, _ = _build_plan_and_clips_from_task( - task_id="test_cap", - downloaded_paths=paths, - mode="one_take", - template_id="tpl_test", - ) - - assert clips[0].duration == 5.0 - assert clips[1].duration == 4.0 - assert clips[2].duration == 3.0 - - def test_clips_not_capped_when_under_max(self): - """clip 时长小于 segment duration_max 时不截断。""" - import tempfile - - from worker_app.tasks.generation import _build_plan_and_clips_from_task - - with tempfile.TemporaryDirectory() as tmpdir: - tmpdir = Path(tmpdir) - paths = [self._make_temp_video(tmpdir)] - - with patch("worker_app.tasks.generation.probe_duration", return_value=3.0): - with patch( - "worker_app.tasks.generation._load_template_segment_durations", - return_value=[5.0], - ): - with patch( - "worker_app.tasks.generation._load_template_clip_configs", - return_value=[], - ): - _, clips, _ = _build_plan_and_clips_from_task( - task_id="test_no_cap", - downloaded_paths=paths, - mode="one_take", - template_id="tpl_test", - ) - - assert clips[0].duration == 3.0 - - def test_no_capping_without_template(self): - """无 template_id 时不截断。""" - import tempfile - - from worker_app.tasks.generation import _build_plan_and_clips_from_task - - with tempfile.TemporaryDirectory() as tmpdir: - tmpdir = Path(tmpdir) - paths = [self._make_temp_video(tmpdir)] - - with patch("worker_app.tasks.generation.probe_duration", return_value=30.0): - _, clips, _ = _build_plan_and_clips_from_task( - task_id="test_no_tpl", - downloaded_paths=paths, - mode="one_take", - template_id="", - ) - - assert clips[0].duration == 30.0 - - def test_partial_segments_only_caps_matching(self): - """segment 数量少于 clip 时,只截断有对应 segment 的 clip。""" - import tempfile - - from worker_app.tasks.generation import _build_plan_and_clips_from_task - - with tempfile.TemporaryDirectory() as tmpdir: - tmpdir = Path(tmpdir) - paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)] - - with patch("worker_app.tasks.generation.probe_duration", return_value=20.0): - with patch( - "worker_app.tasks.generation._load_template_segment_durations", - return_value=[5.0], # only 1 segment for 3 clips - ): - with patch( - "worker_app.tasks.generation._load_template_clip_configs", - return_value=[], - ): - _, clips, _ = _build_plan_and_clips_from_task( - task_id="test_partial", - downloaded_paths=paths, - mode="one_take", - template_id="tpl_test", - ) - - assert clips[0].duration == 5.0 # capped - assert clips[1].duration == 20.0 # not capped (no matching segment) - assert clips[2].duration == 20.0 # not capped diff --git a/tests/unit/test_oneclick_gen_p0_fixes.py b/tests/unit/test_oneclick_gen_p0_fixes.py index 73cf4f1fe..05aa2b60c 100755 --- a/tests/unit/test_oneclick_gen_p0_fixes.py +++ b/tests/unit/test_oneclick_gen_p0_fixes.py @@ -1,7 +1,6 @@ """P0/P1 修复单元测试 — 一键生成 P0 问题 + P1 校验. 覆盖: - P0-1: _download_library_assets 双模式查询(asset_library_id / project_id) P0-2: OSS 上传失败抛异常 + URL 可访问性校验 P0-3: FFmpeg 失败时完整 stderr 日志 P1: template_id 存在性校验 + asset_ids 归属校验 @@ -26,130 +25,6 @@ if str(_WORKER_ROOT) not in sys.path: # ── P0-1: _download_library_assets ──────────────────────────────────────────── -class TestDownloadLibraryAssets: - """P0-1: 素材下载双模式 + 错误处理.""" - - def _make_asset(self, id_: str, file_url: str, project_id: str = "p1", library_id: str = "lib1"): - mock = MagicMock() - mock.id = id_ - mock.file_url = file_url - mock.name = f"asset_{id_}" - mock.project_id = project_id - mock.asset_library_id = library_id - return mock - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_asset_library_mode(self, mock_download, mock_session_factory): - """素材库模式:按 asset_library_id 查询.""" - from worker_app.tasks.generation import _download_library_assets - - session = MagicMock() - mock_session_factory.return_value = session - query = MagicMock() - session.query.return_value = query - filter_result = MagicMock() - query.filter.return_value = filter_result - in_filter = MagicMock() - filter_result.filter.return_value = in_filter - assets = [self._make_asset("a1", "video/a1.mp4")] - in_filter.order_by.return_value.all.return_value = assets - - mock_download.return_value = True - - with patch("worker_app.tasks.generation.AssetModel", create=True): - result = _download_library_assets( - Path("/tmp/test"), - asset_library_id="lib1", - ) - - assert len(result) == 1 - mock_download.assert_called_once() - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_project_mode(self, mock_download, mock_session_factory): - """项目级模式:asset_library_id 为空时按 project_id 查询.""" - from worker_app.tasks.generation import _download_library_assets - - session = MagicMock() - mock_session_factory.return_value = session - query = MagicMock() - session.query.return_value = query - filter_result = MagicMock() - query.filter.return_value = filter_result - proj_filter = MagicMock() - filter_result.filter.return_value = proj_filter - assets = [self._make_asset("a1", "video/a1.mp4", project_id="proj1")] - proj_filter.order_by.return_value.all.return_value = assets - - mock_download.return_value = True - - result = _download_library_assets( - Path("/tmp/test"), - project_id="proj1", - ) - - assert len(result) == 1 - - def test_both_empty_raises(self): - """asset_library_id 和 project_id 都为空时抛 ValueError.""" - from worker_app.tasks.generation import _download_library_assets - - with pytest.raises(ValueError, match="至少需要提供一个"): - _download_library_assets(Path("/tmp/test")) - - @patch("worker_app.tasks.generation.SessionLocal") - def test_no_assets_found_raises(self, mock_session_factory): - """查不到素材时抛 RuntimeError.""" - from worker_app.tasks.generation import _download_library_assets - - session = MagicMock() - mock_session_factory.return_value = session - query = MagicMock() - session.query.return_value = query - filter_result = MagicMock() - query.filter.return_value = filter_result - in_filter = MagicMock() - filter_result.filter.return_value = in_filter - in_filter.order_by.return_value.all.return_value = [] - - with pytest.raises(RuntimeError, match="未找到视频素材"): - _download_library_assets( - Path("/tmp/test"), - asset_library_id="lib1", - ) - - @patch("worker_app.tasks.generation.SessionLocal") - @patch("worker_app.tasks.generation.download_asset") - def test_all_asset_ids_fail_raises(self, mock_download, mock_session_factory): - """指定 asset_ids 但全部下载失败时抛 RuntimeError.""" - from worker_app.tasks.generation import _download_library_assets - - session = MagicMock() - mock_session_factory.return_value = session - query = MagicMock() - session.query.return_value = query - filter_result = MagicMock() - query.filter.return_value = filter_result - id_filter = MagicMock() - filter_result.filter.return_value = id_filter - assets = [self._make_asset("a1", "video/a1.mp4")] - id_filter.order_by.return_value.all.return_value = assets - - mock_download.return_value = False # 全部下载失败 - - with pytest.raises(RuntimeError, match="素材下载失败"): - _download_library_assets( - Path("/tmp/test"), - asset_library_id="lib1", - asset_ids=["a1"], - ) - - -# ── P0-2: OSS 上传 + URL 校验 ──────────────────────────────────────────────── - - class TestOSSUploadAndVerify: """P0-2: OSS 上传失败抛异常 + URL 可访问性校验.""" @@ -483,125 +358,3 @@ class TestTemplateClipEffectMapping: assert result == {} -class TestTemplatePlanConfigLoading: - """验证从模板加载 plan 级配置(BGM、字幕、标题)的逻辑。""" - - def _mock_template( - self, - title_config=None, - subtitle_config=None, - bgm_config=None, - is_active=True, - ): - template = MagicMock() - template.id = "tmpl_001" - template.name = "Test Template" - template.is_active = is_active - template.title_config = title_config or {} - template.subtitle_config = subtitle_config or {} - template.bgm_config = bgm_config or {} - return template - - def _mock_session(self, template): - session = MagicMock() - - # EditTemplateModel 查询返回 None(走旧模板系统 fallback) - edit_query = MagicMock() - edit_filter = MagicMock() - edit_query.filter.return_value = edit_filter - edit_filter.first.return_value = None - - # TemplateModel 查询返回 template(旧模板系统) - old_query = MagicMock() - old_filter = MagicMock() - old_query.filter.return_value = old_filter - old_filter.first.return_value = template - - def _query_side_effect(model): - name = getattr(model, "__name__", "") - if "EditTemplate" in name: - return edit_query - return old_query - - session.query.side_effect = _query_side_effect - return session - - def test_load_template_config_assembles_three_fields(self): - """模板的三个独立字段正确组装成 plan.config 格式。""" - from worker_app.tasks.generation import _load_template_plan_config - - title_cfg = {"enabled": True, "text": "我的标题", "font_size": 36} - subtitle_cfg = {"enabled": True, "auto_generated": True, "language": "zh"} - bgm_cfg = {"enabled": True, "preset_id": "bgm-001", "volume": 0.5} - - template = self._mock_template( - title_config=title_cfg, - subtitle_config=subtitle_cfg, - bgm_config=bgm_cfg, - ) - session = self._mock_session(template) - - with patch("worker_app.tasks.generation.SessionLocal", return_value=session): - result = _load_template_plan_config("tmpl_001") - - assert result["title"] == title_cfg - assert result["subtitle"] == subtitle_cfg - assert result["bgm"] == bgm_cfg - - def test_load_template_config_empty_template_returns_empty(self): - """模板三个字段都为空时返回空 dict。""" - from worker_app.tasks.generation import _load_template_plan_config - - template = self._mock_template() - session = self._mock_session(template) - - with patch("worker_app.tasks.generation.SessionLocal", return_value=session): - result = _load_template_plan_config("tmpl_001") - - assert result == {} - - def test_load_template_config_only_bgm(self): - """只有 BGM 配置时只返回 bgm 字段。""" - from worker_app.tasks.generation import _load_template_plan_config - - bgm_cfg = {"enabled": True, "audio_url": "https://example.com/bgm.mp3"} - template = self._mock_template(bgm_config=bgm_cfg) - session = self._mock_session(template) - - with patch("worker_app.tasks.generation.SessionLocal", return_value=session): - result = _load_template_plan_config("tmpl_001") - - assert "bgm" in result - assert result["bgm"] == bgm_cfg - assert "title" not in result - assert "subtitle" not in result - - def test_load_template_config_empty_template_id(self): - """空 template_id 直接返回空 dict。""" - from worker_app.tasks.generation import _load_template_plan_config - - result = _load_template_plan_config("") - assert result == {} - - result = _load_template_plan_config(None) - assert result == {} - - def test_load_template_config_not_found_returns_empty(self): - """模板不存在时返回空 dict,不抛异常。""" - from worker_app.tasks.generation import _load_template_plan_config - - session = MagicMock() - - def _query_side_effect(model): - q = MagicMock() - f = MagicMock() - q.filter.return_value = f - f.first.return_value = None - return q - - session.query.side_effect = _query_side_effect - - with patch("worker_app.tasks.generation.SessionLocal", return_value=session): - result = _load_template_plan_config("tmpl_nonexist") - - assert result == {} diff --git a/tests/unit/test_preview_title_config.py b/tests/unit/test_preview_title_config.py deleted file mode 100644 index fffbd5cfc..000000000 --- a/tests/unit/test_preview_title_config.py +++ /dev/null @@ -1,190 +0,0 @@ -"""Tests for preview title_config feature. - -验证预览 API 的 title_config 字段和 Worker 的标题配置解析逻辑。 -""" - -from __future__ import annotations - -import json - -import pytest - - -class TestPreviewTitleConfigSchema: - """测试 CreatePreviewGenerationTaskRequest 的 title_config 字段.""" - - def test_title_config_default_empty(self): - """title_config 默认为空 dict.""" - from app.schemas.generation_task import CreatePreviewGenerationTaskRequest - - req = CreatePreviewGenerationTaskRequest( - template_id="test_template", - asset_ids=["asset1"], - ) - assert req.title_config == {} - - def test_title_config_with_text(self): - """传入标题文本.""" - from app.schemas.generation_task import CreatePreviewGenerationTaskRequest - - req = CreatePreviewGenerationTaskRequest( - template_id="test_template", - asset_ids=["asset1"], - title_config={"text": "测试标题"}, - ) - assert req.title_config["text"] == "测试标题" - - def test_title_config_with_full_style(self): - """传入完整标题样式配置.""" - from app.schemas.generation_task import CreatePreviewGenerationTaskRequest - - config = { - "text": "我的视频标题", - "font": "思源黑体", - "font_size": 48, - "font_color": "#ffffff", - "position": "top", - "bold": True, - "stroke": 2, - "shadow": True, - } - req = CreatePreviewGenerationTaskRequest( - template_id="test_template", - asset_ids=["asset1"], - title_config=config, - ) - assert req.title_config["text"] == "我的视频标题" - assert req.title_config["font_size"] == 48 - assert req.title_config["position"] == "top" - - -class TestCommandTitleConfig: - """测试 CreateGenerationTaskCommand 的 title_config 字段.""" - - def test_command_has_title_config(self): - """Command 包含 title_config 字段.""" - from packages.application.generation_tasks import CreateGenerationTaskCommand - - cmd = CreateGenerationTaskCommand( - title_config={"text": "hello", "font_size": 32}, - ) - assert cmd.title_config["text"] == "hello" - assert cmd.title_config["font_size"] == 32 - - def test_command_title_config_default_empty(self): - """Command 的 title_config 默认为空 dict.""" - from packages.application.generation_tasks import CreateGenerationTaskCommand - - cmd = CreateGenerationTaskCommand() - assert cmd.title_config == {} - - -class TestWorkerTitleConfigParsing: - """测试 Worker 渲染时的标题配置解析逻辑.""" - - def test_json_format_parsing(self): - """JSON 格式的 custom_title 能正确解析.""" - config = {"text": "测试标题", "font_size": 48, "font_color": "#ff0000"} - custom_title = json.dumps(config, ensure_ascii=False) - - ct_stripped = custom_title.strip() - parsed = None - if ct_stripped.startswith("{"): - try: - parsed = json.loads(ct_stripped) - except (json.JSONDecodeError, ValueError): - parsed = None - - assert parsed is not None - assert parsed["text"] == "测试标题" - assert parsed["font_size"] == 48 - - def test_plain_text_fallback(self): - """纯文本的 custom_title 不触发 JSON 解析.""" - custom_title = "简单的标题文字" - - ct_stripped = custom_title.strip() - parsed = None - if ct_stripped.startswith("{"): - try: - parsed = json.loads(ct_stripped) - except (json.JSONDecodeError, ValueError): - parsed = None - - assert parsed is None - - def test_invalid_json_fallback(self): - """无效 JSON 的 custom_title 降级为纯文本.""" - custom_title = "{invalid json" - - ct_stripped = custom_title.strip() - parsed = None - if ct_stripped.startswith("{"): - try: - parsed = json.loads(ct_stripped) - except (json.JSONDecodeError, ValueError): - parsed = None - - assert parsed is None - - def test_json_without_text_skipped(self): - """JSON 格式但缺少 text 字段时,跳过标题注入.""" - config = {"font_size": 48} - custom_title = json.dumps(config, ensure_ascii=False) - - ct_stripped = custom_title.strip() - parsed = json.loads(ct_stripped) - title_text = (parsed.get("text") or "").strip() - - assert title_text == "" - - def test_style_key_mapping(self): - """前端字段名正确映射到 ASS 字段名.""" - config = { - "text": "标题", - "font_size": 48, - "font_color": "#ffffff", - "font_preset": "思源黑体", - } - - style_keys = ["font", "font_size", "font_color", "position", "bold", "stroke", "shadow", "font_preset"] - title_cfg = {} - for key in style_keys: - if key in config and config[key] is not None: - mapped_key = { - "font_size": "size", - "font_color": "color", - "font_preset": "font", - }.get(key, key) - title_cfg[mapped_key] = config[key] - - assert title_cfg["size"] == 48 - assert title_cfg["color"] == "#ffffff" - assert title_cfg["font"] == "思源黑体" - - -class TestPreviewRouteTitleConfigPassing: - """测试预览路由正确序列化 title_config 到 custom_title.""" - - def test_title_config_serialization(self): - """title_config 序列化为 JSON 字符串.""" - title_config = { - "text": "我的标题", - "font_size": 32, - "font_color": "#d4a843", - } - serialized = json.dumps(title_config, ensure_ascii=False) - - parsed = json.loads(serialized) - assert parsed["text"] == "我的标题" - assert parsed["font_size"] == 32 - - def test_empty_title_config_produces_empty_string(self): - """空 title_config 时 custom_title 为空字符串.""" - title_config = {} - title_text = (title_config.get("text") or "").strip() - custom_title_value = "" - if title_text: - custom_title_value = json.dumps(title_config, ensure_ascii=False) - - assert custom_title_value == "" diff --git a/tests/unit/test_preview_title_render.py b/tests/unit/test_preview_title_render.py deleted file mode 100644 index c6206e3f9..000000000 --- a/tests/unit/test_preview_title_render.py +++ /dev/null @@ -1,256 +0,0 @@ -"""预览视频标题渲染修复测试 — 覆盖3个断点。 - -断点1: generate_video() → _render_video() 传递 custom_title -断点2: _render_video() 解析 custom_title 并注入 virtual_plan.config["title"] -断点3: generate_ass_from_timeline() ASR路径也渲染标题 -""" - -import json -import time -from pathlib import Path -from unittest.mock import MagicMock, PropertyMock, patch - -import pytest - -# ── 断点2: _render_video 标题注入 ───────────────────────────────────────────── - - -class TestRenderVideoCustomTitleInjection: - """验证 _render_video 正确接收并注入 custom_title 到 virtual_plan.config['title']。""" - - @pytest.fixture - def sample_custom_title(self): - """模拟前端发送的 custom_title JSON(含 font_size/font_color)。""" - return json.dumps( - { - "text": "测试标题", - "font": "思源黑体", - "font_size": 30, - "font_color": "#FF0000", - "position": "top", - "bold": True, - "stroke": True, - "shadow": False, - }, - ensure_ascii=False, - ) - - def _call_render_video_with_capture(self, custom_title, template_config=None, tmp_path=None): - """调用 _render_video,在 RenderAdapter 处中断并捕获 virtual_plan.config。""" - from worker_app.tasks.generation import _render_video - - captured_config = {} - - class FakePlan: - def __init__(self): - self.config = {} - self.id = "test-plan" - - fake_plan = FakePlan() - - def capture_and_raise(*args, **kwargs): - # 此时 title 已注入到 fake_plan.config - captured_config.update(fake_plan.config or {}) - raise RuntimeError("STOP_HERE") - - with ( - patch("worker_app.tasks.generation._build_plan_and_clips_from_task") as mock_build, - patch("worker_app.tasks.generation._load_template_plan_config", return_value=template_config), - patch("worker_app.tasks.generation.time.monotonic", side_effect=[0.0, 1.0]), - patch("video_processing.render_adapter.RenderAdapter") as mock_adapter_cls, - ): - - mock_build.return_value = (fake_plan, [], {}) - mock_adapter_cls.side_effect = capture_and_raise - - with pytest.raises(RuntimeError, match="STOP_HERE"): - _render_video( - task_id="test-task", - downloaded_videos=[tmp_path / "v1.mp4"] if tmp_path else [Path("/tmp/v1.mp4")], - voice_path=None, - editing_mode=MagicMock(value="one_take"), - project_id="proj-1", - template_id="tpl-1", - user_id="user-1", - temp_path=tmp_path or Path("/tmp"), - output_name="test_output", - resolution="1280x720", - bgm_config={}, - voice_ids=[], - custom_title=custom_title, - ) - - return captured_config - - def test_custom_title_injected_into_plan_config(self, sample_custom_title, tmp_path): - """custom_title JSON 应被解析并注入 virtual_plan.config['title']。""" - config = self._call_render_video_with_capture(sample_custom_title, tmp_path=tmp_path) - - assert "title" in config - title_cfg = config["title"] - assert title_cfg["text"] == "测试标题" - # 字段归一化: font_size → size - assert title_cfg["size"] == 30 - # 字段归一化: font_color → color - assert title_cfg["color"] == "#FF0000" - - def test_custom_title_overrides_template_title(self, sample_custom_title, tmp_path): - """用户自定义标题应覆盖模板默认标题。""" - template_config = {"title": {"text": "模板默认标题", "size": 24}} - config = self._call_render_video_with_capture( - sample_custom_title, template_config=template_config, tmp_path=tmp_path - ) - - # 用户标题应覆盖模板标题 - assert config["title"]["text"] == "测试标题" - assert config["title"]["size"] == 30 - - def test_empty_custom_title_no_injection(self, tmp_path): - """空 custom_title 不应注入 title 字段。""" - config = self._call_render_video_with_capture("", tmp_path=tmp_path) - assert "title" not in config - - def test_malformed_custom_title_gracefully_ignored(self, tmp_path): - """非法 JSON 不应崩溃,应跳过注入。""" - config = self._call_render_video_with_capture("{invalid json!!!", tmp_path=tmp_path) - assert "title" not in config - - -# ── 断点3: generate_ass_from_timeline ASR路径支持标题 ────────────────────────── - - -class TestGenerateAssFromTimelineWithTitle: - """验证 generate_ass_from_timeline 在有标题时生成包含 TitleStyle 的 ASS。""" - - def test_title_included_in_ass_output(self, tmp_path): - """有 title_text 时,ASS 输出应包含 TitleStyle 和标题事件。""" - from video_processing.subtitle_generator import generate_ass_from_timeline - - from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline - - timeline = SubtitleTimeline( - segments=[ - SubtitleSegment(start=0.0, end=2.0, text="你好世界"), - ] - ) - - ass_path = tmp_path / "test.ass" - result = generate_ass_from_timeline( - ass_path, - timeline, - video_width=1280, - video_height=720, - video_duration=10.0, - subtitle_config={"font": "思源黑体", "size": 24}, - title_text="我的标题", - title_config={"font": "思源黑体", "size": 36, "color": "#FFFFFF", "position": "top"}, - ) - - content = result.read_text(encoding="utf-8") - # 应包含 TitleStyle - assert "TitleStyle" in content - # 应包含标题文本 - assert "我的标题" in content - # 也应包含 ASR 字幕 - assert "你好世界" in content - - def test_no_title_no_title_style(self, tmp_path): - """无标题时,ASS 输出不应包含 TitleStyle。""" - from video_processing.subtitle_generator import generate_ass_from_timeline - - from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline - - timeline = SubtitleTimeline( - segments=[ - SubtitleSegment(start=0.0, end=2.0, text="只有字幕"), - ] - ) - - ass_path = tmp_path / "test.ass" - result = generate_ass_from_timeline( - ass_path, - timeline, - video_width=1280, - video_height=720, - video_duration=10.0, - subtitle_config={}, - title_text="", - title_config={}, - ) - - content = result.read_text(encoding="utf-8") - assert "TitleStyle" not in content - assert "只有字幕" in content - - def test_title_field_normalization_in_ass(self, tmp_path): - """前端字段名 font_size/font_color 应被正确归一化。""" - from video_processing.subtitle_generator import generate_ass_from_timeline - - from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline - - timeline = SubtitleTimeline(segments=[SubtitleSegment(start=0.0, end=2.0, text="test")]) - - ass_path = tmp_path / "test.ass" - result = generate_ass_from_timeline( - ass_path, - timeline, - video_width=1280, - video_height=720, - video_duration=10.0, - subtitle_config={}, - title_text="归一化测试", - title_config={ - "font_size": 30, # 前端字段名 - "font_color": "#FF0000", # 前端字段名 - "position": "top", - }, - ) - - content = result.read_text(encoding="utf-8") - assert "TitleStyle" in content - assert "归一化测试" in content - - def test_title_boolean_stroke_shadow_compat(self, tmp_path): - """boolean stroke/shadow 应被兼容处理。""" - from video_processing.subtitle_generator import generate_ass_from_timeline - - from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline - - timeline = SubtitleTimeline(segments=[SubtitleSegment(start=0.0, end=2.0, text="test")]) - - ass_path = tmp_path / "test.ass" - result = generate_ass_from_timeline( - ass_path, - timeline, - video_width=1280, - video_height=720, - video_duration=10.0, - subtitle_config={}, - title_text="描边测试", - title_config={ - "size": 36, - "stroke": True, # boolean - "shadow": False, # boolean - }, - ) - - content = result.read_text(encoding="utf-8") - assert "TitleStyle" in content - assert "描边测试" in content - - -# ── 断点1: _render_video 签名包含 custom_title ──────────────────────────────── - - -class TestRenderVideoSignature: - """验证 _render_video 函数签名正确。""" - - def test_custom_title_parameter_exists(self): - """_render_video 应有 custom_title 参数,默认空字符串。""" - import inspect - - from worker_app.tasks.generation import _render_video - - sig = inspect.signature(_render_video) - assert "custom_title" in sig.parameters - assert sig.parameters["custom_title"].default == "" diff --git a/tests/unit/test_preview_voice_ids_fallback.py b/tests/unit/test_preview_voice_ids_fallback.py deleted file mode 100644 index acac09d1a..000000000 --- a/tests/unit/test_preview_voice_ids_fallback.py +++ /dev/null @@ -1,151 +0,0 @@ -"""Tests for voice_ids fallback in _download_all_assets. - -When voice_library_id is empty but voice_ids is non-empty, the Worker -should fallback to voice_ids[0] as the audio asset_id. -""" - -from pathlib import Path -from unittest.mock import MagicMock, patch - -import pytest - - -class TestDownloadAllAssetsVoiceIdsFallback: - """_download_all_assets 配音下载 fallback 逻辑测试。""" - - @patch("worker_app.tasks.generation._download_voice_asset") - @patch("worker_app.tasks.generation._download_library_assets") - def test_voice_library_id_takes_priority(self, mock_download_videos, mock_download_voice, tmp_path): - """voice_library_id 存在时优先使用,不 fallback 到 voice_ids。""" - from worker_app.tasks.generation import _download_all_assets - - mock_download_videos.return_value = [tmp_path / "v1.mp4"] - mock_download_voice.return_value = True - - videos, audio = _download_all_assets( - temp_path=tmp_path, - asset_library_id="lib-1", - project_id="proj-1", - task_asset_ids=["a1"], - voice_library_id="voice-lib-123", - task_id="task-1", - voice_ids=["voice-ids-456"], - ) - - assert audio is not None - mock_download_voice.assert_called_once() - call_args = mock_download_voice.call_args - assert call_args[0][0] == "voice-lib-123" # first positional arg - - @patch("worker_app.tasks.generation._download_voice_asset") - @patch("worker_app.tasks.generation._download_library_assets") - def test_fallback_to_voice_ids_when_voice_library_id_empty( - self, mock_download_videos, mock_download_voice, tmp_path - ): - """voice_library_id 为空时 fallback 到 voice_ids[0]。""" - from worker_app.tasks.generation import _download_all_assets - - mock_download_videos.return_value = [tmp_path / "v1.mp4"] - mock_download_voice.return_value = True - - videos, audio = _download_all_assets( - temp_path=tmp_path, - asset_library_id="lib-1", - project_id="proj-1", - task_asset_ids=["a1"], - voice_library_id="", # 空字符串 - task_id="task-2", - voice_ids=["voice-asset-789"], - ) - - assert audio is not None - mock_download_voice.assert_called_once() - call_args = mock_download_voice.call_args - assert call_args[0][0] == "voice-asset-789" - - @patch("worker_app.tasks.generation._download_voice_asset") - @patch("worker_app.tasks.generation._download_library_assets") - def test_no_audio_when_both_empty(self, mock_download_videos, mock_download_voice, tmp_path): - """voice_library_id 和 voice_ids 都为空时,不下载音频。""" - from worker_app.tasks.generation import _download_all_assets - - mock_download_videos.return_value = [tmp_path / "v1.mp4"] - - videos, audio = _download_all_assets( - temp_path=tmp_path, - asset_library_id="lib-1", - project_id="proj-1", - task_asset_ids=["a1"], - voice_library_id="", - task_id="task-3", - voice_ids=[], - ) - - assert audio is None - mock_download_voice.assert_not_called() - - @patch("worker_app.tasks.generation._download_voice_asset") - @patch("worker_app.tasks.generation._download_library_assets") - def test_no_audio_when_voice_ids_none(self, mock_download_videos, mock_download_voice, tmp_path): - """voice_ids 为 None 时,不触发 fallback。""" - from worker_app.tasks.generation import _download_all_assets - - mock_download_videos.return_value = [tmp_path / "v1.mp4"] - - videos, audio = _download_all_assets( - temp_path=tmp_path, - asset_library_id="lib-1", - project_id="proj-1", - task_asset_ids=["a1"], - voice_library_id="", - task_id="task-4", - voice_ids=None, - ) - - assert audio is None - mock_download_voice.assert_not_called() - - @patch("worker_app.tasks.generation._download_voice_asset") - @patch("worker_app.tasks.generation._download_library_assets") - def test_voice_library_id_empty_string_fallback(self, mock_download_videos, mock_download_voice, tmp_path): - """voice_library_id 为空字符串且 voice_ids 有多个元素时,取第一个。""" - from worker_app.tasks.generation import _download_all_assets - - mock_download_videos.return_value = [tmp_path / "v1.mp4"] - mock_download_voice.return_value = True - - videos, audio = _download_all_assets( - temp_path=tmp_path, - asset_library_id="lib-1", - project_id="proj-1", - task_asset_ids=["a1"], - voice_library_id="", - task_id="task-5", - voice_ids=["first-id", "second-id", "third-id"], - ) - - assert audio is not None - call_args = mock_download_voice.call_args - assert call_args[0][0] == "first-id" - - @patch("worker_app.tasks.generation._download_voice_asset") - @patch("worker_app.tasks.generation._download_library_assets") - def test_backward_compat_no_voice_ids_param(self, mock_download_videos, mock_download_voice, tmp_path): - """不传 voice_ids 参数时,行为与之前一致(向后兼容)。""" - from worker_app.tasks.generation import _download_all_assets - - mock_download_videos.return_value = [tmp_path / "v1.mp4"] - mock_download_voice.return_value = True - - # 不传 voice_ids - videos, audio = _download_all_assets( - temp_path=tmp_path, - asset_library_id="lib-1", - project_id="proj-1", - task_asset_ids=["a1"], - voice_library_id="voice-lib-999", - task_id="task-6", - ) - - assert audio is not None - mock_download_voice.assert_called_once_with("voice-lib-999", tmp_path / "voice.mp3") diff --git a/tests/unit/test_render_subtitles_pure.py b/tests/unit/test_render_subtitles_pure.py index 2b4c53e04..af335da11 100644 --- a/tests/unit/test_render_subtitles_pure.py +++ b/tests/unit/test_render_subtitles_pure.py @@ -280,7 +280,7 @@ class TestGenerateAssSubtitles: content = output.read_text(encoding="utf-8") assert "有字幕" in content - def test_custom_title_color(self, tmp_path): + def test_title_color_overlay(self, tmp_path): """自定义标题颜色.""" output = tmp_path / "color.ass" result = generate_ass_subtitles( diff --git a/tests/unit/test_three_bugs_fix.py b/tests/unit/test_three_bugs_fix.py index 23cc0ef12..0b76cf8dc 100644 --- a/tests/unit/test_three_bugs_fix.py +++ b/tests/unit/test_three_bugs_fix.py @@ -214,7 +214,6 @@ class TestPlanIdFallbackExecution: t.output_width = 1280 t.output_height = 720 t.cover_url = "" - t.custom_title = "" t.title_config = {} t.logs = "[]" t.status = "pending" @@ -255,7 +254,6 @@ class TestPlanIdFallbackExecution: req.output_width = 0 req.output_height = 0 req.cover_url = "" - req.custom_title = "" req.title_config = {} req.project_id = None req.asset_library_id = None @@ -295,7 +293,6 @@ class TestPlanIdFallbackExecution: output_width=1280, output_height=720, cover_url="", - custom_title="", title_config={}, logs="[]", status="pending", diff --git a/tests/unit/test_worker_cover_meta_and_status.py b/tests/unit/test_worker_cover_meta_and_status.py index 9d6fa1ed3..a9baa66f8 100644 --- a/tests/unit/test_worker_cover_meta_and_status.py +++ b/tests/unit/test_worker_cover_meta_and_status.py @@ -46,13 +46,10 @@ class TestAssetAnalysesDoesNotOverwriteStatus: "Use an independent session to update only extra_meta." ) - def test_asset_analyses_uses_independent_session(self): - """asset_analyses 持久化必须用独立 session 查询最新模型再提交。""" + def test_no_stale_repo_update(self): + """旧路径的 _repo.update(gen_task) 模式已随 DEPRECATED 代码一起删除。""" source = _read_source() - assert "_meta_session" in source - assert "GenerationTaskModel" in source - # 必须只更新 extra_meta 字段 - assert 'existing["asset_analyses"]' in source + assert "_repo.update(gen_task)" not in source class TestGenerationTaskModelOrmAttribute: -- 2.54.0 From e8c3d181b1218b0c1f3d33805544da9e9a2b24f2 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 24 Aug 2026 05:47:41 +0000 Subject: [PATCH 2/4] style: auto-format with black + isort + prettier [skip ci-format-check] --- apps/worker/worker_app/tasks/generation.py | 6 +++++- tests/unit/test_1294_preview_voice_injection.py | 2 -- tests/unit/test_oneclick_gen_p0_fixes.py | 2 -- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 1ba1bf399..d94719733 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -158,6 +158,7 @@ from video_processing.oss_helpers import ( upload_to_oss, ) + def _download_voice_asset(voice_library_id: str, local_path: Path) -> bool: """下载配音文件。 @@ -304,6 +305,7 @@ def _verify_url_accessible( logger.warning("URL 可访问性校验最终失败: url=%s error=%s", url, last_error) return False + def _validate_template_exists(template_id: str) -> None: """校验 template_id 是否存在且可用。 @@ -347,6 +349,8 @@ def _validate_template_exists(template_id: str) -> None: raise ValueError(f"模板不存在或已禁用: template_id={template_id}") finally: session.close() + + def _load_task_info(task_id: str) -> dict | None: """从数据库加载 GenerationTask 元数据。 @@ -387,6 +391,7 @@ def _load_task_info(task_id: str) -> dict | None: finally: session.close() + def _upload_and_record( task_id: str, output_path: Path, @@ -861,7 +866,6 @@ def generate_video(self, task_id: str) -> dict: except OSError: logger.warning("[task_id=%s] 清理临时配音文件失败: %s", task_id, voiceover_tmp_path) - else: logger.error( "[task_id=%s] source_edit_plan_id 为空,无法渲染。所有任务必须通过预览 API 创建并关联 EditPlan。", diff --git a/tests/unit/test_1294_preview_voice_injection.py b/tests/unit/test_1294_preview_voice_injection.py index 7401e7d98..7158ae985 100644 --- a/tests/unit/test_1294_preview_voice_injection.py +++ b/tests/unit/test_1294_preview_voice_injection.py @@ -73,5 +73,3 @@ class TestLoadTaskInfoVoiceIds: result = _load_task_info("test_task_id") assert result["voice_ids"] == [] - - diff --git a/tests/unit/test_oneclick_gen_p0_fixes.py b/tests/unit/test_oneclick_gen_p0_fixes.py index 05aa2b60c..147d41dc2 100755 --- a/tests/unit/test_oneclick_gen_p0_fixes.py +++ b/tests/unit/test_oneclick_gen_p0_fixes.py @@ -356,5 +356,3 @@ class TestTemplateClipEffectMapping: result = _extract_intro_outro_from_clip_configs(clip_configs) assert result == {} - - -- 2.54.0 From 70c822e54e01ee9e7facad364bfc30c7df543534 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 24 Aug 2026 14:00:11 +0800 Subject: [PATCH 3/4] =?UTF-8?q?fix:=20=E6=B8=85=E7=90=86=E5=88=A0=E9=99=A4?= =?UTF-8?q?=E6=97=A7=E8=B7=AF=E5=BE=84=E5=90=8E=E7=9A=84=20unused=20import?= =?UTF-8?q?s/variables=20=E5=92=8C=E5=AD=A4=E5=84=BF=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - generation.py: 移除 9 个 unused import (json, tempfile, Any, VirtualClip 等) - generation.py: 注释 3 个 unused variable (asset_library_id, voice_library_id, output_name) - generation_preview.py: 移除 unused import json - test_oneclick_gen_p0_fixes.py: 删除 TestTemplateClipEffectMapping (10个测试) 测试的 _apply_template_clip_effects/_extract_intro_outro_from_clip_configs 已随旧路径删除 单测 13637 passed, 0 failed --- apps/api/app/api/routes/generation_preview.py | 1 - apps/worker/worker_app/tasks/generation.py | 16 +- tests/unit/test_oneclick_gen_p0_fixes.py | 189 ------------------ 3 files changed, 3 insertions(+), 203 deletions(-) diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index 6ae78cb6d..4209df1a3 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -5,7 +5,6 @@ from __future__ import annotations -import json import logging from app.auth import AuthenticatedUser, get_current_user diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index d94719733..2a06eab23 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -13,25 +13,15 @@ from __future__ import annotations -import json import logging -import tempfile import time from pathlib import Path -from typing import Any from video_processing.ffmpeg_utils import probe_duration from worker_app.celery_app import celery_app from worker_app.db import SessionLocal -from worker_app.tasks.generation_plan_builder import VirtualClip as _VirtualClip -from worker_app.tasks.generation_plan_builder import VirtualPlan as _VirtualPlan -from worker_app.tasks.generation_plan_builder import apply_template_clip_effects as _apply_template_clip_effects from worker_app.tasks.generation_plan_builder import build_error_info as _build_error_info -from worker_app.tasks.generation_plan_builder import ( - extract_intro_outro_from_clip_configs as _extract_intro_outro_from_clip_configs, -) -from packages.domain.bgm_utils import merge_bgm_config OUTPUT_WIDTH = 1280 OUTPUT_HEIGHT = 720 @@ -644,8 +634,8 @@ def generate_video(self, task_id: str) -> dict: return {"status": "failed", "error": f"generation task {task_id} not found"} project_id = task_info["project_id"] - asset_library_id = task_info["asset_library_id"] - voice_library_id = task_info["voice_library_id"] + # asset_library_id, voice_library_id 已在 _render_from_edit_plan 内部重新获取 + # (不再需要在 generate_video 顶层解包) template_id = task_info["template_id"] task_asset_ids = task_info["task_asset_ids"] batch_id = task_info["batch_id"] @@ -679,7 +669,7 @@ def generate_video(self, task_id: str) -> dict: except ValueError: editing_mode = EditingMode.ONE_TAKE - output_name = f"generated-{task_id}.mp4" + # output_name 不再使用,渲染路径内部自行处理文件名 try: if template_id: diff --git a/tests/unit/test_oneclick_gen_p0_fixes.py b/tests/unit/test_oneclick_gen_p0_fixes.py index 147d41dc2..065e12e0c 100755 --- a/tests/unit/test_oneclick_gen_p0_fixes.py +++ b/tests/unit/test_oneclick_gen_p0_fixes.py @@ -167,192 +167,3 @@ class TestP1Validations: # ── P1: 一键生成 clip 级效果层映射 ─────────────────────────────────────────── -class TestTemplateClipEffectMapping: - """P1: 模板 clip 级效果层映射到一键生成素材 clips.""" - - def _make_virtual_clip(self, idx: int, clip_type: str = "main", config: dict | None = None): - from dataclasses import dataclass, field - - _clip_type_val = clip_type - - @dataclass - class FakeClip: - id: str = f"vc_{idx:03d}" - plan_id: str = "task_001" - clip_type: str = _clip_type_val - order: int = idx - asset_id: str = f"asset_{idx}" - duration: float = 5.0 - transition_effect: str = "cut" - transition_duration: float = 0.0 - playback_speed: float = 1.0 - config: dict = field(default_factory=dict) - - return FakeClip(config=config or {}) - - def _make_template_clip_config(self, clip_type: str = "main", transition: str = "cut", config: dict | None = None): - mock = MagicMock() - mock.clip_type = clip_type - mock.transition_effect = transition - mock.config = config or {} - mock.default_duration = 3.0 - mock.text_template = "" - return mock - - def test_transition_effect_mapped(self): - """转场效果正确映射到素材 clips.""" - from worker_app.tasks.generation import _apply_template_clip_effects - - clips = [self._make_virtual_clip(i) for i in range(3)] - clip_configs = [ - self._make_template_clip_config("main", transition="fade"), - self._make_template_clip_config("main", transition="dissolve"), - ] - - _apply_template_clip_effects(clips, clip_configs, "one_take") - - # 前两个按顺序映射,第三个用最后一个模板配置 - assert clips[0].transition_effect == "fade" - assert clips[1].transition_effect == "dissolve" - assert clips[2].transition_effect == "dissolve" # 复用最后一个 - - def test_color_grade_mapped(self): - """滤镜配置正确映射到 clip.config.color_grade.""" - from worker_app.tasks.generation import _apply_template_clip_effects - - clips = [self._make_virtual_clip(i) for i in range(2)] - clip_configs = [ - self._make_template_clip_config( - "main", config={"color_grade": {"enabled": True, "filter": "vintage", "brightness": 0.1}} - ), - ] - - _apply_template_clip_effects(clips, clip_configs, "one_take") - - assert clips[0].config["color_grade"]["filter"] == "vintage" - assert clips[0].config["color_grade"]["brightness"] == 0.1 - # 第二个素材复用第一个模板配置 - assert clips[1].config["color_grade"]["filter"] == "vintage" - - def test_existing_config_preserved(self): - """已有 clip.config 内容(如 role)被保留.""" - from worker_app.tasks.generation import _apply_template_clip_effects - - clips = [self._make_virtual_clip(0, config={"role": "b_roll"})] - clip_configs = [ - self._make_template_clip_config("main", config={"color_grade": {"enabled": True, "filter": "warm"}}), - ] - - _apply_template_clip_effects(clips, clip_configs, "voice_over") - - assert clips[0].config["role"] == "b_roll" # 保留原有配置 - assert clips[0].config["color_grade"]["filter"] == "warm" # 新增滤镜配置 - - def test_empty_clip_configs_no_change(self): - """空模板配置时 clips 保持不变.""" - from worker_app.tasks.generation import _apply_template_clip_effects - - clips = [self._make_virtual_clip(i) for i in range(2)] - _apply_template_clip_effects(clips, [], "one_take") - - assert clips[0].transition_effect == "cut" - assert clips[1].transition_effect == "cut" - - def test_cut_transition_not_overwritten(self): - """模板转场为 cut 时不覆盖(保持默认).""" - from worker_app.tasks.generation import _apply_template_clip_effects - - clips = [self._make_virtual_clip(0)] - clips[0].transition_effect = "fade" # 已有非默认值 - clip_configs = [ - self._make_template_clip_config("main", transition="cut"), - ] - - _apply_template_clip_effects(clips, clip_configs, "one_take") - - # 模板是 cut 时,保留原有值(避免无意义覆盖) - assert clips[0].transition_effect == "fade" - - def test_transition_duration_mapped(self): - """转场时长(transition_duration)从模板 config 正确映射到 clip.""" - from worker_app.tasks.generation import _apply_template_clip_effects - - clips = [self._make_virtual_clip(i) for i in range(3)] - clip_configs = [ - self._make_template_clip_config("main", transition="fade", config={"transition_duration": 0.8}), - self._make_template_clip_config("main", transition="dissolve", config={"transition_duration": 1.2}), - ] - - _apply_template_clip_effects(clips, clip_configs, "one_take") - - # 前两个按顺序映射,第三个复用最后一个 - assert clips[0].transition_effect == "fade" - assert clips[0].transition_duration == 0.8 - assert clips[1].transition_effect == "dissolve" - assert clips[1].transition_duration == 1.2 - assert clips[2].transition_effect == "dissolve" - assert clips[2].transition_duration == 1.2 - - def test_transition_duration_ignored_for_cut(self): - """模板转场为 cut 时,transition_duration 不生效(保持默认0).""" - from worker_app.tasks.generation import _apply_template_clip_effects - - clips = [self._make_virtual_clip(0)] - clip_configs = [ - self._make_template_clip_config("main", transition="cut", config={"transition_duration": 0.5}), - ] - - _apply_template_clip_effects(clips, clip_configs, "one_take") - - # cut 转场不映射,transition_duration 也不应用 - assert clips[0].transition_duration == 0.0 - - def test_transition_duration_invalid_value_skipped(self): - """transition_duration 为无效值时安全跳过.""" - from worker_app.tasks.generation import _apply_template_clip_effects - - clips = [self._make_virtual_clip(0)] - clip_configs = [ - self._make_template_clip_config("main", transition="fade", config={"transition_duration": "abc"}), - ] - - _apply_template_clip_effects(clips, clip_configs, "one_take") - - assert clips[0].transition_effect == "fade" - assert clips[0].transition_duration == 0.0 # 无效值保持默认 - - def test_intro_outro_extracted(self): - """intro/outro 类型 clip_config 正确提取为 plan 级 intro_outro 配置.""" - from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs - - clip_configs = [ - self._make_template_clip_config("intro", config={"intro_type": "text", "intro_text_color": "#ffffff"}), - self._make_template_clip_config("main"), - self._make_template_clip_config("outro", config={"outro_type": "follow", "outro_follow_text": "关注我们"}), - ] - # 设置 intro/outro 的 text_template - clip_configs[0].text_template = "精彩视频" - clip_configs[0].default_duration = 2.5 - - result = _extract_intro_outro_from_clip_configs(clip_configs) - - assert result["has_intro"] is True - assert result["intro_type"] == "text" - assert result["intro_text"] == "精彩视频" - assert result["intro_duration"] == 2.5 - assert result["intro_text_color"] == "#ffffff" - assert result["has_outro"] is True - assert result["outro_type"] == "follow" - assert result["outro_follow_text"] == "关注我们" - - def test_intro_outro_empty_when_none(self): - """没有 intro/outro 时返回空 dict.""" - from worker_app.tasks.generation import _extract_intro_outro_from_clip_configs - - clip_configs = [ - self._make_template_clip_config("main"), - self._make_template_clip_config("main"), - ] - - result = _extract_intro_outro_from_clip_configs(clip_configs) - assert result == {} -- 2.54.0 From 9fae480e4be627ff7092cae3291deb819233e75b Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 24 Aug 2026 06:04:36 +0000 Subject: [PATCH 4/4] style: auto-format with black + isort + prettier [skip ci-format-check] --- apps/worker/worker_app/tasks/generation.py | 1 - tests/unit/test_oneclick_gen_p0_fixes.py | 2 -- 2 files changed, 3 deletions(-) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 2a06eab23..b5f1f9ea0 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -22,7 +22,6 @@ from worker_app.celery_app import celery_app from worker_app.db import SessionLocal from worker_app.tasks.generation_plan_builder import build_error_info as _build_error_info - OUTPUT_WIDTH = 1280 OUTPUT_HEIGHT = 720 OUTPUT_FPS = 25.0 diff --git a/tests/unit/test_oneclick_gen_p0_fixes.py b/tests/unit/test_oneclick_gen_p0_fixes.py index 065e12e0c..5b8417aa4 100755 --- a/tests/unit/test_oneclick_gen_p0_fixes.py +++ b/tests/unit/test_oneclick_gen_p0_fixes.py @@ -165,5 +165,3 @@ class TestP1Validations: # ── P1: 一键生成 clip 级效果层映射 ─────────────────────────────────────────── - - -- 2.54.0