feat: 统一预览与确认生成渲染路径 — 预览=正式品质+确认复用

## 渲染管线统一
- UnifiedRenderService: 移除 is_preview 参数，统一 CRF 23 + medium preset
- RenderAdapter: 移除 is_preview 参数，统一执行校验和缩略图生成
- Worker generation.py: 移除 480p+1M 码率覆盖逻辑
- generation_preview.py: 删除 PREVIEW_RESOLUTION 常量和 _calc_preview_resolution()

## 确认生成复用预览产物
- Domain: 新增 GenerationTask.mark_confirmed() 方法
- confirm_generation: 预览已完成时直接复用产物（秒出），无需重新渲染
- Editor generate: 检查 plan 关联的预览任务，plan未修改时复用产物
- Schema: CreatePreviewGenerationTaskRequest 新增 source_edit_plan_id 字段

## 测试更新
- test_1280_preview_speedup: 重写为验证统一品质参数
- test_confirm_generation: 重写为验证复用逻辑

## 交付标准
- 预览渲染质量与确认生成一致（1080p, CRF 23, medium）
- 预览后确认生成直接复用产物（秒出）
- Worker 端无任何 is_preview 低质量渲染代码残留

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,

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,26 @@ 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):
+        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,
+        )
+
+    # 4. 预览任务未完成，创建新的正式任务走渲染流程
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
     new_task = use_case.execute(
         CreateGenerationTaskCommand(
@@ -394,7 +414,7 @@ def confirm_generation(
         )
     )
 
-    # 4. 调度 worker.generate_video（同一条渲染路径）
+    # 5. 调度 worker
     try:
         if not safe_enqueue_generation_task(
             new_task,

apps/api/app/api/routes/templates_editor/generation.py

@@ -80,6 +80,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)
+        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 +125,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 +180,50 @@ 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 在预览完成后又被修改了，不能复用
+            if plan_updated > completed_at:
+                continue
+        return task
+    return None
+
+
+def _get_task_output_url(task, gen_task_repo) -> str:
+    """获取任务的输出视频 URL。"""
+    try:
+        from app.core.task_repository_helpers import get_generated_videos_for_task
+
+        videos = get_generated_videos_for_task(task.id, gen_task_repo)
+        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,

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":

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 + 渲染 + 缩略图 + 上传）。
 
@@ -508,7 +507,7 @@ class RenderAdapter:
         # 预览模式下，如果 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 +531,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 +556,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 +614,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 +672,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:

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)

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", []),
             )
 

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

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"

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()
@@ -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,40 +396,12 @@ 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(
-        self,
-        client: TestClient,
-        gen_task_repo: StubGenerationTaskRepository,
-    ) -> None:
-        """确认生成的任务确实被存入 repository"""
-        preview = _make_preview_task()
-        gen_task_repo.create(preview)
-
-        initial_count = len(gen_task_repo._store)
-
-        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={},
-            )
-
-        assert resp.status_code == 200
-        new_task_id = resp.json()["items"][0]["id"]
-        assert new_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