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

@@ -209,39 +209,101 @@ def generate_cover(
         ) from e
 
     # 统一封面管道：优先从 GenerationTask.cover_url 读取渲染后视频抽帧的封面
+    # 多步查找 cover_url，和查找视频 URL 一样的 fallback 逻辑
     if body.cover_type in ("ai_frame", "ai_regenerate"):
-        # 尝试从 GenerationTask 读取已持久化的封面 URL
+        cover_url_from_task = None
+        gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
+
+        # 步骤 A：通过 generation_task_id 直接查找
         generation_task_id = (plan.config or {}).get("generation_task_id", "")
         if generation_task_id:
             try:
-                gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
                 task = gen_task_repo.get(generation_task_id)
                 if task and getattr(task, "cover_url", ""):
-                    cover_data = {
-                        "type": "ai_frame",
-                        "image_url": task.cover_url,
-                        "frame_time": 0.0,
-                        "confidence": 0.95,
-                    }
+                    cover_url_from_task = task.cover_url
                     logger.info(
-                        "[封面生成] 使用统一管道封面: plan_id=%s task_id=%s url=%s",
+                        "[封面生成] 统一管道封面(步骤A-direct): plan_id=%s task_id=%s url=%s",
                         plan_id,
                         generation_task_id,
-                        task.cover_url[:80],
+                        cover_url_from_task[:80],
                     )
-                    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"]})
-                    return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)
             except Exception:
                 logger.warning(
-                    "[封面生成] 读取 GenerationTask.cover_url 失败: plan_id=%s task_id=%s",
+                    "[封面生成] 步骤A读取 cover_url 失败: plan_id=%s task_id=%s",
                     plan_id,
                     generation_task_id,
                     exc_info=True,
                 )
 
+        # 步骤 B：通过 source_edit_plan_id 查找关联预览任务的 cover_url
+        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", ""):
+                        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],
+                        )
+                        break
+            except Exception:
+                logger.warning(
+                    "[封面生成] 步骤B查找 cover_url 失败: plan_id=%s",
+                    plan_id,
+                    exc_info=True,
+                )
+
+        # 步骤 C：通过 user+template 查找最近的已完成预览任务的 cover_url
+        if not cover_url_from_task:
+            try:
+                preview_tasks = gen_task_repo.list_latest_completed_preview(
+                    user_id=str(current_user.user.id),
+                    template_id=template_id,
+                )
+                for pt in preview_tasks:
+                    if getattr(pt, "cover_url", ""):
+                        cover_url_from_task = pt.cover_url
+                        logger.info(
+                            "[封面生成] 统一管道封面(步骤C-user+template): plan_id=%s task_id=%s url=%s",
+                            plan_id,
+                            pt.id,
+                            cover_url_from_task[:80],
+                        )
+                        break
+            except Exception:
+                logger.warning(
+                    "[封面生成] 步骤C查找 cover_url 失败: plan_id=%s template_id=%s",
+                    plan_id,
+                    template_id,
+                    exc_info=True,
+                )
+
+        if cover_url_from_task:
+            cover_data = {
+                "type": "ai_frame",
+                "image_url": cover_url_from_task,
+                "frame_time": 0.0,
+                "confidence": 0.95,
+            }
+            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"]})
+            return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)
+
+        logger.warning(
+            "[封面生成] 统一管道未找到 cover_url: plan_id=%s",
+            plan_id,
+        )
+        # ai_frame/ai_regenerate 类型必须从渲染管道获取，不再回退到 AI 服务
+        raise HTTPException(
+            status_code=400,
+            detail="封面尚未生成，请先重新生成预览视频以触发封面自动提取",
+        )
+
     from packages.shared.ai_service import run_generate_cover
 
     try:

tests/unit/test_generation_cover.py

@@ -154,11 +154,12 @@ class TestUnifiedCoverPipelineEndpoint:
             assert result.cover["image_url"] == "https://oss.example.com/rendered/plan-1/cover.jpg"
             assert result.cover["type"] == "ai_frame"
 
