apps/api/app/api/routes/generation_cover.py

@@ -49,6 +49,10 @@ class GenerateCoverRequest(BaseModel):
         ge=0.0,
         description="手动选帧时间点（秒），仅 cover_type=manual 时有效",
     )
+    cover_url: Optional[str] = Field(
+        default=None,
+        description="上传的封面图片 URL，仅 cover_type=upload 时有效",
+    )
 
 
 class GenerateCoverResponse(BaseModel):
@@ -80,6 +84,29 @@ def generate_cover(
     _, plan_svc = services
     plan = plan_svc.get_plan_or_raise(plan_id)
 
+    # ── upload 类型：直接保存前端上传的封面图片，不需要预览视频 ──────
+    if body.cover_type == "upload":
+        if not body.cover_url:
+            raise HTTPException(
+                status_code=400,
+                detail="cover_type=upload 时必须提供 cover_url",
+            )
+        cover_data = {
+            "type": "upload",
+            "image_url": body.cover_url,
+        }
+        current_config = dict(plan.config) if plan.config else {}
+        current_config["cover"] = cover_data
+        normalized = normalize_plan_config(current_config)
+        plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
+        logger.info(
+            "封面上传完成: plan_id=%s cover_url=%s by user=%s",
+            plan_id,
+            body.cover_url[:80] if body.cover_url else "",
+            current_user.user.id,
+        )
+        return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)
+
     # ── 3 步查找预览视频 URL ──────────────────────────────────────────
     # 第一步：从 plan.config 读取
     logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id)

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

@@ -8,8 +8,9 @@
 
 from __future__ import annotations
 
+import json
 import logging
