Merge pull request 'fix: 确认生成 API 改为复用 worker.generate_video 渲染路径' (#1308) from fix/confirm-generation-api into main

- 新增 ConfirmGenerationRequest schema 和 POST /tasks/{task_id}/confirm 端点
- 确认生成复用 worker.generate_video 完整渲染路径（不再走 edit_plan_generation.py）
- 默认分辨率 1080x1920（竖屏），支持自定义封面和标题
- 新增字段: is_preview, source_task_id, output_width, output_height, cover_url, custom_title
- 贯穿 Schema → Domain → Application → DB Model → Repository → Worker 全链路
- 更新 task_center.py 和 generation_tasks.py 的重试逻辑传递新字段
- Alembic 迁移 034: 为 generation_tasks 表添加 6 个新列
- 7 个单元测试覆盖确认生成的正常/异常流程
- 使用 getattr 带默认值确保向后兼容

alembic/versions/034_add_confirm_generation_fields.py

@@ -0,0 +1,60 @@
+"""Add confirm generation fields
+
+Revision ID: 034
+Revises: 033
+Create Date: 2026-07-08
+
+确认生成 API 改造：为 generation_tasks 表添加 is_preview、source_task_id、
+output_width、output_height、cover_url、custom_title 字段。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "034"
+down_revision = "033"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generation_tasks",
+        sa.Column("is_preview", sa.Boolean(), nullable=False, server_default=sa.true()),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("source_task_id", sa.String(32), nullable=False, server_default=""),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("output_width", sa.Integer(), nullable=False, server_default=sa.text("1280")),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("output_height", sa.Integer(), nullable=False, server_default=sa.text("720")),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("cover_url", sa.String(1000), nullable=False, server_default=""),
+    )
+    op.add_column(
+        "generation_tasks",
+        sa.Column("custom_title", sa.String(500), nullable=False, server_default=""),
+    )
+    op.create_index(
+        op.f("ix_generation_tasks_source_task_id"),
+        "generation_tasks",
+        ["source_task_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_generation_tasks_source_task_id"), table_name="generation_tasks")
+    op.drop_column("generation_tasks", "custom_title")
+    op.drop_column("generation_tasks", "cover_url")
+    op.drop_column("generation_tasks", "output_height")
+    op.drop_column("generation_tasks", "output_width")
+    op.drop_column("generation_tasks", "source_task_id")
+    op.drop_column("generation_tasks", "is_preview")

apps/api/app/api/routes/generation_tasks.py

@@ -17,6 +17,7 @@ from app.schemas.generated_video import (
 )
 from app.schemas.generation_task import (
     BatchGenerationTaskResponse,
+    ConfirmGenerationRequest,
     CreateGenerationTaskRequest,
     GenerationTaskResponse,
     ListGenerationTasksResponse,
@@ -56,6 +57,12 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
         source_edit_plan_id=task.source_edit_plan_id or "",
         asset_select_mode=getattr(task, "asset_select_mode", ""),
         batch_id=getattr(task, "batch_id", ""),
+        is_preview=getattr(task, "is_preview", True),
+        source_task_id=getattr(task, "source_task_id", ""),
+        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", ""),
         status=task.status,
         progress=task.progress,
         result_count=task.result_count,
@@ -222,6 +229,12 @@ def create_generation_task(
                 source_edit_plan_id=request.source_edit_plan_id,
                 asset_select_mode=request.asset_select_mode,
                 batch_id=batch_id,
+                is_preview=request.is_preview,
+                source_task_id=request.source_task_id,
+                output_width=request.output_width,
+                output_height=request.output_height,
+                cover_url=request.cover_url,
+                custom_title=request.custom_title,
             )
         )
         celery_app.send_task("worker.generate_video", args=[task.id])
@@ -231,6 +244,63 @@ def create_generation_task(
     return BatchGenerationTaskResponse(items=items, total=len(items))
 
 
+@router.post("/tasks/{task_id}/confirm", response_model=BatchGenerationTaskResponse)
+def confirm_generation(
+    task_id: str,
+    request: ConfirmGenerationRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    generation_task_repository: Any = Depends(get_generation_task_repository),
+    project_repository: Any = Depends(get_project_repository),
+) -> BatchGenerationTaskResponse:
+    """确认生成 — 基于预览任务创建正式生成任务。
+
+    查找预览任务，复制其配置，创建新的正式生成任务（is_preview=False），
+    使用高分辨率，复用 worker.generate_video 渲染路径。
+    """
+    # 1. 查找源预览任务
+    source_task = generation_task_repository.get(task_id)
+    if source_task is None:
+        raise HTTPException(status_code=404, detail=f"Preview task {task_id} not found")
+
+    # 2. 权限检查
+    if source_task.created_by_user_id and source_task.created_by_user_id != authenticated_user.user.id:
+        raise HTTPException(status_code=403, detail="Access denied to this task")
+    if source_task.project_id:
+        _check_project_access(source_task.project_id, authenticated_user.user.id, project_repository)
+
+    # 3. 创建正式生成任务，复制预览任务的配置
+    use_case = CreateGenerationTaskUseCase(generation_task_repository)
+    new_task = use_case.execute(
+        CreateGenerationTaskCommand(
+            project_id=source_task.project_id,
+            asset_library_id=source_task.asset_library_id,
+            strategy_id=source_task.strategy_id,
+            voice_library_id=source_task.voice_library_id,
+            template_id=source_task.template_id,
+            asset_ids=source_task.asset_ids,
+            title_ids=source_task.title_ids,
+            voice_ids=source_task.voice_ids,
+            created_by_user_id=authenticated_user.user.id,
+            source_edit_plan_id=source_task.source_edit_plan_id or "",
+            asset_select_mode=source_task.asset_select_mode,
+            is_preview=False,
+            source_task_id=task_id,
+            output_width=request.output_width,
+            output_height=request.output_height,
+            cover_url=request.cover_url,
+            custom_title=request.custom_title,
+        )
+    )
+
+    # 4. 调度 worker.generate_video（同一条渲染路径）
+    celery_app.send_task("worker.generate_video", args=[new_task.id])
+
+    return BatchGenerationTaskResponse(
+        items=[_to_generation_task_response(new_task)],
+        total=1,
+    )
+
+
 @router.get("/tasks", response_model=ListGenerationTasksResponse)
 def list_generation_tasks(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -306,6 +376,12 @@ def retry_generation_task(
             created_by_user_id=authenticated_user.user.id,
             source_edit_plan_id=task.source_edit_plan_id or "",
             asset_select_mode=getattr(task, "asset_select_mode", ""),
+            is_preview=getattr(task, "is_preview", True),
+            source_task_id=getattr(task, "source_task_id", ""),
+            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", ""),
         )
     )
     celery_app.send_task("worker.generate_video", args=[retried.id])

apps/api/app/api/routes/task_center.py

@@ -151,6 +151,12 @@ def retry_task_by_id(
             title_ids=task.title_ids,
             voice_ids=task.voice_ids,
             created_by_user_id=authenticated_user.user.id,
+            is_preview=getattr(task, "is_preview", True),
+            source_task_id=getattr(task, "source_task_id", ""),
+            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", ""),
         )
     )
     celery_app.send_task("worker.generate_video", args=[retried.id])
@@ -233,6 +239,12 @@ def retry_project_task(
                 title_ids=task.title_ids,
                 voice_ids=task.voice_ids,
                 created_by_user_id=authenticated_user.user.id,
+                is_preview=getattr(task, "is_preview", True),
+                source_task_id=getattr(task, "source_task_id", ""),
+                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", ""),
             )
         )
         celery_app.send_task("worker.generate_video", args=[retried.id])

apps/api/app/schemas/generation_task.py

@@ -1,6 +1,15 @@
 from pydantic import BaseModel, Field, model_validator
 
 
+class ConfirmGenerationRequest(BaseModel):
+    """确认生成请求体 — 基于预览任务创建正式生成任务"""
+
+    output_width: int = Field(default=1080, description="输出视频宽度")
+    output_height: int = Field(default=1920, description="输出视频高度")
+    cover_url: str = Field(default="", description="自定义封面图片 URL")
+    custom_title: str = Field(default="", description="自定义视频标题")
+
+
 class CreateGenerationTaskRequest(BaseModel):
     """创建生成任务请求。
 
@@ -31,6 +40,13 @@ class CreateGenerationTaskRequest(BaseModel):
     asset_select_count: int = Field(
         default=0, ge=0, le=100, description="选取数量，0表示全部（仅 random/smart 模式有效）"
     )
+    # ── 预览 / 确认生成 ──
+    is_preview: bool = Field(default=True, description="是否为预览任务")
+    source_task_id: str = Field(default="", description="来源预览任务 ID（确认生成时传入）")
+    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":
@@ -58,6 +74,12 @@ class GenerationTaskResponse(BaseModel):
     source_edit_plan_id: str = ""
     asset_select_mode: str = ""
     batch_id: str = ""
+    is_preview: bool = True
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
     status: str
     progress: float
     result_count: int

apps/worker/worker_app/tasks/generation.py

@@ -96,7 +96,9 @@ def _probe_duration(local_path: Path) -> float:
         return OUTPUT_DURATION_SECONDS
 
 
-def _create_fallback_clip(output_path: Path, title: str) -> None:
+def _create_fallback_clip(
+    output_path: Path, title: str, width: int = OUTPUT_WIDTH, height: int = OUTPUT_HEIGHT
+) -> None:
     """创建 fallback 视频（无素材时）"""
     safe_title = title.replace(":", "\\:").replace("'", "\\'")[:80]
     _run_ffmpeg(
@@ -106,7 +108,7 @@ def _create_fallback_clip(output_path: Path, title: str) -> None:
             "-f",
             "lavfi",
             "-i",
-            f"color=c=#111827:s={OUTPUT_WIDTH}x{OUTPUT_HEIGHT}:d={OUTPUT_DURATION_SECONDS}:r={int(OUTPUT_FPS)}",
+            f"color=c=#111827:s={width}x{height}:d={OUTPUT_DURATION_SECONDS}:r={int(OUTPUT_FPS)}",
             "-vf",
             f"drawtext=text='{safe_title}':fontcolor=white:fontsize=48:x=(w-text_w)/2:y=(h-text_h)/2",
             "-c:v",
@@ -204,6 +206,8 @@ def _process_with_editing_mode(
     audio_path: Optional[str],
     mode: str,
     output_path: Path,
+    output_width: int = OUTPUT_WIDTH,
+    output_height: int = OUTPUT_HEIGHT,
 ) -> None:
     """根据剪辑模式处理视频"""
     from video_processing.editing_modes import (
@@ -215,8 +219,8 @@ def _process_with_editing_mode(
 
     config = EditingModeConfig(
         mode=EditingMode(mode),
-        output_width=OUTPUT_WIDTH,
-        output_height=OUTPUT_HEIGHT,
+        output_width=output_width,
+        output_height=output_height,
         output_fps=int(OUTPUT_FPS),
         pip_position=PIPPosition.TOP_RIGHT,
         pip_scale=0.25,
@@ -262,6 +266,11 @@ def generate_video(self, task_id: str) -> dict:
         mode = gen_task.strategy_id or "one_take"
         task_asset_ids = list(gen_task.asset_ids or [])
         batch_id = getattr(gen_task, "batch_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
+        cover_url = getattr(gen_task, "cover_url", "") or ""
+        custom_title = getattr(gen_task, "custom_title", "") or ""
     finally:
         session.close()
 
@@ -293,9 +302,13 @@ def generate_video(self, task_id: str) -> dict:
                     audio_path=audio_path,
                     mode=editing_mode.value,
                     output_path=output_path,
+                    output_width=output_width,
+                    output_height=output_height,
                 )
             else:
-                _create_fallback_clip(output_path, f"Generated Video {task_id[:8]}")
+                _create_fallback_clip(
+                    output_path, f"Generated Video {task_id[:8]}", width=output_width, height=output_height
+                )
 
             file_size = output_path.stat().st_size
             duration = _probe_duration(output_path)
@@ -324,6 +337,8 @@ def generate_video(self, task_id: str) -> dict:
                 duration=duration,
                 video_path=str(output_path),
                 mode=editing_mode.value,
+                width=output_width,
+                height=output_height,
             )
 
             return {
@@ -332,8 +347,8 @@ def generate_video(self, task_id: str) -> dict:
                 "output_path": str(output_path),
                 "file_size": file_size,
                 "duration": duration,
-                "width": OUTPUT_WIDTH,
-                "height": OUTPUT_HEIGHT,
+                "width": output_width,
+                "height": output_height,
                 "mode": editing_mode.value,
             }
     except Exception as error:
@@ -355,6 +370,8 @@ def _create_video_record_and_dedup(
     duration: float,
     video_path: str,
     mode: str,
+    width: int = OUTPUT_WIDTH,
+    height: int = OUTPUT_HEIGHT,
 ) -> None:
     """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。"""
     from uuid import uuid4
@@ -378,8 +395,8 @@ def _create_video_record_and_dedup(
             file_url=file_url,
             file_size=file_size,
             duration=duration,
-            width=OUTPUT_WIDTH,
-            height=OUTPUT_HEIGHT,
+            width=width,
+            height=height,
             fps=OUTPUT_FPS,
             status="completed",
             generation_params={"mode": mode},

docs/schema-metadata-snapshot.json

@@ -1533,6 +1533,54 @@
           "type": "VARCHAR(32)",
           "unique": false
         },
+        {
+          "index": false,
+          "name": "is_preview",
+          "nullable": false,
+          "primary_key": false,
+          "type": "BOOLEAN",
+          "unique": false
+        },
+        {
+          "index": true,
+          "name": "source_task_id",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(32)",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "output_width",
+          "nullable": false,
+          "primary_key": false,
+          "type": "INTEGER",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "output_height",
+          "nullable": false,
+          "primary_key": false,
+          "type": "INTEGER",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "cover_url",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(1000)",
+          "unique": false
+        },
+        {
+          "index": false,
+          "name": "custom_title",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(500)",
+          "unique": false
+        },
         {
           "index": false,
           "name": "metadata",
@@ -1593,6 +1641,13 @@
           "name": "ix_generation_tasks_source_edit_plan_id",
           "unique": false
         },
+        {
+          "columns": [
+            "source_task_id"
+          ],
+          "name": "ix_generation_tasks_source_task_id",
+          "unique": false
+        },
         {
           "columns": [
             "status"

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -27,6 +27,12 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         source_edit_plan_id=model.source_edit_plan_id or "",
         asset_select_mode=model.asset_select_mode or "",
         batch_id=model.batch_id or "",
+        is_preview=getattr(model, "is_preview", True),
+        source_task_id=getattr(model, "source_task_id", "") or "",
+        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 "",
         created_at=model.created_at,
     )
 
@@ -56,6 +62,12 @@ class SQLAlchemyGenerationTaskRepository:
             source_edit_plan_id=task.source_edit_plan_id or None,
             asset_select_mode=task.asset_select_mode or "",
             batch_id=task.batch_id or "",
+            is_preview=task.is_preview,
+            source_task_id=task.source_task_id or "",
+            output_width=task.output_width,
+            output_height=task.output_height,
+            cover_url=task.cover_url or "",
+            custom_title=task.custom_title or "",
             created_at=task.created_at,
         )
         self.session.add(model)
@@ -129,5 +141,11 @@ class SQLAlchemyGenerationTaskRepository:
         model.source_edit_plan_id = task.source_edit_plan_id or None
         model.asset_select_mode = task.asset_select_mode or ""
         model.batch_id = task.batch_id or ""
+        model.is_preview = task.is_preview
+        model.source_task_id = task.source_task_id or ""
+        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 ""
         self.session.commit()
         return task

packages/adapters/sqlalchemy_impl/models.py

@@ -255,6 +255,12 @@ class GenerationTaskModel(Base):
     source_edit_plan_id = Column(String(32), nullable=True, index=True)
     asset_select_mode = Column(String(20), nullable=False, default="")
     batch_id = Column(String(32), nullable=False, default="", index=True)
+    is_preview = Column(Boolean, nullable=False, default=True)
+    source_task_id = Column(String(32), nullable=False, default="", index=True)
+    output_width = Column(Integer, nullable=False, default=1280)
+    output_height = Column(Integer, nullable=False, default=720)
+    cover_url = Column(String(1000), nullable=False, default="")
+    custom_title = Column(String(500), nullable=False, default="")
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 

packages/application/generation_tasks.py

@@ -21,6 +21,12 @@ class CreateGenerationTaskCommand:
     source_edit_plan_id: str = ""
     asset_select_mode: str = ""
     batch_id: str = ""
+    is_preview: bool = True
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
 
 
 class CreateGenerationTaskUseCase:
@@ -48,6 +54,12 @@ class CreateGenerationTaskUseCase:
             source_edit_plan_id=command.source_edit_plan_id,
             asset_select_mode=command.asset_select_mode,
             batch_id=command.batch_id,
+            is_preview=command.is_preview,
+            source_task_id=command.source_task_id,
+            output_width=command.output_width,
+            output_height=command.output_height,
+            cover_url=command.cover_url,
+            custom_title=command.custom_title,
         )
         return self.generation_task_repository.create(task)
 

packages/domain/generation_task.py

@@ -45,6 +45,12 @@ class GenerationTask:
     created_by_user_id: str = ""
     asset_select_mode: str = ""
     batch_id: str = ""
+    is_preview: bool = True
+    source_task_id: str = ""
+    output_width: int = 1280
+    output_height: int = 720
+    cover_url: str = ""
+    custom_title: str = ""
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
     @classmethod
@@ -63,6 +69,12 @@ class GenerationTask:
         source_edit_plan_id: str = "",
         asset_select_mode: str = "",
         batch_id: str = "",
+        is_preview: bool = True,
+        source_task_id: str = "",
+        output_width: int = 1280,
+        output_height: int = 720,
+        cover_url: str = "",
+        custom_title: str = "",
     ) -> "GenerationTask":
         if not project_id.strip() and not template_id.strip():
             raise ValueError("project_id 或 template_id 至少需要提供一个")
@@ -82,4 +94,10 @@ class GenerationTask:
             source_edit_plan_id=source_edit_plan_id.strip(),
             asset_select_mode=asset_select_mode,
             batch_id=batch_id,
+            is_preview=is_preview,
+            source_task_id=source_task_id,
+            output_width=output_width,
+            output_height=output_height,
+            cover_url=cover_url,
+            custom_title=custom_title,
         )

tests/unit/test_confirm_generation.py

@@ -0,0 +1,379 @@
+"""确认生成 API 单元测试.
+
+覆盖 POST /tasks/{task_id}/confirm 端点：
+  - 正常确认流程
+  - 预览任务不存在 → 404
+  - 权限不足 → 403
+  - is_preview=False 及分辨率正确
+  - cover_url 和 custom_title 正确传递
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from dataclasses import dataclass, field
+from datetime import datetime, 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
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
+
+from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
+
+# ── Stub Repository ──────────────────────────────────────────────────────────
+
+
+class StubGenerationTaskRepository:
+    """内存中模拟 GenerationTask 仓储"""
+
+    def __init__(self) -> None:
+        self._store: dict[str, Any] = {}
+
+    def create(self, task: Any) -> Any:
+        self._store[task.id] = task
+        return task
+
+    def get(self, task_id: str) -> Optional[Any]:
+        return self._store.get(task_id)
+
+    def update(self, task: Any) -> Any:
+        if task.id not in self._store:
+            raise ValueError(f"GenerationTask {task.id} not found")
+        self._store[task.id] = task
+        return task
+
+    def list_by_project(self, project_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.project_id == project_id]
+
+    def list_by_user(self, user_id: str) -> list[Any]:
+        return [t for t in self._store.values() if t.created_by_user_id == user_id]
+
+    def count_by_user(self, user_id: str) -> int:
+        return len([t for t in self._store.values() if t.created_by_user_id == user_id])
+
+    def list_recent_by_user(self, user_id: str, limit: int = 5) -> list[Any]:
+        items = [t for t in self._store.values() if t.created_by_user_id == user_id]
+        items.sort(key=lambda t: t.created_at, reverse=True)
+        return items[:limit]
+
+    def list_by_source_edit_plan(self, plan_id: str) -> list[Any]:
+        return [t for t in self._store.values() if (t.source_edit_plan_id or "") == plan_id]
+
+
+# ── Stub Project Repository ──────────────────────────────────────────────────
+
+
+@dataclass
+class FakeProject:
+    id: str = "project-001"
+    owner_user_id: str = "user-001"
+    shared_users: list[str] = field(default_factory=list)
+    name: str = "Test Project"
+
+    def can_access(self, user_id: str) -> bool:
+        return user_id == self.owner_user_id or user_id in self.shared_users
+
+
+class StubProjectRepository:
+    def __init__(self) -> None:
+        self._projects: dict[str, FakeProject] = {}
+
+    def add(self, project: FakeProject) -> None:
+        self._projects[project.id] = project
+
+    def find_by_id(self, project_id: str) -> Optional[FakeProject]:
+        return self._projects.get(project_id)
+
+
+# ── Fixtures ─────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeUser:
+    id: str = "user-001"
+    email: str = "test@example.com"
+
+
+@dataclass
+class FakeAuthenticatedUser:
+    user: FakeUser = field(default_factory=FakeUser)
+    session_id: str | None = None
+    token_type: str | None = None
+
+
+@pytest.fixture
+def gen_task_repo() -> StubGenerationTaskRepository:
+    return StubGenerationTaskRepository()
+
+
+@pytest.fixture
+def project_repo() -> StubProjectRepository:
+    repo = StubProjectRepository()
+    repo.add(FakeProject())
+    return repo
+
+
+@pytest.fixture
+def app(
+    gen_task_repo: StubGenerationTaskRepository,
+    project_repo: StubProjectRepository,
+) -> FastAPI:
+    """构建测试 FastAPI 应用，注入 Stub Repository"""
+    from app.api.routes.generation_tasks import router
+    from app.auth import get_current_user
+    from app.dependencies import (
+        get_asset_library_repository,
+        get_asset_repository,
+        get_generated_video_repository,
+        get_generation_task_repository,
+        get_project_repository,
+    )
+
+    test_app = FastAPI()
+    test_app.include_router(router, prefix="/api/v1")
+
+    def override_get_current_user():
+        return FakeAuthenticatedUser()
+
+    def override_get_generation_task_repository():
+        return gen_task_repo
+
+    def override_get_project_repository():
+        return project_repo
+
+    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()
+
+    yield test_app
+    test_app.dependency_overrides.clear()
+
+
+@pytest.fixture
+def client(app: FastAPI) -> TestClient:
+    return TestClient(app)
+
+
+def _make_preview_task(**kwargs: Any) -> GenerationTask:
+    """创建预览任务"""
+    defaults = dict(
+        id="preview-task-001",
+        project_id="project-001",
+        asset_library_id="library-001",
+        strategy_id="one_take",
+        voice_library_id="",
+        template_id="",
+        asset_ids=["asset-1"],
+        title_ids=[],
+        voice_ids=[],
+        status=GenerationTaskStatus.COMPLETED,
+        progress=100.0,
+        result_count=1,
+        error_message="",
+        created_by_user_id="user-001",
+        source_edit_plan_id="",
+        asset_select_mode="all",
+        is_preview=True,
+        source_task_id="",
+        output_width=1280,
+        output_height=720,
+        cover_url="",
+        custom_title="",
+    )
+    defaults.update(kwargs)
+    return GenerationTask(**defaults)
+
+
+# ── Tests ────────────────────────────────────────────────────────────────────
+
+
+class TestConfirmGeneration:
+    def test_confirm_success(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """正常确认流程：预览任务存在、权限正确 → 创建正式任务"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            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["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"]
+
+        # 验证 celery 任务被调度
+        mock_celery.send_task.assert_called_once()
+        call_args = mock_celery.send_task.call_args
+        assert call_args[0][0] == "worker.generate_video"
+
+    def test_confirm_not_found(self, client: TestClient) -> None:
+        """预览任务不存在 → 404"""
+        resp = client.post(
+            "/api/v1/tasks/nonexistent-task/confirm",
+            json={"output_width": 1080, "output_height": 1920},
+        )
+        assert resp.status_code == 404
+        assert "not found" in resp.json()["detail"]
+
+    def test_confirm_access_denied(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """权限不足 → 403"""
+        preview = _make_preview_task(created_by_user_id="other-user-999")
+        gen_task_repo.create(preview)
+
+        resp = client.post(
+            f"/api/v1/tasks/{preview.id}/confirm",
+            json={"output_width": 1080, "output_height": 1920},
+        )
+        assert resp.status_code == 403
+        assert "denied" in resp.json()["detail"].lower() or "Access" in resp.json()["detail"]
+
+    def test_confirm_preserves_config(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """确认后的任务 is_preview=False，分辨率已更新，其余配置从预览任务复制"""
+        preview = _make_preview_task(
+            voice_library_id="voice-001",
+            template_id="tmpl-001",
+            title_ids=["title-1", "title-2"],
+            voice_ids=["voice-a"],
+        )
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            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"]
+        assert item["voice_ids"] == ["voice-a"]
+
+    def test_confirm_cover_and_title(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """cover_url 和 custom_title 正确传递"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            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]
+        assert item["cover_url"] == "https://cdn.example.com/my-cover.png"
+        assert item["custom_title"] == "测试视频标题"
+
+    def test_confirm_default_resolution(
+        self,
+        client: TestClient,
+        gen_task_repo: StubGenerationTaskRepository,
+    ) -> None:
+        """不传分辨率时使用默认值 1080x1920"""
+        preview = _make_preview_task()
+        gen_task_repo.create(preview)
+
+        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            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.celery_app") as mock_celery:
+            mock_celery.send_task = MagicMock()
+            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