-    def test_cover_url_fallback_when_no_cover_url(self):
-        """当 GenerationTask 没有 cover_url 时，跳过统一管道走 run_generate_cover。"""
+    def test_cover_url_all_fallbacks_fail_returns_400(self):
+        """当所有步骤都找不到 cover_url 时，返回 400 而非 500。"""
         from unittest.mock import MagicMock, patch
 
         from app.api.routes.generation_cover import GenerateCoverRequest
+        from fastapi import HTTPException
 
         mock_plan = MagicMock()
         mock_plan.config = {
@@ -168,7 +169,6 @@ class TestUnifiedCoverPipelineEndpoint:
 
         mock_plan_svc = MagicMock()
         mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-
         mock_template_svc = MagicMock()
 
         # Task has no cover_url
@@ -181,42 +181,150 @@ class TestUnifiedCoverPipelineEndpoint:
 
         with (
             patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
-            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
             patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
         ):
             mock_repo = MagicMock()
             mock_repo.get.return_value = mock_task
+            # No tasks found by source_edit_plan_id or user+template
+            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 service
             mock_storage_svc = MagicMock()
             mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-2/video.mp4"
             mock_storage_getter.return_value = mock_storage_svc
 
-            mock_run.return_value = {"type": "ai_frame", "image_url": "https://fallback.com/cover.jpg"}
+            from app.api.routes.generation_cover import generate_cover
+
+            with pytest.raises(HTTPException) as exc_info:
+                generate_cover(
+                    body=body,
+                    template_id="template-2",
+                    plan_id="plan-2",
+                    services=(mock_template_svc, mock_plan_svc),
+                    db=mock_db,
+                    current_user=MagicMock(),
+                )
+
+            assert exc_info.value.status_code == 400
+            assert "封面尚未生成" in exc_info.value.detail
+
+    def test_cover_url_found_via_source_edit_plan(self):
+        """步骤B：通过 source_edit_plan_id 找到预览任务的 cover_url。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        # No generation_task_id, so step A is skipped
+        mock_plan.config = {
+            "rendered_storage_key": "rendered/plan-x/video.mp4",
+        }
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+
+        # Preview task found by source_edit_plan_id with cover_url
+        mock_preview_task = MagicMock()
+        mock_preview_task.id = "preview-task-abc"
+        mock_preview_task.status = "completed"
+        mock_preview_task.cover_url = "https://oss.example.com/rendered/preview/cover.jpg"
+
+        mock_db = MagicMock()
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.list_by_source_edit_plan.return_value = [mock_preview_task]
+            mock_repo_cls.return_value = mock_repo
+
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/preview/cover.jpg"}
+            }
 
             from app.api.routes.generation_cover import generate_cover
 
             result = generate_cover(
                 body=body,
-                template_id="template-2",
-                plan_id="plan-2",
+                template_id="template-x",
+                plan_id="plan-x",
                 services=(mock_template_svc, mock_plan_svc),
                 db=mock_db,
                 current_user=MagicMock(),
             )
 
-            # 应该走 run_generate_cover 回退
-            mock_run.assert_called_once()
-            assert result.cover["image_url"] == "https://fallback.com/cover.jpg"
+            assert result.cover["image_url"] == "https://oss.example.com/rendered/preview/cover.jpg"
+            mock_repo.list_by_source_edit_plan.assert_called_once_with("plan-x")
+
+    def test_cover_url_found_via_user_template(self):
+        """步骤C：通过 user+template 找到预览任务的 cover_url。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {
+            "rendered_storage_key": "rendered/plan-y/video.mp4",
+        }
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+
+        # Step B finds nothing, step C finds a task
+        mock_preview_task = MagicMock()
+        mock_preview_task.id = "preview-task-def"
+        mock_preview_task.status = "completed"
+        mock_preview_task.cover_url = "https://oss.example.com/rendered/user-template-cover.jpg"
+
+        mock_current_user = MagicMock()
+        mock_current_user.user.id = "user-123"
+        mock_db = MagicMock()
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.list_by_source_edit_plan.return_value = []
+            mock_repo.list_latest_completed_preview.return_value = [mock_preview_task]
+            mock_repo_cls.return_value = mock_repo
+
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/user-template-cover.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="template-y",
+                plan_id="plan-y",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=mock_current_user,
+            )
+
+            assert result.cover["image_url"] == "https://oss.example.com/rendered/user-template-cover.jpg"
+            mock_repo.list_latest_completed_preview.assert_called_once_with(
+                user_id="user-123",
+                template_id="template-y",
+            )
 
 
 class TestSourceEditPlanFallback:
     """测试步骤 2.5：通过 source_edit_plan_id 查找预览视频兜底逻辑。"""
 
     def test_step25_finds_video_by_source_edit_plan_id(self):