-from typing import Any
+from typing import Any, Optional
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
@@ -45,6 +46,7 @@ from ._fallback import (
 from .dependencies import _check_queue_limits, get_draft_plan_id, get_editor_services
 from .schemas import (
     ClipStatusItem,
+    EditPlanGenerateRequest,
     EditPlanGenerateResponse,
     EditPlanGenerationsResponse,
     EditPlanGenerationStatusResponse,
@@ -57,6 +59,7 @@ router = APIRouter(tags=["Template Editor"])
 @router.post("/generate", response_model=EditPlanGenerateResponse)
 def generate_editor_draft(
     template_id: str,
+    request: Optional[EditPlanGenerateRequest] = None,
     plan_id: str = Depends(get_draft_plan_id),
     services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
     db: Session = Depends(get_db_session),
@@ -65,6 +68,7 @@ def generate_editor_draft(
     asset_repo: Any = Depends(get_asset_repository),
 ) -> EditPlanGenerateResponse:
     """触发模板草稿渲染生成"""
+    req = request or EditPlanGenerateRequest()
     _, plan_svc = services
     plan_check = plan_svc.get_plan_or_raise(plan_id)
 
@@ -85,6 +89,36 @@ def generate_editor_draft(
     # 检查是否可复用已完成的预览产物（预览品质已与正式一致）
     gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
     reusable_task = _find_reusable_preview_task(gen_task_repo, plan_id, plan_check)
+    if reusable_task:
+        # 复用预览产物：标记为正式产出，跳过渲染
+        # 如果前端传了 title_config，需要创建新任务（因为预览任务的 custom_title 可能不同）
+        title_config_reuse = req.title_config or {}
+        title_text_reuse = (title_config_reuse.get("text") or "").strip()
+        existing_custom_title = getattr(reusable_task, "custom_title", "") or ""
+        if title_text_reuse and existing_custom_title:
+            # 如果新标题和已有标题不同，不能复用，走新建任务流程
+            new_title_json = json.dumps(title_config_reuse, ensure_ascii=False)
+            if new_title_json != existing_custom_title:
+                logger.info(
+                    "[模板生成] 标题已变更，跳过复用: task_id=%s",
+                    reusable_task.id,
+                )
+                reusable_task = None
+        elif title_text_reuse and not existing_custom_title:
+            # 原来没标题，现在有标题，不能复用
+            logger.info(
+                "[模板生成] 新增标题，跳过复用: task_id=%s",
+                reusable_task.id,
+            )
+            reusable_task = None
+        elif not title_text_reuse and existing_custom_title:
+            # 原来有标题，现在移除了，不能复用
+            logger.info(
+                "[模板生成] 移除标题，跳过复用: task_id=%s",
+                reusable_task.id,
+            )
+            reusable_task = None
+
     if reusable_task:
         # 复用预览产物：标记为正式产出，跳过渲染
         reusable_task.mark_confirmed()
@@ -133,6 +167,21 @@ def generate_editor_draft(
         gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo)
         plan = plan_svc.get_plan_or_raise(plan_id)
         config_asset_ids = (plan.config or {}).get("asset_ids", [])
+        # 从 plan config 读取封面 URL（由 generate-cover 保存）
+        cover_url_from_config = (plan.config or {}).get("cover", {}).get("image_url", "")
+
+        # 处理标题配置：序列化 title_config 为 JSON 存入 custom_title
+        title_config = req.title_config or {}
+        title_text = (title_config.get("text") or "").strip()
+        custom_title_value = ""
+        if title_text:
+            custom_title_value = json.dumps(title_config, ensure_ascii=False)
+            logger.info(
+                "[模板生成] 标题配置: text=%s, config_keys=%s",
+                title_text[:30],
+                list(title_config.keys()),
+            )
+
         gen_task = gen_task_use_case.execute(
             CreateGenerationTaskCommand(
                 project_id=plan.project_id or "",
@@ -140,6 +189,8 @@ def generate_editor_draft(
                 created_by_user_id=current_user.user.id,
                 source_edit_plan_id=plan_id,
                 asset_ids=list(config_asset_ids) if config_asset_ids else [],
+                cover_url=cover_url_from_config,
+                custom_title=custom_title_value,
             ),
         )
 

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

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import re as _re
-from typing import Any, List, Optional
+from typing import Any, Dict, List, Optional
 
 from app.schemas.generation_task import GenerationTaskResponse
 from pydantic import BaseModel, Field, validator
@@ -44,6 +44,14 @@ class EditPlanGenerationStatusResponse(BaseModel):
     clips: List[ClipStatusItem]
 
 
+class EditPlanGenerateRequest(BaseModel):
+    """模板编辑器触发生成请求体"""
+    title_config: Optional[Dict[str, Any]] = Field(
+        default_factory=dict,
+        description="标题配置（可选），渲染时烧录到视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow",
+    )
+
+
 class EditPlanGenerateResponse(BaseModel):
     """剪辑计划触发生成响应体"""
 

tests/unit/test_generate_title_and_cover.py

@@ -0,0 +1,369 @@
+"""Tests for /generate endpoint — custom_title and cover_url passing."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+class TestGenerateEndpointTitleAndCover:
+    """测试 /generate 端点传递 custom_title 和 cover_url。"""
+
+    def test_generate_passes_cover_url_from_plan_config(self):
+        """从 plan.config.cover.image_url 读取封面 URL 传递给生成任务。"""
+        from app.api.routes.templates_editor.generation import generate_editor_draft
+        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
+
+        mock_plan = MagicMock()
+        mock_plan.id = "plan-123"
+        mock_plan.project_id = "project-1"
+        mock_plan.template_id = "template-1"
+        mock_plan.status = MagicMock(value="editing")
+        mock_plan.config = {
+            "clips": [{"id": "c1"}],
+            "asset_ids": ["a1"],
+            "cover": {"type": "upload", "image_url": "https://oss.example.com/uploaded/cover.jpg"},
+        }
+        mock_plan.updated_at = None
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_plan_svc.can_generate.return_value = (True, "")
+        mock_plan_svc.mark_clips_ready.return_value = 1
+        mock_template_svc = MagicMock()
+
+        mock_gen_task = MagicMock()
+        mock_gen_task.id = "task-new"
+        mock_gen_task.project_id = "project-1"
+
+        mock_current_user = MagicMock()
+        mock_current_user.user.id = "user-1"
+
+        body = EditPlanGenerateRequest()  # No title_config
+
+        with (
+            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
+            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
+            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
+            patch("app.api.routes.templates_editor.generation.celery_app"),
+            patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-123"),
+        ):
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+
+            mock_usecase = MagicMock()
+            mock_usecase.execute.return_value = mock_gen_task
+            mock_usecase_cls.return_value = mock_usecase
+
+            mock_plan_svc.transition_status = MagicMock()
+            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+            result = generate_editor_draft(
+                template_id="template-1",
+                request=body,
+                plan_id="plan-123",
+                services=(mock_template_svc, mock_plan_svc),
+                db=MagicMock(),
+                current_user=mock_current_user,
+                asset_library_repo=MagicMock(),
+                asset_repo=MagicMock(),
+            )
+
+            # Verify cover_url was passed to CreateGenerationTaskCommand
+            call_args = mock_usecase.execute.call_args
+            command = call_args[0][0]
+            assert command.cover_url == "https://oss.example.com/uploaded/cover.jpg"
+            assert command.custom_title == ""
+
+    def test_generate_passes_custom_title_from_title_config(self):
+        """前端传 title_config 时，序列化为 JSON 存入 custom_title。"""
+        from app.api.routes.templates_editor.generation import generate_editor_draft
+        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
+
+        mock_plan = MagicMock()
+        mock_plan.id = "plan-456"
+        mock_plan.project_id = "project-1"
+        mock_plan.template_id = "template-1"
+        mock_plan.status = MagicMock(value="editing")
+        mock_plan.config = {
+            "clips": [{"id": "c1"}],
+            "asset_ids": ["a1"],
+            "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/cover.jpg"},
+        }
+        mock_plan.updated_at = None
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_plan_svc.can_generate.return_value = (True, "")
+        mock_plan_svc.mark_clips_ready.return_value = 1
+        mock_template_svc = MagicMock()
+
+        mock_gen_task = MagicMock()
+        mock_gen_task.id = "task-title"
+
+        mock_current_user = MagicMock()
+        mock_current_user.user.id = "user-1"
+
+        title_config = {
+            "text": "测试标题",
+            "font_size": 36,
+            "font_color": "#ffffff",
+            "position": "center",
+        }
+        body = EditPlanGenerateRequest(title_config=title_config)
+
+        with (
+            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
+            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
+            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
+            patch("app.api.routes.templates_editor.generation.celery_app"),
+        ):
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+
+            mock_usecase = MagicMock()
+            mock_usecase.execute.return_value = mock_gen_task
+            mock_usecase_cls.return_value = mock_usecase
+
+            mock_plan_svc.transition_status = MagicMock()
+            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+            result = generate_editor_draft(
+                template_id="template-1",
+                request=body,
+                plan_id="plan-456",
+                services=(mock_template_svc, mock_plan_svc),
+                db=MagicMock(),
+                current_user=mock_current_user,
+                asset_library_repo=MagicMock(),
+                asset_repo=MagicMock(),
+            )
+
+            # Verify custom_title was serialized to JSON
+            call_args = mock_usecase.execute.call_args
+            command = call_args[0][0]
+            parsed_title = json.loads(command.custom_title)
+            assert parsed_title["text"] == "测试标题"
+            assert parsed_title["font_size"] == 36
+            assert command.cover_url == "https://oss.example.com/cover.jpg"
+
+    def test_generate_empty_title_config_passes_empty_custom_title(self):
+        """title_config 为空时 custom_title 为空字符串。"""
+        from app.api.routes.templates_editor.generation import generate_editor_draft
+        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
+
+        mock_plan = MagicMock()
+        mock_plan.id = "plan-789"
+        mock_plan.project_id = "project-1"
+        mock_plan.template_id = "template-1"
+        mock_plan.status = MagicMock(value="editing")
+        mock_plan.config = {"clips": [{"id": "c1"}], "asset_ids": ["a1"]}
+        mock_plan.updated_at = None
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_plan_svc.can_generate.return_value = (True, "")
+        mock_plan_svc.mark_clips_ready.return_value = 1
+        mock_template_svc = MagicMock()
+
+        mock_gen_task = MagicMock()
+        mock_gen_task.id = "task-no-title"
+
+        body = EditPlanGenerateRequest()  # No title_config
+
+        with (
+            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
+            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
+            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
+            patch("app.api.routes.templates_editor.generation.celery_app"),
+        ):
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+
+            mock_usecase = MagicMock()
+            mock_usecase.execute.return_value = mock_gen_task
+            mock_usecase_cls.return_value = mock_usecase
+
+            mock_plan_svc.transition_status = MagicMock()
+            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+            result = generate_editor_draft(
+                template_id="template-1",
+                request=body,
+                plan_id="plan-789",
+                services=(mock_template_svc, mock_plan_svc),
+                db=MagicMock(),
+                current_user=MagicMock(),
+                asset_library_repo=MagicMock(),
+                asset_repo=MagicMock(),
+            )
+
+            call_args = mock_usecase.execute.call_args
+            command = call_args[0][0]
+            assert command.custom_title == ""
+
+
+class TestGenerateEndpointRequestSchema:
+    """测试 EditPlanGenerateRequest schema。"""
+
+    def test_schema_default_empty_title_config(self):
+        """默认 title_config 为空 dict。"""
+        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
+
+        req = EditPlanGenerateRequest()
+        assert req.title_config == {}
+
+    def test_schema_accepts_title_config(self):
+        """可以传入标题配置。"""
+        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
+
+        req = EditPlanGenerateRequest(title_config={"text": "我的标题", "font_size": 48})
+        assert req.title_config["text"] == "我的标题"
+        assert req.title_config["font_size"] == 48
+
+
+class TestGenerateTitleChangeSkipsReuse:
+    """测试标题变更时跳过预览产物复用。"""
+
+    def _make_mocks(self, custom_title=""):
+        mock_plan = MagicMock()
+        mock_plan.id = "plan-reuse"
+        mock_plan.project_id = "project-1"
+        mock_plan.template_id = "template-1"
+        mock_plan.status = MagicMock(value="editing")
+        mock_plan.config = {"clips": [{"id": "c1"}], "asset_ids": ["a1"]}
+        mock_plan.updated_at = None
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_plan_svc.can_generate.return_value = (True, "")
+        mock_plan_svc.mark_clips_ready.return_value = 1
+        mock_template_svc = MagicMock()
+
+        reusable_task = MagicMock()
+        reusable_task.id = "task-reusable"
+        reusable_task.is_completed = True
+        reusable_task.is_preview = True
+        reusable_task.custom_title = custom_title
+        reusable_task.project_id = "project-1"
+        reusable_task.source_edit_plan_id = "plan-reuse"
+
+        mock_new_task = MagicMock()
+        mock_new_task.id = "task-new"
+
+        return mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task
+
+    def test_title_removed_skips_reuse(self):
+        """原来有标题，现在移除了 → 跳过复用，创建新任务。"""
+        from app.api.routes.templates_editor.generation import generate_editor_draft
+        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
+
+        mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task = self._make_mocks(
+            custom_title='{"text": "旧标题"}'
+        )
+
+        body = EditPlanGenerateRequest()  # No title_config → title removed
+
+        with (
+            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
+            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=reusable_task),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
+            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
+            patch("app.api.routes.templates_editor.generation.celery_app"),
+            patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-reuse"),
+        ):
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+
+            mock_usecase = MagicMock()
+            mock_usecase.execute.return_value = mock_new_task
+            mock_usecase_cls.return_value = mock_usecase
+
+            mock_plan_svc.transition_status = MagicMock()
+            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+            result = generate_editor_draft(
+                template_id="template-1",
+                request=body,
+                plan_id="plan-reuse",
+                services=(mock_template_svc, mock_plan_svc),
+                db=MagicMock(),
+                current_user=MagicMock(),
+                asset_library_repo=MagicMock(),
+                asset_repo=MagicMock(),
+            )
+
+            # 应该创建新任务而不是复用
+            mock_usecase.execute.assert_called_once()
+            # 不应该 mark_confirmed 在 reusable_task 上
+            reusable_task.mark_confirmed.assert_not_called()
+
+    def test_title_changed_skips_reuse(self):
+        """标题变更 → 跳过复用。"""
+        import json
+
+        from app.api.routes.templates_editor.generation import generate_editor_draft
+        from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest
+
+        mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task = self._make_mocks(
+            custom_title=json.dumps({"text": "旧标题", "font_size": 36}, ensure_ascii=False)
+        )
+
+        body = EditPlanGenerateRequest(title_config={"text": "新标题", "font_size": 48})
+
+        with (
+            patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls,
+            patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=reusable_task),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]),
+            patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"),
+            patch("app.api.routes.templates_editor.generation._check_queue_limits"),
+            patch("app.api.routes.templates_editor.generation.celery_app"),
+            patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-reuse"),
+        ):
+            mock_repo = MagicMock()
+            mock_repo_cls.return_value = mock_repo
+
+            mock_usecase = MagicMock()
+            mock_usecase.execute.return_value = mock_new_task
+            mock_usecase_cls.return_value = mock_usecase
+
+            mock_plan_svc.transition_status = MagicMock()
+            mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+            result = generate_editor_draft(
+                template_id="template-1",
+                request=body,
+                plan_id="plan-reuse",
+                services=(mock_template_svc, mock_plan_svc),
+                db=MagicMock(),
+                current_user=MagicMock(),
+                asset_library_repo=MagicMock(),
+                asset_repo=MagicMock(),
+            )
+
+            mock_usecase.execute.assert_called_once()
+            reusable_task.mark_confirmed.assert_not_called()

tests/unit/test_generation_cover.py

@@ -552,3 +552,138 @@ class TestStrayLoggerRemoved:
         assert (
             "logger.info(\n                            plan_id," not in source
         ), "Stray logger.info(plan_id, generation_task_id) should be removed"
+
+
+class TestUploadCoverType:
+    """测试 cover_type=upload 封面上传功能。"""
+
+    def test_upload_cover_saves_url_directly(self):
+        """cover_type=upload 时直接保存 cover_url，不需要预览视频。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {}
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+
+        mock_current_user = MagicMock()
+        mock_current_user.user.id = "user-upload"
+
+        body = GenerateCoverRequest(
+            cover_type="upload",
+            cover_url="https://oss.example.com/uploaded/cover.jpg",
+        )
+
+        with patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize:
+            mock_normalize.return_value = {
+                "cover": {"type": "upload", "image_url": "https://oss.example.com/uploaded/cover.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="template-upload",
+                plan_id="plan-upload",
+                services=(mock_template_svc, mock_plan_svc),
+                db=MagicMock(),
+                current_user=mock_current_user,
+            )
+
+            assert result.plan_id == "plan-upload"
+            assert result.cover["type"] == "upload"
+            assert result.cover["image_url"] == "https://oss.example.com/uploaded/cover.jpg"
+
+            # 验证 plan config 被更新
+            mock_plan_svc.update_plan_config.assert_called_once()
+            call_args = mock_plan_svc.update_plan_config.call_args
+            assert call_args[0][0] == "plan-upload"
+            assert call_args[0][1]["cover"]["type"] == "upload"
+
+    def test_upload_cover_without_url_returns_400(self):
+        """cover_type=upload 但未提供 cover_url 时返回 400。"""
+        from unittest.mock import MagicMock
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+        from fastapi import HTTPException
+
+        mock_plan = MagicMock()
+        mock_plan.config = {}
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+
+        body = GenerateCoverRequest(cover_type="upload")
+        # cover_url is None by default
+
+        import pytest
+        from app.api.routes.generation_cover import generate_cover
+
+        with pytest.raises(HTTPException) as exc_info:
+            generate_cover(
+                body=body,
+                template_id="template-upload",
+                plan_id="plan-upload",
+                services=(mock_template_svc, mock_plan_svc),
+                db=MagicMock(),
+                current_user=MagicMock(),
+            )
+
+        assert exc_info.value.status_code == 400
+        assert "cover_url" in exc_info.value.detail
+
+    def test_upload_cover_schema_has_cover_url_field(self):
+        """GenerateCoverRequest schema 包含 cover_url 字段。"""
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        req = GenerateCoverRequest(cover_type="upload", cover_url="https://example.com/img.jpg")
+        assert req.cover_url == "https://example.com/img.jpg"
+        assert req.cover_type == "upload"
+
+        # 默认值为 None
+        req2 = GenerateCoverRequest()
+        assert req2.cover_url is None
+
+    def test_upload_cover_does_not_require_preview_video(self):
+        """cover_type=upload 时不查找预览视频，即使 plan.config 为空也不报错。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {}  # 没有 rendered_storage_key
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+
+        body = GenerateCoverRequest(
+            cover_type="upload",
+            cover_url="https://oss.example.com/uploaded/my-cover.png",
+        )
+
+        with patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize:
+            mock_normalize.return_value = {
+                "cover": {"type": "upload", "image_url": "https://oss.example.com/uploaded/my-cover.png"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            # 不应该抛出 "请先生成预览视频" 的异常
+            result = generate_cover(
+                body=body,
+                template_id="template-1",
+                plan_id="plan-no-preview",
+                services=(mock_template_svc, mock_plan_svc),
+                db=MagicMock(),
+                current_user=MagicMock(),
+            )
+
+            assert result.cover["image_url"] == "https://oss.example.com/uploaded/my-cover.png"
+            # 验证没有调用任何预览视频查找逻辑
+            # (normalize_plan_config 是唯一被调用的外部函数)