From e09ad9f4b738220f85a04e9005c55d6cdccdbea8 Mon Sep 17 00:00:00 2001
From: xiaoxia <xiaoxia@xiaoxiajianji.com>
Date: Fri, 28 Aug 2026 22:58:45 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E5=B0=81=E9=9D=A2=E9=80=89=E5=B8=A7?=
 =?UTF-8?q?=E8=BE=93=E5=85=A5=E6=BA=90=E6=94=B9=E4=B8=BA=E6=9C=80=E7=BB=88?=
 =?UTF-8?q?=E6=88=90=E7=89=87=EF=BC=8C=E9=A2=84=E8=A7=88=E7=89=87=E6=AE=B5?=
 =?UTF-8?q?=E4=BD=9C=E4=B8=BA=E5=85=BC=E5=AE=B9=E5=9B=9E=E9=80=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- GenerateCoverRequest 新增可选字段 generated_video_id / video_url，
  与前端 PR #1533 契约对齐：前端在「选择封面」步骤传入确认生成产出的
  最终视频 ID 或 URL，后端据此定位最终成片文件并抽帧
- 视频源查找优先级调整：
  0a. 请求体 generated_video_id → GeneratedVideo.file_url
  0b. 请求体 video_url → 直接使用
  1. plan.config.rendered_storage_key
  2. plan.config.generation_task_id
  3. source_edit_plan 关联的 is_preview=False 已完成任务（最终成片优先）
  4. source_edit_plan 关联的 is_preview=True 预览任务（回退）
  5. user+template 最近预览任务（兜底）
- cover_url 查找同步调整：优先最终成片任务的 cover_url，再回退预览
- MediaKit 的 strategy/max_frames/轮询/重试/降级逻辑全部不变
- 权限校验：通过 generated_video_id 关联的 generation_task 校验归属
- 新增 5 个单测覆盖：schema 字段、video_id 抽帧、video_url 抽帧、
  最终成片优先于预览、权限校验 403
---
 apps/api/app/api/routes/generation_cover.py | 322 ++++++++++++++------
 tests/unit/test_generation_cover.py         | 301 ++++++++++++++++++
 2 files changed, 533 insertions(+), 90 deletions(-)

diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py
index ef4400615..b42283c52 100644
--- a/apps/api/app/api/routes/generation_cover.py
+++ b/apps/api/app/api/routes/generation_cover.py
@@ -1,7 +1,7 @@
 """封面生成路由 — Generation 模块.
 
 端点：
-- POST /generate-cover     AI 生成封面（从预览视频中抽帧）
+- POST /generate-cover     AI 生成封面（从最终成片视频中抽帧，兼容预览片段回退）
 
 挂载路径: /api/v1/generation/generate-cover
 """
@@ -51,6 +51,16 @@ class GenerateCoverRequest(BaseModel):
         default=None,
         description="上传的封面图片 URL，仅 cover_type=upload 时有效",
     )
+    generated_video_id: Optional[str] = Field(
+        default=None,
+        description="确认生成产出的最终视频 ID。传入后封面从该视频文件抽帧，"
+        "而非预览片段。",
+    )
+    video_url: Optional[str] = Field(
+        default=None,
+        description="最终视频 URL（兜底）。当 generated_video_id 不可用时，"
+        "直接从此 URL 对应的视频抽帧。",
+    )
 
 
 class GenerateCoverResponse(BaseModel):