-        """当步骤1和步骤2都找不到时，步骤2.5通过source_edit_plan_id找到预览视频。"""
-        from unittest.mock import MagicMock, call, patch
+        """当步骤1和步骤2都找不到时，步骤2.5通过source_edit_plan_id找到预览视频和封面。"""
+        from unittest.mock import MagicMock, patch
 
         from app.api.routes.generation_cover import GenerateCoverRequest
 
@@ -228,11 +336,12 @@ class TestSourceEditPlanFallback:
         mock_plan_svc.get_plan_or_raise.return_value = mock_plan
         mock_template_svc = MagicMock()
 
-        # Mock preview task found by source_edit_plan_id
+        # Mock preview task found by source_edit_plan_id — with cover_url
         mock_preview_task = MagicMock()
         mock_preview_task.id = "preview-task-789"
         mock_preview_task.status = "completed"
         mock_preview_task.is_preview = True
+        mock_preview_task.cover_url = "https://oss.example.com/rendered/cover.jpg"
 
         # Mock generated video
         mock_video = MagicMock()
@@ -247,13 +356,10 @@ class TestSourceEditPlanFallback:
             patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo,
             patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
             patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
-            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
-            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
         ):
             mock_repo = MagicMock()
-            # Step 2: no generation_task_id, so get() is not called
             mock_repo.get.return_value = None
-            # Step 2.5: found by source_edit_plan_id
+            # Both video lookup (step 2.5) and cover_url lookup (step B) use this
             mock_repo.list_by_source_edit_plan.return_value = [mock_preview_task]
             mock_repo_cls.return_value = mock_repo
 
@@ -261,14 +367,9 @@ class TestSourceEditPlanFallback:
             mock_usecase.execute.return_value = [mock_video]
             mock_usecase_cls.return_value = mock_usecase
 
-            mock_normalize.return_value = {"cover": {"type": "ai_frame", "image_url": ""}}
-
-            mock_storage_svc = MagicMock()
-            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-x/video.mp4"
-            mock_storage_getter.return_value = mock_storage_svc
-
-            # cover_type=ai_frame but no cover_url on task -> fallback to run_generate_cover
-            mock_run.return_value = {"type": "ai_frame", "image_url": "https://cover.com/final.jpg"}
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/cover.jpg"}
+            }
 
             from app.api.routes.generation_cover import generate_cover
 
@@ -281,10 +382,10 @@ class TestSourceEditPlanFallback:
                 current_user=MagicMock(),
             )
 
-            # Verify step 2.5 was called
-            mock_repo.list_by_source_edit_plan.assert_called_once_with("plan-x")
-            # Step 3 should NOT be called since step 2.5 found the video
-            mock_repo.list_latest_completed_preview.assert_not_called()
+            # Verify step 2.5 was called for video
+            mock_repo.list_by_source_edit_plan.assert_called_with("plan-x")
+            # Cover was found via unified pipeline step B
+            assert result.cover["image_url"] == "https://oss.example.com/rendered/cover.jpg"
 
     def test_step25_skips_non_completed_or_non_preview_tasks(self):
         """步骤2.5跳过非completed或非is_preview的任务，继续到步骤3。"""
@@ -304,18 +405,22 @@ class TestSourceEditPlanFallback:
         mock_task_failed.id = "task-failed"
         mock_task_failed.status = "failed"
         mock_task_failed.is_preview = True
+        mock_task_failed.cover_url = ""
 
         # Task that is not preview
         mock_task_full = MagicMock()
         mock_task_full.id = "task-full"
         mock_task_full.status = "completed"
         mock_task_full.is_preview = False
+        mock_task_full.cover_url = ""
 
         mock_db = MagicMock()
 
-        # Step 3 fallback finds a valid preview task
+        # Step 3 fallback finds a valid preview task WITH cover_url
         mock_step3_task = MagicMock()
         mock_step3_task.id = "step3-task"
