diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index db03026e9..7e4771e8f 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -45,7 +45,6 @@ logger = logging.getLogger(__name__) router = APIRouter() -PREVIEW_RESOLUTION = "854x480" # 模板 mode → 视频比例映射 _TEMPLATE_MODE_TO_RATIO = { @@ -55,24 +54,7 @@ _TEMPLATE_MODE_TO_RATIO = { } -def _calc_preview_resolution(video_ratio: str = "") -> str: - """根据视频比例计算预览分辨率(短边 480,长边按比例)。 - - 支持的比例:16:9, 9:16, 1:1, 4:3, 3:4, 其他默认 16:9。 - """ - ratio_map = { - "16:9": "854x480", - "9:16": "480x854", - "1:1": "480x480", - "4:3": "640x480", - "3:4": "480x640", - } - return ratio_map.get(video_ratio.strip(), PREVIEW_RESOLUTION) - - -def _infer_video_ratio_from_template( - template_id: str, db: Session, user_id: str = "" -) -> str: +def _infer_video_ratio_from_template(template_id: str, db: Session, user_id: str = "") -> str: """从模板 mode 推断视频比例,前端未传 video_ratio 时使用。 Returns: @@ -102,9 +84,7 @@ def _infer_video_ratio_from_template( return "" -def _resolve_strategy_id_from_template( - template_id: str, db: Session, user_id: str = "" -) -> str: +def _resolve_strategy_id_from_template(template_id: str, db: Session, user_id: str = "") -> str: """从模板读取 editing_mode / mode 作为 strategy_id。 优先查新模板系统(EditTemplate.editing_mode),fallback 旧模板(Template.mode)。 @@ -236,7 +216,7 @@ def _to_preview_response(task, generated_videos: list | None = None) -> PreviewG status=task.status.value if hasattr(task.status, "value") else str(task.status), progress=float(task.progress or 0.0), is_preview=bool(getattr(task, "is_preview", True)), - resolution=getattr(task, "resolution", PREVIEW_RESOLUTION) or PREVIEW_RESOLUTION, + resolution=getattr(task, "resolution", "") or "", video_url=video_url, duration=duration, file_size=file_size, @@ -260,7 +240,7 @@ def create_preview_generation_task( ) -> PreviewGenerationTaskResponse: """创建预览生成任务。 - 预览为完整时长的低清版(480p + 低码率),效果与正式生成一致,仅清晰度降低。 + 预览渲染品质与正式生成一致(1080p, CRF 23, medium preset),确认生成时可直接复用预览产物。 Args: request: 预览任务创建请求(template_id + asset_ids 等) @@ -318,11 +298,11 @@ def create_preview_generation_task( title_ids=list(request.title_ids), voice_ids=list(request.voice_ids), created_by_user_id=user_id, - source_edit_plan_id="", + source_edit_plan_id=request.source_edit_plan_id, asset_select_mode="", batch_id="", video_title=request.video_title, - resolution=_calc_preview_resolution(video_ratio), + resolution="", bgm_config=request.bgm_config or {}, auto_retry_enabled=False, auto_retry_max=0, diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 6268536e1..b290310b6 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -352,10 +352,11 @@ def confirm_generation( generation_task_repository: Any = Depends(get_generation_task_repository), project_repository: Any = Depends(get_project_repository), ) -> BatchGenerationTaskResponse: - """确认生成 — 基于预览任务创建正式生成任务。 + """确认生成 -- 复用预览渲染产物(预览与正式品质一致)。 - 查找预览任务,复制其配置,创建新的正式生成任务(is_preview=False), - 使用高分辨率,复用 worker.generate_video 渲染路径。 + 预览已使用 1080p / CRF 23 / medium 渲染,品质与正式生成一致。 + 确认时直接将预览任务标记为正式产出,无需重新渲染,实现秒出。 + 仅当预览任务未完成时,才创建新的正式任务走渲染流程。 """ # 1. 查找源预览任务 source_task = generation_task_repository.get(task_id) @@ -368,7 +369,43 @@ def confirm_generation( if source_task.project_id: check_project_access(source_task.project_id, authenticated_user.user.id, project_repository) - # 3. 创建正式生成任务,复制预览任务的配置 + # 3. 如果预览任务已完成,检查分辨率一致性后复用产物(秒出) + if source_task.is_completed and getattr(source_task, "is_preview", False): + # 校验请求的分辨率是否与预览实际渲染的分辨率一致 + req_w = request.output_width or 0 + req_h = request.output_height or 0 + src_w = getattr(source_task, "output_width", 0) or 0 + src_h = getattr(source_task, "output_height", 0) or 0 + resolution_match = (req_w == 0 or req_w == src_w) and (req_h == 0 or req_h == src_h) + + 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, + ) + generation_task_repository.update(source_task) + logger.info( + "[确认生成] 复用预览产物: task_id=%s, user_id=%s", + task_id, + authenticated_user.user.id, + ) + return BatchGenerationTaskResponse( + items=[_to_generation_task_response(source_task)], + total=1, + ) + # 分辨率不一致,跳过复用,走新建任务流程 + logger.info( + "[确认生成] 分辨率不一致,跳过复用: task_id=%s, src=%sx%s, req=%sx%s", + task_id, + src_w, + src_h, + req_w, + req_h, + ) + + # 4. 预览任务未完成,创建新的正式任务走渲染流程 use_case = CreateGenerationTaskUseCase(generation_task_repository) new_task = use_case.execute( CreateGenerationTaskCommand( @@ -394,7 +431,7 @@ def confirm_generation( ) ) - # 4. 调度 worker.generate_video(同一条渲染路径) + # 5. 调度 worker try: if not safe_enqueue_generation_task( new_task, diff --git a/apps/api/app/api/routes/templates_editor/generation.py b/apps/api/app/api/routes/templates_editor/generation.py index b4e152290..175dcd5b3 100755 --- a/apps/api/app/api/routes/templates_editor/generation.py +++ b/apps/api/app/api/routes/templates_editor/generation.py @@ -18,6 +18,7 @@ from app.dependencies import ( get_asset_library_repository, get_asset_repository, get_db_session, + get_generated_video_repository, ) from app.schemas.generation_task import GenerationTaskResponse from app.services.edit_plan_service import EditPlanService @@ -28,6 +29,7 @@ from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.generation_task_repository import ( SQLAlchemyGenerationTaskRepository, ) +from packages.application.generated_videos import ListGeneratedVideosByTaskUseCase from packages.application.generation_tasks import ( CreateGenerationTaskCommand, CreateGenerationTaskUseCase, @@ -80,6 +82,40 @@ def generate_editor_draft( user_id=str(current_user.user.id), ) + # 检查是否可复用已完成的预览产物(预览品质已与正式一致) + gen_task_repo = SQLAlchemyGenerationTaskRepository(db) + reusable_task = _find_reusable_preview_task(gen_task_repo, plan_id, plan_check) + if reusable_task: + # 复用预览产物:标记为正式产出,跳过渲染 + reusable_task.mark_confirmed() + gen_task_repo.update(reusable_task) + + # 将产物 URL 写入 plan config + rendered_url = _get_task_output_url(reusable_task, gen_task_repo, db) + plan_svc.update_plan_config( + plan_id, + { + "generation_task_id": reusable_task.id, + "rendered_url": rendered_url, + }, + ) + plan_svc.transition_status(plan_id, EditPlanStatus.COMPLETED) + + updated_plan = plan_svc.get_plan_or_raise(plan_id) + logger.info( + "模板编辑器复用预览产物: template_id=%s plan_id=%s task_id=%s by user=%s", + template_id, + plan_id, + reusable_task.id, + current_user.user.id, + ) + return EditPlanGenerateResponse( + plan_id=plan_id, + plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status, + generation_task_id=reusable_task.id, + clip_count=len((plan_check.config or {}).get("clips", [])), + ) + # 检查是否可生成(含最后防线自动修复 + 诊断日志) try: can_gen, reason = plan_svc.can_generate(plan_id) @@ -91,7 +127,6 @@ def generate_editor_draft( try: clip_count = plan_svc.mark_clips_ready(plan_id) - gen_task_repo = SQLAlchemyGenerationTaskRepository(db) user_id = current_user.user.id _check_queue_limits(gen_task_repo, user_id) @@ -147,6 +182,53 @@ def generate_editor_draft( ) from _e +def _find_reusable_preview_task(gen_task_repo, plan_id: str, plan) -> "object | None": + """查找该 plan 关联的已完成预览任务,判断是否可复用。 + + 复用条件: + 1. 存在 source_edit_plan_id == plan_id 的已完成预览任务 + 2. plan 在预览完成后未被修改(updated_at <= 预览完成时间) + + Returns: + 可复用的 GenerationTask,或 None + """ + try: + tasks = gen_task_repo.list_by_source_edit_plan(plan_id) + except Exception: + return None + + for task in tasks: + if not getattr(task, "is_preview", False): + continue + if not task.is_completed: + continue + # 检查 plan 是否在预览完成后被修改 + completed_at = getattr(task, "completed_at", None) + if completed_at and hasattr(plan, "updated_at"): + plan_updated = plan.updated_at + # 如果 plan.updated_at 为空,无法判断是否修改过,跳过 + if plan_updated is None: + continue + # 如果 plan 在预览完成后又被修改了,不能复用 + if plan_updated > completed_at: + continue + return task + return None + + +def _get_task_output_url(task, gen_task_repo, db) -> str: + """获取任务的输出视频 URL。""" + try: + video_repo = get_generated_video_repository(db) + use_case = ListGeneratedVideosByTaskUseCase(video_repo) + videos = use_case.execute(task.id) + if videos: + return getattr(videos[0], "file_url", "") or "" + except Exception: + pass + return "" + + @router.get("/generation-status", response_model=EditPlanGenerationStatusResponse) def get_editor_generation_status( template_id: str, diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py index 3d6015276..9b95a57ea 100755 --- a/apps/api/app/schemas/generation_task.py +++ b/apps/api/app/schemas/generation_task.py @@ -154,7 +154,7 @@ class CreatePreviewGenerationTaskRequest(BaseModel): """创建预览生成任务请求。 仅支持模板模式:template_id + asset_ids 等素材 ID 列表。 - 预览为完整时长低清版(480p + 低码率)。 + 预览渲染品质与正式生成一致(1080p, CRF 23, medium preset)。 """ template_id: str @@ -177,6 +177,10 @@ class CreatePreviewGenerationTaskRequest(BaseModel): le=10, description="预览视频生成数量,范围 1-10,默认 1", ) + source_edit_plan_id: str = Field( + default="", + description="关联的编辑计划ID(可选),用于确认生成时复用预览产物", + ) @model_validator(mode="after") def _check_template_id(self) -> "CreatePreviewGenerationTaskRequest": diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index 42ced1315..8697d943e 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -485,7 +485,6 @@ class RenderAdapter: rendered_clip_ids: list[str] | None = None, failed_clip_ids: list[str] | None = None, voiceover_audio_path: str | None = None, - is_preview: bool = False, ) -> RenderAdapterResult: """执行统一渲染核心流程(BGM + ASR + 渲染 + 缩略图 + 上传)。 @@ -505,10 +504,8 @@ class RenderAdapter: self._report_progress(progress_cb, 40.0, "执行视频渲染") # 2. 初始化 ASR - # 预览模式下,如果 plan.config 中存在 voice_id,仍需初始化 ASR 以支持配音 plan_config = plan.config or {} - has_voice_id = bool(plan_config.get("voice_id")) - asr_service = None if (is_preview and not has_voice_id) else self._get_asr_service() + asr_service = self._get_asr_service() # 3. 读取输出分辨率 export_config = plan_config.get("export", {}) or {} @@ -532,27 +529,23 @@ class RenderAdapter: bgm_path=bgm_path, asr_service=asr_service, voiceover_audio_path=voiceover_audio_path, - is_preview=is_preview, ) result = render_svc.render() - # 4.5 渲染后校验输出完整性(预览模式跳过,节省耗时) - if is_preview: - logger.info("[render-adapter] 预览模式:跳过输出校验") - else: - validation = validate_video_output(result.output_path) - if not validation.valid: - logger.error( - "[render-adapter] 渲染输出校验失败: plan_id=%s job_id=%s error=%s", - plan_id, - job_id, - validation.error_message, - ) - return RenderAdapterResult( - success=False, - error_message=f"渲染输出校验失败: {validation.error_message}", - error_detail=validation.error_message, - ) + # 4.5 渲染后校验输出完整性 + validation = validate_video_output(result.output_path) + if not validation.valid: + logger.error( + "[render-adapter] 渲染输出校验失败: plan_id=%s job_id=%s error=%s", + plan_id, + job_id, + validation.error_message, + ) + return RenderAdapterResult( + success=False, + error_message=f"渲染输出校验失败: {validation.error_message}", + error_detail=validation.error_message, + ) self._report_progress(progress_cb, 80.0, "上传渲染结果") # 5. 上传结果 @@ -561,20 +554,19 @@ class RenderAdapter: self._report_progress(progress_cb, 90.0, "生成封面缩略图") - # 6. 生成缩略图(预览模式跳过,节省耗时) + # 6. 生成封面缩略图 thumbnail_url = "" - if not is_preview: - try: - from video_processing.thumbnail_generator import generate_and_upload_thumbnail + try: + from video_processing.thumbnail_generator import generate_and_upload_thumbnail - thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg" - thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key) - except Exception as thumb_err: - logger.warning( - "[render-adapter] 缩略图生成失败(不影响主流程): plan_id=%s error=%s", - plan_id, - thumb_err, - ) + thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg" + thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key) + except Exception as thumb_err: + logger.warning( + "[render-adapter] 缩略图生成失败(不影响主流程): plan_id=%s error=%s", + plan_id, + thumb_err, + ) self._report_progress(progress_cb, 100.0, "渲染完成") @@ -620,7 +612,6 @@ class RenderAdapter: work_dir: Path | None = None, progress_cb: ProgressCallback | None = None, voiceover_audio_path: str | None = None, - is_preview: bool = False, ) -> RenderAdapterResult: """使用内存中的 plan/clips/asset_path_map 直接渲染。 @@ -679,7 +670,6 @@ class RenderAdapter: job_id=job_id, progress_cb=progress_cb, voiceover_audio_path=voiceover_audio_path, - is_preview=is_preview, ) except subprocess.CalledProcessError as exc: diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 153d004ed..d386f829c 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -151,7 +151,6 @@ class UnifiedRenderService: asr_service: Any = None, # ASRService 实例,用于自动生成字幕 bgm_path: str | None = None, # BGM 本地文件路径 voiceover_audio_path: str | None = None, # 配音素材库音频本地路径 - is_preview: bool = False, # 预览模式:ultrafast 编码 + 跳过非必要步骤 ): self.plan = plan self.clips = clips @@ -164,7 +163,6 @@ class UnifiedRenderService: self.asr_service = asr_service self.bgm_path = bgm_path self.voiceover_audio_path = voiceover_audio_path - self.is_preview = is_preview self._transition_engine = TransitionEngine(default_duration=transition_duration) self._speed_engine = SpeedEngine() self._asr_timeline_cache: Any = None # ASR 字幕结果缓存,避免重复调用 @@ -1250,9 +1248,9 @@ class UnifiedRenderService: "-c:v", "libx264", "-crf", - "28" if self.is_preview else "23", + "23", "-preset", - "ultrafast" if self.is_preview else "medium", + "medium", "-pix_fmt", "yuv420p", "-movflags", @@ -1776,9 +1774,9 @@ class UnifiedRenderService: "-c:v", "libx264", "-crf", - "28" if self.is_preview else "23", + "23", "-preset", - "ultrafast" if self.is_preview else "medium", + "medium", "-pix_fmt", "yuv420p", "-movflags", @@ -1787,11 +1785,10 @@ class UnifiedRenderService: ] logger.info( - "执行渲染: plan_id=%s inputs=%d output=%s preview=%s", + "执行渲染: plan_id=%s inputs=%d output=%s", self.plan.id, input_args.count("-i"), output_path, - self.is_preview, ) try: run_ffmpeg(command) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 96fced8fc..f82fe67db 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1039,7 +1039,6 @@ def _load_task_info(task_id: str) -> dict | None: "video_title": getattr(gen_task, "video_title", "") or "", "resolution": getattr(gen_task, "resolution", "") or "", "bgm_config": dict(getattr(gen_task, "bgm_config", {}) or {}), - "is_preview": bool(getattr(gen_task, "is_preview", False)), "source_task_id": getattr(gen_task, "source_task_id", "") or "", "output_width": getattr(gen_task, "output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH, "output_height": getattr(gen_task, "output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT, @@ -1123,7 +1122,6 @@ def _render_video( output_name: str, resolution: str = "", bgm_config: dict | None = None, - is_preview: bool = False, voice_ids: list[str] | None = None, ) -> tuple[Path, float]: """渲染视频(含配音混音)。 @@ -1131,8 +1129,6 @@ def _render_video( 使用 RenderAdapter 统一渲染入口,复用 BGM/ASR/分辨率/缩略图逻辑。 Args: - is_preview: 是否为预览生成,若是则强制 480p + 低码率 - Returns: (output_path, render_duration) """ @@ -1179,18 +1175,7 @@ def _render_video( # 注意:必须拷贝字典,避免预览模式修改污染源对象(模板配置) plan_cfg = dict(virtual_plan.config or {}) export_cfg = dict(plan_cfg.get("export", {}) or {}) - if is_preview: - # 预览模式:短边 480p + 低码率,但尊重视频比例(竖屏模板不应强制横屏) - preview_res = resolution if resolution else "854x480" - export_cfg["resolution"] = preview_res - export_cfg["bitrate"] = "1M" - logger.info( - "[task_id=%s] [渲染] 预览模式:分辨率=%s, 码率=%s", - task_id, - preview_res, - "1M", - ) - elif resolution: + if resolution: # 用户在 API 调用时指定的分辨率优先级最高 export_cfg["resolution"] = resolution elif not export_cfg.get("resolution"): @@ -1239,7 +1224,6 @@ def _render_video( job_id=task_id, work_dir=temp_path, voiceover_audio_path=voice_path, - is_preview=is_preview, ) finally: db.close() @@ -1474,7 +1458,6 @@ def generate_video(self, task_id: str) -> dict: output_name=output_name, resolution=_resolved_resolution, bgm_config=task_info.get("bgm_config", {}), - is_preview=task_info.get("is_preview", False), voice_ids=task_info.get("voice_ids", []), ) diff --git a/packages/domain/generation_task.py b/packages/domain/generation_task.py index 1ba58132d..631f26bdb 100755 --- a/packages/domain/generation_task.py +++ b/packages/domain/generation_task.py @@ -296,6 +296,30 @@ class GenerationTask: self.transition_to(GenerationTaskStatus.CANCELLED) self.completed_at = datetime.now(timezone.utc) + def mark_confirmed( + self, + *, + cover_url: str = "", + custom_title: str = "", + output_width: int = 0, + output_height: int = 0, + ) -> None: + """将预览任务确认为正式产出。 + + 预览渲染品质已与正式生成一致(1080p, CRF 23, medium), + 确认时直接复用已有产物,无需重新渲染。 + """ + 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: + self.output_height = output_height + self.updated_at = datetime.now(timezone.utc) + # ── 日志辅助 ──────────────────────────────────────────────────────────── _MAX_LOGS = 200 diff --git a/tests/unit/test_1280_preview_speedup.py b/tests/unit/test_1280_preview_speedup.py index c315c8106..1511352fd 100644 --- a/tests/unit/test_1280_preview_speedup.py +++ b/tests/unit/test_1280_preview_speedup.py @@ -1,12 +1,10 @@ -"""#1280 预览视频生成加速 — 单元测试。 +"""统一渲染管线 — 预览与确认生成使用相同品质参数。 验证点: -1. UnifiedRenderService.is_preview 参数正确传递 -2. 预览模式使用 ultrafast preset + crf 28 -3. RenderAdapter.render_from_memory 正确传递 is_preview -4. 预览模式跳过 ASR 初始化 -5. 预览模式跳过输出校验和缩略图 -6. generation.py 并行下载逻辑 +1. UnifiedRenderService 不再有 is_preview 参数 +2. 所有渲染统一使用 medium preset + CRF 23 +3. RenderAdapter 统一执行校验和缩略图生成 +4. generation.py 并行下载逻辑(保留) """ from __future__ import annotations @@ -18,13 +16,24 @@ from unittest.mock import MagicMock, patch import pytest -# ── 1. UnifiedRenderService is_preview 参数 ── +# ── 1. UnifiedRenderService 无 is_preview 参数 ── -class TestUnifiedRenderServicePreviewFlag: - """is_preview 参数正确传递和存储。""" +class TestUnifiedRenderServiceNoPreviewParam: + """UnifiedRenderService 构造函数不再接受 is_preview 参数。""" - def test_default_is_preview_false(self): + def test_constructor_has_no_is_preview(self): + import inspect + + from video_processing.unified_render_service import UnifiedRenderService + + sig = inspect.signature(UnifiedRenderService.__init__) + param_names = list(sig.parameters.keys()) + assert ( + "is_preview" not in param_names + ), f"is_preview should be removed from UnifiedRenderService.__init__, found params: {param_names}" + + def test_no_is_preview_attribute(self): from video_processing.unified_render_service import UnifiedRenderService svc = UnifiedRenderService( @@ -33,38 +42,16 @@ class TestUnifiedRenderServicePreviewFlag: asset_path_map={}, work_dir=Path(tempfile.mkdtemp()), ) - assert svc.is_preview is False - - def test_is_preview_true(self): - from video_processing.unified_render_service import UnifiedRenderService - - svc = UnifiedRenderService( - plan=MagicMock(id="test"), - clips=[], - asset_path_map={}, - work_dir=Path(tempfile.mkdtemp()), - is_preview=True, - ) - assert svc.is_preview is True - - def test_is_preview_false_explicit(self): - from video_processing.unified_render_service import UnifiedRenderService - - svc = UnifiedRenderService( - plan=MagicMock(id="test"), - clips=[], - asset_path_map={}, - work_dir=Path(tempfile.mkdtemp()), - is_preview=False, - ) - assert svc.is_preview is False + assert not hasattr( + svc, "is_preview" + ), "UnifiedRenderService should not have is_preview attribute after unification" -# ── 2. 预览模式 FFmpeg 参数 ── +# ── 2. FFmpeg 参数统一为 medium + CRF 23 ── -class TestPreviewFFmpegPreset: - """预览模式使用 ultrafast preset + crf 28。""" +class TestUnifiedFFmpegPreset: + """所有渲染统一使用 medium preset + CRF 23。""" def _make_clip(self): from video_processing.unified_render_service import ResolvedClip @@ -84,46 +71,7 @@ class TestPreviewFFmpegPreset: ) @patch("video_processing.unified_render_service.run_ffmpeg") - def test_execute_ffmpeg_preview_uses_ultrafast(self, mock_run): - from video_processing.unified_render_service import ( - RenderLayer, - UnifiedRenderService, - ) - - plan = MagicMock() - plan.id = "test_plan" - plan.config = {"export": {"resolution": "854x480"}} - - clip = self._make_clip() - - svc = UnifiedRenderService( - plan=plan, - clips=[clip], - asset_path_map={"a1": Path("/tmp/fake.mp4")}, - work_dir=Path(tempfile.mkdtemp()), - output_width=854, - output_height=480, - is_preview=True, - ) - - layers = [RenderLayer(role="main", clips=[clip])] - filter_complex, input_args = svc._build_filter_complex(layers) - output_path = Path(tempfile.mkdtemp()) / "out.mp4" - svc._execute_ffmpeg(filter_complex, input_args, output_path) - - mock_run.assert_called_once() - cmd = mock_run.call_args[0][0] - - # Check preset is ultrafast - preset_idx = cmd.index("-preset") - assert cmd[preset_idx + 1] == "ultrafast", f"Expected ultrafast, got {cmd[preset_idx + 1]}" - - # Check crf is 28 - crf_idx = cmd.index("-crf") - assert cmd[crf_idx + 1] == "28", f"Expected crf 28, got {cmd[crf_idx + 1]}" - - @patch("video_processing.unified_render_service.run_ffmpeg") - def test_execute_ffmpeg_normal_uses_medium(self, mock_run): + def test_execute_ffmpeg_uses_medium_crf23(self, mock_run): from video_processing.unified_render_service import ( RenderLayer, UnifiedRenderService, @@ -140,9 +88,8 @@ class TestPreviewFFmpegPreset: clips=[clip], asset_path_map={"a1": Path("/tmp/fake.mp4")}, work_dir=Path(tempfile.mkdtemp()), - output_width=1280, - output_height=720, - is_preview=False, + output_width=1920, + output_height=1080, ) layers = [RenderLayer(role="main", clips=[clip])] @@ -153,137 +100,73 @@ class TestPreviewFFmpegPreset: mock_run.assert_called_once() cmd = mock_run.call_args[0][0] + # Check preset is medium (no conditional) preset_idx = cmd.index("-preset") - assert cmd[preset_idx + 1] == "medium" + assert cmd[preset_idx + 1] == "medium", f"Expected medium, got {cmd[preset_idx + 1]}" + # Check crf is 23 (no conditional) crf_idx = cmd.index("-crf") - assert cmd[crf_idx + 1] == "23" + assert cmd[crf_idx + 1] == "23", f"Expected crf 23, got {cmd[crf_idx + 1]}" -# ── 3. RenderAdapter passes is_preview ── +# ── 3. RenderAdapter 统一执行校验和缩略图 ── -class TestRenderAdapterPreviewPassthrough: - """RenderAdapter 正确传递 is_preview 参数。""" +class TestRenderAdapterUnifiedPostProcess: + """RenderAdapter 不再跳过校验和缩略图。""" + + def test_render_adapter_no_is_preview_param(self): + import inspect - def test_render_from_memory_passes_is_preview(self): from video_processing.render_adapter import RenderAdapter - db = MagicMock() - adapter = RenderAdapter(db) + # Check render_from_memory signature + sig = inspect.signature(RenderAdapter.render_from_memory) + param_names = list(sig.parameters.keys()) + assert ( + "is_preview" not in param_names + ), f"is_preview should be removed from render_from_memory, found params: {param_names}" - plan = MagicMock() - plan.id = "test_plan" - plan.config = {"export": {"resolution": "854x480"}} - - clip = MagicMock() - clip.id = "c1" - - with patch.object(adapter, "_do_render") as mock_do_render: - mock_do_render.return_value = MagicMock( - success=True, - output_path=Path("/tmp/out.mp4"), - thumbnail_url="", - duration=5.0, - file_size=1000, - width=854, - height=480, - output_url="https://oss/test.mp4", - rendered_clip_ids=["c1"], - failed_clip_ids=[], - ) - - adapter.render_from_memory( - plan=plan, - clips=[clip], - asset_path_map={"a1": Path("/tmp/fake.mp4")}, - is_preview=True, - ) - - mock_do_render.assert_called_once() - _, kwargs = mock_do_render.call_args - assert kwargs.get("is_preview") is True - - -# ── 4. 预览模式跳过 ASR ── - - -class TestPreviewSkipsASR: - """预览模式跳过 ASR 初始化。""" - - def test_render_method_source_has_asr_skip(self): - """_do_render 在 is_preview=True 时不调用 _get_asr_service。""" + def test_no_preview_skip_validation_in_source(self): + """渲染适配器源码中不再包含预览跳过校验的逻辑。""" with open("apps/worker/video_processing/render_adapter.py") as f: source = f.read() - assert ( - "None if (is_preview and not has_voice_id) else self._get_asr_service()" in source - ), "Should skip ASR initialization in preview mode unless voice_id is provided" + assert "预览模式:跳过输出校验" not in source, "Should not skip validation in any mode" + assert "if not is_preview:" not in source, "Thumbnail should always be generated" -# ── 5. 并行下载逻辑 ── +# ── 4. generation.py 不再有 is_preview 覆盖逻辑 ── -class TestParallelDownload: - """generation.py 并行下载素材。""" +class TestWorkerGenerationNoPreviewOverride: + """Worker generation.py 不再覆盖预览分辨率为 480p。""" - def test_parallel_download_uses_thread_pool(self): + def test_no_480p_override(self): + with open("apps/worker/worker_app/tasks/generation.py") as f: + source = f.read() + + 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" - def test_parallel_download_preserves_order(self): - with open("apps/worker/worker_app/tasks/generation.py") as f: + +# ── 5. generation_preview.py 不再有 PREVIEW_RESOLUTION ── + + +class TestPreviewNoLowQualityConstants: + """预览 API 不再定义低质量常量。""" + + def test_no_preview_resolution_constant(self): + with open("apps/api/app/api/routes/generation_preview.py") as f: source = f.read() - assert "sorted(results_map.keys())" in source, "Should sort results by original index" - - -# ── 6. generation.py _render_video passes is_preview ── - - -class TestRenderVideoPassesPreview: - """_render_video 正确传递 is_preview 到 render_from_memory。""" - - def test_render_video_passes_is_preview(self): - with open("apps/worker/worker_app/tasks/generation.py") as f: - source = f.read() - - assert "is_preview=is_preview" in source, "Should pass is_preview to render_from_memory" - - -# ── 7. Preview mode skips thumbnail and validation ── - - -class TestPreviewSkipsThumbnailAndValidation: - """预览模式跳过缩略图生成和输出校验。""" - - def test_render_adapter_skips_thumbnail_in_preview(self): - with open("apps/worker/video_processing/render_adapter.py") as f: - source = f.read() - - assert "if not is_preview:" in source, "Thumbnail should be conditional on is_preview" - - def test_render_adapter_skips_validation_in_preview(self): - with open("apps/worker/video_processing/render_adapter.py") as f: - source = f.read() - - assert "预览模式:跳过输出校验" in source, "Should skip validation in preview mode" - - -# ── 8. Pass-through rendering uses ultrafast in preview ── - - -class TestPassThroughPreviewPreset: - """直通渲染在预览模式也使用 ultrafast。""" - - def test_pass_through_has_preview_preset(self): - with open("apps/worker/video_processing/unified_render_service.py") as f: - source = f.read() - - # The pass_through method should also use ultrafast for preview - # Count occurrences of "ultrafast" - should be at least 2 (execute_ffmpeg + pass_through) - count = source.count('"ultrafast" if self.is_preview') - assert count >= 2, f"Expected at least 2 ultrafast preset usages, found {count}" + assert "PREVIEW_RESOLUTION" not in source, "PREVIEW_RESOLUTION constant should be removed" + assert "_calc_preview_resolution" not in source, "_calc_preview_resolution function should be removed" diff --git a/tests/unit/test_1294_preview_voice_injection.py b/tests/unit/test_1294_preview_voice_injection.py index 3b27160e7..7f3c75709 100644 --- a/tests/unit/test_1294_preview_voice_injection.py +++ b/tests/unit/test_1294_preview_voice_injection.py @@ -169,7 +169,6 @@ class TestRenderVideoVoiceInjection: temp_path=Path("/tmp"), output_name="test_output.mp4", resolution="854x480", - is_preview=True, voice_ids=["voice_abc"], ) diff --git a/tests/unit/test_confirm_generation.py b/tests/unit/test_confirm_generation.py index 72f70856d..5091c8a5a 100644 --- a/tests/unit/test_confirm_generation.py +++ b/tests/unit/test_confirm_generation.py @@ -1,10 +1,10 @@ """确认生成 API 单元测试. 覆盖 POST /tasks/{task_id}/confirm 端点: - - 正常确认流程 + - 预览任务已完成 → 直接复用(mark_confirmed),秒出 + - 预览任务未完成 → 创建新任务走渲染流程 - 预览任务不存在 → 404 - 权限不足 → 403 - - is_preview=False 及分辨率正确 - cover_url 和 custom_title 正确传递 """ @@ -164,7 +164,6 @@ def app( test_app.dependency_overrides[get_current_user] = override_get_current_user test_app.dependency_overrides[get_generation_task_repository] = override_get_generation_task_repository test_app.dependency_overrides[get_project_repository] = override_get_project_repository - # Stubs for repositories not used by confirm endpoint but required by router test_app.dependency_overrides[get_asset_library_repository] = lambda: MagicMock() test_app.dependency_overrides[get_asset_repository] = lambda: MagicMock() test_app.dependency_overrides[get_generated_video_repository] = lambda: MagicMock() @@ -199,8 +198,8 @@ def _make_preview_task(**kwargs: Any) -> GenerationTask: asset_select_mode="all", is_preview=True, source_task_id="", - output_width=1280, - output_height=720, + output_width=1080, + output_height=1920, cover_url="", custom_title="", video_title="", @@ -214,45 +213,97 @@ def _make_preview_task(**kwargs: Any) -> GenerationTask: # ── Tests ──────────────────────────────────────────────────────────────────── -class TestConfirmGeneration: - def test_confirm_success( +class TestConfirmGenerationReuse: + """确认生成复用预览产物。""" + + def test_confirm_reuses_completed_preview( self, client: TestClient, gen_task_repo: StubGenerationTaskRepository, ) -> None: - """正常确认流程:预览任务存在、权限正确 → 创建正式任务""" + """预览已完成 → 直接复用,返回同一个任务,不创建新任务""" preview = _make_preview_task() gen_task_repo.create(preview) - with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True) as mock_enqueue: - resp = client.post( - f"/api/v1/tasks/{preview.id}/confirm", - json={ - "output_width": 1080, - "output_height": 1920, - "cover_url": "https://example.com/cover.jpg", - "custom_title": "我的视频", - }, - ) + initial_count = len(gen_task_repo._store) + + resp = client.post( + f"/api/v1/tasks/{preview.id}/confirm", + json={ + "output_width": 1080, + "output_height": 1920, + "cover_url": "https://example.com/cover.jpg", + "custom_title": "我的视频", + }, + ) assert resp.status_code == 200 data = resp.json() assert data["total"] == 1 item = data["items"][0] + + # 返回的是同一个任务(复用) + assert item["id"] == preview.id + # is_preview 变为 False assert item["is_preview"] is False - assert item["source_task_id"] == preview.id + # 分辨率更新 assert item["output_width"] == 1080 assert item["output_height"] == 1920 + # 封面和标题更新 assert item["cover_url"] == "https://example.com/cover.jpg" assert item["custom_title"] == "我的视频" - # 复制了预览任务的配置 - assert item["project_id"] == "project-001" - assert item["asset_library_id"] == "library-001" - assert item["strategy_id"] == "one_take" - assert item["asset_ids"] == ["asset-1"] - # 验证入队函数被调用 - mock_enqueue.assert_called_once() + # 没有创建新任务 + assert len(gen_task_repo._store) == initial_count + + def test_confirm_updates_task_in_repo( + self, + client: TestClient, + gen_task_repo: StubGenerationTaskRepository, + ) -> None: + """确认后的任务在 repo 中被更新""" + preview = _make_preview_task() + gen_task_repo.create(preview) + + resp = client.post( + f"/api/v1/tasks/{preview.id}/confirm", + json={"cover_url": "https://cdn.example.com/cover.png", "custom_title": "测试标题"}, + ) + + assert resp.status_code == 200 + + # 验证 repo 中的任务已被更新 + updated = gen_task_repo.get(preview.id) + 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, + client: TestClient, + gen_task_repo: StubGenerationTaskRepository, + ) -> None: + """预览任务未完成 → 创建新任务走渲染流程""" + preview = _make_preview_task(status=GenerationTaskStatus.RUNNING, progress=50.0) + gen_task_repo.create(preview) + + with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True): + resp = client.post( + f"/api/v1/tasks/{preview.id}/confirm", + json={"output_width": 1080, "output_height": 1920}, + ) + + assert resp.status_code == 200 + item = resp.json()["items"][0] + # 创建了新任务 + assert item["id"] != preview.id + assert item["is_preview"] is False + assert item["source_task_id"] == preview.id + + +class TestConfirmGenerationErrors: + """确认生成的错误处理。""" def test_confirm_not_found(self, client: TestClient) -> None: """预览任务不存在 → 404""" @@ -284,7 +335,7 @@ class TestConfirmGeneration: client: TestClient, gen_task_repo: StubGenerationTaskRepository, ) -> None: - """确认后的任务 is_preview=False,分辨率已更新,其余配置从预览任务复制""" + """确认后任务保留预览任务的全部配置""" preview = _make_preview_task( voice_library_id="voice-001", template_id="tmpl-001", @@ -293,22 +344,20 @@ class TestConfirmGeneration: ) gen_task_repo.create(preview) - with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True): - resp = client.post( - f"/api/v1/tasks/{preview.id}/confirm", - json={"output_width": 1920, "output_height": 1080}, - ) + resp = client.post( + f"/api/v1/tasks/{preview.id}/confirm", + json={"output_width": 1920, "output_height": 1080}, + ) assert resp.status_code == 200 item = resp.json()["items"][0] assert item["is_preview"] is False - assert item["source_task_id"] == preview.id assert item["output_width"] == 1920 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" assert item["title_ids"] == ["title-1", "title-2"] @@ -323,16 +372,15 @@ class TestConfirmGeneration: preview = _make_preview_task() gen_task_repo.create(preview) - with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True): - resp = client.post( - f"/api/v1/tasks/{preview.id}/confirm", - json={ - "output_width": 1080, - "output_height": 1920, - "cover_url": "https://cdn.example.com/my-cover.png", - "custom_title": "测试视频标题", - }, - ) + resp = client.post( + f"/api/v1/tasks/{preview.id}/confirm", + json={ + "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] @@ -348,24 +396,23 @@ class TestConfirmGeneration: preview = _make_preview_task() gen_task_repo.create(preview) - with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True): - resp = client.post( - f"/api/v1/tasks/{preview.id}/confirm", - json={}, - ) + resp = client.post( + f"/api/v1/tasks/{preview.id}/confirm", + json={}, + ) assert resp.status_code == 200 item = resp.json()["items"][0] assert item["output_width"] == 1080 assert item["output_height"] == 1920 - def test_confirm_creates_new_task_in_repo( + def test_confirm_skips_reuse_when_resolution_mismatch( self, client: TestClient, gen_task_repo: StubGenerationTaskRepository, ) -> None: - """确认生成的任务确实被存入 repository""" - preview = _make_preview_task() + """请求的分辨率与预览渲染的分辨率不一致时,跳过复用,创建新任务""" + preview = _make_preview_task(output_width=1080, output_height=1920) gen_task_repo.create(preview) initial_count = len(gen_task_repo._store) @@ -373,15 +420,13 @@ class TestConfirmGeneration: with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True): resp = client.post( f"/api/v1/tasks/{preview.id}/confirm", - json={}, + json={"output_width": 720, "output_height": 1280}, ) assert resp.status_code == 200 - new_task_id = resp.json()["items"][0]["id"] - assert new_task_id != preview.id + item = resp.json()["items"][0] + # 创建了新任务(而非复用) + assert item["id"] != preview.id + assert item["is_preview"] is False + assert item["source_task_id"] == preview.id assert len(gen_task_repo._store) == initial_count + 1 - - new_task = gen_task_repo.get(new_task_id) - assert new_task is not None - assert new_task.is_preview is False - assert new_task.source_task_id == preview.id diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py index ac71c1f1d..1848c92b2 100644 --- a/tests/unit/test_generation_preview.py +++ b/tests/unit/test_generation_preview.py @@ -563,7 +563,6 @@ from unittest.mock import PropertyMock, patch # 导入路由模块和被测函数 from app.api.routes.generation_preview import ( - PREVIEW_RESOLUTION, _to_preview_response, create_preview_generation_task, get_preview_generation_task, @@ -1133,35 +1132,6 @@ class TestSignVideoUrl: assert result == "https://bucket.oss.example.com/video.mp4" -class TestCalcPreviewResolution: - """_calc_preview_resolution 单元测试。""" - - def test_portrait_9_16(self): - from app.api.routes.generation_preview import _calc_preview_resolution - - assert _calc_preview_resolution("9:16") == "480x854" - - def test_landscape_16_9(self): - from app.api.routes.generation_preview import _calc_preview_resolution - - assert _calc_preview_resolution("16:9") == "854x480" - - def test_square_1_1(self): - from app.api.routes.generation_preview import _calc_preview_resolution - - assert _calc_preview_resolution("1:1") == "480x480" - - def test_unknown_defaults_to_landscape(self): - from app.api.routes.generation_preview import _calc_preview_resolution - - assert _calc_preview_resolution("unknown") == "854x480" - - def test_empty_defaults_to_landscape(self): - from app.api.routes.generation_preview import _calc_preview_resolution - - assert _calc_preview_resolution("") == "854x480" - - class TestInferVideoRatioFromTemplate: """_infer_video_ratio_from_template 单元测试。""" @@ -1239,7 +1209,7 @@ class TestPreviewRouteAutoInfersVideoRatio: """验证预览路由在前端未传 video_ratio 时自动从模板推断。""" def test_auto_infer_pip_resolution(self): - """前端传 video_ratio='',模板 mode=pip → resolution=480x854""" + """前端传 video_ratio='',模板 mode=pip → resolution 为空(使用用户默认导出配置)""" from app.api.routes.generation_preview import create_preview_generation_task from app.schemas.generation_task import CreatePreviewGenerationTaskRequest @@ -1284,7 +1254,7 @@ class TestPreviewRouteAutoInfersVideoRatio: # Verify the resolution passed to CreateGenerationTaskCommand call_args = MockUC.return_value.execute.call_args cmd = call_args[0][0] - assert cmd.resolution == "480x854", f"Expected 480x854, got {cmd.resolution}" + assert cmd.resolution == "", f"Expected empty resolution, got {cmd.resolution}" # ═══════════════════════════════════════════════════════════════════════════════ diff --git a/tests/unit/test_unified_render_reuse.py b/tests/unit/test_unified_render_reuse.py new file mode 100644 index 000000000..6af944b84 --- /dev/null +++ b/tests/unit/test_unified_render_reuse.py @@ -0,0 +1,287 @@ +"""统一渲染路径 — 编辑器预览产物复用逻辑单元测试。 + +覆盖: +- _find_reusable_preview_task: 查找可复用的预览任务 +- _get_task_output_url: 获取任务输出 URL +- 编辑器 generate 接口复用预览产物路径 +""" + +from __future__ import annotations + +import os +import sys +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any, Optional +from unittest.mock import MagicMock, patch + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") +os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api")) + +from packages.domain.generation_task import GenerationTask, GenerationTaskStatus + +# ── Stub Repository ────────────────────────────────────────────────────────── + + +class StubGenTaskRepo: + def __init__(self): + self._store = {} + + def create(self, task): + self._store[task.id] = task + return task + + def get(self, task_id): + return self._store.get(task_id) + + def update(self, task): + self._store[task.id] = task + return task + + def list_by_source_edit_plan(self, plan_id): + return [t for t in self._store.values() if (t.source_edit_plan_id or "") == plan_id] + + +def _make_task(**kwargs): + defaults = dict( + id="task-001", + project_id="proj-1", + asset_library_id="lib-1", + strategy_id="one_take", + voice_library_id="", + template_id="tmpl-1", + asset_ids=["a1"], + title_ids=[], + voice_ids=[], + status=GenerationTaskStatus.COMPLETED, + progress=100.0, + result_count=1, + error_message="", + created_by_user_id="user-1", + source_edit_plan_id="plan-1", + asset_select_mode="all", + is_preview=True, + source_task_id="", + output_width=1920, + output_height=1080, + cover_url="", + custom_title="", + video_title="", + resolution="", + bgm_config={}, + completed_at=datetime.now(timezone.utc), + ) + defaults.update(kwargs) + return GenerationTask(**defaults) + + +def _make_plan(updated_at=None): + plan = MagicMock() + plan.id = "plan-1" + plan.updated_at = updated_at or datetime.now(timezone.utc) + plan.status = MagicMock() + plan.status.value = "editing" + plan.config = {"clips": [{"id": "c1"}, {"id": "c2"}]} + return plan + + +# ── _find_reusable_preview_task ────────────────────────────────────────────── + + +class TestFindReusablePreviewTask: + def test_returns_completed_preview_task(self): + from app.api.routes.templates_editor.generation import _find_reusable_preview_task + + repo = StubGenTaskRepo() + now = datetime.now(timezone.utc) + task = _make_task( + source_edit_plan_id="plan-1", + is_preview=True, + status=GenerationTaskStatus.COMPLETED, + completed_at=now - timedelta(minutes=5), + ) + repo.create(task) + + plan = _make_plan(updated_at=now - timedelta(minutes=10)) + result = _find_reusable_preview_task(repo, "plan-1", plan) + + assert result is not None + assert result.id == "task-001" + + def test_returns_none_when_no_tasks(self): + from app.api.routes.templates_editor.generation import _find_reusable_preview_task + + repo = StubGenTaskRepo() + plan = _make_plan() + result = _find_reusable_preview_task(repo, "plan-1", plan) + assert result is None + + def test_returns_none_when_preview_not_completed(self): + from app.api.routes.templates_editor.generation import _find_reusable_preview_task + + repo = StubGenTaskRepo() + task = _make_task( + source_edit_plan_id="plan-1", + is_preview=True, + status=GenerationTaskStatus.RUNNING, + completed_at=None, + ) + repo.create(task) + + plan = _make_plan() + result = _find_reusable_preview_task(repo, "plan-1", plan) + assert result is None + + def test_returns_none_when_plan_modified_after_preview(self): + from app.api.routes.templates_editor.generation import _find_reusable_preview_task + + repo = StubGenTaskRepo() + now = datetime.now(timezone.utc) + task = _make_task( + source_edit_plan_id="plan-1", + is_preview=True, + status=GenerationTaskStatus.COMPLETED, + completed_at=now - timedelta(minutes=10), + ) + repo.create(task) + + # Plan was updated AFTER preview completed + plan = _make_plan(updated_at=now) + result = _find_reusable_preview_task(repo, "plan-1", plan) + assert result is None + + def test_skips_non_preview_tasks(self): + from app.api.routes.templates_editor.generation import _find_reusable_preview_task + + repo = StubGenTaskRepo() + task = _make_task( + source_edit_plan_id="plan-1", + is_preview=False, # not a preview task + status=GenerationTaskStatus.COMPLETED, + ) + repo.create(task) + + plan = _make_plan() + result = _find_reusable_preview_task(repo, "plan-1", plan) + assert result is None + + def test_handles_repo_exception(self): + from app.api.routes.templates_editor.generation import _find_reusable_preview_task + + repo = MagicMock() + repo.list_by_source_edit_plan.side_effect = Exception("db error") + plan = _make_plan() + result = _find_reusable_preview_task(repo, "plan-1", plan) + assert result is None + + +# ── _get_task_output_url ───────────────────────────────────────────────────── + + +class TestGetTaskOutputUrl: + def test_returns_video_url(self): + from app.api.routes.templates_editor.generation import _get_task_output_url + + task = _make_task() + repo = MagicMock() + db = MagicMock() + + mock_video = MagicMock() + mock_video.file_url = "https://oss.example.com/video.mp4" + + mock_use_case = MagicMock() + mock_use_case.execute.return_value = [mock_video] + + with patch( + "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase", + return_value=mock_use_case, + ): + result = _get_task_output_url(task, repo, db) + + assert result == "https://oss.example.com/video.mp4" + + def test_returns_empty_when_no_videos(self): + from app.api.routes.templates_editor.generation import _get_task_output_url + + task = _make_task() + repo = MagicMock() + db = MagicMock() + + mock_use_case = MagicMock() + mock_use_case.execute.return_value = [] + + with patch( + "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase", + return_value=mock_use_case, + ): + result = _get_task_output_url(task, repo, db) + + assert result == "" + + def test_returns_empty_on_exception(self): + from app.api.routes.templates_editor.generation import _get_task_output_url + + task = _make_task() + repo = MagicMock() + db = MagicMock() + + with patch( + "app.api.routes.templates_editor.generation.ListGeneratedVideosByTaskUseCase", + side_effect=Exception("db error"), + ): + result = _get_task_output_url(task, repo, db) + + assert result == "" + + +# ── mark_confirmed ─────────────────────────────────────────────────────────── + + +class TestMarkConfirmed: + def test_sets_is_preview_false(self): + task = _make_task(is_preview=True) + task.mark_confirmed() + assert task.is_preview is False + + def test_sets_cover_url(self): + task = _make_task() + task.mark_confirmed(cover_url="https://example.com/cover.jpg") + assert task.cover_url == "https://example.com/cover.jpg" + + def test_sets_custom_title(self): + task = _make_task() + task.mark_confirmed(custom_title="My Video") + assert task.custom_title == "My Video" + + def test_sets_output_dimensions(self): + task = _make_task() + task.mark_confirmed(output_width=1080, output_height=1920) + assert task.output_width == 1080 + assert task.output_height == 1920 + + def test_zero_dimensions_not_applied(self): + task = _make_task(output_width=1920, output_height=1080) + task.mark_confirmed(output_width=0, output_height=0) + assert task.output_width == 1920 + assert task.output_height == 1080 + + def test_skips_when_plan_updated_at_is_none(self): + from app.api.routes.templates_editor.generation import _find_reusable_preview_task + + repo = StubGenTaskRepo() + now = datetime.now(timezone.utc) + task = _make_task( + source_edit_plan_id="plan-1", + is_preview=True, + status=GenerationTaskStatus.COMPLETED, + completed_at=now - timedelta(minutes=5), + ) + repo.create(task) + + plan = _make_plan(updated_at=None) + result = _find_reusable_preview_task(repo, "plan-1", plan) + assert result is None