@@ -140,6 +150,41 @@ def _persist_cover_frame(
             Path(tmp_path).unlink(missing_ok=True)
 
 
+
+def _get_task_video_url(db: Session, task_id: str) -> Optional[str]:
+    """从 GenerationTask 关联的 GeneratedVideo 中获取视频 storage_key / URL."""
+    try:
+        video_repo = get_generated_video_repository(db)
+        use_case = ListGeneratedVideosByTaskUseCase(video_repo)
+        videos = use_case.execute(task_id)
+        if videos:
+            return getattr(videos[0], "file_url", "") or ""
+    except Exception:
+        logger.warning("[封面生成] 获取任务视频失败: task_id=%s", task_id, exc_info=True)
+    return None
+
+
+def _resolve_storage_key_to_url(storage_key: str) -> Optional[str]:
+    """将 storage_key 或完整 URL 转换为可访问的裸 URL。"""
+    if not storage_key:
+        return None
+    try:
+        if storage_key.startswith("http"):
+            url = storage_key
+        else:
+            from packages.shared.storage import get_shared_storage_service
+
+            storage_svc = get_shared_storage_service()
+            url = storage_svc.get_url(storage_key)
+        if url:
+            import re as _re
+            url = _re.sub(r"(?<!:)//", "/", url)
+        return url
+    except Exception as e:
+        logger.warning("[封面生成] storage_key 转 URL 失败: key=%s err=%s", storage_key, e)
+        return None
+
+
 @router.post("/generate-cover", response_model=GenerateCoverResponse)
 def generate_cover(
     body: GenerateCoverRequest,
@@ -149,12 +194,16 @@ def generate_cover(
     db: Session = Depends(get_db_session),
     current_user: AuthenticatedUser = Depends(get_current_user),
 ) -> GenerateCoverResponse:
-    """AI 生成封面 — 从预览视频中抽帧.
+    """AI 生成封面 — 优先从最终成片视频中抽帧，回退到预览片段.
 
     流程（串行）：
-    1. 预览视频已渲染完成（通过 3 步查找获取 URL）
-    2. 用裸 URL 让 MediaKit 下载视频并抽帧
-    3. 帧图下载后上传到 OSS covers/ 路径
+    1. 优先使用前端传入的 generation_task_id 定位最终成片任务，
+       或自动查找 plan 关联的已完成最终成片任务（is_preview=False）
+    2. 回退：从预览片段获取视频 URL（兼容旧流程）
+    3. 用裸 URL 让 MediaKit 下载视频并抽帧
+    4. 帧图下载后上传到 OSS covers/ 路径
+
+    MediaKit 的调用方式（strategy / max_frames / 轮询 / 重试 / 降级）不变。
     """
     _, plan_svc = services
     plan = plan_svc.get_plan_or_raise(plan_id)
@@ -182,122 +231,174 @@ def generate_cover(
         )
         return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)
 
-    # ── 3 步查找预览视频 URL ──────────────────────────────────────────
-    # 第一步：从 plan.config 读取
+    # ── 查找用于抽帧的视频 URL ────────────────────────────────────────
+    # 优先级：
+    #   0. 请求体显式传入的 generation_task_id（最终成片任务）
+    #   1. plan.config.rendered_storage_key
+    #   2. plan.config.generation_task_id 对应的任务
+    #   3. source_edit_plan_id 关联的已完成「最终成片」任务（is_preview=False）
+    #   4. source_edit_plan_id 关联的已完成预览任务（is_preview=True，兼容回退）
+    #   5. user + template 最近的已完成预览任务（兜底）
     logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id)
     rendered_storage_key = (plan.config or {}).get("rendered_storage_key", "")
 
-    # 第二步：如果还没有，通过 generation_task_id 查找预览任务的产物
+    # 步骤 0：请求体传入最终视频标识（generated_video_id 或 video_url）
+    if not rendered_storage_key:
+        # 0a：通过 generated_video_id 查找最终成片视频
+        if body.generated_video_id:
+            logger.info(
+                "[封面生成] 步骤0a: 使用 generated_video_id: plan_id=%s video_id=%s",
+                plan_id, body.generated_video_id,
+            )
+            try:
+                from app.dependencies import get_generated_video_repository as _get_gv_repo
+
+                gv_repo = _get_gv_repo(db)
+                gv = gv_repo.get(body.generated_video_id)
+                if gv:
+                    file_url = getattr(gv, "file_url", "") or ""
+                    if file_url:
+                        # 权限校验：通过关联的 generation_task 校验归属
+                        gv_task_id = getattr(gv, "generation_task_id", "") or ""
+                        if gv_task_id:
+                            try:
+                                _repo0 = SQLAlchemyGenerationTaskRepository(db)
+                                task0 = _repo0.get(gv_task_id)
+                                if (
+                                    task0
+                                    and task0.created_by_user_id
+                                    and task0.created_by_user_id != current_user.user.id
+                                ):
+                                    raise HTTPException(status_code=403, detail="无权访问该视频")
+                            except HTTPException:
+                                raise
+                            except Exception:
+                                pass
+                        rendered_storage_key = file_url
+                        logger.info(
+                            "[封面生成] ✅ 步骤0a找到最终成片: plan_id=%s video_id=%s url=%s",
+                            plan_id, body.generated_video_id, file_url[:80],
+                        )
+            except HTTPException:
+                raise
+            except Exception:
+                logger.warning(
+                    "[封面生成] 步骤0a查找视频失败: plan_id=%s video_id=%s",
+                    plan_id, body.generated_video_id, exc_info=True,
+                )
+
+        # 0b：直接使用 video_url（兜底）
+        if not rendered_storage_key and body.video_url:
+            logger.info(
+                "[封面生成] 步骤0b: 使用请求体传入的 video_url: plan_id=%s url=%s",
+                plan_id, body.video_url[:80],
+            )
+            rendered_storage_key = body.video_url
+
+    # 步骤 2：通过 plan.config.generation_task_id 查找
     if not rendered_storage_key:
         generation_task_id = (plan.config or {}).get("generation_task_id", "")
-        logger.info(
-            "[封面生成] 步骤2: 通过 generation_task_id 查找: plan_id=%s task_id=%s", plan_id, generation_task_id
-        )
         if generation_task_id:
+            logger.info(
+                "[封面生成] 步骤2: 通过 plan.config.generation_task_id 查找: plan_id=%s task_id=%s",
+                plan_id, generation_task_id,
+            )
             try:
-                gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
-                task = gen_task_repo.get(generation_task_id)
+                _repo = SQLAlchemyGenerationTaskRepository(db)
+                task = _repo.get(generation_task_id)
                 if task:
-                    video_repo = get_generated_video_repository(db)
-                    use_case = ListGeneratedVideosByTaskUseCase(video_repo)
-                    videos = use_case.execute(task.id)
-                    if videos:
-                        rendered_storage_key = getattr(videos[0], "file_url", "") or ""
+                    rendered_storage_key = _get_task_video_url(db, task.id) or ""
+                    if rendered_storage_key:
                         logger.info(
                             "[封面生成] ✅ 步骤2找到视频: plan_id=%s task_id=%s url=%s",
-                            plan_id,
-                            generation_task_id,
-                            rendered_storage_key[:80],
+                            plan_id, generation_task_id, rendered_storage_key[:80],
                         )
             except Exception:
                 logger.warning(
-                    "封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s",
-                    plan_id,
-                    exc_info=True,
+                    "[封面生成] 步骤2查找失败: plan_id=%s", plan_id, exc_info=True,
                 )
 
-    # 第 2.5 步：通过 plan_id 作为 source_edit_plan_id 查找关联的已完成预览任务
+    # 步骤 3：通过 source_edit_plan_id 查找已完成「最终成片」任务（is_preview=False）
     if not rendered_storage_key:
         try:
-            gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
-            logger.info("[封面生成] 步骤2.5: 通过 source_edit_plan_id 查找: plan_id=%s", plan_id)
-            preview_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
-            for pt in preview_tasks:
-                if getattr(pt, "status", "") == "completed" and getattr(pt, "is_preview", False):
-                    video_repo = get_generated_video_repository(db)
-                    use_case = ListGeneratedVideosByTaskUseCase(video_repo)
-                    videos = use_case.execute(pt.id)
-                    if videos:
-                        rendered_storage_key = getattr(videos[0], "file_url", "") or ""
+            _repo = SQLAlchemyGenerationTaskRepository(db)
+            logger.info("[封面生成] 步骤3: 查找最终成片任务(is_preview=False): plan_id=%s", plan_id)
+            all_tasks = _repo.list_by_source_edit_plan(plan_id)
+            for pt in all_tasks:
+                if (
+                    getattr(pt, "status", "") == "completed"
+                    and not getattr(pt, "is_preview", False)
+                ):
+                    rendered_storage_key = _get_task_video_url(db, pt.id) or ""
+                    if rendered_storage_key:
                         logger.info(
-                            "[封面生成] ✅ 步骤2.5找到视频: plan_id=%s task_id=%s url=%s",
-                            plan_id,
-                            pt.id,
-                            rendered_storage_key[:80],
+                            "[封面生成] ✅ 步骤3找到最终成片: plan_id=%s task_id=%s url=%s",
+                            plan_id, pt.id, rendered_storage_key[:80],
                         )
                         break
         except Exception:
             logger.warning(
-                "封面生成: 通过 source_edit_plan_id 查找预览任务失败: plan_id=%s",
-                plan_id,
-                exc_info=True,
+                "[封面生成] 步骤3查找最终成片失败: plan_id=%s", plan_id, exc_info=True,
             )
 
-    # 第三步：按 user + template 查找最近的已完成预览任务（兜底）
+    # 步骤 4：兼容回退 — 通过 source_edit_plan_id 查找已完成预览任务
     if not rendered_storage_key:
         try:
-            gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
-            logger.info("[封面生成] 步骤3: 通过 user+template 查找: plan_id=%s template_id=%s", plan_id, template_id)
-            preview_tasks = gen_task_repo.list_latest_completed_preview(
+            _repo = SQLAlchemyGenerationTaskRepository(db)
+            logger.info("[封面生成] 步骤4: 回退查找预览任务(is_preview=True): plan_id=%s", plan_id)
+            preview_tasks = _repo.list_by_source_edit_plan(plan_id)
+            for pt in preview_tasks:
+                if (
+                    getattr(pt, "status", "") == "completed"
+                    and getattr(pt, "is_preview", False)
+                ):
+                    rendered_storage_key = _get_task_video_url(db, pt.id) or ""
+                    if rendered_storage_key:
+                        logger.info(
+                            "[封面生成] ✅ 步骤4找到预览视频: plan_id=%s task_id=%s url=%s",
+                            plan_id, pt.id, rendered_storage_key[:80],
+                        )
+                        break
+        except Exception:
+            logger.warning(
+                "[封面生成] 步骤4查找预览任务失败: plan_id=%s", plan_id, exc_info=True,
+            )
+
+    # 步骤 5：按 user + template 查找最近的已完成预览任务（兜底）
+    if not rendered_storage_key:
+        try:
+            _repo = SQLAlchemyGenerationTaskRepository(db)
+            logger.info(
+                "[封面生成] 步骤5: 通过 user+template 查找预览任务: plan_id=%s template_id=%s",
+                plan_id, template_id,
+            )
+            preview_tasks = _repo.list_latest_completed_preview(
                 user_id=str(current_user.user.id),
                 template_id=template_id,
             )
             if preview_tasks:
-                completed_preview = preview_tasks[0]
-                video_repo = get_generated_video_repository(db)
-                use_case = ListGeneratedVideosByTaskUseCase(video_repo)
-                videos = use_case.execute(completed_preview.id)
-                if videos:
-                    rendered_storage_key = getattr(videos[0], "file_url", "") or ""
+                rendered_storage_key = _get_task_video_url(db, preview_tasks[0].id) or ""
+                if rendered_storage_key:
                     logger.info(
-                        "封面视频: 通过 user+template 找到预览任务: plan_id=%s template_id=%s task_id=%s",
-                        plan_id,
-                        template_id,
-                        completed_preview.id,
+                        "[封面生成] ✅ 步骤5找到预览视频: plan_id=%s task_id=%s",
+                        plan_id, preview_tasks[0].id,
                     )
         except Exception:
             logger.warning(
-                "封面警告: user+template 查找预览任务失败: plan_id=%s template_id=%s",
-                plan_id,
-                template_id,
-                exc_info=True,
+                "[封面生成] 步骤5 user+template 查找失败: plan_id=%s", plan_id, exc_info=True,
             )
 
-    # 使用裸 URL（rendered/* 已配置公开读）；找不到渲染视频时不立即报错，
-    # 因为步骤 E 可以直接从源素材抽帧（历史数据或 Worker 抽帧失败时的兜底）
+    # 将 storage_key 转换为可访问 URL；找不到视频时不立即报错，
+    # 因为步骤 E2 可以直接从源素材抽帧（历史数据或 Worker 抽帧失败时的兜底）
     primary_video_url = None
     if rendered_storage_key:
         plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key})
-        try:
-            if rendered_storage_key.startswith("http"):
-                primary_video_url = rendered_storage_key
-            else:
-                from packages.shared.storage import get_shared_storage_service
-
-                storage_svc = get_shared_storage_service()
-                primary_video_url = storage_svc.get_url(rendered_storage_key)
-            if primary_video_url:
-                import re as _re
-
-                primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
-            logger.info(
-                "获取预览视频URL用于封面生成: plan_id=%s url=%s",
-                plan_id,
-                primary_video_url[:80] if primary_video_url else "",
-            )
-        except Exception as e:
-            logger.warning("获取预览视频URL失败: plan_id=%s err=%s", plan_id, e)
-            primary_video_url = None
+        primary_video_url = _resolve_storage_key_to_url(rendered_storage_key)
+        logger.info(
+            "[封面生成] 封面抽帧视频URL: plan_id=%s url=%s",
+            plan_id,
+            primary_video_url[:80] if primary_video_url else "",
+        )
 
     # 统一封面管道：优先从 GenerationTask.cover_url 读取渲染后视频抽帧的封面
     # 多步查找 cover_url，和查找视频 URL 一样的 fallback 逻辑
@@ -326,20 +427,61 @@ def generate_cover(
                     exc_info=True,
                 )
 
-        # 步骤 B：通过 source_edit_plan_id 查找关联预览任务的 cover_url
+        # 步骤 A2：通过 generated_video_id 查找其关联任务的 cover_url
+        if not cover_url_from_task and body.generated_video_id:
+            try:
+                from app.dependencies import get_generated_video_repository as _get_gv_repo
+
+                gv_repo = _get_gv_repo(db)
+                gv = gv_repo.get(body.generated_video_id)
+                if gv:
+                    gv_task_id = getattr(gv, "generation_task_id", "") or ""
+                    if gv_task_id:
+                        task_a2 = gen_task_repo.get(gv_task_id)
+                        if task_a2 and getattr(task_a2, "cover_url", ""):
+                            cover_url_from_task = task_a2.cover_url
+                            logger.info(
+                                "[封面生成] 封面(步骤A2-video-task): plan_id=%s video_id=%s url=%s",
+                                plan_id, body.generated_video_id, cover_url_from_task[:80],
+                            )
+            except Exception:
+                logger.warning(
+                    "[封面生成] 步骤A2读取 cover_url 失败: plan_id=%s video_id=%s",
+                    plan_id, body.generated_video_id, exc_info=True,
+                )
+
+        # 步骤 B：通过 source_edit_plan_id 查找关联任务的 cover_url
+        # 优先最终成片任务(is_preview=False)，其次预览任务
         if not cover_url_from_task:
             try:
-                preview_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
-                for pt in preview_tasks:
-                    if getattr(pt, "status", "") == "completed" and getattr(pt, "cover_url", ""):
+                all_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
+                # 先找最终成片
+                for pt in all_tasks:
+                    if (
+                        getattr(pt, "status", "") == "completed"
+                        and not getattr(pt, "is_preview", False)
+                        and getattr(pt, "cover_url", "")
+                    ):
                         cover_url_from_task = pt.cover_url
                         logger.info(
-                            "[封面生成] 统一管道封面(步骤B-source_plan): plan_id=%s task_id=%s url=%s",
-                            plan_id,
-                            pt.id,
-                            cover_url_from_task[:80],
+                            "[封面生成] 封面(步骤B-final): plan_id=%s task_id=%s url=%s",
+                            plan_id, pt.id, cover_url_from_task[:80],
                         )
                         break
+                # 再找预览
+                if not cover_url_from_task:
+                    for pt in all_tasks:
+                        if (
+                            getattr(pt, "status", "") == "completed"
+                            and getattr(pt, "is_preview", False)
+                            and getattr(pt, "cover_url", "")
+                        ):
+                            cover_url_from_task = pt.cover_url
+                            logger.info(
+                                "[封面生成] 封面(步骤B-preview): plan_id=%s task_id=%s url=%s",
+                                plan_id, pt.id, cover_url_from_task[:80],
+                            )
+                            break
             except Exception:
                 logger.warning(
                     "[封面生成] 步骤B查找 cover_url 失败: plan_id=%s",
diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py
index 1250c33fb..042ddaf30 100644
--- a/tests/unit/test_generation_cover.py
+++ b/tests/unit/test_generation_cover.py
@@ -1200,3 +1200,304 @@ class TestUploadCoverType:
 
             assert exc_info.value.status_code == 400
             mock_mk.extract_frames.assert_not_called()
+
+
+class TestCoverFromFinalVideo:
+    """测试封面从最终成片任务(is_preview=False)获取视频源。"""
+
+    def test_generated_video_fields_in_schema(self):
+        """请求体支持 generated_video_id 和 video_url 字段。"""
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        req = GenerateCoverRequest(
+            generated_video_id="gv-001",
+            video_url="https://example.com/final.mp4",
+        )
+        assert req.generated_video_id == "gv-001"
+        assert req.video_url == "https://example.com/final.mp4"
+
+        # 默认 None
+        req_default = GenerateCoverRequest()
+        assert req_default.generated_video_id is None
+        assert req_default.video_url is None
+
+    def test_cover_uses_final_video_when_generated_video_id_provided(self):
+        """传 generated_video_id 时，从该最终成片视频抽帧。"""
+        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()
+
+        # Generated video
+        mock_gv = MagicMock()
+        mock_gv.file_url = "rendered/final/video.mp4"
+        mock_gv.generation_task_id = "task-final-001"
+
+        # 最终成片任务
+        mock_final_task = MagicMock()
+        mock_final_task.id = "task-final-001"
+        mock_final_task.created_by_user_id = "user-1"
+        mock_final_task.cover_url = ""
+
+        mock_gv_repo = MagicMock()
+        mock_gv_repo.get.return_value = mock_gv
+
+        mock_db = MagicMock()
+        mock_current_user = MagicMock()
+        mock_current_user.user.id = "user-1"
+
+        body = GenerateCoverRequest(
+            cover_type="ai_frame",
+            generated_video_id="gv-final-001",
+        )
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.dependencies.get_generated_video_repository", return_value=mock_gv_repo),
+            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter,
+            patch("app.api.routes.generation_cover._persist_cover_frame",
+                  return_value="https://oss.example.com/covers/final-cover.jpg") as mock_persist,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = mock_final_task
+            mock_repo.list_by_source_edit_plan.return_value = []
+            mock_repo.list_latest_completed_preview.return_value = []
+            mock_repo_cls.return_value = mock_repo
+
+            mock_storage_svc = MagicMock()
+            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/final/video.mp4"
+            mock_storage_getter.return_value = mock_storage_svc
+
+            mock_mk = MagicMock()
+            mock_mk.is_available = True
+            mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}]
+            mock_mk_getter.return_value = mock_mk
+
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/final-cover.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="tpl-1",
+                plan_id="plan-final",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=mock_current_user,
+            )
+
+            assert result.cover["image_url"] == "https://oss.example.com/covers/final-cover.jpg"
+            call_kwargs = mock_mk.extract_frames.call_args.kwargs
+            assert "rendered/final/video.mp4" in call_kwargs["video_url"]
+            assert call_kwargs["strategy"] == "SpecifiedFrames"
+            assert call_kwargs["max_frames"] == 1
+            assert call_kwargs["max_retries"] == 0
+            mock_persist.assert_called_once()
+
+    def test_cover_uses_video_url_directly(self):
+        """传 video_url 时，直接从该 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_db = MagicMock()
+        mock_current_user = MagicMock()
+        mock_current_user.user.id = "user-1"
+
+        body = GenerateCoverRequest(
+            cover_type="ai_frame",
+            video_url="https://oss.example.com/rendered/final/video.mp4",
+        )
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter,
+            patch("app.api.routes.generation_cover._persist_cover_frame",
+                  return_value="https://oss.example.com/covers/c.jpg"),
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            mock_repo.list_by_source_edit_plan.return_value = []
+            mock_repo.list_latest_completed_preview.return_value = []
+            mock_repo_cls.return_value = mock_repo
+
+            mock_mk = MagicMock()
+            mock_mk.is_available = True
+            mock_mk.extract_frames.return_value = [{"image_url": "https://mk/f.jpg"}]
+            mock_mk_getter.return_value = mock_mk
+
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/c.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            generate_cover(
+                body=body, template_id="tpl-1", plan_id="plan-url",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db, current_user=mock_current_user,
+            )
+
+            call_kwargs = mock_mk.extract_frames.call_args.kwargs
+            assert "rendered/final/video.mp4" in call_kwargs["video_url"]
+
+    def test_cover_prefers_final_task_over_preview_in_source_plan(self):
+        """步骤3：source_edit_plan 关联任务中，优先使用 is_preview=False 的最终成片。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {}  # 无 rendered_storage_key / generation_task_id
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+
+        # 一个预览任务 + 一个最终成片任务
+        mock_preview = MagicMock()
+        mock_preview.id = "task-preview"
+        mock_preview.status = "completed"
+        mock_preview.is_preview = True
+        mock_preview.cover_url = ""
+
+        mock_final = MagicMock()
+        mock_final.id = "task-final"
+        mock_final.status = "completed"
+        mock_final.is_preview = False
+        mock_final.cover_url = ""
+
+        mock_video_preview = MagicMock()
+        mock_video_preview.file_url = "rendered/preview/video.mp4"
+        mock_video_final = MagicMock()
+        mock_video_final.file_url = "rendered/final/video.mp4"
+
+        mock_db = MagicMock()
+        mock_current_user = MagicMock()
+        mock_current_user.user.id = "user-1"
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_list_videos,
+            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter,
+            patch("app.api.routes.generation_cover._persist_cover_frame",
+                  return_value="https://oss.example.com/covers/c.jpg"),
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            # list_by_source_edit_plan 返回 [preview, final]，最终成片排在后面
+            mock_repo.list_by_source_edit_plan.return_value = [mock_preview, mock_final]
+            mock_repo.list_latest_completed_preview.return_value = []
+            mock_repo_cls.return_value = mock_repo
+
+            # 根据 task id 返回不同 video
+            def get_videos(task_id):
+                if task_id == "task-final":
+                    return [mock_video_final]
+                return [mock_video_preview]
+            mock_use_case = MagicMock()
+            mock_use_case.execute.side_effect = get_videos
+            mock_list_videos.return_value = mock_use_case
+
+            mock_storage_svc = MagicMock()
+            mock_storage_svc.get_url.side_effect = lambda key: f"https://oss.example.com/{key}"
+            mock_storage_getter.return_value = mock_storage_svc
+
+            mock_mk = MagicMock()
+            mock_mk.is_available = True
+            mock_mk.extract_frames.return_value = [{"image_url": "https://mk/f.jpg"}]
+            mock_mk_getter.return_value = mock_mk
+
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/c.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            generate_cover(
+                body=body,
+                template_id="tpl-1",
+                plan_id="plan-priority",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=mock_current_user,
+            )
+
+            # 应该使用 final video URL 抽帧，而非 preview
+            call_kwargs = mock_mk.extract_frames.call_args.kwargs
+            assert "rendered/final/video.mp4" in call_kwargs["video_url"]
+            assert "rendered/preview" not in call_kwargs["video_url"]
+
+    def test_cover_generated_video_permission_denied(self):
+        """generated_video_id 关联任务属于其他用户时，返回 403。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+        from fastapi import HTTPException
+        import pytest
+
+        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_gv = MagicMock()
+        mock_gv.file_url = "rendered/other/video.mp4"
+        mock_gv.generation_task_id = "task-other"
+
+        mock_other_task = MagicMock()
+        mock_other_task.created_by_user_id = "other-user"
+
+        mock_gv_repo = MagicMock()
+        mock_gv_repo.get.return_value = mock_gv
+
+        mock_db = MagicMock()
+        mock_current_user = MagicMock()
+        mock_current_user.user.id = "user-1"
+
+        body = GenerateCoverRequest(
+            cover_type="ai_frame",
+            generated_video_id="gv-other",
+        )
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.dependencies.get_generated_video_repository", return_value=mock_gv_repo),
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = mock_other_task
+            mock_repo_cls.return_value = mock_repo
+
+            from app.api.routes.generation_cover import generate_cover
+
+            with pytest.raises(HTTPException) as exc_info:
+                generate_cover(
+                    body=body,
+                    template_id="tpl-1",
+                    plan_id="plan-perm",
+                    services=(mock_template_svc, mock_plan_svc),
+                    db=mock_db,
+                    current_user=mock_current_user,
+                )
+            assert exc_info.value.status_code == 403