+        mock_step3_task.status = "completed"
+        mock_step3_task.cover_url = "https://oss.example.com/rendered/step3-cover.jpg"
 
         mock_video = MagicMock()
         mock_video.file_url = "rendered/step3/video.mp4"
@@ -328,13 +433,11 @@ class TestSourceEditPlanFallback:
             patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
             patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
             patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
-            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
         ):
-            mock_run.return_value = {"type": "ai_frame", "image_url": "https://cover.com/step3.jpg"}
             mock_repo = MagicMock()
             mock_repo.get.return_value = None
             mock_repo.list_by_source_edit_plan.return_value = [mock_task_failed, mock_task_full]
-            # Step 3 returns a valid task
+            # Video step 3 and cover step C both use list_latest_completed_preview
             mock_repo.list_latest_completed_preview.return_value = [mock_step3_task]
             mock_repo_cls.return_value = mock_repo
 
@@ -342,7 +445,9 @@ class TestSourceEditPlanFallback:
             mock_usecase.execute.return_value = [mock_video]
             mock_usecase_cls.return_value = mock_usecase
 
-            mock_normalize.return_value = {"cover": {"type": "ai_frame", "image_url": ""}}
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/step3-cover.jpg"}
+            }
 
             mock_storage_svc = MagicMock()
             mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/step3/video.mp4"
@@ -360,11 +465,12 @@ class TestSourceEditPlanFallback:
             )
 
             # Step 2.5 found tasks but none matched -> step 3 should be called
-            mock_repo.list_by_source_edit_plan.assert_called_once()
-            mock_repo.list_latest_completed_preview.assert_called_once()
+            mock_repo.list_by_source_edit_plan.assert_called()
+            mock_repo.list_latest_completed_preview.assert_called()
+            assert result.cover["image_url"] == "https://oss.example.com/rendered/step3-cover.jpg"
 
     def test_step25_exception_does_not_block_step3(self):
-        """步骤2.5异常时不影响步骤3兜底。"""
+        """步骤2.5异常时不影响步骤3兜底（视频和封面都通过步骤3找到）。"""
         from unittest.mock import MagicMock, patch
 
         from app.api.routes.generation_cover import GenerateCoverRequest
@@ -378,6 +484,8 @@ class TestSourceEditPlanFallback:
 
         mock_step3_task = MagicMock()
         mock_step3_task.id = "step3-task"
+        mock_step3_task.status = "completed"
+        mock_step3_task.cover_url = "https://oss.example.com/rendered/step3-cover.jpg"
 
         mock_video = MagicMock()
         mock_video.file_url = "rendered/step3/video.mp4"
@@ -392,14 +500,12 @@ class TestSourceEditPlanFallback:
             patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
             patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
             patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
-            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
         ):
-            mock_run.return_value = {"type": "ai_frame", "image_url": "https://cover.com/step3.jpg"}
             mock_repo = MagicMock()
             mock_repo.get.return_value = None
-            # Step 2.5 raises
+            # Both video step 2.5 and cover step B raise
             mock_repo.list_by_source_edit_plan.side_effect = RuntimeError("db error")
-            # Step 3 succeeds
+            # Step 3 / step C succeeds
             mock_repo.list_latest_completed_preview.return_value = [mock_step3_task]
             mock_repo_cls.return_value = mock_repo
 
@@ -407,7 +513,9 @@ class TestSourceEditPlanFallback:
             mock_usecase.execute.return_value = [mock_video]
             mock_usecase_cls.return_value = mock_usecase
 
-            mock_normalize.return_value = {"cover": {"type": "ai_frame", "image_url": ""}}
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/step3-cover.jpg"}
+            }
 
             mock_storage_svc = MagicMock()
             mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/step3/video.mp4"
@@ -425,7 +533,8 @@ class TestSourceEditPlanFallback:
             )
 
             # Step 3 was called after step 2.5 failed
-            mock_repo.list_latest_completed_preview.assert_called_once()
+            mock_repo.list_latest_completed_preview.assert_called()
+            assert result.cover["image_url"] == "https://oss.example.com/rendered/step3-cover.jpg"
 
 
 class TestStrayLoggerRemoved: