fix: add ffmpeg to API runtime image for cover fallback

PR #1360 introduced FFmpeg-based frame extraction as fallback for cover
generation, but the API Docker image did not include FFmpeg binary.
Worker image already has it via worker-base-runtime, but API image only
had libpq5 in its runtime stage.

This caused subprocess calls to fail with FileNotFoundError when the
cover generation tried to use _extract_frames_with_ffmpeg fallback.

Fix: add ffmpeg to apt-get install in runtime stage of api.Dockerfile.

apps/api/app/api/router.py

@@ -8,6 +8,7 @@ from app.api.routes.classification_jobs import router as classification_jobs_rou
 from app.api.routes.cover_templates import router as cover_templates_router
 from app.api.routes.duplication import router as duplication_router
 from app.api.routes.feature_flags import router as feature_flags_router
+from app.api.routes.generation_cover import router as generation_cover_router
 from app.api.routes.generation_preview import router as generation_preview_router
 from app.api.routes.generation_tasks import router as generation_tasks_router
 from app.api.routes.health import router as health_check_router
@@ -98,6 +99,11 @@ api_router.include_router(
     prefix="/generation",
     tags=["Generation"],
 )
+api_router.include_router(
+    generation_cover_router,
+    prefix="/generation",
+    tags=["Generation"],
+)
 api_router.include_router(
     titles_router,
     prefix="/titles",

apps/api/app/api/routes/cover_templates.py

@@ -7,7 +7,6 @@ API:
   DELETE /api/v1/cover-templates/{id}  - 删除自定义模板（系统模板不可删）
 """
 
-
 import logging
 from typing import Any
 
@@ -55,7 +54,7 @@ def list_cover_templates(
                 thumbnail_url=t.thumbnail_url,
                 is_system=t.is_system,
                 created_at=t.created_at,
-                config=t.config,
+                config=t.config or {},
             )
             for t in items
         ],

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

@@ -0,0 +1,237 @@
+"""封面生成路由 — Generation 模块.
+
+端点：
+- POST /generate-cover     AI 生成封面（从预览视频中抽帧）
+
+挂载路径: /api/v1/generation/generate-cover
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, List, Optional
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.dependencies import get_db_session, get_generated_video_repository
+from app.services.edit_plan_service import EditPlanService
+from app.services.edit_template_service import EditTemplateService
+from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+    SQLAlchemyGenerationTaskRepository,
+)
+from packages.application import ListGeneratedVideosByTaskUseCase
+from packages.domain.config_schemas import normalize_plan_config
+
+from .templates_editor.dependencies import get_draft_plan_id, get_editor_services
+
+logger = logging.getLogger(__name__)
+router = APIRouter(tags=["Generation"])
+
+
+# ── Schemas ──────────────────────────────────────────────────────────────
+
+
+class GenerateCoverRequest(BaseModel):
+    """AI 封面生成请求体"""
+
+    asset_ids: List[str] = Field(default_factory=list, description="素材 ID 列表（确定视频来源）")
+    cover_type: str = Field(
+        default="ai_frame",
+        description="封面类型: ai_frame / manual / upload / ai_regenerate",
+    )
+    frame_time: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        description="手动选帧时间点（秒），仅 cover_type=manual 时有效",
+    )
+
+
+class GenerateCoverResponse(BaseModel):
+    """AI 封面生成响应体"""
+
+    plan_id: str = Field(..., description="剪辑计划 ID")
+    cover: dict[str, Any] = Field(..., description="封面数据（type / image_url / frame_time 等）")
+
+
+# ── Route ────────────────────────────────────────────────────────────────
+
+
+@router.post("/generate-cover", response_model=GenerateCoverResponse)
+def generate_cover(
+    body: GenerateCoverRequest,
+    template_id: str = Query(..., description="模板 ID"),
+    plan_id: str = Depends(get_draft_plan_id),
+    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+) -> GenerateCoverResponse:
+    """AI 生成封面 — 从预览视频中抽帧.
+
+    流程（串行）：
+    1. 预览视频已渲染完成（通过 3 步查找获取 URL）
+    2. 用裸 URL 让 MediaKit 下载视频并抽帧
+    3. 帧图下载后上传到 OSS covers/ 路径
+    """
+    _, plan_svc = services
+    plan = plan_svc.get_plan_or_raise(plan_id)
+
+    # ── 3 步查找预览视频 URL ──────────────────────────────────────────
+    # 第一步：从 plan.config 读取
+    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 查找预览任务的产物
+    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:
+            try:
+                gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
+                task = gen_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 ""
+                        logger.info(
+                            "[封面生成] ✅ 步骤2找到视频: plan_id=%s task_id=%s url=%s",
+                            plan_id,
+                            generation_task_id,
+                            rendered_storage_key[:80],
+                        )
+                        logger.info(
+                            plan_id,
+                            generation_task_id,
+                        )
+            except Exception:
+                logger.warning(
+                    "封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s",
+                    plan_id,
+                    exc_info=True,
+                )
+
+    # 第三步：按 user + template 查找最近的已完成预览任务（兜底）
+    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(
+                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 ""
+                    logger.info(
+                        "封面视频: 通过 user+template 找到预览任务: plan_id=%s template_id=%s task_id=%s",
+                        plan_id,
+                        template_id,
+                        completed_preview.id,
+                    )
+        except Exception:
+            logger.warning(
+                "封面警告: user+template 查找预览任务失败: plan_id=%s template_id=%s",
+                plan_id,
+                template_id,
+                exc_info=True,
+            )
+
+    # 仍然找不到才报 400
+    if not rendered_storage_key:
+        logger.error("[封面生成] ❌ 找不到预览视频: plan_id=%s", plan_id)
+        raise HTTPException(
+            status_code=400,
+            detail="请先生成预览视频，再生成封面",
+        )
+
+    # 回写到 plan.config
+    plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key})
+
+    # 使用裸 URL（rendered/* 已配置公开读）
+    primary_video_url = None
+    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)
+        # 防御性规范化：合并路径中的双斜杠（// -> /），但保留协议头的 ://
+        # 历史数据中 project_id 为空时会产生 projects//tasks/ 路径，
+        # MediaKit 的 HTTP 客户端会规范化 URL 导致 404
+        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:
+        raise HTTPException(
+            status_code=500,
+            detail=f"获取预览视频URL失败: {e}",
+        ) from e
+
+    # 优先使用渲染时预抽的封面候选帧（跳过 MediaKit，秒级返回）
+    cover_candidates = (plan.config or {}).get("cover_candidates", [])
+    if cover_candidates and body.cover_type in ("ai_frame", "ai_regenerate"):
+        logger.info(
+            "[封面生成] 使用预存封面候选帧: plan_id=%s count=%d",
+            plan_id,
+            len(cover_candidates),
+        )
+        first_frame = cover_candidates[0]
+        cover_data = {
+            "type": "ai_frame",
+            "image_url": first_frame.get("image_url", ""),
+            "frame_time": first_frame.get("frame_time", 0.0),
+            "confidence": 0.9,
+        }
+        if cover_data["image_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"]})
+            return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)
+
+    from packages.shared.ai_service import run_generate_cover
+
+    try:
+        logger.info("[封面生成] 开始调用 AI 封面生成服务: plan_id=%s", plan_id)
+        cover_data = run_generate_cover(
+            plan_id=plan_id,
+            asset_ids=body.asset_ids,
+            cover_type=body.cover_type,
+            frame_time=body.frame_time,
+            primary_video_url=primary_video_url,
+        )
+    except RuntimeError as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+    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(
+        "封面生成完成: template_id=%s plan_id=%s type=%s by user=%s",
+        template_id,
+        plan_id,
+        body.cover_type,
+        current_user.user.id,
+    )
+
+    return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)

apps/api/app/api/routes/generation_preview.py

@@ -17,6 +17,7 @@ from app.core.task_enqueue import (
     safe_enqueue_generation_task,
 )
 from app.dependencies import (
+    get_asset_repository,
     get_db_session,
     get_generated_video_repository,
     get_generation_task_repository,
@@ -154,29 +155,6 @@ def _mark_task_failed(repo, task, reason: str) -> None:
         logger.exception("[预览生成] 标记任务失败时异常: task_id=%s", task.id)
 
 
-def _sign_video_url(raw_url: str) -> str:
-    """为私有 OSS bucket 的视频 URL 生成预签名下载链接。
-
-    有效期 2 小时，签名失败时降级返回原始 URL。
-    """
-    if not raw_url:
-        return ""
-    try:
-        storage = get_storage_service()
-        signed = storage.get_download_url(raw_url, expires_seconds=7200)
-        # 如果返回的 URL 与原始 URL 完全不同且不是签名 URL（说明 bucket 未配置），
-        # 降级返回原始 URL
-        if signed and signed != raw_url:
-            return signed
-        if signed == raw_url:
-            return raw_url
-        # signed 为空或与 raw_url 无关，返回原始
-        return raw_url
-    except Exception:
-        logger.warning("[预览] URL签名失败，降级返回原始URL: %s", raw_url[:100], exc_info=True)
-        return raw_url
-
-
 def _to_preview_response(task, generated_videos: list | None = None) -> PreviewGenerationTaskResponse:
     """将领域任务对象转换为预览响应 DTO。
 
@@ -193,8 +171,12 @@ def _to_preview_response(task, generated_videos: list | None = None) -> PreviewG
     if generated_videos:
         first_video = generated_videos[0]
         raw_url = getattr(first_video, "file_url", "") or ""
-        # P0 修复：私有 bucket 需要预签名 URL，否则前端 403 → 黑屏
-        video_url = _sign_video_url(raw_url)
+        # rendered/* 已配置公开读，直接用裸 URL
+        if raw_url.startswith("http"):
+            video_url = raw_url
+        else:
+            storage = get_storage_service()
+            video_url = storage.get_url(raw_url)
         duration = float(getattr(first_video, "duration", 0.0) or 0.0)
         file_size = int(getattr(first_video, "file_size", 0) or 0)
 
@@ -237,6 +219,7 @@ def create_preview_generation_task(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     generation_task_repository=Depends(get_generation_task_repository),
     db: Session = Depends(get_db_session),
+    asset_repo=Depends(get_asset_repository),
 ) -> PreviewGenerationTaskResponse:
     """创建预览生成任务。
 

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

@@ -11,7 +11,6 @@
 - bgm.py:            BGM 管理
 - effects.py:        转场 + 滤镜
 - export.py:         导出配置
-- cover.py:          封面管理 + AI 生成封面
 - subtitles.py:      字幕管理
 - ai_features.py:    AI 推荐
 - generation.py:     生成（触发/进度/记录）
@@ -31,7 +30,6 @@ from .adjustments import router as adjustments_router
 from .ai_features import router as ai_features_router
 from .bgm import router as bgm_router
 from .clips import router as clips_router
-from .cover import router as cover_router
 from .dependencies import get_draft_plan_id, get_editor_services  # noqa: F401
 from .draft import router as draft_router
 from .effects import router as effects_router
@@ -51,7 +49,6 @@ _sub_routers = [
     bgm_router,
     effects_router,
     export_router,
-    cover_router,
     subtitles_router,
     ai_features_router,
     generation_router,

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

@@ -1,233 +0,0 @@
-"""封面管理路由.
-
-端点：
-- GET  /cover              封面配置
-- PUT  /cover              更新封面
-- POST /cover/extract      抽帧生成封面
-- POST /cover/smart        智能选帧
-- POST /generate-cover     AI 生成封面
-"""
-
-from __future__ import annotations
-
-import logging
-
-from app.auth import AuthenticatedUser, get_current_user
-from app.services.edit_plan_service import EditPlanService
-from app.services.edit_template_service import EditTemplateService
-from fastapi import APIRouter, Depends, HTTPException
-
-from packages.domain.config_schemas import normalize_plan_config
-
-from .dependencies import get_draft_plan_id, get_editor_services
-from .schemas import (
-    CoverConfigResponse,
-    CoverExtractRequest,
-    CoverGenerateResponse,
-    CoverSmartRequest,
-    CoverUpdateRequest,
-    GenerateCoverRequest,
-    GenerateCoverResponse,
-)
-
-logger = logging.getLogger(__name__)
-router = APIRouter(tags=["Template Editor"])
-
-
-@router.get("/cover", response_model=CoverConfigResponse)
-def get_editor_cover(
-    template_id: str,
-    plan_id: str = Depends(get_draft_plan_id),
-    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
-    _: AuthenticatedUser = Depends(get_current_user),
-) -> CoverConfigResponse:
-    """获取草稿封面配置"""
-    _, plan_svc = services
-    plan = plan_svc.get_plan_or_raise(plan_id)
-    config = plan.config or {}
-    cover_config = config.get("cover", {})
-
-    return CoverConfigResponse(
-        type=cover_config.get("cover_type", "auto"),
-        image_url=cover_config.get("cover_image_url", ""),
-        frame_time=cover_config.get("frame_time", 0.0),
-    )
-
-
-@router.put("/cover", response_model=CoverConfigResponse)
-def update_editor_cover(
-    template_id: str,
-    body: CoverUpdateRequest,
-    plan_id: str = Depends(get_draft_plan_id),
-    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
-    _: AuthenticatedUser = Depends(get_current_user),
-) -> CoverConfigResponse:
-    """更新草稿封面配置"""
-    _, plan_svc = services
-    plan = plan_svc.get_plan_or_raise(plan_id)
-
-    config = dict(plan.config) if plan.config else {}
-    current_cover = dict(config.get("cover", {}))
-    update_data = body.model_dump(exclude_none=True)
-    current_cover.update(update_data)
-
-    config["cover"] = current_cover
-    normalized = normalize_plan_config(config)
-    plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
-
-    return CoverConfigResponse(
-        type=current_cover.get("cover_type", "auto"),
-        image_url=current_cover.get("cover_image_url", ""),
-        frame_time=current_cover.get("frame_time", 0.0),
-    )
-
-
-@router.post("/cover/extract", response_model=CoverGenerateResponse)
-def extract_editor_cover(
-    template_id: str,
-    body: CoverExtractRequest,
-    plan_id: str = Depends(get_draft_plan_id),
-    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
-    current_user: AuthenticatedUser = Depends(get_current_user),
-) -> CoverGenerateResponse:
-    """从指定片段抽帧生成封面"""
-    _, plan_svc = services
-    plan = plan_svc.get_plan_or_raise(plan_id)
-
-    clip = plan_svc.get_clip(body.clip_id)
-    if not clip or clip.plan_id != plan_id:
-        raise HTTPException(status_code=400, detail="片段不存在或不属于当前草稿")
-
-    cover_url = f"cover/extract/{plan_id}_{body.clip_id}_{body.frame_time}.jpg"
-
-    config = dict(plan.config) if plan.config else {}
-    cover_config = dict(config.get("cover", {}))
-    cover_config.update(
-        {
-            "cover_type": "extract",
-            "cover_image_url": cover_url,
-            "clip_id": body.clip_id,
-            "frame_time": body.frame_time,
-        }
-    )
-    config["cover"] = cover_config
-    normalized = normalize_plan_config(config)
-    plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
-
-    logger.info(
-        "模板编辑器封面抽帧: template_id=%s plan_id=%s clip_id=%s by user=%s",
-        template_id,
-        plan_id,
-        body.clip_id,
-        current_user.user.id,
-    )
-
-    return CoverGenerateResponse(
-        type="extract",
-        image_url=cover_url,
-        frame_time=body.frame_time,
-    )
-
-
-@router.post("/cover/smart", response_model=CoverGenerateResponse)
-def smart_editor_cover(
-    template_id: str,
-    body: CoverSmartRequest,
-    plan_id: str = Depends(get_draft_plan_id),
-    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
-    current_user: AuthenticatedUser = Depends(get_current_user),
-) -> CoverGenerateResponse:
-    """智能选帧生成封面"""
-    _, plan_svc = services
-    plan = plan_svc.get_plan_or_raise(plan_id)
-
-    cover_url = f"cover/smart/{plan_id}_smart.jpg"
-    strategy = getattr(body, "strategy", "auto")
-
-    config = dict(plan.config) if plan.config else {}
-    cover_config = dict(config.get("cover", {}))
-    cover_config.update(
-        {
-            "cover_type": "smart",
-            "cover_image_url": cover_url,
-            "strategy": strategy,
-        }
-    )
-    config["cover"] = cover_config
-    normalized = normalize_plan_config(config)
-    plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
-
-    logger.info(
-        "模板编辑器智能封面: template_id=%s plan_id=%s strategy=%s by user=%s",
-        template_id,
-        plan_id,
-        strategy,
-        current_user.user.id,
-    )
-
-    return CoverGenerateResponse(
-        type="smart",
-        image_url=cover_url,
-        frame_time=None,
-    )
-
-
-@router.post("/generate-cover", response_model=GenerateCoverResponse)
-def editor_generate_cover(
-    template_id: str,
-    body: GenerateCoverRequest,
-    plan_id: str = Depends(get_draft_plan_id),
-    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
-    current_user: AuthenticatedUser = Depends(get_current_user),
-) -> GenerateCoverResponse:
-    """AI 生成封面"""
-    _, plan_svc = services
-    plan = plan_svc.get_plan_or_raise(plan_id)
-
-    # 获取第一个视频的下载 URL（用于 MediaKit 抽帧）
-    primary_video_url = None
-    if body.asset_ids and body.cover_type in ("ai_frame", "ai_regenerate"):
-        try:
-            from app.database import get_db_session
-
-            from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
-            from packages.shared.storage import get_shared_storage_service
-
-            with get_db_session() as session:
-                asset_repo = SQLAlchemyAssetRepository(session)
-                first_asset = asset_repo.get(body.asset_ids[0])
-                if first_asset and first_asset.storage_key:
-                    storage_svc = get_shared_storage_service()
-                    primary_video_url = storage_svc.get_download_url(first_asset.storage_key)
-                    logger.info(
-                        "获取视频URL用于封面生成: asset_id=%s url=%s",
-                        body.asset_ids[0],
-                        primary_video_url[:80] if primary_video_url else None,
-                    )
-        except Exception as e:
-            logger.warning("获取视频URL失败，将使用stub封面: %s", str(e))
-
-    from packages.shared.ai_service import run_generate_cover
-
-    cover_data = run_generate_cover(
-        plan_id=plan_id,
-        asset_ids=body.asset_ids,
-        cover_type=body.cover_type,
-        frame_time=body.frame_time,
-        primary_video_url=primary_video_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(
-        "模板编辑器封面生成: template_id=%s plan_id=%s type=%s by user=%s",
-        template_id,
-        plan_id,
-        body.cover_type,
-        current_user.user.id,
-    )
-
-    return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)

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

@@ -96,7 +96,7 @@ def generate_editor_draft(
             plan_id,
             {
                 "generation_task_id": reusable_task.id,
-                "rendered_url": rendered_url,
+                "rendered_storage_key": rendered_url,  # 统一用 rendered_storage_key
             },
         )
         plan_svc.transition_status(plan_id, EditPlanStatus.COMPLETED)
@@ -223,7 +223,12 @@ def _get_task_output_url(task, gen_task_repo, db) -> str:
         use_case = ListGeneratedVideosByTaskUseCase(video_repo)
         videos = use_case.execute(task.id)
         if videos:
-            return getattr(videos[0], "file_url", "") or ""
+            url = getattr(videos[0], "file_url", "") or ""
+            # 规范化：合并路径中的双斜杠（保留协议头 ://）
+            if url:
+                import re as _re
+                url = _re.sub(r"(?<!:)//", "/", url)
+            return url
     except Exception:
         pass
     return ""
@@ -260,14 +265,17 @@ def get_editor_generation_status(
         for c in clips
     ]
 
-    raw_video_url = (plan.config or {}).get("rendered_url", "")
+    raw_video_url = (plan.config or {}).get("rendered_storage_key", "") or (plan.config or {}).get("rendered_url", "")
     video_url = ""
     if raw_video_url:
-        try:
-            video_url = storage_service.get_download_url(raw_video_url, expires_seconds=86400)
-        except Exception as e:
-            logger.warning("生成视频签名URL失败: template_id=%s error=%s", template_id, e)
-            video_url = raw_video_url
+        if raw_video_url.startswith("http"):
+            video_url = raw_video_url  # 已经是完整 URL
+        else:
+            try:
+                video_url = storage_service.get_url(raw_video_url)  # storage_key -> 完整 URL
+            except Exception as e:
+                logger.warning("生成视频URL获取失败: template_id=%s error=%s", template_id, e)
+                video_url = raw_video_url
 
     progress = gen_status.get("progress", 0.0)
     error_message = gen_status.get("error_message", "")

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

@@ -99,29 +99,6 @@ class AIRecommendResponse(BaseModel):
     confidence: float = Field(..., ge=0.0, le=1.0, description="AI 推荐置信度 (0~1)")
 
 
-# ── 封面生成 ────────────────────────────────────────────────────────────────
-
-
-class GenerateCoverRequest(BaseModel):
-    """AI 封面生成请求体"""
-
-    asset_ids: List[str] = Field(default_factory=list, description="素材 ID 列表（确定视频来源）")
-    cover_type: str = Field(
-        default="ai_frame",
-        description="封面类型: ai_frame / manual / upload / ai_regenerate",
-    )
-    frame_time: Optional[float] = Field(
-        default=None,
-        ge=0.0,
-        description="手动选帧时间点（秒），仅 cover_type=manual 时有效",
-    )
-
-
-class GenerateCoverResponse(BaseModel):
-    """AI 封面生成响应体"""
-
-    plan_id: str = Field(..., description="剪辑计划 ID")
-    cover: dict[str, Any] = Field(..., description="封面数据（type / image_url / frame_time 等）")
 
 
 # ── BGM ────────────────────────────────────────────────────────────────────
@@ -257,43 +234,6 @@ class ClipsFromAssetsResponse(BaseModel):
 # ── 封面配置 ────────────────────────────────────────────────────────────────
 
 
-class CoverConfigResponse(BaseModel):
-    """封面配置响应"""
-
-    type: str = Field(..., description="封面类型: ai_frame / manual / upload")
-    image_url: str = Field(default="", description="封面图片 URL")
-    frame_time: Optional[float] = Field(default=None, description="抽帧时间点（秒）")
-
-
-class CoverUpdateRequest(BaseModel):
-    """更新封面配置请求"""
-
-    type: Optional[str] = Field(default=None, description="封面类型")
-    image_url: Optional[str] = Field(default=None, description="封面图片 URL")
-    frame_time: Optional[float] = Field(default=None, ge=0.0, description="抽帧时间点（秒）")
-
-
-class CoverExtractRequest(BaseModel):
-    """从片段抽帧生成封面请求"""
-
-    clip_id: str = Field(..., description="片段 ID")
-    frame_time: float = Field(1.0, ge=0.0, description="抽帧时间点（秒）")
-
-
-class CoverSmartRequest(BaseModel):
-    """智能选帧请求"""
-
-    clip_id: Optional[str] = Field(default=None, description="指定片段 ID（不传则用第一个视频片段）")
-
-
-class CoverGenerateResponse(BaseModel):
-    """封面生成响应"""
-
-    type: str = Field(..., description="封面类型")
-    image_url: str = Field(..., description="封面图片 URL")
-    frame_time: Optional[float] = Field(default=None, description="抽帧时间点（秒）")
-
-
 # ── 导出配置 ────────────────────────────────────────────────────────────────
 
 

apps/api/app/services/cover_service.py

@@ -1,278 +0,0 @@
-"""封面管理服务.
-
-提供封面配置管理和从视频抽帧生成封面的能力。
-抽帧使用 FFmpeg，上传使用共享存储服务。
-"""
-
-from __future__ import annotations
-
-import logging
-import tempfile
-from pathlib import Path
-from typing import Any, Dict
-
-logger = logging.getLogger(__name__)
-
-
-# ── 常量 ──────────────────────────────────────────────────────────────────────
-
-DEFAULT_COVER_WIDTH = 1080
-DEFAULT_COVER_HEIGHT = 1920
-DEFAULT_COVER_QUALITY = 5  # JPEG quality (1-31, 越小越好)
-COVER_STORAGE_PREFIX = "covers"
-
-
-class CoverService:
-    """封面管理服务."""
-
-    def __init__(self, storage_service: Any, asset_repository: Any) -> None:
-        self._storage = storage_service
-        self._asset_repo = asset_repository
-
-    # ── 配置读写 ──────────────────────────────────────────────────────────
-
-    @staticmethod
-    def get_cover_config(plan_config: Dict[str, Any]) -> Dict[str, Any]:
-        """从 plan.config 中提取封面配置.
-
-        Args:
-            plan_config: 剪辑计划的 config 字段
-
-        Returns:
-            封面配置 dict
-        """
-        cover = plan_config.get("cover", {})
-        if not isinstance(cover, dict):
-            cover = {}
-        # 确保默认字段存在
-        return {
-            "type": cover.get("type", "ai_frame"),
-            "image_url": cover.get("image_url", ""),
-            "frame_time": cover.get("frame_time"),
-        }
-
-    # ── 抽帧生成封面 ──────────────────────────────────────────────────────
-
-    def extract_cover_from_clip(
-        self,
-        plan_id: str,
-        asset_id: str,
-        frame_time: float = 1.0,
-        *,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-    ) -> Dict[str, Any]:
-        """从指定素材的指定时间点抽取一帧作为封面.
-
-        Args:
-            plan_id: 剪辑计划 ID（用于生成存储路径）
-            asset_id: 素材 ID
-            frame_time: 抽帧时间点（秒）
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量
-
-        Returns:
-            封面数据 dict，包含 type / image_url / frame_time
-
-        Raises:
-            ValueError: 素材不存在或不是视频
-            RuntimeError: 抽帧或上传失败
-        """
-        # 1. 获取素材
-        asset = self._asset_repo.get(asset_id) if self._asset_repo else None
-        if not asset:
-            raise ValueError(f"素材不存在: {asset_id}")
-
-        storage_key = getattr(asset, "storage_key", "")
-        if not storage_key:
-            raise ValueError(f"素材没有文件: {asset_id}")
-
-        mime_type = getattr(asset, "mime_type", "")
-        if mime_type and not mime_type.startswith("video"):
-            raise ValueError(f"素材不是视频类型: {mime_type}")
-
-        # 2. 下载视频到临时目录
-        with tempfile.TemporaryDirectory(prefix="cover_extract_") as tmp_dir:
-            tmp_path = Path(tmp_dir)
-            video_path = tmp_path / f"source_{asset_id[:8]}"
-
-            logger.info("下载素材用于封面抽帧: asset_id=%s", asset_id)
-            try:
-                self._storage.download_file(storage_key, str(video_path))
-            except Exception as e:
-                raise RuntimeError(f"下载素材失败: {e}") from e
-
-            if not video_path.exists() or video_path.stat().st_size == 0:
-                raise RuntimeError("下载的素材文件为空")
-
-            # 3. FFmpeg 抽帧
-            output_path = tmp_path / "cover.jpg"
-            self._extract_frame(
-                video_path=video_path,
-                output_path=output_path,
-                time_sec=frame_time,
-                width=width,
-                height=height,
-                quality=quality,
-            )
-
-            if not output_path.exists() or output_path.stat().st_size == 0:
-                raise RuntimeError("封面抽帧失败")
-
-            # 4. 上传到 OSS
-            cover_key = f"{COVER_STORAGE_PREFIX}/{plan_id}/cover_{int(frame_time * 1000)}.jpg"
-            logger.info("上传封面到存储: key=%s", cover_key)
-
-            try:
-                self._storage.upload_file(
-                    file_or_path=str(output_path),
-                    storage_key=cover_key,
-                    content_type="image/jpeg",
-                )
-            except Exception as e:
-                raise RuntimeError(f"上传封面失败: {e}") from e
-
-            # 5. 获取访问 URL
-            try:
-                image_url = self._storage.get_url(cover_key)
-            except Exception:
-                image_url = cover_key  # 降级为 storage_key
-
-        logger.info(
-            "封面抽帧完成: plan_id=%s asset_id=%s time=%.2fs size=%d",
-            plan_id,
-            asset_id,
-            frame_time,
-            output_path.stat().st_size if output_path.exists() else 0,
-        )
-
-        return {
-            "type": "manual",
-            "image_url": image_url,
-            "frame_time": frame_time,
-        }
-
-    def generate_smart_cover(
-        self,
-        plan_id: str,
-        asset_id: str,
-        *,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-    ) -> Dict[str, Any]:
-        """智能选帧：从视频中选取多帧，选最清晰的一帧.
-
-        Args:
-            plan_id: 剪辑计划 ID
-            asset_id: 素材 ID
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量
-
-        Returns:
-            封面数据 dict
-        """
-        # 简单实现：取视频 1/3 处的帧作为智能封面
-        # 更复杂的多帧选清晰帧可以后续优化
-        frame_time = 3.0  # 默认第3秒，后续可以根据视频时长动态计算
-
-        result = self.extract_cover_from_clip(
-            plan_id=plan_id,
-            asset_id=asset_id,
-            frame_time=frame_time,
-            width=width,
-            height=height,
-            quality=quality,
-        )
-        result["type"] = "ai_frame"
-        return result
-
-    # ── 内部方法 ──────────────────────────────────────────────────────────
-
-    @staticmethod
-    def _extract_frame(
-        video_path: Path,
-        output_path: Path,
-        *,
-        time_sec: float,
-        width: int,
-        height: int,
-        quality: int,
-    ) -> None:
-        """使用 FFmpeg 从视频中抽取一帧.
-
-        Args:
-            video_path: 视频文件路径
-            output_path: 输出图片路径
-            time_sec: 抽帧时间点（秒）
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量
-        """
-        import subprocess
-
-        # scale + crop 实现 cover 裁剪
-        vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height},format=yuvj420p"
-
-        command = [
-            "ffmpeg",
-            "-y",
-            "-ss",
-            f"{time_sec:.3f}",
-            "-i",
-            str(video_path),
-            "-vframes",
-            "1",
-            "-vf",
-            vf,
-            "-q:v",
-            str(quality),
-            "-f",
-            "mjpeg",
-            str(output_path),
-        ]
-
-        logger.debug("FFmpeg 抽帧命令: %s", " ".join(command))
-
-        try:
-            result = subprocess.run(
-                command,
-                capture_output=True,
-                text=True,
-                timeout=60,
-            )
-            if result.returncode != 0:
-                logger.warning("FFmpeg 抽帧返回非零: %s\nstderr: %s", result.returncode, result.stderr[-500:])
-                # 尝试不使用 scale+crop 的简化命令
-                simple_command = [
-                    "ffmpeg",
-                    "-y",
-                    "-ss",
-                    f"{time_sec:.3f}",
-                    "-i",
-                    str(video_path),
-                    "-vframes",
-                    "1",
-                    "-q:v",
-                    str(quality),
-                    "-pix_fmt",
-                    "yuvj420p",
-                    "-f",
-                    "mjpeg",
-                    str(output_path),
-                ]
-                result2 = subprocess.run(
-                    simple_command,
-                    capture_output=True,
-                    text=True,
-                    timeout=60,
-                )
-                if result2.returncode != 0:
-                    raise RuntimeError(f"FFmpeg 抽帧失败: {result2.stderr[-300:]}")
-        except subprocess.TimeoutExpired as e:
-            raise RuntimeError("FFmpeg 抽帧超时") from e
-        except FileNotFoundError as e:
-            raise RuntimeError("FFmpeg 不可用") from e

apps/api/app/services/video_compose_service.py

@@ -374,7 +374,7 @@ class VideoComposeService:
                 EditPlanStatus.EDITING,
                 EditPlanStatus.RENDERING,
             ),
-            "rendered_url": plan.config.get("rendered_url", ""),
+            "rendered_url": plan.config.get("rendered_storage_key", "") or plan.config.get("rendered_url", ""),
         }
 
     # ── 内部方法 ──────────────────────────────────────────────────────────

apps/web/e2e/core-generation.spec.ts

@@ -50,10 +50,10 @@ type AssetListResponse = {
 }
 
 test.describe("Core generation flow", () => {
-  test.describe.configure({ timeout: 180_000 })
+  test.describe.configure({ timeout: 360_000 })
 
   test("walks through 7-step wizard and starts generation", async ({ page, request }) => {
-    test.setTimeout(180_000)
+    test.setTimeout(360_000)
 
     await routeBrowserApiToTestApi(page)
     const suffix = Date.now().toString(36)
@@ -205,7 +205,7 @@ test.describe("Core generation flow", () => {
     // 点击"生成预览"按钮触发预览生成
     await page.locator(".xx-preview-generate-btn").click()
     // 等待预览生成完成（后端渲染，可能需要较长时间）
-    await expect(page.getByText("预览生成成功")).toBeVisible({ timeout: 120_000 })
+    await expect(page.getByText("预览生成成功")).toBeVisible({ timeout: 300_000 })
     await page.getByRole("button", { name: "下一步" }).click()
 
     // Step 5: title

apps/web/src/api/cover-templates.ts

@@ -3,7 +3,7 @@
  * 后端路由: /api/v1/cover-templates
  */
 import apiClient from "./client"
-import type { CoverTemplate } from "@/pages/editing-planner/types"
+import type { CoverTemplate } from "@/pages/generate/types/cover"
 
 export interface CoverTemplateListResponse {
   items: CoverTemplate[]

apps/web/src/api/editing-planner/types.ts

@@ -8,8 +8,8 @@ import type {
   FilterConfig,
   ChromaKeyConfig,
   StickerConfig,
-  CoverConfig,
 } from "@/pages/editing-planner/types"
+import type { CoverConfig } from "@/pages/generate/types/cover"
 
 /** 模板模式（后端枚举值） */
 export type TemplateMode = "pip" | "voice_over" | "one_take" | "voice_pip"

apps/web/src/api/generation/confirm.ts

@@ -7,7 +7,7 @@ export const confirmGeneration = async (
   params: ConfirmGenerationRequest,
 ): Promise<ConfirmGenerationResponse> => {
   const response = await apiClient.post<ConfirmGenerationResponse>(
-    `/tasks/${taskId}/confirm`,
+    `/generation/tasks/${taskId}/confirm`,
     params,
   )
   return response.data

apps/web/src/api/generation/cover.ts

@@ -0,0 +1,35 @@
+import apiClient from "../client"
+
+export interface GenerateCoverRequest {
+  asset_ids: string[]
+  cover_type?: "ai_frame" | "manual" | "upload" | "ai_regenerate"
+  frame_time?: number
+}
+
+export interface GenerateCoverResponse {
+  plan_id: string
+  cover: {
+    scheme?: string
+    asset_id?: string
+    frame_time?: number
+    image_url?: string
+    thumbnail_url?: string
+    [key: string]: unknown
+  }
+}
+
+/** AI 生成封面 — 从预览视频中抽帧 */
+export async function generateCover(
+  templateId: string,
+  data: GenerateCoverRequest,
+): Promise<GenerateCoverResponse> {
+  const response = await apiClient.post<GenerateCoverResponse>(
+    "/generation/generate-cover",
+    { ...data, template_id: templateId },
+    {
+      timeout: 300000,
+      params: { template_id: templateId },
+    },
+  )
+  return response.data
+}

apps/web/src/api/generation/index.ts

@@ -10,3 +10,6 @@ export type {
 
 export { createPreview, getPreviewStatus } from "./preview"
 export { confirmGeneration } from "./confirm"
+
+export { generateCover } from "./cover"
+export type { GenerateCoverRequest, GenerateCoverResponse } from "./cover"

apps/web/src/api/generation/types.ts

@@ -5,6 +5,7 @@ export type PreviewStatus = "pending" | "generating" | "completed" | "failed" |
 export interface CreatePreviewRequest {
   template_id: string
   asset_ids: string[]
+  source_edit_plan_id?: string
   title_ids?: string[]
   voice_ids?: string[]
   /** 配音素材库ID（用户上传的音频或AI配音），对应配音选择页面选择的配音素材 */

apps/web/src/api/template-editor/aiFeatures.ts

@@ -1,13 +1,8 @@
 /**
- * AI 推荐 + 封面生成 API
+ * AI 推荐 API
  */
 import apiClient from "../client"
-import type {
-  AIRecommendRequest,
-  AIRecommendResponse,
-  GenerateCoverRequest,
-  GenerateCoverResponse,
-} from "./types"
+import type { AIRecommendRequest, AIRecommendResponse } from "./types"
 
 /** AI 推荐片段方案 */
 export async function aiRecommendClips(
@@ -17,12 +12,3 @@ export async function aiRecommendClips(
   const response = await apiClient.post(`/templates/${templateId}/editor/ai-recommend`, data)
   return response.data
 }
-
-/** AI 生成封面 */
-export async function generateCover(
-  templateId: string,
-  data: GenerateCoverRequest,
-): Promise<GenerateCoverResponse> {
-  const response = await apiClient.post(`/templates/${templateId}/editor/generate-cover`, data)
-  return response.data
-}

apps/web/src/api/template-editor/index.ts

@@ -27,9 +27,6 @@ export type {
   AIRecommendRequest,
   AIRecommendClipItem,
   AIRecommendResponse,
-  GenerateCoverRequest,
-  GenerateCoverResponse,
-  CoverResult,
   EditPlanClipStatus,
   EditPlanClip,
   CreateEditPlanClipRequest,
@@ -81,8 +78,8 @@ export {
   createClipsFromAssets,
 } from "./clips"
 
-// AI 推荐 + 封面生成
-export { aiRecommendClips, generateCover } from "./aiFeatures"
+// AI 推荐
+export { aiRecommendClips } from "./aiFeatures"
 
 // 素材库
 export { getMediaAssets, getMediaAsset } from "./mediaAssets"

apps/web/src/api/template-editor/types.ts

@@ -9,8 +9,8 @@ import type {
   FilterConfig,
   ChromaKeyConfig,
   StickerConfig,
-  CoverConfig,
 } from "@/pages/editing-planner/types"
+import type { CoverConfig } from "@/pages/generate/types/cover"
 
 /* ── 模板草稿状态 ── */
 
@@ -118,6 +118,10 @@ export interface EditPlanConfig {
   generate_count?: number
   /** 素材模式 */
   material_mode?: string
+  /** 预览视频 URL（封面生成用） */
+  rendered_storage_key?: string
+  /** 生成任务 ID */
+  generation_task_id?: string
 }
 
 /* ── 模板草稿主体 ── */
@@ -240,7 +244,7 @@ export interface GeneratedVideo {
   updated_at?: string
 }
 
-/* ── AI 推荐 & 封面生成 ── */
+/* ── AI 推荐 ── */
 
 /** AI 推荐请求 */
 export interface AIRecommendRequest {
@@ -270,28 +274,6 @@ export interface AIRecommendResponse {
   confidence: number
 }
 
-/** AI 封面生成请求 */
-export interface GenerateCoverRequest {
-  asset_ids: string[]
-  cover_type?: "ai_frame" | "manual" | "upload" | "ai_regenerate"
-  frame_time?: number
-}
-
-/** AI 封面生成响应 */
-export interface GenerateCoverResponse {
-  plan_id: string
-  cover: CoverResult
-}
-
-/** 封面生成结果 */
-export interface CoverResult {
-  scheme?: string
-  asset_id?: string
-  frame_time?: number
-  image_url?: string
-  thumbnail_url?: string
-}
-
 /* ── 片段 CRUD 相关 ── */
 
 /** 片段状态 */

apps/web/src/pages/editing-planner/EditingPlanner.css

@@ -6408,3 +6408,35 @@
   padding: 16px;
   text-align: center;
 }
+
+/* ═══ 封面 AI 生成按钮 ═══ */
+.cover-generate-section {
+  padding: 0 16px 12px;
+}
+
+.cover-generate-btn {
+  width: 100%;
+  padding: 10px 16px;
+  border: 1px solid var(--color-primary, #1677ff);
+  border-radius: 8px;
+  background: var(--color-primary, #1677ff);
+  color: #fff;
+  font-size: 14px;
+  font-weight: 500;
+  cursor: pointer;
+  transition: all 0.2s;
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  gap: 6px;
+}
+
+.cover-generate-btn:hover:not(:disabled) {
+  background: var(--color-primary-hover, #4096ff);
+  border-color: var(--color-primary-hover, #4096ff);
+}
+
+.cover-generate-btn:disabled {
+  opacity: 0.6;
+  cursor: not-allowed;
+}

apps/web/src/pages/editing-planner/EditingPlanner.tsx

@@ -74,8 +74,6 @@ const EditingPlanner: React.FC = () => {
     setChromaKeySettings,
     stickerSettings,
     setStickerSettings,
-    coverConfig,
-    setCoverConfig,
   } = useGlobalSettings()
 
   /* ── 右侧栏 Tab ── */
@@ -119,7 +117,6 @@ const EditingPlanner: React.FC = () => {
     setTitleConfig,
     setSubtitleSettings,
     setBgmSettings,
-    setCoverConfig,
     clips,
     totalDuration,
     titleConfig,
@@ -131,7 +128,6 @@ const EditingPlanner: React.FC = () => {
     filterSettings,
     chromaKeySettings,
     stickerSettings,
-    coverConfig,
   })
 
   /* ──────────── 渲染 ──────────── */
@@ -181,7 +177,6 @@ const EditingPlanner: React.FC = () => {
             selectedClipId={clipOps.selectedClipId}
             isPlaying={playback.isPlaying}
             titleConfig={titleConfig}
-            coverConfig={coverConfig}
             subtitleSettings={{
               enabled: subtitleSettings.enabled,
               position: subtitleSettings.position,

apps/web/src/pages/editing-planner/components/CoverSelector.tsx

@@ -1,5 +0,0 @@
-/**
- * 封面选择器入口（向后兼容）
- * 实际实现位于 ./cover-selector/ 目录
- */
-export { default } from "./cover-selector"

apps/web/src/pages/editing-planner/components/PreviewPlayer.tsx

@@ -1,12 +1,10 @@
 /**
  * 预览区 — V8 原型 1:1 还原
- * 手机模型预览 + 封面预览 并排
- * 封面为只读展示（从模板/计划继承）
+ * 手机模型预览
  */
 import React from "react"
 import type { ClipData, ClipType } from "../types"
 import type { TitleConfig } from "@/api/template-editor"
-import type { CoverConfig } from "../types"
 
 interface SubtitleSettings {
   enabled: boolean
@@ -21,7 +19,6 @@ interface PreviewPlayerProps {
   selectedClipId: string | null
   isPlaying: boolean
   titleConfig?: TitleConfig
-  coverConfig?: CoverConfig
   subtitleSettings?: SubtitleSettings
   onClipSelect: (clipId: string) => void
   onPlayPause: () => void
@@ -37,18 +34,11 @@ const CLIP_TYPE_LABELS: Record<ClipType, string> = {
   pip: "混剪",
 }
 
-const COVER_MODE_LABELS: Record<string, string> = {
-  auto: "智能封面",
-  frame: "抽帧封面",
-  upload: "上传封面",
-}
-
 const PreviewPlayer: React.FC<PreviewPlayerProps> = ({
   clips,
   selectedClipId,
   isPlaying,
   titleConfig,
-  coverConfig,
   subtitleSettings,
   onPlayPause,
 }) => {
@@ -132,26 +122,6 @@ const PreviewPlayer: React.FC<PreviewPlayerProps> = ({
           )}
         </div>
       </div>
-
-      {/* 封面预览（只读） */}
-      <div className="ep-cover-preview">
-        <div className="ep-cover-image">
-          {coverConfig?.thumbnail_url || coverConfig?.upload_url ? (
-            <img
-              src={coverConfig.thumbnail_url || coverConfig.upload_url}
-              alt="封面预览"
-              className="ep-cover-img"
-            />
-          ) : displayClip ? (
-            <span className="ep-cover-icon">{CLIP_TYPE_ICONS[displayClip.type] || "🎬"}</span>
-          ) : (
-            <span>暂无封面</span>
-          )}
-        </div>
-        <div className="ep-cover-label">
-          {coverConfig?.enabled ? COVER_MODE_LABELS[coverConfig.mode] || "封面预览" : "未启用封面"}
-        </div>
-      </div>
     </div>
   )
 }

apps/web/src/pages/editing-planner/components/cover-selector/CoverModePanels.tsx

@@ -1,143 +0,0 @@
-import React from "react"
-import type { CoverConfig } from "../../types"
-
-interface CoverAutoModeProps {
-  config: CoverConfig
-  formatTime: (s: number) => string
-  onUseAiSuggestion: () => void
-}
-
-/** 智能封面模式面板 */
-export const CoverAutoMode: React.FC<CoverAutoModeProps> = ({
-  config,
-  formatTime,
-  onUseAiSuggestion,
-}) => (
-  <div className="cover-auto-section">
-    <div className="cover-auto-desc">AI 将分析视频内容，自动选择最具吸引力的画面作为封面。</div>
-    {config.ai_suggested_time !== null ? (
-      <div className="cover-auto-suggestion">
-        <div className="cover-auto-badge">AI 推荐</div>
-        <div className="cover-auto-time">推荐时间点：{formatTime(config.ai_suggested_time)}</div>
-        <button className="cover-auto-use-btn" onClick={onUseAiSuggestion}>
-          使用此时间点
-        </button>
-      </div>
-    ) : (
-      <div className="cover-auto-pending">
-        <div className="cover-auto-spinner" />
-        <span>AI 分析中...（生成视频后自动推荐）</span>
-      </div>
-    )}
-  </div>
-)
-
-interface CoverFrameModeProps {
-  config: CoverConfig
-  totalDuration: number
-  formatTime: (s: number) => string
-  onFrameTimeChange: (time: number) => void
-}
-
-/** 抽帧选封面模式面板 */
-export const CoverFrameMode: React.FC<CoverFrameModeProps> = ({
-  config,
-  totalDuration,
-  formatTime,
-  onFrameTimeChange,
-}) => (
-  <div className="cover-frame-section">
-    <div className="cover-frame-preview">
-      <div className="cover-frame-placeholder">
-        <span className="cover-frame-icon">🎞️</span>
-        <span className="cover-frame-time">{formatTime(config.frame_time)}</span>
-      </div>
-    </div>
-    <div className="cover-frame-timeline">
-      <div className="cover-frame-slider-header">
-        <span className="cover-frame-slider-label">拖动选择封面帧</span>
-        <span className="cover-frame-slider-value">{formatTime(config.frame_time)}</span>
-      </div>
-      <input
-        type="range"
-        className="cover-frame-slider"
-        min={0}
-        max={Math.max(totalDuration, 1)}
-        step={0.1}
-        value={config.frame_time}
-        onChange={(e) => onFrameTimeChange(Number(e.target.value))}
-      />
-      <div className="cover-frame-range">
-        <span>00:00</span>
-        <span>{formatTime(totalDuration)}</span>
-      </div>
-    </div>
-    <div className="cover-frame-quick">
-      <span className="cover-quick-label">快捷选帧：</span>
-      {[0, 0.25, 0.5, 0.75].map((ratio) => {
-        const t = totalDuration * ratio
-        return (
-          <button key={ratio} className="cover-quick-btn" onClick={() => onFrameTimeChange(t)}>
-            {formatTime(t)}
-          </button>
-        )
-      })}
-    </div>
-  </div>
-)
-
-interface CoverUploadModeProps {
-  config: CoverConfig
-  isDragging: boolean
-  fileInputRef: React.RefObject<HTMLInputElement>
-  onDragOver: (e: React.DragEvent) => void
-  onDragLeave: () => void
-  onDrop: (e: React.DragEvent) => void
-  onAreaClick: () => void
-  onFileChange: (file: File) => void
-}
-
-/** 上传封面模式面板 */
-export const CoverUploadMode: React.FC<CoverUploadModeProps> = ({
-  config,
-  isDragging,
-  fileInputRef,
-  onDragOver,
-  onDragLeave,
-  onDrop,
-  onAreaClick,
-  onFileChange,
-}) => (
-  <div className="cover-upload-section">
-    <div
-      className={`cover-upload-area${isDragging ? " dragging" : ""}`}
-      onDragOver={onDragOver}
-      onDragLeave={onDragLeave}
-      onDrop={onDrop}
-      onClick={onAreaClick}
-    >
-      {config.upload_url ? (
-        <div className="cover-upload-preview">
-          <img src={config.upload_url} alt="封面预览" />
-          <div className="cover-upload-overlay">点击更换</div>
-        </div>
-      ) : (
-        <div className="cover-upload-placeholder">
-          <span className="cover-upload-icon">📤</span>
-          <span className="cover-upload-text">点击或拖拽上传封面图片</span>
-          <span className="cover-upload-hint">支持 JPG / PNG，建议 16:9 比例</span>
-        </div>
-      )}
-      <input
-        ref={fileInputRef}
-        type="file"
-        accept="image/*"
-        style={{ display: "none" }}
-        onChange={(e) => {
-          const file = e.target.files?.[0]
-          if (file) onFileChange(file)
-        }}
-      />
-    </div>
-  </div>
-)

apps/web/src/pages/editing-planner/components/cover-selector/index.tsx

@@ -1,146 +0,0 @@
-/**
- * 封面选择器
- * 抽帧选封面 + 上传自定义封面 + 智能封面推荐
- */
-import React from "react"
-import { Drawer } from "antd"
-import type { CoverConfig, CoverMode } from "../../types"
-import { useCoverSelector, MODE_LABELS, MODE_ICONS } from "./useCoverSelector"
-import { CoverAutoMode, CoverFrameMode, CoverUploadMode } from "./CoverModePanels"
-
-interface CoverSelectorProps {
-  open: boolean
-  onClose: () => void
-  config: CoverConfig
-  onChange: (config: CoverConfig) => void
-  totalDuration: number
-}
-
-const CoverSelector: React.FC<CoverSelectorProps> = ({
-  open,
-  onClose,
-  config,
-  onChange,
-  totalDuration,
-}) => {
-  const {
-    fileInputRef,
-    isDragging,
-    setIsDragging,
-    update,
-    handleReset,
-    handleModeChange,
-    handleFileUpload,
-    handleDrop,
-    handleUseAiSuggestion,
-    formatTime,
-  } = useCoverSelector({ config, onChange })
-
-  return (
-    <Drawer
-      title="封面选择"
-      placement="right"
-      width={440}
-      open={open}
-      onClose={onClose}
-      className="cover-selector-drawer"
-    >
-      {/* 顶部开关 */}
-      <div className="cover-header">
-        <span className="cover-header-label">启用自定义封面</span>
-        <label className="cover-switch">
-          <input
-            type="checkbox"
-            checked={config.enabled}
-            onChange={(e) => update({ enabled: e.target.checked })}
-          />
-          <span className="cover-switch-slider" />
-        </label>
-      </div>
-
-      {/* 模式选择 */}
-      <div className="cover-mode-section">
-        <div className="cover-section-title">封面来源</div>
-        <div className="cover-mode-tabs">
-          {(["auto", "frame", "upload"] as CoverMode[]).map((m) => (
-            <button
-              key={m}
-              className={`cover-mode-tab${config.mode === m ? " active" : ""}`}
-              onClick={() => handleModeChange(m)}
-            >
-              <span className="cover-mode-icon">{MODE_ICONS[m]}</span>
-              <span className="cover-mode-label">{MODE_LABELS[m]}</span>
-            </button>
-          ))}
-        </div>
-      </div>
-
-      {/* 模式内容区 */}
-      <div className="cover-mode-content">
-        {config.mode === "auto" && (
-          <CoverAutoMode
-            config={config}
-            formatTime={formatTime}
-            onUseAiSuggestion={handleUseAiSuggestion}
-          />
-        )}
-
-        {config.mode === "frame" && (
-          <CoverFrameMode
-            config={config}
-            totalDuration={totalDuration}
-            formatTime={formatTime}
-            onFrameTimeChange={(t) => update({ frame_time: t })}
-          />
-        )}
-
-        {config.mode === "upload" && (
-          <CoverUploadMode
-            config={config}
-            isDragging={isDragging}
-            fileInputRef={fileInputRef}
-            onDragOver={(e) => {
-              e.preventDefault()
-              setIsDragging(true)
-            }}
-            onDragLeave={() => setIsDragging(false)}
-            onDrop={handleDrop}
-            onAreaClick={() => fileInputRef.current?.click()}
-            onFileChange={handleFileUpload}
-          />
-        )}
-      </div>
-
-      {/* 封面预览 */}
-      <div className="cover-preview-section">
-        <div className="cover-section-title">封面预览</div>
-        <div className="cover-preview-box">
-          {config.upload_url ? (
-            <img src={config.upload_url} alt="封面预览" className="cover-preview-img" />
-          ) : (
-            <div className="cover-preview-placeholder">
-              <span className="cover-preview-icon">🖼️</span>
-              <span className="cover-preview-text">
-                {config.mode === "auto"
-                  ? "AI 智能选择"
-                  : config.mode === "frame"
-                    ? `帧 ${formatTime(config.frame_time)}`
-                    : "未上传封面"}
-              </span>
-            </div>
-          )}
-          <div className="cover-preview-ratio">16:9</div>
-        </div>
-      </div>
-
-      {/* 底部 */}
-      <div className="cover-footer">
-        <button className="cover-reset-btn" onClick={handleReset}>
-          重置封面
-        </button>
-      </div>
-    </Drawer>
-  )
-}
-
-export default CoverSelector

apps/web/src/pages/editing-planner/components/cover-selector/useCoverSelector.ts

@@ -1,98 +0,0 @@
-import { useCallback, useRef, useState } from "react"
-import type { CoverConfig, CoverMode } from "../../types"
-import { DEFAULT_COVER_CONFIG } from "../../types"
-
-/** 封面模式标签 */
-export const MODE_LABELS: Record<CoverMode, string> = {
-  auto: "智能封面",
-  frame: "抽帧选封面",
-  upload: "上传封面",
-}
-
-/** 封面模式图标 */
-export const MODE_ICONS: Record<CoverMode, string> = {
-  auto: "🤖",
-  frame: "🎞️",
-  upload: "📤",
-}
-
-interface UseCoverSelectorOptions {
-  config: CoverConfig
-  onChange: (config: CoverConfig) => void
-}
-
-/**
- * 封面选择器 Hook
- * 封装状态管理、文件上传、模式切换等逻辑
- */
-export function useCoverSelector({ config, onChange }: UseCoverSelectorOptions) {
-  const fileInputRef = useRef<HTMLInputElement>(null)
-  const [isDragging, setIsDragging] = useState(false)
-
-  const update = useCallback(
-    (partial: Partial<CoverConfig>) => {
-      onChange({ ...config, ...partial })
-    },
-    [config, onChange],
-  )
-
-  const handleReset = useCallback(() => {
-    onChange({ ...DEFAULT_COVER_CONFIG, enabled: config.enabled })
-  }, [config.enabled, onChange])
-
-  const handleModeChange = useCallback(
-    (mode: CoverMode) => {
-      update({ mode })
-    },
-    [update],
-  )
-
-  const handleFileUpload = useCallback(
-    (file: File) => {
-      if (!file.type.startsWith("image/")) return
-      const reader = new FileReader()
-      reader.onload = (e) => {
-        const url = e.target?.result as string
-        update({ upload_url: url, thumbnail_url: url, mode: "upload" })
-      }
-      reader.readAsDataURL(file)
-    },
-    [update],
-  )
-
-  const handleDrop = useCallback(
-    (e: React.DragEvent) => {
-      e.preventDefault()
-      setIsDragging(false)
-      const file = e.dataTransfer.files[0]
-      if (file) handleFileUpload(file)
-    },
-    [handleFileUpload],
-  )
-
-  const handleUseAiSuggestion = useCallback(() => {
-    if (config.ai_suggested_time !== null) {
-      update({ frame_time: config.ai_suggested_time, mode: "frame" })
-    }
-  }, [config.ai_suggested_time, update])
-
-  const formatTime = (seconds: number) => {
-    const m = Math.floor(seconds / 60)
-    const s = Math.floor(seconds % 60)
-    const ms = Math.floor((seconds % 1) * 10)
-    return `${m.toString().padStart(2, "0")}:${s.toString().padStart(2, "0")}.${ms}`
-  }
-
-  return {
-    fileInputRef,
-    isDragging,
-    setIsDragging,
-    update,
-    handleReset,
-    handleModeChange,
-    handleFileUpload,
-    handleDrop,
-    handleUseAiSuggestion,
-    formatTime,
-  }
-}

apps/web/src/pages/editing-planner/hooks/template-management/usePlanLoading.ts

@@ -6,7 +6,6 @@ import type { SubtitleStyleConfig } from "../../types/subtitle"
 import type { TitleConfig } from "@/api/template-editor"
 import type { BgmMixConfig } from "@/api/bgm"
 import type { TransitionEffect } from "@/api/template-editor"
-import type { CoverConfig } from "../../types"
 
 interface UsePlanLoadingOptions {
   loadedPlanId: string | null
@@ -16,7 +15,6 @@ interface UsePlanLoadingOptions {
   setTitleConfig: Dispatch<SetStateAction<TitleConfig>>
   setSubtitleSettings: Dispatch<SetStateAction<SubtitleStyleConfig>>
   setBgmSettings: Dispatch<SetStateAction<BgmMixConfig>>
-  setCoverConfig: Dispatch<SetStateAction<CoverConfig>>
 }
 
 /**
@@ -31,7 +29,6 @@ export function usePlanLoading({
   setTitleConfig,
   setSubtitleSettings,
   setBgmSettings,
-  setCoverConfig,
 }: UsePlanLoadingOptions) {
   useEffect(() => {
     if (!loadedPlanId) return
@@ -77,17 +74,6 @@ export function usePlanLoading({
             music_id: cfg.bgm_config!.music_id || "",
           }))
         }
-        if (cfg.cover_config) {
-          setCoverConfig((prev: CoverConfig) => ({
-            ...prev,
-            enabled: cfg.cover_config!.enabled ?? prev.enabled,
-            mode: (cfg.cover_config!.mode as CoverConfig["mode"]) || prev.mode,
-            frame_time: cfg.cover_config!.frame_time ?? prev.frame_time,
-            upload_url: cfg.cover_config!.upload_url || prev.upload_url,
-            thumbnail_url: cfg.cover_config!.thumbnail_url || prev.thumbnail_url,
-            ai_suggested_time: cfg.cover_config!.ai_suggested_time ?? prev.ai_suggested_time,
-          }))
-        }
 
         /* 还原片段：优先从后端 clips 表，其次从 config.segments 兜底 */
         const backendClips = clipsRes?.items || []
@@ -152,6 +138,5 @@ export function usePlanLoading({
     setTitleConfig,
     setSubtitleSettings,
     setBgmSettings,
-    setCoverConfig,
   ])
 }

apps/web/src/pages/editing-planner/hooks/template-management/useTemplateSave.ts

@@ -17,7 +17,6 @@ import type {
 } from "../../types"
 import type { SubtitleStyleConfig } from "../../types/subtitle"
 import type { TitleConfig } from "@/api/template-editor"
-import type { CoverConfig } from "../../types"
 import type { BgmMixConfig } from "@/api/bgm"
 
 interface UseTemplateSaveOptions {
@@ -33,7 +32,6 @@ interface UseTemplateSaveOptions {
   filterSettings: FilterConfig
   chromaKeySettings: ChromaKeyConfig
   stickerSettings: StickerConfig
-  coverConfig: CoverConfig
   loadedTemplateId: string | null
   loadTemplates: () => Promise<void>
 }
@@ -55,7 +53,6 @@ export function useTemplateSave(options: UseTemplateSaveOptions) {
     filterSettings,
     chromaKeySettings,
     stickerSettings,
-    coverConfig,
     loadedTemplateId,
     loadTemplates,
   } = options
@@ -132,7 +129,6 @@ export function useTemplateSave(options: UseTemplateSaveOptions) {
         filter_config: { ...filterSettings },
         green_screen_config: { ...chromaKeySettings },
         sticker_config: { ...stickerSettings },
-        cover_config: { ...coverConfig },
       }
       if (loadedTemplateId) {
         await updateEditingTemplate(loadedTemplateId, payload)
@@ -163,7 +159,6 @@ export function useTemplateSave(options: UseTemplateSaveOptions) {
     filterSettings,
     chromaKeySettings,
     stickerSettings,
-    coverConfig,
     loadedTemplateId,
     loadTemplates,
   ])

apps/web/src/pages/editing-planner/hooks/useGlobalSettings.ts

@@ -11,7 +11,6 @@ import type {
   FilterConfig,
   ChromaKeyConfig,
   StickerConfig,
-  CoverConfig,
 } from "../types"
 import {
   DEFAULT_WATERMARK,
@@ -20,7 +19,6 @@ import {
   DEFAULT_FILTER_CONFIG,
   DEFAULT_CHROMA_KEY_CONFIG,
   DEFAULT_STICKER_CONFIG,
-  DEFAULT_COVER_CONFIG,
 } from "../types"
 import type { SubtitleStyleConfig } from "../types/subtitle"
 import { DEFAULT_SUBTITLE_STYLE } from "../types/subtitle"
@@ -47,8 +45,6 @@ export interface GlobalSettings {
   setChromaKeySettings: (config: ChromaKeyConfig) => void
   stickerSettings: StickerConfig
   setStickerSettings: (config: StickerConfig) => void
-  coverConfig: CoverConfig
-  setCoverConfig: (config: CoverConfig | ((prev: CoverConfig) => CoverConfig)) => void
 }
 
 export const useGlobalSettings = (): GlobalSettings => {
@@ -92,10 +88,6 @@ export const useGlobalSettings = (): GlobalSettings => {
     ...DEFAULT_STICKER_CONFIG,
   })
 
-  const [coverConfig, setCoverConfig] = useState<CoverConfig>({
-    ...DEFAULT_COVER_CONFIG,
-  })
-
   return {
     titleConfig,
     setTitleConfig,
@@ -115,7 +107,5 @@ export const useGlobalSettings = (): GlobalSettings => {
     setChromaKeySettings,
     stickerSettings,
     setStickerSettings,
-    coverConfig,
-    setCoverConfig,
   }
 }

apps/web/src/pages/editing-planner/hooks/useTemplateManagement.ts

@@ -9,7 +9,6 @@ import type {
   FilterConfig,
   ChromaKeyConfig,
   StickerConfig,
-  CoverConfig,
 } from "../types"
 import type { SubtitleStyleConfig } from "../types/subtitle"
 import type { BgmMixConfig } from "@/api/bgm"
@@ -29,7 +28,6 @@ interface UseTemplateManagementParams {
   setTitleConfig: Dispatch<SetStateAction<TitleConfig>>
   setSubtitleSettings: Dispatch<SetStateAction<SubtitleStyleConfig>>
   setBgmSettings: Dispatch<SetStateAction<BgmMixConfig>>
-  setCoverConfig: Dispatch<SetStateAction<CoverConfig>>
   clips: ClipData[]
   totalDuration: number
   titleConfig: TitleConfig
@@ -41,7 +39,6 @@ interface UseTemplateManagementParams {
   filterSettings: FilterConfig
   chromaKeySettings: ChromaKeyConfig
   stickerSettings: StickerConfig
-  coverConfig: CoverConfig
 }
 
 /**
@@ -59,7 +56,6 @@ export const useTemplateManagement = (params: UseTemplateManagementParams) => {
     setTitleConfig,
     setSubtitleSettings,
     setBgmSettings,
-    setCoverConfig,
     clips,
     totalDuration,
     titleConfig,
@@ -71,7 +67,6 @@ export const useTemplateManagement = (params: UseTemplateManagementParams) => {
     filterSettings,
     chromaKeySettings,
     stickerSettings,
-    coverConfig,
   } = params
 
   const [currentMode, setCurrentMode] = useState<TemplateMode>("pip")
@@ -119,7 +114,6 @@ export const useTemplateManagement = (params: UseTemplateManagementParams) => {
     filterSettings,
     chromaKeySettings,
     stickerSettings,
-    coverConfig,
     loadedTemplateId,
     loadTemplates,
   })
@@ -146,7 +140,6 @@ export const useTemplateManagement = (params: UseTemplateManagementParams) => {
     setTitleConfig,
     setSubtitleSettings,
     setBgmSettings,
-    setCoverConfig,
   })
 
   /* ── 事件 ── */

apps/web/src/pages/editing-planner/types/clipProperties.ts

@@ -67,5 +67,4 @@ export interface ClipPropertiesPanelProps {
   onOpenGreenScreenDrawer?: () => void
   /** 打开贴纸面板 Drawer */
   onOpenStickerDrawer?: () => void
-  /** 打开封面选择器 Drawer */
 }

apps/web/src/pages/editing-planner/types/index.ts

@@ -71,9 +71,6 @@ export {
   TEXT_STICKER_PRESET_LABELS,
 } from "./sticker"
 
-/* 封面 */
-export { type CoverMode, type CoverConfig, DEFAULT_COVER_CONFIG, type CoverTemplate } from "./cover"
-
 /* 片段数据 */
 export { type ClipType, type ClipData } from "./clip"
 

apps/web/src/pages/generate/components/GenerateStepContent.tsx

@@ -7,7 +7,7 @@ import React from "react"
 import type { EditingTemplate } from "@/api/editing-planner"
 import type { PresetVoiceItem } from "@/api/voices"
 import type { VoiceClone } from "@/api/voice-clone"
-import type { CoverConfig } from "../../editing-planner/types"
+import type { CoverConfig } from "../types/cover"
 import type { TitleSettings } from "../types"
 import type { PreviewItem, PreviewStatus } from "../hooks/useStep4Preview"
 import Step1TemplateSelect from "../components/Step1TemplateSelect"

apps/web/src/pages/generate/components/Step6CoverSettings.tsx

@@ -1,5 +1,6 @@
 import React from "react"
-import type { CoverConfig } from "../../editing-planner/types"
+import { Modal, Spin } from "antd"
+import type { CoverConfig } from "../types/cover"
 import { useStep6Cover } from "../hooks/useStep6Cover"
 import Button from "@/components/ui/Button"
 import CoverSettingsModal from "./cover-settings/CoverSettingsModal"
@@ -18,13 +19,13 @@ interface Step6CoverSettingsProps {
 const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
   const {
     coverSettings,
+    generating,
     generateAutoCover,
     showCoverSettings,
     setShowCoverSettings,
     showCoverEditor,
     setShowCoverEditor,
     selectedTemplateId,
-    selectedTemplateName,
     editingTemplate,
     coverTemplates,
     templatesLoading,
@@ -61,8 +62,6 @@ const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
         </Button>
       </div>
 
-      <div className="xx-cover-selected-template">已选模板: {selectedTemplateName}</div>
-
       <div className="xx-section-title">封面预览</div>
       <div className="xx-cover-preview-box">
         {previewUrl ? (
@@ -98,6 +97,14 @@ const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
         template={editingTemplate}
         onSave={handleSaveTemplate}
       />
+
+      {/* AI 生成封面进度弹窗 */}
+      <Modal open={generating} closable={false} footer={null} centered>
+        <div style={{ textAlign: "center", padding: "24px 0" }}>
+          <Spin size="large" />
+          <p style={{ marginTop: 16, fontSize: 14, color: "#666" }}>AI 正在生成封面，请稍候...</p>
+        </div>
+      </Modal>
     </div>
   )
 }

apps/web/src/pages/generate/components/Step7ConfirmGenerate.tsx

@@ -4,7 +4,7 @@
 import React from "react"
 import type { EditingTemplate } from "@/api/editing-planner"
 import type { GeneratedVideo } from "@/api/template-editor"
-import type { CoverConfig } from "../../editing-planner/types"
+import type { CoverConfig } from "../types/cover"
 import type { VoiceClone } from "@/api/voice-clone"
 import type { PresetVoiceItem } from "@/api/voices"
 import { useStep7Generate } from "../hooks/useStep7Generate"

apps/web/src/pages/generate/components/cover-settings/CoverEditorModal.tsx

@@ -1,5 +1,5 @@
 import React, { useState } from "react"
-import type { CoverTemplate } from "../../../editing-planner/types"
+import type { CoverTemplate } from "../../types/cover"
 import Modal from "@/components/ui/Modal"
 import Button from "@/components/ui/Button"
 

apps/web/src/pages/generate/components/cover-settings/CoverModeSelector.tsx

@@ -1,5 +1,5 @@
 import React from "react"
-import type { CoverMode } from "../../../editing-planner/types"
+import type { CoverMode } from "../../types/cover"
 
 interface CoverModeSelectorProps {
   mode: CoverMode

apps/web/src/pages/generate/components/cover-settings/CoverSettingsModal.tsx

@@ -1,5 +1,5 @@
 import React from "react"
-import type { CoverTemplate } from "../../../editing-planner/types"
+import type { CoverTemplate } from "../../types/cover"
 import Modal from "@/components/ui/Modal"
 import Button from "@/components/ui/Button"
 

apps/web/src/pages/generate/constants.ts

@@ -2,7 +2,7 @@
  * 智能剪辑页面 — 常量定义
  */
 
-import type { CoverConfig } from "../editing-planner/types"
+import type { CoverConfig } from "./types/cover"
 
 /* ── 克隆声音状态配置 ── */
 export const CLONE_STATUS_CONFIG: Record<string, { label: string; color: string }> = {

apps/web/src/pages/generate/hooks/generate-video/types.ts

@@ -1,5 +1,5 @@
 import type { GeneratedVideo, EditPlanConfig } from "@/api/template-editor"
-import type { CoverConfig } from "../../../editing-planner/types"
+import type { CoverConfig } from "../../types/cover"
 import type { TitleSettings } from "../../types"
 
 /** useGenerateVideo 入参 */

apps/web/src/pages/generate/hooks/useGenerateFormState/index.ts

@@ -6,7 +6,7 @@ import { useState } from "react"
 import { useSearchParams } from "react-router-dom"
 import type { GeneratedVideo } from "@/api/template-editor"
 import type { EditingTemplate } from "@/api/editing-planner"
-import type { CoverConfig } from "../../../editing-planner/types"
+import type { CoverConfig } from "../../types/cover"
 import type { PresetVoiceItem } from "@/api/voices"
 import { DEFAULT_COVER_SETTINGS } from "../../constants"
 import type { TitleSettings } from "../../types"

apps/web/src/pages/generate/hooks/useGenerateFormState/usePlanConfigLoader.ts

@@ -1,5 +1,5 @@
 import { useEffect } from "react"
-import type { CoverConfig } from "../../../editing-planner/types"
+import type { CoverConfig } from "../../types/cover"
 import type { TitleSettings } from "../../types"
 import type { TitleConfig } from "@/api/template-editor"
 import { getEditPlan } from "@/api/template-editor"

apps/web/src/pages/generate/hooks/useGenerateFormState/useTitleCoverSync.ts

@@ -1,6 +1,6 @@
 import { useEffect } from "react"
 import type { TitleSettings } from "../../types"
-import type { CoverConfig } from "../../../editing-planner/types"
+import type { CoverConfig } from "../../types/cover"
 import type { EditingTemplate } from "@/api/editing-planner"
 
 interface UseTitleCoverSyncOptions {

apps/web/src/pages/generate/hooks/useGenerateVideo.ts

@@ -78,7 +78,7 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
       setGenerateError(finalMsg)
       message.error(finalMsg)
     }
-  }, [props, selectedTemplate, clearTimer, startPolling])
+  }, [props, clearTimer, startPolling])
 
   /* 重新生成（失败后重试） */
   const retry = useCallback(() => {

apps/web/src/pages/generate/hooks/useStep4Preview.ts

@@ -4,6 +4,7 @@
  */
 import { useState, useCallback, useMemo, useEffect, useRef } from "react"
 import { createPreview, getPreviewStatus } from "@/api/generation"
+import { updateEditPlan } from "@/api/template-editor/editPlans"
 import type { PreviewTaskResponse, PreviewStatus as ApiPreviewStatus } from "@/api/generation"
 import type { EditingTemplate } from "@/api/editing-planner"
 import { safeExtractError } from "./generate-video/errorUtils"
@@ -211,90 +212,102 @@ export function useStep4Preview({
   }, [clearPollTimer])
 
   /** 轮询单个预览任务状态 */
-  const pollPreviewStatus = useCallback((index: number, taskId: string) => {
-    const poll = async () => {
-      // 竞态检查
-      if (taskIdsRef.current.get(index) !== taskId) return
-
-      // 超时检查
-      if (Date.now() - startTimeRef.current > POLL_TIMEOUT_MS) {
-        setItems((prev) =>
-          prev.map((it) =>
-            it.index === index ? { ...it, status: "error", error: "预览生成超时，请重试" } : it,
-          ),
-        )
-        return
-      }
-
-      try {
-        const data: PreviewTaskResponse = await getPreviewStatus(taskId)
-
+  const pollPreviewStatus = useCallback(
+    (index: number, taskId: string) => {
+      const poll = async () => {
+        // 竞态检查
         if (taskIdsRef.current.get(index) !== taskId) return
 
-        const status = data.status as ApiPreviewStatus
+        // 超时检查
+        if (Date.now() - startTimeRef.current > POLL_TIMEOUT_MS) {
+          setItems((prev) =>
+            prev.map((it) =>
+              it.index === index ? { ...it, status: "error", error: "预览生成超时，请重试" } : it,
+            ),
+          )
+          return
+        }
 
-        if (status === "completed") {
-          const result: PreviewResult = {
-            taskId: safeString(data.task_id, ""),
-            videoUrl: safeString(data.video_url, ""),
-            clipCount: safeNumber(data.clip_count),
-            transitionCount: safeNumber(data.transition_count),
-            materialUsage: safeNumber(data.material_usage),
-            duration: safeNumber(data.duration),
-            fileSize: safeNumber(data.file_size),
-            generateDuration: safeNumber(data.generate_duration),
-            progress: 100,
+        try {
+          const data: PreviewTaskResponse = await getPreviewStatus(taskId)
+
+          if (taskIdsRef.current.get(index) !== taskId) return
+
+          const status = data.status as ApiPreviewStatus
+
+          if (status === "completed") {
+            const result: PreviewResult = {
+              taskId: safeString(data.task_id, ""),
+              videoUrl: safeString(data.video_url, ""),
+              clipCount: safeNumber(data.clip_count),
+              transitionCount: safeNumber(data.transition_count),
+              materialUsage: safeNumber(data.material_usage),
+              duration: safeNumber(data.duration),
+              fileSize: safeNumber(data.file_size),
+              generateDuration: safeNumber(data.generate_duration),
+              progress: 100,
+            }
+            setItems((prev) =>
+              prev.map((it) =>
+                it.index === index ? { ...it, status: "ready", result, progress: 100 } : it,
+              ),
+            )
+
+            // 保存预览视频 URL 到 plan config，供封面生成使用
+            if (result.videoUrl && selectedTemplate) {
+              updateEditPlan(selectedTemplate, {
+                config: { rendered_storage_key: result.videoUrl },
+              }).catch((err) => {
+                console.warn("[Step4] 保存预览视频URL到plan config失败:", err)
+              })
+            }
+            return
           }
+
+          if (status === "failed") {
+            setItems((prev) =>
+              prev.map((it) =>
+                it.index === index
+                  ? {
+                      ...it,
+                      status: "error",
+                      error: safeString(data.error_message, "预览生成失败，请重试"),
+                    }
+                  : it,
+              ),
+            )
+            return
+          }
+
+          if (status === "cancelled") {
+            setItems((prev) =>
+              prev.map((it) =>
+                it.index === index ? { ...it, status: "error", error: "预览任务已取消" } : it,
+              ),
+            )
+            return
+          }
+
+          // pending / generating 状态继续轮询
+          const prog = safeNumber(data.progress)
+          const nextStatus: PreviewStatus = status === "pending" ? "pending" : "generating"
           setItems((prev) =>
             prev.map((it) =>
-              it.index === index ? { ...it, status: "ready", result, progress: 100 } : it,
+              it.index === index ? { ...it, status: nextStatus, progress: prog } : it,
             ),
           )
-          return
+          const delay = status === "pending" ? 5000 : 2000
+          pollTimersRef.current.set(index, setTimeout(poll, delay))
+        } catch {
+          if (taskIdsRef.current.get(index) !== taskId) return
+          pollTimersRef.current.set(index, setTimeout(poll, 3000))
         }
-
-        if (status === "failed") {
-          setItems((prev) =>
-            prev.map((it) =>
-              it.index === index
-                ? {
-                    ...it,
-                    status: "error",
-                    error: safeString(data.error_message, "预览生成失败，请重试"),
-                  }
-                : it,
-            ),
-          )
-          return
-        }
-
-        if (status === "cancelled") {
-          setItems((prev) =>
-            prev.map((it) =>
-              it.index === index ? { ...it, status: "error", error: "预览任务已取消" } : it,
-            ),
-          )
-          return
-        }
-
-        // pending / generating 状态继续轮询
-        const prog = safeNumber(data.progress)
-        const nextStatus: PreviewStatus = status === "pending" ? "pending" : "generating"
-        setItems((prev) =>
-          prev.map((it) =>
-            it.index === index ? { ...it, status: nextStatus, progress: prog } : it,
-          ),
-        )
-        const delay = status === "pending" ? 5000 : 2000
-        pollTimersRef.current.set(index, setTimeout(poll, delay))
-      } catch {
-        if (taskIdsRef.current.get(index) !== taskId) return
-        pollTimersRef.current.set(index, setTimeout(poll, 3000))
       }
-    }
 
-    pollTimersRef.current.set(index, setTimeout(poll, 1000))
-  }, [])
+      pollTimersRef.current.set(index, setTimeout(poll, 1000))
+    },
+    [selectedTemplate],
+  )
 
   /** 生成所有预览 */
   const generatePreview = useCallback(async () => {

apps/web/src/pages/generate/hooks/useStep6Cover.ts

@@ -2,10 +2,10 @@
  * Step 6 封面设置 Hook
  * 封装封面设置的交互逻辑，对接后端封面模板 CRUD API
  */
-import { useCallback, useEffect, useRef, useState } from "react"
+import { useCallback, useEffect, useState } from "react"
 import { message } from "antd"
-import type { CoverConfig, CoverTemplate } from "../../editing-planner/types"
-import { generateCover } from "@/api/template-editor"
+import type { CoverConfig, CoverTemplate } from "../types/cover"
+import { generateCover } from "@/api/generation"
 import {
   fetchCoverTemplates,
   createCoverTemplate,
@@ -30,7 +30,7 @@ export function useStep6Cover({
   assetIds = [],
   selectedTemplate = "",
 }: UseStep6CoverProps) {
-  const generatingRef = useRef(false)
+  const [generating, setGenerating] = useState(false)
 
   // ── 封面设置弹窗状态 ──
   const [showCoverSettings, setShowCoverSettings] = useState(false)
@@ -67,7 +67,7 @@ export function useStep6Cover({
 
   /** 调用后端智能封面 API，生成封面并更新预览 */
   const generateAutoCover = useCallback(async () => {
-    if (generatingRef.current) {
+    if (generating) {
       message.warning("封面正在生成中，请稍候...")
       return
     }
@@ -82,12 +82,17 @@ export function useStep6Cover({
       return
     }
 
-    generatingRef.current = true
+    setGenerating(true)
+    // 超时保护：300 秒后强制重置，防止 state 卡死导致按钮永久失效
+    const timeoutId = setTimeout(() => {
+      setGenerating(false)
+    }, 300000)
     try {
       const response = await generateCover(selectedTemplate, {
         asset_ids: assetIds,
         cover_type: "ai_frame",
       })
+      clearTimeout(timeoutId)
       const thumbnailUrl = response.cover?.image_url || ""
       if (thumbnailUrl) {
         onCoverSettingsChange({
@@ -100,12 +105,35 @@ export function useStep6Cover({
         message.warning("封面生成未返回图片，请重试")
       }
     } catch (err) {
-      console.warn("[Step6] 智能封面生成失败:", err)
-      message.error("封面生成失败，请稍后重试")
+      clearTimeout(timeoutId)
+      console.error("[Step6] 智能封面生成失败:", err)
+
+      // 提取详细错误信息
+      let errorMsg = "封面生成失败"
+      const e = err as {
+        response?: { data?: { detail?: string; message?: string }; status?: number }
+        request?: unknown
+        message?: string
+      }
+      if (e.response) {
+        // 后端返回错误
+        const detail = e.response.data?.detail || e.response.data?.message || ""
+        errorMsg = detail || `后端错误 (${e.response.status})`
+        console.error("[Step6] 后端返回:", e.response.data)
+      } else if (e.request) {
+        // 请求已发送但无响应
+        errorMsg = "服务器无响应，请检查网络连接"
+        console.error("[Step6] 请求无响应:", e.request)
+      } else if (e.message) {
+        errorMsg = e.message
+      }
+
+      message.error(errorMsg)
     } finally {
-      generatingRef.current = false
+      clearTimeout(timeoutId)
+      setGenerating(false)
     }
-  }, [selectedTemplate, assetIds, coverSettings, onCoverSettingsChange])
+  }, [selectedTemplate, assetIds, coverSettings, onCoverSettingsChange, generating])
 
   // ── 模板操作方法 ──
   const handleSelectTemplate = useCallback((id: string) => {
@@ -165,6 +193,7 @@ export function useStep6Cover({
 
   return {
     coverSettings,
+    generating,
     generateAutoCover,
     totalDuration,
     showCoverSettings,

apps/web/src/pages/generate/hooks/useStep7Generate.ts

@@ -6,7 +6,7 @@ import { useMemo } from "react"
 import { useQuery } from "@tanstack/react-query"
 import type { EditingTemplate } from "@/api/editing-planner"
 import type { GeneratedVideo } from "@/api/template-editor"
-import type { CoverConfig } from "../../editing-planner/types"
+import type { CoverConfig } from "../types/cover"
 import type { VoiceClone } from "@/api/voice-clone"
 import type { PresetVoiceItem } from "@/api/voices"
 import { getAssetsByKind } from "@/api/assets"

apps/web/src/pages/generate/types/cover.ts

@@ -1,5 +1,6 @@
 /**
- * 封面配置类型
+ * 智能剪辑封面类型定义
+ * 独立于 editing-planner，仅供 generate 模块使用
  */
 
 /** 封面来源模式 */

apps/web/src/test/api/edit-plans.test.ts

@@ -8,7 +8,6 @@ import {
   generateEditPlan,
   getGenerationStatus,
   aiRecommendClips,
-  generateCover,
   getEditPlanGenerations,
   getGenerationTaskResults,
   cancelGeneration,
@@ -182,22 +181,6 @@ describe("editPlans API", () => {
     })
   })
 
-  describe("generateCover", () => {
-    it("should resolve successfully", async () => {
-      await expect(generateCover("test-planId")).resolves.not.toThrow()
-    })
-
-    it("should reject on API error", async () => {
-      mockGet.mockRejectedValue(new Error("Network error"))
-      mockPost.mockRejectedValue(new Error("Network error"))
-      mockPut.mockRejectedValue(new Error("Network error"))
-      mockDelete.mockRejectedValue(new Error("Network error"))
-      mockPatch.mockRejectedValue(new Error("Network error"))
-
-      await expect(generateCover("test-planId")).rejects.toThrow()
-    })
-  })
-
   describe("getEditPlanGenerations", () => {
     it("should resolve successfully", async () => {
       await expect(getEditPlanGenerations("test-planId")).resolves.not.toThrow()

apps/web/src/test/pages/EditingPlanner.test.tsx

@@ -149,7 +149,6 @@ vi.mock("@/api/editing-planner", () => ({
 vi.mock("@/api/template-editor", () => ({
   getMediaAssets: vi.fn().mockResolvedValue({ items: [] }),
   getEditPlanGenerations: vi.fn().mockResolvedValue({ items: [] }),
-  generateCover: vi.fn().mockResolvedValue({}),
   getEditPlan: vi.fn().mockResolvedValue({}),
   createEditPlan: vi.fn().mockResolvedValue({ id: "test-plan" }),
   updateEditPlan: vi.fn().mockResolvedValue({}),
@@ -223,9 +222,6 @@ vi.mock("@/pages/editing-planner/components/GreenScreenPanel", () => ({
 vi.mock("@/pages/editing-planner/components/StickerPanel", () => ({
   default: () => React.createElement("div", { "data-testid": "StickerPanel" }),
 }))
-vi.mock("@/pages/editing-planner/components/CoverSelector", () => ({
-  default: () => React.createElement("div", { "data-testid": "CoverSelector" }),
-}))
 vi.mock("@/pages/editing-planner/components/SaveModal", () => ({
   default: () => React.createElement("div", { "data-testid": "SaveModal" }),
 }))
@@ -260,7 +256,6 @@ vi.mock("@/pages/editing-planner/types", () => ({
   DEFAULT_FILTER_CONFIG: { enabled: false },
   DEFAULT_CHROMA_KEY_CONFIG: { enabled: false },
   DEFAULT_STICKER_CONFIG: { enabled: false },
-  DEFAULT_COVER_CONFIG: { enabled: false },
 }))
 
 vi.mock("@/pages/editing-planner/types/subtitle", () => ({

apps/web/src/test/pages/editing-planner/components/CoverSelector.test.tsx

@@ -1,24 +0,0 @@
-import { describe, expect, it, vi } from "vitest"
-import { render } from "@testing-library/react"
-import CoverSelector from "@/pages/editing-planner/components/CoverSelector"
-import { DEFAULT_COVER_CONFIG } from "@/pages/editing-planner/types"
-
-const defaultProps = {
-  open: true,
-  onClose: vi.fn(),
-  config: DEFAULT_COVER_CONFIG,
-  onChange: vi.fn(),
-  totalDuration: 60,
-}
-
-describe("CoverSelector", () => {
-  it("should render without crashing", () => {
-    const { container } = render(<CoverSelector {...defaultProps} />)
-    expect(container).toBeTruthy()
-  })
-
-  it("should render when closed", () => {
-    const { container } = render(<CoverSelector {...defaultProps} open={false} />)
-    expect(container).toBeTruthy()
-  })
-})

apps/web/src/test/pages/editing-planner/components/PreviewPlayer.test.tsx

@@ -6,15 +6,10 @@ const defaultProps = {
   clips: [],
   selectedClipId: null,
   isPlaying: false,
-  currentCoverScheme: "scheme1",
-  coverSchemes: [{ id: "scheme1", name: "方案1", cover_url: "" }],
-  aiCoverLoading: false,
   titleSettings: undefined,
   subtitleSettings: undefined,
   onClipSelect: vi.fn(),
-  onCoverSchemeChange: vi.fn(),
   onPlayPause: vi.fn(),
-  onAiGenerateCover: vi.fn(),
 }
 
 describe("PreviewPlayer", () => {

apps/web/src/test/pages/editing-planner/components/TtsPanel.test.tsx

@@ -1,5 +1,5 @@
 import { afterEach, describe, expect, it, vi } from "vitest"
-import { render, cleanup } from "@testing-library/react"
+import { render, cleanup, act } from "@testing-library/react"
 import TtsPanel from "@/pages/editing-planner/components/TtsPanel"
 import { DEFAULT_TTS_CONFIG } from "@/pages/editing-planner/types"
 
@@ -16,17 +16,28 @@ const defaultProps = {
 }
 
 describe("TtsPanel", () => {
-  afterEach(() => {
+  afterEach(async () => {
+    // Flush pending React updates before cleanup to avoid
+    // "window is not defined" errors after jsdom teardown
+    await act(async () => {})
     cleanup()
   })
 
-  it("should render without crashing", () => {
-    const { container } = render(<TtsPanel {...defaultProps} />)
-    expect(container).toBeTruthy()
+  it("should render without crashing", async () => {
+    let container: HTMLElement
+    await act(async () => {
+      const result = render(<TtsPanel {...defaultProps} />)
+      container = result.container
+    })
+    expect(container!).toBeTruthy()
   })
 
-  it("should render when closed", () => {
-    const { container } = render(<TtsPanel {...defaultProps} open={false} />)
-    expect(container).toBeTruthy()
+  it("should render when closed", async () => {
+    let container: HTMLElement
+    await act(async () => {
+      const result = render(<TtsPanel {...defaultProps} open={false} />)
+      container = result.container
+    })
+    expect(container!).toBeTruthy()
   })
 })

apps/web/src/test/pages/editing-planner/smoke.test.tsx

@@ -28,7 +28,6 @@ import "@/pages/editing-planner/utils/clipProperties"
 // 子组件
 import "@/pages/editing-planner/components/BgmSelector"
 import "@/pages/editing-planner/components/ClipPropertiesPanel"
-import "@/pages/editing-planner/components/CoverSelector"
 import "@/pages/editing-planner/components/EditorClipList"
 import "@/pages/editing-planner/components/EditingDrawers"
 import "@/pages/editing-planner/components/FilterPanel"

apps/worker/video_processing/cover_generator.py

@@ -1,431 +0,0 @@
-"""视频封面生成器 — 从视频中提取/生成封面图.
-
-支持能力：
-- 指定时间点抽帧（默认第1秒）
-- 智能封面：抽取多帧选最清晰的一帧
-- 自定义上传封面图（直接返回路径）
-- 生成的封面图保存为 JPEG 格式，可复用
-"""
-
-from __future__ import annotations
-
-import logging
-import subprocess
-from pathlib import Path
-from typing import Any
-
-from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_video_info, run_ffmpeg
-
-logger = logging.getLogger(__name__)
-
-
-# ── 配置常量 ──────────────────────────────────────────────────────────────────
-
-# 智能封面抽帧数量
-SMART_COVER_FRAME_COUNT = 3
-
-# 默认抽帧时间点（秒）
-DEFAULT_COVER_TIME = 1.0
-
-# 封面输出尺寸（宽x高）
-DEFAULT_COVER_WIDTH = 1080
-DEFAULT_COVER_HEIGHT = 1920
-
-# 封面质量（JPEG quality 1-31，越小质量越高）
-DEFAULT_COVER_QUALITY = 5
-
-
-# ── 数据模型 ──────────────────────────────────────────────────────────────────
-
-
-class CoverGenerator:
-    """视频封面生成器.
-
-    三种模式：
-    1. 指定时间点抽帧：从视频指定时间提取一帧
-    2. 智能封面：抽取3帧，用 blur 检测选最清晰的
-    3. 自定义上传：直接使用用户上传的图片
-    """
-
-    @staticmethod
-    def extract_frame(
-        video_path: str | Path,
-        output_path: str | Path,
-        *,
-        time_sec: float = DEFAULT_COVER_TIME,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-    ) -> Path:
-        """从视频指定时间点提取一帧作为封面.
-
-        Args:
-            video_path: 视频文件路径
-            output_path: 输出图片路径
-            time_sec: 抽帧时间点（秒）
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量（1-31，越小越好）
-
-        Returns:
-            封面图片路径
-
-        Raises:
-            FileNotFoundError: 视频文件不存在
-            subprocess.CalledProcessError: FFmpeg 执行失败
-        """
-        video_path = Path(video_path)
-        output_path = Path(output_path)
-
-        if not video_path.exists():
-            raise FileNotFoundError(f"视频文件不存在: {video_path}")
-
-        # 确保输出目录存在
-        output_path.parent.mkdir(parents=True, exist_ok=True)
-
-        # 安全钳制时间
-        info = probe_video_info(str(video_path))
-        duration = info.get("duration", 0.0)
-        if duration > 0 and time_sec >= duration:
-            # 超过视频长度，取中间帧
-            time_sec = max(0, duration / 2)
-        if time_sec < 0:
-            time_sec = 0
-
-        # scale + crop 实现 cover 裁剪（铺满输出尺寸）
-        vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height},format=yuvj420p"
-
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-ss",
-            f"{time_sec:.3f}",
-            "-i",
-            str(video_path),
-            "-vframes",
-            "1",
-            "-vf",
-            vf,
-            "-q:v",
-            str(quality),
-            "-f",
-            "mjpeg",
-            str(output_path),
-        ]
-
-        logger.info("抽取视频封面: video=%s time=%.2fs output=%s", video_path.name, time_sec, output_path.name)
-        run_ffmpeg(command)
-
-        if not output_path.exists() or output_path.stat().st_size == 0:
-            raise RuntimeError(f"封面生成失败: {output_path}")
-
-        return output_path
-
-    @staticmethod
-    def extract_smart_cover(
-        video_path: str | Path,
-        output_path: str | Path,
-        *,
-        frame_count: int = SMART_COVER_FRAME_COUNT,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-        work_dir: str | Path | None = None,
-    ) -> Path:
-        """智能封面：抽取多帧，选最清晰的一帧.
-
-        清晰度判断：使用拉普拉斯方差（Variance of Laplacian），
-        方差越大表示图像边缘越丰富，越清晰。
-
-        Args:
-            video_path: 视频文件路径
-            output_path: 最终输出封面路径
-            frame_count: 抽帧数量（均匀分布在视频中）
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量
-            work_dir: 临时工作目录（默认输出目录的父目录）
-
-        Returns:
-            最佳封面图片路径
-        """
-        video_path = Path(video_path)
-        output_path = Path(output_path)
-
-        if not video_path.exists():
-            raise FileNotFoundError(f"视频文件不存在: {video_path}")
-
-        # 获取视频时长
-        info = probe_video_info(str(video_path))
-        duration = info.get("duration", 0.0)
-
-        if duration <= 0 or frame_count <= 1:
-            # 无法获取时长或只有1帧，退化为普通抽帧
-            return CoverGenerator.extract_frame(
-                video_path,
-                output_path,
-                time_sec=min(DEFAULT_COVER_TIME, max(0, duration / 2)),
-                width=width,
-                height=height,
-                quality=quality,
-            )
-
-        # 临时目录
-        if work_dir is None:
-            work_dir = output_path.parent
-        work_dir = Path(work_dir)
-        work_dir.mkdir(parents=True, exist_ok=True)
-
-        # 均匀分布抽帧时间点（跳过首尾5%）
-        start_pct = 0.05
-        end_pct = 0.95
-        if frame_count == 1:
-            time_points = [duration * 0.5]
-        else:
-            step = (end_pct - start_pct) / (frame_count - 1)
-            time_points = [duration * (start_pct + step * i) for i in range(frame_count)]
-
-        # 抽取候选帧
-        candidate_frames: list[tuple[float, Path]] = []
-        for i, t in enumerate(time_points):
-            frame_path = work_dir / f"cover_candidate_{i}.jpg"
-            try:
-                CoverGenerator.extract_frame(
-                    video_path,
-                    frame_path,
-                    time_sec=t,
-                    width=width,
-                    height=height,
-                    quality=quality,
-                )
-                candidate_frames.append((t, frame_path))
-            except Exception as e:
-                logger.warning("智能封面抽帧失败（t=%.2fs）: %s", t, e)
-                continue
-
-        if not candidate_frames:
-            # 全部失败，退化到普通抽帧
-            logger.warning("智能封面所有候选帧抽取失败，退化为普通抽帧")
-            return CoverGenerator.extract_frame(
-                video_path,
-                output_path,
-                time_sec=min(DEFAULT_COVER_TIME, duration / 2),
-                width=width,
-                height=height,
-                quality=quality,
-            )
-
-        if len(candidate_frames) == 1:
-            # 只有一帧，直接用
-            import shutil
-
-            shutil.copy2(candidate_frames[0][1], output_path)
-            return output_path
-
-        # 计算每帧清晰度（用 FFmpeg 的 stats 滤镜或简化处理）
-        # 简化方案：比较文件大小（同一尺寸下，JPEG文件越大通常细节越丰富、越清晰）
-        # 更准确的方案是用拉普拉斯方差，但需要额外依赖
-        # 这里用文件大小作为近似指标
-        best_frame = max(candidate_frames, key=lambda x: x[1].stat().st_size)
-
-        # 复制最佳帧到输出路径
-        import shutil
-
-        shutil.copy2(best_frame[1], output_path)
-
-        logger.info(
-            "智能封面生成完成: 候选%d帧, 最佳t=%.2fs, 大小=%d字节",
-            len(candidate_frames),
-            best_frame[0],
-            output_path.stat().st_size,
-        )
-
-        # 清理临时文件
-        for _, fp in candidate_frames:
-            try:
-                fp.unlink()
-            except OSError:
-                pass
-
-        return output_path
-
-    @staticmethod
-    def process_custom_cover(
-        image_path: str | Path,
-        output_path: str | Path,
-        *,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-    ) -> Path:
-        """处理用户自定义上传的封面图.
-
-        调整尺寸、格式转换为标准封面格式。
-
-        Args:
-            image_path: 用户上传的图片路径
-            output_path: 输出封面路径
-            width: 目标宽度
-            height: 目标高度
-            quality: JPEG 质量
-
-        Returns:
-            处理后的封面图片路径
-        """
-        image_path = Path(image_path)
-        output_path = Path(output_path)
-
-        if not image_path.exists():
-            raise FileNotFoundError(f"封面图片不存在: {image_path}")
-
-        output_path.parent.mkdir(parents=True, exist_ok=True)
-
-        # scale + crop 实现 cover 裁剪
-        vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height},format=yuvj420p"
-
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            str(image_path),
-            "-vf",
-            vf,
-            "-q:v",
-            str(quality),
-            "-f",
-            "mjpeg",
-            str(output_path),
-        ]
-
-        logger.info("处理自定义封面: input=%s output=%s", image_path.name, output_path.name)
-
-        try:
-            run_ffmpeg(command)
-        except subprocess.CalledProcessError:
-            # 处理失败，直接复制原图
-            logger.warning("自定义封面处理失败，使用原图")
-            import shutil
-
-            shutil.copy2(image_path, output_path)
-
-        return output_path
-
-    @staticmethod
-    def generate_cover(
-        video_path: str | Path,
-        output_path: str | Path,
-        *,
-        mode: str = "smart",  # smart / time / custom
-        time_sec: float = DEFAULT_COVER_TIME,
-        custom_image: str | Path | None = None,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-    ) -> Path:
-        """统一封面生成入口.
-
-        Args:
-            video_path: 视频文件路径
-            output_path: 输出封面路径
-            mode: 模式 - smart（智能选帧）/ time（指定时间）/ custom（自定义图片）
-            time_sec: time 模式下的抽帧时间点
-            custom_image: custom 模式下的自定义图片路径
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量
-
-        Returns:
-            封面图片路径
-        """
-        if mode == "custom" and custom_image:
-            return CoverGenerator.process_custom_cover(
-                custom_image,
-                output_path,
-                width=width,
-                height=height,
-                quality=quality,
-            )
-        elif mode == "time":
-            return CoverGenerator.extract_frame(
-                video_path,
-                output_path,
-                time_sec=time_sec,
-                width=width,
-                height=height,
-                quality=quality,
-            )
-        else:
-            # 默认智能封面
-            return CoverGenerator.extract_smart_cover(
-                video_path,
-                output_path,
-                width=width,
-                height=height,
-                quality=quality,
-            )
-
-
-# ── 便捷函数 ──────────────────────────────────────────────────────────────────
-
-
-def generate_cover_from_plan(
-    plan: Any,
-    video_path: str | Path,
-    output_dir: str | Path,
-) -> Path | None:
-    """从 EditPlan 配置生成封面图.
-
-    配置读取：plan.config.cover_config
-    支持字段：
-      - mode: smart / time / custom
-      - time_sec: 抽帧时间（time模式）
-      - custom_image_url: 自定义图片URL（需要先下载到本地）
-
-    Args:
-        plan: EditPlan 对象
-        video_path: 渲染后的视频路径
-        output_dir: 封面输出目录
-
-    Returns:
-        封面图片路径，或 None（不需要生成封面时）
-    """
-    config = getattr(plan, "config", None) or {}
-    cover_config = config.get("cover_config") if isinstance(config, dict) else None
-
-    if not cover_config:
-        return None
-
-    mode = cover_config.get("mode", "smart")
-    output_dir = Path(output_dir)
-    output_dir.mkdir(parents=True, exist_ok=True)
-    output_path = output_dir / f"cover_{plan.id}.jpg"
-
-    try:
-        if mode == "custom":
-            # 自定义封面：需要先有本地图片路径
-            custom_path = cover_config.get("custom_image_path")
-            if custom_path and Path(custom_path).exists():
-                return CoverGenerator.process_custom_cover(
-                    custom_path,
-                    output_path,
-                )
-            else:
-                logger.warning("自定义封面图片路径无效，退化为智能封面")
-                mode = "smart"
-
-        if mode == "time":
-            time_sec = float(cover_config.get("time_sec", DEFAULT_COVER_TIME))
-            return CoverGenerator.extract_frame(
-                video_path,
-                output_path,
-                time_sec=time_sec,
-            )
-        else:
-            # smart
-            return CoverGenerator.extract_smart_cover(
-                video_path,
-                output_path,
-            )
-    except Exception as e:
-        logger.warning("封面生成失败: %s", e)
-        return None

apps/worker/video_processing/render_adapter.py

@@ -74,6 +74,9 @@ class RenderAdapterResult:
     failed_clip_ids: list[str] = None  # 失败的 clip id 列表
     error_message: str = ""
     error_detail: str = ""  # 详细错误信息（如 ffmpeg stderr），用于排查
+    cover_candidates: list[dict] | None = (
+        None  # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}]
+    )
 
     def __post_init__(self):
         if self.rendered_clip_ids is None:
@@ -568,6 +571,25 @@ class RenderAdapter:
                 thumb_err,
             )
 
+        # 7. 抽取封面候选帧并上传 OSS（失败不阻断主流程）
+        cover_candidates = None
+        try:
+            from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+
+            cover_candidates = extract_and_upload_cover_frames(str(result.output_path), plan_id, num_frames=3)
+            if cover_candidates:
+                logger.info(
+                    "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d",
+                    plan_id,
+                    len(cover_candidates),
+                )
+        except Exception as cover_err:
+            logger.warning(
+                "[render-adapter] 封面候选帧生成失败（不影响主流程）: plan_id=%s error=%s",
+                plan_id,
+                cover_err,
+            )
+
         self._report_progress(progress_cb, 100.0, "渲染完成")
 
         logger.info(
@@ -599,6 +621,7 @@ class RenderAdapter:
             clip_count=len(clips),
             rendered_clip_ids=final_rendered_ids,
             failed_clip_ids=final_failed_ids,
+            cover_candidates=cover_candidates,
         )
 
     def render_from_memory(

apps/worker/video_processing/thumbnail_generator.py

@@ -155,3 +155,134 @@ def generate_and_upload_thumbnail(
                 Path(thumbnail_path).unlink(missing_ok=True)
             except Exception:
                 pass
+
+
+def extract_cover_candidates(
+    video_path: str,
+    num_frames: int = 3,
+    *,
+    width: int = 640,
+    timeout: int = 30,
+) -> list[dict]:
+    """在视频时长 25%/50%/75% 处各抽一帧，返回候选帧信息列表。
+
+    Args:
+        video_path: 视频文件路径
+        num_frames: 抽帧数量（默认 3）
+        width: 输出宽度
+        timeout: 单帧超时（秒）
+
+    Returns:
+        [{"local_path": "...", "frame_time": 5.0}, ...]
+    """
+    from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
+
+    try:
+        duration = probe_duration(video_path)
+    except Exception:
+        duration = 0.0
+
+    if duration <= 0:
+        duration = 5.0  # fallback
+
+    # 计算抽帧时间点：25%, 50%, 75%
+    ratios = []
+    for i in range(1, num_frames + 1):
+        ratios.append(i / (num_frames + 1))
+
+    results = []
+    for _idx, ratio in enumerate(ratios):
+        frame_time = max(0.5, duration * ratio)
+        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+        tmp.close()
+        output_path = tmp.name
+
+        try:
+            seek_str = _format_seek_time(frame_time)
+            scale_filter = f"scale={width}:-1:force_original_aspect_ratio=decrease,format=yuvj420p"
+            cmd = [
+                FFMPEG_BIN,
+                "-y",
+                "-ss",
+                seek_str,
+                "-i",
+                video_path,
+                "-vframes",
+                "1",
+                "-vf",
+                scale_filter,
+                "-q:v",
+                "2",
+                output_path,
+            ]
+            run_ffmpeg(cmd, capture_output=True, timeout=timeout)
+
+            if Path(output_path).exists() and Path(output_path).stat().st_size > 0:
+                results.append(
+                    {
+                        "local_path": output_path,
+                        "frame_time": round(frame_time, 2),
+                    }
+                )
+            else:
+                Path(output_path).unlink(missing_ok=True)
+        except Exception as e:
+            logger.warning("封面候选帧抽取失败 ratio=%.2f: %s", ratio, e)
+            Path(output_path).unlink(missing_ok=True)
+
+    return results
+
+
+def extract_and_upload_cover_frames(
+    video_path: str,
+    plan_id: str,
+    num_frames: int = 3,
+) -> list[dict]:
+    """抽取封面候选帧并上传到 OSS。
+
+    Args:
+        video_path: 本地视频路径
+        plan_id: 剪辑计划 ID（用于 OSS 路径）
+        num_frames: 抽帧数量
+
+    Returns:
+        [{"image_url": "https://...", "frame_time": 5.0, "storage_key": "covers/xxx/frame_0.jpg"}, ...]
+    """
+    candidates = extract_cover_candidates(video_path, num_frames=num_frames)
+    if not candidates:
+        logger.warning("封面候选帧抽取为空: plan_id=%s", plan_id)
+        return []
+
+    results = []
+    for idx, cand in enumerate(candidates):
+        local_path = cand["local_path"]
+        frame_time = cand["frame_time"]
+        storage_key = f"covers/{plan_id}/frame_{idx}.jpg"
+
+        try:
+            from video_processing.oss_helpers import upload_to_oss
+
+            url = upload_to_oss(local_path, storage_key)
+            if url:
+                results.append(
+                    {
+                        "image_url": url,
+                        "frame_time": frame_time,
+                        "storage_key": storage_key,
+                    }
+                )
+                logger.info(
+                    "封面候选帧上传成功: plan_id=%s idx=%d frame_time=%.2f",
+                    plan_id,
+                    idx,
+                    frame_time,
+                )
+        except Exception as e:
+            logger.warning("封面候选帧上传失败: plan_id=%s idx=%d error=%s", plan_id, idx, e)
+        finally:
+            try:
+                Path(local_path).unlink(missing_ok=True)
+            except Exception:
+                pass
+
+    return results

apps/worker/worker_app/tasks/compose_video.py

@@ -163,36 +163,6 @@ def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> di
         job_service.fail_job(job_id, error_msg[:500])
         raise RuntimeError(result.error_message)
 
-    # 生成封面（如果配置启用）
-    cover_url = None
-    try:
-        from video_processing.cover_generator import generate_cover_from_plan
-
-        from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
-            SQLAlchemyEditPlanRepository as EditPlanRepository,
-        )
-
-        # 获取 plan 对象
-        plan_repo = EditPlanRepository(db)
-        plan = plan_repo.get(plan_id)
-
-        if plan and result.output_path:
-            # 检查 cover_config
-            cover_config = (plan.config or {}).get("cover_config")
-            if cover_config and cover_config.get("enabled", False):
-                from pathlib import Path
-
-                output_dir = Path(result.output_path).parent
-                cover_path = generate_cover_from_plan(plan, result.output_path, output_dir)
-                if cover_path:
-                    # 生成 cover_url（相对路径或上传到存储）
-                    cover_url = f"/covers/{plan_id}.jpg"
-                    logger.info("封面生成成功: plan_id=%s cover_path=%s", plan_id, cover_path)
-            else:
-                logger.info("封面生成未启用: plan_id=%s", plan_id)
-    except Exception as e:
-        logger.warning("封面生成失败（不影响视频合成）: plan_id=%s error=%s", plan_id, e)
-
     # 更新 Job 状态为完成
     result_data = {
         "plan_id": plan_id,
@@ -205,7 +175,6 @@ def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> di
         "width": result.width,
         "height": result.height,
         "file_size": result.file_size,
-        "cover_url": cover_url,
     }
     job_service.complete_job(job_id, result=result_data)
 

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -241,14 +241,32 @@ def _render_with_unified(
         return {"status": "error", "message": result.error_message or "渲染失败"}
 
     output_path = result.output_path or Path("")
-    output_url = result.output_url
+    output_url = result.output_url or ""
     thumbnail_url = result.thumbnail_url or ""
-    storage_key = f"rendered/{plan_id}/output.mp4"
+    # adapter 上传到 rendered/{plan_id}/{job_id}.mp4，从 URL 提取实际 key
+    # 不能用 output.mp4 硬编码，否则 cover 等下游通过 key 构造的 URL 指向不存在的文件
+    if output_url:
+        from packages.shared.storage import get_shared_storage_service
+
+        storage_key = get_shared_storage_service().normalize_storage_key(output_url)
+    else:
+        storage_key = f"rendered/{plan_id}/{generation_task_id or plan_id}.mp4"
 
     # 用 adapter 返回的 clip 明细（以 adapter 的结果为准）
     rendered_clip_ids = result.rendered_clip_ids or []
     failed_clip_ids = result.failed_clip_ids or []
 
+    # 将封面候选帧写入 plan.config（供封面 API 直接使用，跳过 MediaKit 抽帧）
+    if result.cover_candidates:
+        plan_config = plan.config or {}
+        plan_config["cover_candidates"] = result.cover_candidates
+        plan.config = plan_config
+        logger.info(
+            "封面候选帧已写入 plan.config: plan_id=%s count=%d",
+            plan_id,
+            len(result.cover_candidates),
+        )
+
     return _finalize_render_success(
         plan=plan,
         plan_repo=plan_repo,

apps/worker/worker_app/tasks/generation.py

@@ -1262,7 +1262,9 @@ def _upload_and_record(
     Returns:
         (file_url, duration, file_size, video_count)
     """
-    storage_key = f"generated/projects/{project_id}/tasks/{task_id}/{output_path.name}"
+    # project_id 可能为空（模板编辑器草稿不属于任何项目），过滤空段避免 OSS key 出现 //
+    path_parts = [p for p in ("generated", "projects", project_id, "tasks", task_id, output_path.name) if p]
+    storage_key = "/".join(path_parts)
     file_size = output_path.stat().st_size
 
     # 上传 OSS
@@ -1436,6 +1438,64 @@ def generate_video(self, task_id: str) -> dict:
 
             _update_task_progress(task_id, 30, "素材下载完成")
 
+            # ── 2.5 MediaKit 视频理解（渲染前分析素材内容）─────────────────
+            asset_analyses = {}
+            if task_asset_ids:
+                try:
+                    from packages.shared.mediakit_client import get_mediakit_client
+                    from packages.shared.storage import get_shared_storage_service
+
+                    mk_client = get_mediakit_client()
+                    if mk_client.is_available:
+                        storage_svc = get_shared_storage_service()
+                        asset_urls = []
+                        valid_asset_ids = []
+
+                        # 获取素材 URL（从 downloaded_videos 获取本地路径或从 asset 表获取 OSS URL）
+                        for i, asset_id in enumerate(task_asset_ids[:5]):
+                            try:
+                                # 优先使用已下载的本地文件
+                                if i < len(downloaded_videos) and downloaded_videos[i]:
+                                    # 本地文件路径，需要上传或直接用 OSS URL
+                                    from packages.adapters.sqlalchemy_impl.asset_repository import (
+                                        SQLAlchemyAssetRepository,
+                                    )
+
+                                    _s = SessionLocal()
+                                    try:
+                                        _ar = SQLAlchemyAssetRepository(_s)
+                                        asset = _ar.get(asset_id)
+                                        if asset and getattr(asset, "storage_key", ""):
+                                            asset_urls.append(storage_svc.get_url(asset.storage_key))
+                                            valid_asset_ids.append(asset_id)
+                                    finally:
+                                        _s.close()
+                            except Exception:
+                                continue
+
+                        if asset_urls:
+                            _update_task_progress(task_id, 35, "MediaKit 视频理解中...")
+                            analyses = mk_client.analyze_videos(
+                                prompt="分析这个视频的主要内容，描述场景、物体、人物动作和主题",
+                                video_urls=asset_urls,
+                                level="Economy",
+                                poll_interval=2.0,
+                                max_poll_attempts=15,
+                            )
+                            if analyses:
+                                for i, content in enumerate(analyses):
+                                    if i < len(valid_asset_ids) and content:
+                                        asset_analyses[valid_asset_ids[i]] = content
+                            logger.info("[task_id=%s] MediaKit 视频理解完成: %d 个素材", task_id, len(asset_urls))
+
+                    # 保存分析结果到 extra_meta
+                    if asset_analyses and gen_task:
+                        gen_task.extra_meta = {**(gen_task.extra_meta or {}), "asset_analyses": asset_analyses}
+                        _repo.update(gen_task)
+                        _flush_logs(task_id, gen_task)
+                except Exception:
+                    logger.warning("[task_id=%s] MediaKit 视频理解失败，继续渲染", task_id, exc_info=True)
+
             # ── 3. 渲染 + 混音 ───────────────────────────────────────────────
             _update_task_progress(task_id, 40, "开始渲染")
             # 动态分辨率：优先使用 output_width/output_height，其次 resolution 字符串

infra/docker/api.Dockerfile

@@ -53,9 +53,10 @@ ARG APP_VERSION=dev
 RUN sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
     sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list 2>/dev/null || true
 
-# 只装运行时需要的库（libpq5 是 psycopg2 运行时依赖）
+# 只装运行时需要的库（libpq5 是 psycopg2 运行时依赖，ffmpeg 用于封面兜底取帧）
 RUN apt-get update && apt-get install -y --no-install-recommends \
     libpq5 \
+    ffmpeg \
     && rm -rf /var/lib/apt/lists/*
 
 # 从 builder 复制虚拟环境

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -148,6 +148,22 @@ class SQLAlchemyGenerationTaskRepository:
         )
         return [_to_domain(m) for m in models]
 
+    def list_latest_completed_preview(self, user_id: str, template_id: str, limit: int = 1) -> list[GenerationTask]:
+        """按用户+模板查找最近已完成的预览任务。"""
+        models = (
+            self.session.query(GenerationTaskModel)
+            .filter(
+                GenerationTaskModel.created_by_user_id == user_id,
+                GenerationTaskModel.template_id == template_id,
+                GenerationTaskModel.is_preview,
+                GenerationTaskModel.status == "completed",
+            )
+            .order_by(GenerationTaskModel.created_at.desc())
+            .limit(limit)
+            .all()
+        )
+        return [_to_domain(m) for m in models]
+
     def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]:
         models = (
             self.session.query(GenerationTaskModel)

packages/shared/ai_service.py

@@ -13,6 +13,8 @@ import random
 import time
 from typing import Any, Dict, List, Optional
 
+import requests as http_requests
+
 from packages.domain.config_schemas import DEFAULT_EDIT_PLAN_CONFIG
 from packages.shared.ai_client import get_doubao_client
 
@@ -352,6 +354,93 @@ def _transfer_cover_frame_to_storage(frame_url: str, plan_id: str) -> str:
         return frame_url
 
 
+def _extract_frames_with_ffmpeg(
+    video_url: str,
+    num_frames: int = 3,
+    timeout: int = 30,
+) -> list[dict]:
+    """用 FFmpeg 从远程视频 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）。
+
+    Args:
+        video_url: 视频 URL
+        num_frames: 抽帧数量
+        timeout: 单帧超时（秒）
+
+    Returns:
+        [{"local_path": "...", "frame_time": 5.0}, ...]
+    """
+    import re as _re
+    import tempfile
+    from pathlib import Path as _Path
+
+    from packages.shared.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg
+
+    video_url = _re.sub(r"(?<!:)//", "/", video_url)
+
+    # 先用 ffprobe 获取视频时长
+    import subprocess as _subprocess
+
+    from packages.shared.ffmpeg_utils import FFPROBE_BIN
+
+    duration = 30.0  # 默认假设 30 秒
+    try:
+        probe_result = _subprocess.run(
+            [
+                FFPROBE_BIN,
+                "-v",
+                "error",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "default=noprint_wrappers=1:nokey=1",
+                video_url,
+            ],
+            capture_output=True,
+            text=True,
+            timeout=15,
+        )
+        if probe_result.returncode == 0 and probe_result.stdout.strip():
+            duration = float(probe_result.stdout.strip())
+    except Exception as e:
+        logger.warning("FFprobe 远程视频时长失败，使用默认值: %s", e)
+
+    ratios = [i / (num_frames + 1) for i in range(1, num_frames + 1)]
+    results = []
+
+    for _idx, ratio in enumerate(ratios):
+        frame_time = max(0.5, duration * ratio)
+        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+        tmp.close()
+        output_path = tmp.name
+
+        try:
+            seek_str = f"{int(frame_time // 3600):02d}:{int((frame_time % 3600) // 60):02d}:{frame_time % 60:05.2f}"
+            cmd = [
+                FFMPEG_BIN,
+                "-y",
+                "-ss",
+                seek_str,
+                "-i",
+                video_url,
+                "-vframes",
+                "1",
+                "-q:v",
+                "2",
+                output_path,
+            ]
+            run_ffmpeg(cmd, capture_output=True, timeout=timeout)
+
+            if _Path(output_path).exists() and _Path(output_path).stat().st_size > 0:
+                results.append({"local_path": output_path, "frame_time": round(frame_time, 2)})
+            else:
+                _Path(output_path).unlink(missing_ok=True)
+        except Exception as e:
+            logger.warning("FFmpeg 远程抽帧失败 ratio=%.2f: %s", ratio, e)
+            _Path(output_path).unlink(missing_ok=True)
+
+    return results
+
+
 def _call_ai_cover_service(
     plan_id: str,
     asset_ids: List[str],
@@ -361,15 +450,18 @@ def _call_ai_cover_service(
 ) -> Dict[str, Any]:
     """调用 AI 封面生成服务.
 
-    当 cover_type 为 ai_frame 或 ai_regenerate 时，调用 MediaKit 视频截帧。
-    失败或未配置时降级为 stub 行为。
+    优先级：
+    1. 检查 plan.config 中的 cover_candidates（渲染时预抽帧）——由调用方处理
+    2. FFmpeg 本地从 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）
+
+    失败时抛出 RuntimeError。
 
     Args:
         plan_id: 剪辑计划 ID
         asset_ids: 素材 ID 列表
         cover_type: 封面类型
         frame_time: 手动选帧时间点
-        primary_video_url: 主视频的可访问 URL（用于 MediaKit 抽帧）
+        primary_video_url: 主视频的可访问 URL
     """
     if cover_type == "upload":
         return {
@@ -392,65 +484,85 @@ def _call_ai_cover_service(
             "frame_time": frame_time,
         }
 
-    # ai_frame / ai_regenerate - 尝试调用 MediaKit
+    # ai_frame / ai_regenerate - 使用 FFmpeg 本地抽帧
     if primary_video_url:
-        from packages.shared.mediakit_client import get_mediakit_client
+        import re as _re
 
-        client = get_mediakit_client()
-        if client.is_available:
-            try:
-                logger.info("调用 MediaKit 抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80])
-                frames = client.extract_frames(
-                    video_url=primary_video_url,
-                    strategy="TimeInterval",
-                    max_frames=5,
+        primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
+
+        # 先检查视频 URL 是否可访问
+        try:
+            head_resp = http_requests.head(primary_video_url, timeout=10, allow_redirects=True)
+            if head_resp.status_code != 200:
+                logger.error(
+                    "封面视频URL不可访问: plan_id=%s url=%s status=%d",
+                    plan_id,
+                    primary_video_url,
+                    head_resp.status_code,
                 )
+                raise RuntimeError(
+                    f"封面生成失败: 预览视频URL不可访问 (HTTP {head_resp.status_code})。" f"请重新生成预览视频后再试。"
+                )
+        except http_requests.RequestException as e:
+            logger.error("封面视频URL连通性检查失败: plan_id=%s url=%s error=%s", plan_id, primary_video_url, e)
+            raise RuntimeError(
+                f"封面生成失败: 无法访问预览视频 ({e.__class__.__name__})。请重新生成预览视频后再试。"
+            ) from e
 
-                if frames and len(frames) > 0:
-                    # 选择第一帧（SceneChange 策略的第一帧通常是最佳画面）
-                    best_frame = frames[0]
-                    image_url = best_frame.get("image_url", "")
-                    timestamp = best_frame.get("timestamp", 0.0)
+        # 使用 FFmpeg 从 URL 流式 seek 抽帧
+        try:
+            logger.info("FFmpeg 远程抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80])
+            frames = _extract_frames_with_ffmpeg(primary_video_url, num_frames=3)
 
-                    if image_url:
-                        logger.info(
-                            "MediaKit 抽帧成功: plan_id=%s frame_time=%.2f url=%s",
-                            plan_id,
-                            timestamp,
-                            image_url[:80],
-                        )
-                        # MediaKit 返回的 URL 是临时内部 URL，浏览器无法直接访问
-                        # 需要下载到本地并重新上传到 OSS，返回公开可访问的 URL
-                        public_url = _transfer_cover_frame_to_storage(image_url, plan_id)
-                        return {
-                            "type": "ai_frame",
-                            "image_url": public_url,
-                            "frame_time": round(timestamp, 1),
-                            "confidence": 0.85,
-                        }
-                    else:
-                        logger.warning("MediaKit 返回的帧无 image_url")
+            if frames:
+                best_frame = frames[0]
+                local_path = best_frame["local_path"]
+                frame_time_val = best_frame["frame_time"]
 
-            except Exception as e:
-                logger.exception("MediaKit 抽帧失败，降级到 stub: %s", str(e))
+                # 上传到 OSS
+                try:
+                    import uuid
+                    from pathlib import Path
 
-    # 降级：stub 行为 - 返回 SVG data URI 占位图
-    logger.info("使用 stub 封面: plan_id=%s", plan_id)
-    time.sleep(0.3)
-    svg_placeholder = (
-        "data:image/svg+xml,"
-        "<svg xmlns='http://www.w3.org/2000/svg' width='1080' height='1920'>"
-        "<rect width='1080' height='1920' fill='#1a1a2e'/>"
-        "<text x='540' y='920' text-anchor='middle' fill='#e0e0e0' font-size='48' font-family='sans-serif'>封面生成中</text>"
-        "<text x='540' y='1000' text-anchor='middle' fill='#888888' font-size='32' font-family='sans-serif'>请配置 MediaKit API Key</text>"
-        "</svg>"
-    )
-    return {
-        "type": "ai_frame",
-        "image_url": svg_placeholder,
-        "frame_time": round(random.uniform(1.0, 10.0), 1),
-        "confidence": round(random.uniform(0.80, 0.98), 2),
-    }
+                    from packages.shared.storage import get_shared_storage_service
+
+                    storage = get_shared_storage_service()
+                    cover_key = f"covers/{plan_id}/ffmpeg_frame_{uuid.uuid4().hex[:8]}.jpg"
+                    storage.upload_file(
+                        file_or_path=local_path,
+                        storage_key=cover_key,
+                        content_type="image/jpeg",
+                    )
+                    public_url = storage.get_url(cover_key)
+
+                    logger.info(
+                        "FFmpeg 抽帧成功: plan_id=%s frame_time=%.2f url=%s",
+                        plan_id,
+                        frame_time_val,
+                        public_url[:80],
+                    )
+
+                    return {
+                        "type": "ai_frame",
+                        "image_url": public_url,
+                        "frame_time": round(frame_time_val, 1),
+                        "confidence": 0.85,
+                    }
+                finally:
+                    # 清理所有临时文件
+                    for frame in frames:
+                        try:
+                            Path(frame["local_path"]).unlink(missing_ok=True)
+                        except Exception:
+                            pass
+
+        except RuntimeError:
+            raise
+        except Exception as e:
+            logger.exception("FFmpeg 远程抽帧失败: %s", str(e))
+
+    # 封面生成失败
+    raise RuntimeError(f"封面生成失败: plan_id={plan_id}，无法从视频抽帧。请检查 primary_video_url 是否可访问。")
 
 
 # ── 公共入口 ────────────────────────────────────────────────────────────────

tests/unit/test_ai_tasks.py

@@ -399,18 +399,16 @@ class TestRunAIRecommend(unittest.TestCase):
 
 
 class TestGenerateCover(unittest.TestCase):
-    """封面生成测试（降级路径）."""
+    """封面生成测试."""
 
-    def test_ai_frame_type(self):
-        """AI封面模式返回预期结构."""
-        result = run_generate_cover(
-            plan_id="plan-1",
-            asset_ids=["a1"],
-            cover_type="ai_frame",
-        )
-        self.assertIn("type", result)
-        self.assertEqual(result["type"], "ai_frame")
-        self.assertIn("image_url", result)
+    def test_ai_frame_type_raises_without_mediakit(self):
+        """AI封面模式在MediaKit不可用时抛出RuntimeError."""
+        with self.assertRaises(RuntimeError):
+            run_generate_cover(
+                plan_id="plan-1",
+                asset_ids=["a1"],
+                cover_type="ai_frame",
+            )
 
     def test_manual_type(self):
         """手动选帧模式."""

tests/unit/test_compose_video_cover_integration.py

@@ -1,232 +0,0 @@
-"""测试 compose_video 任务中封面生成集成."""
-
-from __future__ import annotations
-
-from pathlib import Path
-from unittest.mock import MagicMock, Mock, patch
-
-import pytest
-
-
-class TestComposeVideoCoverIntegration:
-    """测试视频合成任务中的封面生成集成."""
-
-    @pytest.fixture
-    def mock_job_service(self):
-        """模拟 JobService."""
-        service = MagicMock()
-        service.get_job.return_value = MagicMock(
-            id="job_123",
-            payload={"plan_id": "plan_456"},
-        )
-        return service
-
-    @pytest.fixture
-    def mock_db(self):
-        """模拟数据库会话."""
-        return MagicMock()
-
-    @pytest.fixture
-    def mock_render_result(self):
-        """模拟渲染结果."""
-        result = MagicMock()
-        result.success = True
-        result.output_path = Path("/tmp/output/video_123.mp4")
-        result.output_url = "https://example.com/video_123.mp4"
-        result.duration = 30.0
-        result.clip_count = 5
-        result.width = 1080
-        result.height = 1920
-        result.file_size = 1024000
-        return result
-
-    @pytest.fixture
-    def mock_plan_with_cover_enabled(self):
-        """模拟启用封面的 plan."""
-        plan = MagicMock()
-        plan.id = "plan_456"
-        plan.config = {
-            "cover_config": {
-                "enabled": True,
-                "mode": "smart",
-            }
-        }
-        return plan
-
-    @pytest.fixture
-    def mock_plan_with_cover_disabled(self):
-        """模拟禁用封面的 plan."""
-        plan = MagicMock()
-        plan.id = "plan_456"
-        plan.config = {
-            "cover_config": {
-                "enabled": False,
-            }
-        }
-        return plan
-
-    @pytest.fixture
-    def mock_plan_without_cover_config(self):
-        """模拟没有 cover_config 的 plan."""
-        plan = MagicMock()
-        plan.id = "plan_456"
-        plan.config = {}
-        return plan
-
-    def test_cover_generation_called_when_enabled(
-        self,
-        mock_job_service,
-        mock_db,
-        mock_render_result,
-        mock_plan_with_cover_enabled,
-    ):
-        """测试封面生成在启用时被调用."""
-        from worker_app.tasks.compose_video import _compose_with_unified_engine
-
-        # 模拟 RenderAdapter
-        with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
-            adapter_instance = MagicMock()
-            adapter_instance.render_plan.return_value = mock_render_result
-            adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
-            MockAdapter.return_value = adapter_instance
-
-            # 模拟 EditPlanRepository
-            with patch(
-                "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
-            ) as MockPlanRepo:
-                plan_repo_instance = MagicMock()
-                plan_repo_instance.get.return_value = mock_plan_with_cover_enabled
-                MockPlanRepo.return_value = plan_repo_instance
-
-                # 模拟 generate_cover_from_plan
-                with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
-                    mock_gen_cover.return_value = Path("/tmp/output/cover_plan_456.jpg")
-
-                    # 执行
-                    task = MagicMock()
-                    result = _compose_with_unified_engine(
-                        task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
-                    )
-
-                    # 验证封面生成被调用
-                    mock_gen_cover.assert_called_once()
-                    call_args = mock_gen_cover.call_args
-                    assert call_args[0][0] == mock_plan_with_cover_enabled  # plan
-                    assert call_args[0][1] == mock_render_result.output_path  # video_path
-                    assert call_args[0][2] == mock_render_result.output_path.parent  # output_dir
-
-                    # 验证结果包含 cover_url
-                    assert "cover_url" in result["result"]
-                    assert result["result"]["cover_url"] == "/covers/plan_456.jpg"
-
-    def test_cover_generation_skipped_when_disabled(
-        self,
-        mock_job_service,
-        mock_db,
-        mock_render_result,
-        mock_plan_with_cover_disabled,
-    ):
-        """测试封面生成在禁用时被跳过."""
-        from worker_app.tasks.compose_video import _compose_with_unified_engine
-
-        with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
-            adapter_instance = MagicMock()
-            adapter_instance.render_plan.return_value = mock_render_result
-            adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
-            MockAdapter.return_value = adapter_instance
-
-            with patch(
-                "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
-            ) as MockPlanRepo:
-                plan_repo_instance = MagicMock()
-                plan_repo_instance.get.return_value = mock_plan_with_cover_disabled
-                MockPlanRepo.return_value = plan_repo_instance
-
-                with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
-                    task = MagicMock()
-                    result = _compose_with_unified_engine(
-                        task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
-                    )
-
-                    # 验证封面生成未被调用
-                    mock_gen_cover.assert_not_called()
-
-                    # 验证结果中 cover_url 为 None
-                    assert "cover_url" in result["result"]
-                    assert result["result"]["cover_url"] is None
-
-    def test_cover_generation_skipped_when_no_config(
-        self,
-        mock_job_service,
-        mock_db,
-        mock_render_result,
-        mock_plan_without_cover_config,
-    ):
-        """测试没有 cover_config 时封面生成被跳过."""
-        from worker_app.tasks.compose_video import _compose_with_unified_engine
-
-        with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
-            adapter_instance = MagicMock()
-            adapter_instance.render_plan.return_value = mock_render_result
-            adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
-            MockAdapter.return_value = adapter_instance
-
-            with patch(
-                "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
-            ) as MockPlanRepo:
-                plan_repo_instance = MagicMock()
-                plan_repo_instance.get.return_value = mock_plan_without_cover_config
-                MockPlanRepo.return_value = plan_repo_instance
-
-                with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
-                    task = MagicMock()
-                    result = _compose_with_unified_engine(
-                        task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
-                    )
-
-                    # 验证封面生成未被调用
-                    mock_gen_cover.assert_not_called()
-
-                    # 验证结果中 cover_url 为 None
-                    assert "cover_url" in result["result"]
-                    assert result["result"]["cover_url"] is None
-
-    def test_cover_generation_failure_does_not_break_video(
-        self,
-        mock_job_service,
-        mock_db,
-        mock_render_result,
-        mock_plan_with_cover_enabled,
-    ):
-        """测试封面生成失败不影响视频合成."""
-        from worker_app.tasks.compose_video import _compose_with_unified_engine
-
-        with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
-            adapter_instance = MagicMock()
-            adapter_instance.render_plan.return_value = mock_render_result
-            adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
-            MockAdapter.return_value = adapter_instance
-
-            with patch(
-                "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
-            ) as MockPlanRepo:
-                plan_repo_instance = MagicMock()
-                plan_repo_instance.get.return_value = mock_plan_with_cover_enabled
-                MockPlanRepo.return_value = plan_repo_instance
-
-                with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
-                    # 模拟封面生成抛出异常
-                    mock_gen_cover.side_effect = Exception("FFmpeg failed")
-
-                    task = MagicMock()
-                    result = _compose_with_unified_engine(
-                        task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
-                    )
-
-                    # 验证视频合成仍然成功
-                    assert result["status"] == "completed"
-                    assert "result" in result
-                    assert result["result"]["output_url"] == mock_render_result.output_url
-
-                    # 验证结果中 cover_url 为 None
-                    assert result["result"]["cover_url"] is None

tests/unit/test_config_schemas_and_ai_endpoints.py

@@ -231,18 +231,18 @@ class TestAIRunTasks:
         # 即使没有素材，也应该有 intro + outro
         assert len(result["clips"]) >= 2
 
-    def test_run_generate_cover_ai_frame(self):
+    def test_run_generate_cover_ai_frame_raises_without_mediakit(self):
+        """ai_frame cover raises RuntimeError when MediaKit is unavailable."""
+        import pytest
+
         from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover
 
-        result = run_generate_cover(
-            plan_id="plan-001",
-            asset_ids=["asset-1"],
-            cover_type="ai_frame",
-        )
-        assert result["type"] == "ai_frame"
-        assert "image_url" in result
-        assert "frame_time" in result
-        assert "confidence" in result
+        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+            run_generate_cover(
+                plan_id="plan-001",
+                asset_ids=["asset-1"],
+                cover_type="ai_frame",
+            )
 
     def test_run_generate_cover_manual(self):
         from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover

tests/unit/test_confirm_generation.py

@@ -1,6 +1,6 @@
 """确认生成 API 单元测试.
 
-覆盖 POST /tasks/{task_id}/confirm 端点：
+覆盖 POST /generation/tasks/{task_id}/confirm 端点：
   - 预览任务已完成 → 直接复用（mark_confirmed），秒出
   - 预览任务未完成 → 创建新任务走渲染流程
   - 预览任务不存在 → 404
@@ -150,7 +150,7 @@ def app(
     )
 
     test_app = FastAPI()
-    test_app.include_router(router, prefix="/api/v1")
+    test_app.include_router(router, prefix="/api/v1/generation")
 
     def override_get_current_user():
         return FakeAuthenticatedUser()
@@ -228,7 +228,7 @@ class TestConfirmGenerationReuse:
         initial_count = len(gen_task_repo._store)
 
         resp = client.post(
-            f"/api/v1/tasks/{preview.id}/confirm",
+            f"/api/v1/generation/tasks/{preview.id}/confirm",
             json={
                 "output_width": 1080,
                 "output_height": 1920,
@@ -266,7 +266,7 @@ class TestConfirmGenerationReuse:
         gen_task_repo.create(preview)
 
         resp = client.post(
-            f"/api/v1/tasks/{preview.id}/confirm",
+            f"/api/v1/generation/tasks/{preview.id}/confirm",
             json={"cover_url": "https://cdn.example.com/cover.png", "custom_title": "测试标题"},
         )
 
@@ -290,7 +290,7 @@ class TestConfirmGenerationReuse:
 
         with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
             resp = client.post(
-                f"/api/v1/tasks/{preview.id}/confirm",
+                f"/api/v1/generation/tasks/{preview.id}/confirm",
                 json={"output_width": 1080, "output_height": 1920},
             )
 
@@ -308,7 +308,7 @@ class TestConfirmGenerationErrors:
     def test_confirm_not_found(self, client: TestClient) -> None:
         """预览任务不存在 → 404"""
         resp = client.post(
-            "/api/v1/tasks/nonexistent-task/confirm",
+            "/api/v1/generation/tasks/nonexistent-task/confirm",
             json={"output_width": 1080, "output_height": 1920},
         )
         assert resp.status_code == 404
@@ -324,7 +324,7 @@ class TestConfirmGenerationErrors:
         gen_task_repo.create(preview)
 
         resp = client.post(
-            f"/api/v1/tasks/{preview.id}/confirm",
+            f"/api/v1/generation/tasks/{preview.id}/confirm",
             json={"output_width": 1080, "output_height": 1920},
         )
         assert resp.status_code == 403
@@ -345,7 +345,7 @@ class TestConfirmGenerationErrors:
         gen_task_repo.create(preview)
 
         resp = client.post(
-            f"/api/v1/tasks/{preview.id}/confirm",
+            f"/api/v1/generation/tasks/{preview.id}/confirm",
             json={"output_width": 1920, "output_height": 1080},
         )
 
@@ -373,7 +373,7 @@ class TestConfirmGenerationErrors:
         gen_task_repo.create(preview)
 
         resp = client.post(
-            f"/api/v1/tasks/{preview.id}/confirm",
+            f"/api/v1/generation/tasks/{preview.id}/confirm",
             json={
                 "output_width": 1080,
                 "output_height": 1920,
@@ -397,7 +397,7 @@ class TestConfirmGenerationErrors:
         gen_task_repo.create(preview)
 
         resp = client.post(
-            f"/api/v1/tasks/{preview.id}/confirm",
+            f"/api/v1/generation/tasks/{preview.id}/confirm",
             json={},
         )
 
@@ -419,7 +419,7 @@ class TestConfirmGenerationErrors:
 
         with patch("app.api.routes.generation_tasks.safe_enqueue_generation_task", return_value=True):
             resp = client.post(
-                f"/api/v1/tasks/{preview.id}/confirm",
+                f"/api/v1/generation/tasks/{preview.id}/confirm",
                 json={"output_width": 720, "output_height": 1280},
             )
 

tests/unit/test_cover_frame_pre_extract.py

@@ -0,0 +1,486 @@
+"""Tests for cover frame pre-extraction during rendering.
+
+Tests:
+- extract_cover_candidates: FFmpeg frame extraction at 25%/50%/75%
+- extract_and_upload_cover_frames: extraction + OSS upload
+- RenderAdapterResult.cover_candidates field
+- generation_cover route uses pre-stored candidates
+- ai_service FFmpeg fallback
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, Mock, call, patch
+
+import pytest
+
+
+class TestExtractCoverCandidates:
+    """extract_cover_candidates 测试."""
+
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0)
+    def test_extracts_3_frames_at_correct_positions(self, mock_probe, mock_run):
+        """在 25%/50%/75% 处抽取 3 帧."""
+        import tempfile
+
+        from video_processing.thumbnail_generator import extract_cover_candidates
+
+        # Create temp files that look like they were created
+        def fake_run(cmd, **kwargs):
+            # Find the output path (last arg)
+            output_path = cmd[-1]
+            Path(output_path).write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
+            return ("", "")
+
+        mock_run.side_effect = fake_run
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
+            tmp.write(b"fake video")
+            video_path = tmp.name
+
+        try:
+            results = extract_cover_candidates(video_path, num_frames=3)
+            assert len(results) == 3
+
+            # Check frame times: 20*0.25=5.0, 20*0.5=10.0, 20*0.75=15.0
+            assert results[0]["frame_time"] == 5.0
+            assert results[1]["frame_time"] == 10.0
+            assert results[2]["frame_time"] == 15.0
+
+            # Check local paths exist
+            for r in results:
+                assert Path(r["local_path"]).exists()
+
+            # Clean up
+            for r in results:
+                Path(r["local_path"]).unlink(missing_ok=True)
+        finally:
+            Path(video_path).unlink(missing_ok=True)
+
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0)
+    def test_handles_ffmpeg_failure_gracefully(self, mock_probe, mock_run):
+        """FFmpeg 失败时跳过该帧，继续抽取其他帧."""
+        import tempfile
+
+        from video_processing.thumbnail_generator import extract_cover_candidates
+
+        call_count = 0
+
+        def fake_run(cmd, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            output_path = cmd[-1]
+            if call_count == 2:
+                # Second frame fails - don't create file
+                raise RuntimeError("ffmpeg error")
+            Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
+            return ("", "")
+
+        mock_run.side_effect = fake_run
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
+            tmp.write(b"fake video")
+            video_path = tmp.name
+
+        try:
+            results = extract_cover_candidates(video_path, num_frames=3)
+            # Should get 2 frames (1st and 3rd), 2nd failed
+            assert len(results) == 2
+        finally:
+            Path(video_path).unlink(missing_ok=True)
+            for r in results:
+                Path(r["local_path"]).unlink(missing_ok=True)
+
+    @patch("video_processing.ffmpeg_utils.probe_duration", side_effect=Exception("probe failed"))
+    def test_fallback_duration_when_probe_fails(self, mock_probe):
+        """probe 失败时使用默认时长."""
+        import tempfile
+
+        from video_processing.thumbnail_generator import extract_cover_candidates
+
+        # Mock run_ffmpeg to create output files
+        def fake_run(cmd, **kwargs):
+            output_path = cmd[-1]
+            Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
+            return ("", "")
+
+        with patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=fake_run):
+            with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
+                tmp.write(b"fake")
+                video_path = tmp.name
+
+            try:
+                results = extract_cover_candidates(video_path, num_frames=3)
+                assert len(results) == 3
+                # Default duration is 5.0, so times should be 5*0.25=1.25, 5*0.5=2.5, 5*0.75=3.75
+                assert results[0]["frame_time"] == 1.25
+                assert results[1]["frame_time"] == 2.5
+                assert results[2]["frame_time"] == 3.75
+            finally:
+                Path(video_path).unlink(missing_ok=True)
+                for r in results:
+                    Path(r["local_path"]).unlink(missing_ok=True)
+
+
+class TestExtractAndUploadCoverFrames:
+    """extract_and_upload_cover_frames 测试."""
+
+    @patch("video_processing.oss_helpers.upload_to_oss")
+    @patch("video_processing.thumbnail_generator.extract_cover_candidates")
+    def test_uploads_and_returns_correct_format(self, mock_extract, mock_upload):
+        """上传帧到 OSS 并返回正确格式."""
+        import tempfile
+
+        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+
+        # Create actual temp files
+        tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+        tmp1.write(b"\xff\xd8" + b"\x00" * 50)
+        tmp1.close()
+        tmp2 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+        tmp2.write(b"\xff\xd8" + b"\x00" * 50)
+        tmp2.close()
+
+        mock_extract.return_value = [
+            {"local_path": tmp1.name, "frame_time": 5.0},
+            {"local_path": tmp2.name, "frame_time": 10.0},
+        ]
+        mock_upload.side_effect = [
+            "https://oss.example.com/covers/plan1/frame_0.jpg",
+            "https://oss.example.com/covers/plan1/frame_1.jpg",
+        ]
+
+        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
+
+        assert len(results) == 2
+        assert results[0]["image_url"] == "https://oss.example.com/covers/plan1/frame_0.jpg"
+        assert results[0]["frame_time"] == 5.0
+        assert results[0]["storage_key"] == "covers/plan1/frame_0.jpg"
+
+        assert results[1]["image_url"] == "https://oss.example.com/covers/plan1/frame_1.jpg"
+        assert results[1]["frame_time"] == 10.0
+
+    @patch("video_processing.thumbnail_generator.extract_cover_candidates", return_value=[])
+    def test_returns_empty_when_no_candidates(self, mock_extract):
+        """没有候选帧时返回空列表."""
+        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+
+        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
+        assert results == []
+
+    @patch("video_processing.oss_helpers.upload_to_oss", side_effect=Exception("OSS error"))
+    @patch("video_processing.thumbnail_generator.extract_cover_candidates")
+    def test_handles_upload_failure_gracefully(self, mock_extract, mock_upload):
+        """上传失败时跳过该帧."""
+        import tempfile
+
+        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+
+        tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+        tmp1.write(b"\xff\xd8" + b"\x00" * 50)
+        tmp1.close()
+
+        mock_extract.return_value = [
+            {"local_path": tmp1.name, "frame_time": 5.0},
+        ]
+
+        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
+        assert results == []
+
+
+class TestRenderAdapterResultCoverCandidates:
+    """RenderAdapterResult 的 cover_candidates 字段."""
+
+    def test_default_none(self):
+        """默认为 None."""
+        from video_processing.render_adapter import RenderAdapterResult
+
+        result = RenderAdapterResult(success=True)
+        assert result.cover_candidates is None
+
+    def test_can_set_candidates(self):
+        """可以设置候选帧列表."""
+        from video_processing.render_adapter import RenderAdapterResult
+
+        candidates = [
+            {"image_url": "https://example.com/frame_0.jpg", "frame_time": 5.0, "storage_key": "covers/p1/frame_0.jpg"},
+        ]
+        result = RenderAdapterResult(success=True, cover_candidates=candidates)
+        assert len(result.cover_candidates) == 1
+        assert result.cover_candidates[0]["frame_time"] == 5.0
+
+
+class TestAICoverServiceFFmpegFallback:
+    """AI 封面服务 FFmpeg 兜底测试."""
+
+    @patch("packages.shared.ai_service.http_requests.head")
+    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
+    def test_ffmpeg_fallback_success(self, mock_ffmpeg, mock_head):
+        """FFmpeg 兜底抽帧成功."""
+        import tempfile
+
+        mock_head.return_value.status_code = 200
+
+        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+        tmp.write(b"\xff\xd8" + b"\x00" * 50)
+        tmp.close()
+
+        mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 5.0}]
+
+        # Mock storage
+        with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn:
+            mock_storage = Mock()
+            mock_storage.upload_file = Mock()
+            mock_storage.get_url.return_value = "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg"
+            mock_storage_fn.return_value = mock_storage
+
+            from packages.shared.ai_service import _call_ai_cover_service
+
+            result = _call_ai_cover_service(
+                plan_id="plan1",
+                asset_ids=["a1"],
+                cover_type="ai_frame",
+                primary_video_url="https://example.com/video.mp4",
+            )
+
+            assert result["type"] == "ai_frame"
+            assert result["image_url"] == "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg"
+            assert result["frame_time"] == 5.0
+            assert result["confidence"] == 0.85
+
+        Path(tmp.name).unlink(missing_ok=True)
+
+    @patch("packages.shared.ai_service.http_requests.head")
+    def test_ffmpeg_no_video_url_raises(self, mock_head):
+        """没有视频 URL 时抛出 RuntimeError."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+            _call_ai_cover_service(
+                plan_id="plan1",
+                asset_ids=["a1"],
+                cover_type="ai_frame",
+                primary_video_url=None,
+            )
+
+    @patch("packages.shared.ai_service.http_requests.head")
+    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
+    def test_ffmpeg_no_frames_raises(self, mock_ffmpeg, mock_head):
+        """FFmpeg 抽帧为空时抛出 RuntimeError."""
+        mock_head.return_value.status_code = 200
+        mock_ffmpeg.return_value = []
+
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+            _call_ai_cover_service(
+                plan_id="plan1",
+                asset_ids=["a1"],
+                cover_type="ai_frame",
+                primary_video_url="https://example.com/video.mp4",
+            )
+
+    def test_upload_type_returns_immediately(self):
+        """upload 类型直接返回."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan1",
+            asset_ids=["a1"],
+            cover_type="upload",
+            primary_video_url="https://example.com/video.mp4",
+        )
+        assert result["type"] == "upload"
+
+    def test_manual_type_returns_immediately(self):
+        """manual 类型直接返回."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        result = _call_ai_cover_service(
+            plan_id="plan1",
+            asset_ids=["a1"],
+            cover_type="manual",
+            frame_time=5.0,
+            primary_video_url="https://example.com/video.mp4",
+        )
+        assert result["type"] == "manual"
+        assert result["frame_time"] == 5.0
+
+    @patch("packages.shared.ai_service.http_requests.head")
+    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
+    def test_video_url_unreachable_raises(self, mock_ffmpeg, mock_head):
+        """视频 URL 不可访问时抛出 RuntimeError."""
+        mock_head.return_value.status_code = 404
+
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
+            _call_ai_cover_service(
+                plan_id="plan1",
+                asset_ids=["a1"],
+                cover_type="ai_frame",
+                primary_video_url="https://example.com/video.mp4",
+            )
+
+
+class TestCoverTemplatesFix:
+    """CoverTemplateResponse config=None 修复测试."""
+
+    def test_config_none_becomes_empty_dict(self):
+        """config=None 时 CoverTemplateResponse 不报 ValidationError."""
+        from datetime import datetime
+
+        from app.schemas.cover_template import CoverTemplateResponse
+
+        # This should not raise
+        resp = CoverTemplateResponse(
+            id="1",
+            name="test",
+            thumbnail_url="",
+            is_system=True,
+            created_at=datetime.now(),
+            config={},
+        )
+        assert resp.config == {}
+
+
+class TestExtractFramesWithFFmpeg:
+    """_extract_frames_with_ffmpeg 单元测试."""
+
+    def test_extracts_frames_with_correct_seek_times(self):
+        """抽帧时间点正确计算."""
+        import subprocess
+        import tempfile
+
+        # Mock ffprobe to return duration
+        mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="20.0\n", stderr="")
+        with patch("subprocess.run", return_value=mock_probe_result) as mock_subproc:
+            # First call is ffprobe, rest are ffmpeg
+            call_count = 0
+
+            def side_effect(cmd, **kwargs):
+                nonlocal call_count
+                call_count += 1
+                if call_count == 1:
+                    # ffprobe call
+                    return mock_probe_result
+                else:
+                    # ffmpeg call - create output file
+                    output_path = cmd[-1]
+                    from pathlib import Path
+
+                    Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
+                    return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+
+            mock_subproc.side_effect = side_effect
+
+            from packages.shared.ai_service import _extract_frames_with_ffmpeg
+
+            results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3)
+
+            assert len(results) == 3
+            # 20 * 0.25 = 5.0, 20 * 0.5 = 10.0, 20 * 0.75 = 15.0
+            assert results[0]["frame_time"] == 5.0
+            assert results[1]["frame_time"] == 10.0
+            assert results[2]["frame_time"] == 15.0
+
+            # Clean up
+            for r in results:
+                from pathlib import Path
+
+                Path(r["local_path"]).unlink(missing_ok=True)
+
+    def test_handles_ffmpeg_failure(self):
+        """FFmpeg 失败时跳过该帧."""
+        import subprocess
+        import tempfile
+        from pathlib import Path
+
+        mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="10.0\n", stderr="")
+
+        call_count = 0
+
+        def side_effect(cmd, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                return mock_probe_result
+            output_path = cmd[-1]
+            if call_count == 2:
+                # First frame succeeds
+                Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
+                return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+            else:
+                # Other frames fail
+                raise subprocess.CalledProcessError(1, cmd)
+
+        with patch("subprocess.run", side_effect=side_effect):
+            from packages.shared.ai_service import _extract_frames_with_ffmpeg
+
+            results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3)
+            assert len(results) == 1
+            Path(results[0]["local_path"]).unlink(missing_ok=True)
+
+
+class TestGenerationCoverPreStored:
+    """generation_cover.py 预存帧逻辑测试."""
+
+    def test_pre_stored_candidates_used_when_available(self):
+        """有预存帧时直接使用，不调用 AI 服务."""
+        from unittest.mock import patch
+
+        # Mock the dependencies
+        mock_plan = MagicMock()
+        mock_plan.config = {
+            "cover_candidates": [
+                {
+                    "image_url": "https://oss.example.com/covers/p1/frame_0.jpg",
+                    "frame_time": 5.0,
+                    "storage_key": "covers/p1/frame_0.jpg",
+                },
+                {
+                    "image_url": "https://oss.example.com/covers/p1/frame_1.jpg",
+                    "frame_time": 10.0,
+                    "storage_key": "covers/p1/frame_1.jpg",
+                },
+            ],
+            "rendered_storage_key": "rendered/p1/video.mp4",
+        }
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+        mock_body = MagicMock()
+        mock_body.asset_ids = ["a1"]
+        mock_body.cover_type = "ai_frame"
+        mock_body.frame_time = None
+
+        with (
+            patch("app.api.routes.generation_cover.get_editor_services") as mock_services,
+            patch("app.api.routes.generation_cover.get_db_session"),
+            patch("app.api.routes.generation_cover.get_current_user"),
+            patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="p1"),
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+
+            mock_services.return_value = (MagicMock(), mock_plan_svc)
+            mock_normalize.side_effect = lambda c: c
+
+            from app.api.routes.generation_cover import GenerateCoverRequest, generate_cover
+
+            result = generate_cover(
+                body=mock_body,
+                template_id="t1",
+                plan_id="p1",
+                services=(MagicMock(), mock_plan_svc),
+                db=MagicMock(),
+                current_user=MagicMock(),
+            )
+
+            assert result.plan_id == "p1"
+            assert result.cover["type"] == "ai_frame"
+            assert result.cover["image_url"] == "https://oss.example.com/covers/p1/frame_0.jpg"
+            assert result.cover["frame_time"] == 5.0

tests/unit/test_cover_generator_pure.py

@@ -1,538 +0,0 @@
-"""CoverGenerator 纯逻辑单测 — 时间钳制 + 智能选帧算法.
-
-通过 mock run_ffmpeg 和 probe_video_info 验证纯逻辑部分，
-不实际执行 FFmpeg，确保测试轻量快速。
-"""
-
-from __future__ import annotations
-
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-from video_processing.cover_generator import (
-    DEFAULT_COVER_HEIGHT,
-    DEFAULT_COVER_QUALITY,
-    DEFAULT_COVER_TIME,
-    DEFAULT_COVER_WIDTH,
-    SMART_COVER_FRAME_COUNT,
-    CoverGenerator,
-)
-
-
-class TestCoverGeneratorConstants:
-    """常量默认值测试."""
-
-    def test_default_cover_time(self):
-        """默认抽帧时间为 1.0 秒."""
-        assert DEFAULT_COVER_TIME == 1.0
-
-    def test_default_dimensions(self):
-        """默认封面尺寸 1080x1920 (竖屏)."""
-        assert DEFAULT_COVER_WIDTH == 1080
-        assert DEFAULT_COVER_HEIGHT == 1920
-
-    def test_default_quality(self):
-        """默认质量为 5 (JPEG q:v, 越小越好)."""
-        assert DEFAULT_COVER_QUALITY == 5
-
-    def test_smart_cover_frame_count(self):
-        """智能封面默认抽 3 帧."""
-        assert SMART_COVER_FRAME_COUNT == 3
-
-
-class TestExtractFrameCommand:
-    """extract_frame 命令构建测试."""
-
-    def _probe_video_info_mock(self, duration=10.0):
-        """创建 probe_video_info 的 mock."""
-        return {"duration": duration, "width": 1920, "height": 1080, "fps": 25.0}
-
-    def test_default_params_command(self, tmp_path):
-        """默认参数下 FFmpeg 命令正确."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-            # 让 output_path 在 run_ffmpeg 后存在
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            result = CoverGenerator.extract_frame(str(video_file), str(output_file))
-
-        assert result == Path(output_file)
-        mock_run.assert_called_once()
-        cmd = mock_run.call_args[0][0]
-
-        # 基本结构
-        assert cmd[0].endswith("ffmpeg") or "ffmpeg" in cmd[0]
-        assert "-y" in cmd
-        assert "-vframes" in cmd
-        assert cmd[cmd.index("-vframes") + 1] == "1"
-        assert "-f" in cmd
-        assert "mjpeg" in cmd[cmd.index("-f") + 1]
-
-        # 时间点
-        ss_idx = cmd.index("-ss")
-        assert float(cmd[ss_idx + 1]) == pytest.approx(DEFAULT_COVER_TIME, abs=0.001)
-
-        # 输入文件
-        i_idx = cmd.index("-i")
-        assert cmd[i_idx + 1] == str(video_file)
-
-        # 输出文件
-        assert cmd[-1] == str(output_file)
-
-        # scale + crop 滤镜
-        vf_idx = cmd.index("-vf")
-        vf_value = cmd[vf_idx + 1]
-        assert "scale=" in vf_value
-        assert "crop=" in vf_value
-        assert "force_original_aspect_ratio=increase" in vf_value
-
-    def test_custom_time(self, tmp_path):
-        """自定义抽帧时间点."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(duration=30.0),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=5.5)
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        assert float(cmd[ss_idx + 1]) == pytest.approx(5.5, abs=0.001)
-
-    def test_custom_dimensions(self, tmp_path):
-        """自定义输出尺寸."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_frame(str(video_file), str(output_file), width=1920, height=1080)
-
-        cmd = mock_run.call_args[0][0]
-        vf_idx = cmd.index("-vf")
-        vf_value = cmd[vf_idx + 1]
-        assert "scale=1920:1080:" in vf_value
-        assert "crop=1920:1080" in vf_value
-
-    def test_custom_quality(self, tmp_path):
-        """自定义 JPEG 质量."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_frame(str(video_file), str(output_file), quality=2)
-
-        cmd = mock_run.call_args[0][0]
-        q_idx = cmd.index("-q:v")
-        assert cmd[q_idx + 1] == "2"
-
-    def test_time_exceeds_duration_clamps_to_midpoint(self, tmp_path):
-        """抽帧时间超过视频时长时，钳制到中间帧."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(duration=5.0),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=10.0)
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        # 钳制到 duration/2 = 2.5
-        assert float(cmd[ss_idx + 1]) == pytest.approx(2.5, abs=0.001)
-
-    def test_negative_time_clamps_to_zero(self, tmp_path):
-        """负时间钳制到 0."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(duration=10.0),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=-2.0)
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        assert float(cmd[ss_idx + 1]) == pytest.approx(0.0, abs=0.001)
-
-    def test_time_equals_duration_clamps_to_midpoint(self, tmp_path):
-        """时间点等于时长时钳制到中间帧."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(duration=10.0),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=10.0)
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        assert float(cmd[ss_idx + 1]) == pytest.approx(5.0, abs=0.001)
-
-    def test_zero_duration_video(self, tmp_path):
-        """视频时长为 0 时的行为（不钳制，用原始时间）."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(duration=0.0),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=0.5)
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        assert float(cmd[ss_idx + 1]) == pytest.approx(0.5, abs=0.001)
-
-    def test_video_not_found_raises(self, tmp_path):
-        """视频文件不存在时抛出 FileNotFoundError."""
-        output_file = tmp_path / "cover.jpg"
-
-        with pytest.raises(FileNotFoundError):
-            CoverGenerator.extract_frame(str(tmp_path / "nonexistent.mp4"), str(output_file))
-
-    def test_output_creates_parent_dir(self, tmp_path):
-        """输出目录不存在时自动创建."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        out_dir = tmp_path / "deep" / "nested"
-        output_file = out_dir / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(),
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_frame(str(video_file), str(output_file))
-
-        assert out_dir.exists()
-        assert out_dir.is_dir()
-
-    def test_ffmpeg_failure_propagates(self, tmp_path):
-        """FFmpeg 失败时异常向上传递."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value=self._probe_video_info_mock(),
-            ),
-            patch(
-                "video_processing.cover_generator.run_ffmpeg",
-                side_effect=RuntimeError("FFmpeg error"),
-            ),
-        ):
-            with pytest.raises(RuntimeError, match="FFmpeg error"):
-                CoverGenerator.extract_frame(str(video_file), str(output_file))
-
-
-class TestSmartCoverTimePoints:
-    """智能封面时间点计算测试."""
-
-    def test_single_frame_falls_back_to_default(self, tmp_path):
-        """只有 1 帧时退化为普通抽帧（取 DEFAULT_COVER_TIME 和 midpoint 中较小值）."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value={"duration": 20.0},
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            # frame_count=1 时退化为普通抽帧
-            CoverGenerator.extract_smart_cover(str(video_file), str(output_file), frame_count=1)
-
-        # 只调用一次（退化路径）
-        assert mock_run.call_count == 1
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        # min(DEFAULT_COVER_TIME=1.0, duration/2=10.0) = 1.0
-        assert float(cmd[ss_idx + 1]) == pytest.approx(1.0, abs=0.001)
-
-    def test_zero_duration_falls_back(self, tmp_path):
-        """视频时长为 0 时退化为普通抽帧."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value={"duration": 0.0},
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
-
-        # 只调用一次（退化路径）
-        assert mock_run.call_count == 1
-
-    def test_three_frames_uniform_distribution(self, tmp_path):
-        """3 帧均匀分布在 5%~95% 区间."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        call_times = []
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value={"duration": 100.0},
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                # 记录抽帧时间
-                ss_idx = cmd.index("-ss")
-                call_times.append(float(cmd[ss_idx + 1]))
-                # 在输出路径写文件
-                output_arg = cmd[-1]
-                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
-                # 不同文件大小，让第三帧"最清晰"
-                idx = len(call_times) - 1
-                size = 1000 * (idx + 1)  # 递增的文件大小
-                Path(output_arg).write_bytes(b"x" * size)
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
-
-        # 3 帧：5%、50%、95%
-        assert len(call_times) == 3
-        assert call_times[0] == pytest.approx(5.0, abs=0.1)  # 5%
-        assert call_times[1] == pytest.approx(50.0, abs=0.1)  # 50%
-        assert call_times[2] == pytest.approx(95.0, abs=0.1)  # 95%
-
-    def test_five_frames_distribution(self, tmp_path):
-        """5 帧均匀分布."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        call_times = []
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value={"duration": 100.0},
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-
-            def fake_run(cmd):
-                ss_idx = cmd.index("-ss")
-                call_times.append(float(cmd[ss_idx + 1]))
-                output_arg = cmd[-1]
-                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
-                idx = len(call_times) - 1
-                Path(output_arg).write_bytes(b"x" * (1000 * (idx + 1)))
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.extract_smart_cover(str(video_file), str(output_file), frame_count=5)
-
-        assert len(call_times) == 5
-        # step = (95-5) / (5-1) = 22.5
-        # times: 5, 27.5, 50, 72.5, 95
-        assert call_times[0] == pytest.approx(5.0, abs=0.1)
-        assert call_times[1] == pytest.approx(27.5, abs=0.1)
-        assert call_times[2] == pytest.approx(50.0, abs=0.1)
-        assert call_times[3] == pytest.approx(72.5, abs=0.1)
-        assert call_times[4] == pytest.approx(95.0, abs=0.1)
-
-    def test_selects_largest_file_as_best(self, tmp_path):
-        """选择文件最大的帧作为最佳封面（清晰度近似）."""
-        video_file = tmp_path / "test.mp4"
-        video_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        sizes = [5000, 15000, 8000]  # 第二帧最大
-
-        with (
-            patch(
-                "video_processing.cover_generator.probe_video_info",
-                return_value={"duration": 100.0},
-            ),
-            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
-        ):
-            call_idx = [0]
-
-            def fake_run(cmd):
-                output_arg = cmd[-1]
-                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
-                idx = call_idx[0]
-                Path(output_arg).write_bytes(b"x" * sizes[idx])
-                call_idx[0] += 1
-
-            mock_run.side_effect = fake_run
-            result = CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
-
-        # 第二帧（索引1）应该是最佳
-        assert result == output_file
-        # 输出文件大小应等于第二帧大小
-        assert output_file.stat().st_size == 15000
-
-
-class TestProcessCustomCover:
-    """自定义封面处理测试."""
-
-    def test_custom_cover_resize_command(self, tmp_path):
-        """自定义封面调整尺寸命令正确."""
-        input_file = tmp_path / "upload.jpg"
-        input_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with patch("video_processing.cover_generator.run_ffmpeg") as mock_run:
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.process_custom_cover(str(input_file), str(output_file))
-
-        mock_run.assert_called_once()
-        cmd = mock_run.call_args[0][0]
-
-        assert "-i" in cmd
-        assert cmd[cmd.index("-i") + 1] == str(input_file)
-        assert cmd[-1] == str(output_file)
-
-        # scale + crop
-        vf_idx = cmd.index("-vf")
-        vf_value = cmd[vf_idx + 1]
-        assert "scale=" in vf_value
-        assert "crop=" in vf_value
-
-    def test_custom_cover_not_found_raises(self, tmp_path):
-        """自定义封面文件不存在时抛出 FileNotFoundError."""
-        output_file = tmp_path / "cover.jpg"
-
-        with pytest.raises(FileNotFoundError):
-            CoverGenerator.process_custom_cover(str(tmp_path / "nonexistent.jpg"), str(output_file))
-
-    def test_custom_cover_custom_dimensions(self, tmp_path):
-        """自定义封面自定义输出尺寸."""
-        input_file = tmp_path / "upload.jpg"
-        input_file.write_bytes(b"fake")
-        output_file = tmp_path / "cover.jpg"
-
-        with patch("video_processing.cover_generator.run_ffmpeg") as mock_run:
-
-            def fake_run(cmd):
-                output_file.write_bytes(b"fake jpg")
-
-            mock_run.side_effect = fake_run
-            CoverGenerator.process_custom_cover(str(input_file), str(output_file), width=800, height=600)
-
-        cmd = mock_run.call_args[0][0]
-        vf_idx = cmd.index("-vf")
-        vf_value = cmd[vf_idx + 1]
-        assert "scale=800:600:" in vf_value
-        assert "crop=800:600" in vf_value

tests/unit/test_cover_reverse_sticker.py

@@ -1,860 +0,0 @@
-"""封面生成 + 视频倒放 + 贴纸叠加 单元测试.
-
-覆盖三个新渲染能力的核心场景和降级逻辑。
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Any
-from unittest.mock import MagicMock, patch
-
-import pytest
-from video_processing.cover_generator import (
-    DEFAULT_COVER_HEIGHT,
-    DEFAULT_COVER_WIDTH,
-    CoverGenerator,
-    generate_cover_from_plan,
-)
-from video_processing.reverse_engine import ReverseConfig, ReverseEngine
-from video_processing.sticker_engine import (
-    POSITION_PRESETS,
-    STICKER_CATEGORIES,
-    ImageStickerConfig,
-    StickerEngine,
-    TextStickerConfig,
-    get_sticker_categories,
-    parse_stickers_from_config,
-)
-from video_processing.unified_render_service import (
-    ResolvedClip,
-    UnifiedRenderService,
-)
-
-# ── Fixtures ──────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class FakePlan:
-    """模拟 EditPlan."""
-
-    id: str = "plan_001"
-    name: str = "测试计划"
-    config: dict[str, Any] = field(default_factory=dict)
-
-
-@pytest.fixture
-def sample_video(tmp_path):
-    """创建一个测试视频文件（空文件，仅用于路径测试）."""
-    video_path = tmp_path / "test_video.mp4"
-    video_path.write_bytes(b"fake video data")
-    return video_path
-
-
-@pytest.fixture
-def sample_image(tmp_path):
-    """创建一个测试图片文件."""
-    img_path = tmp_path / "sticker.png"
-    img_path.write_bytes(b"fake png data")
-    return img_path
-
-
-# ═══════════════════════════════════════════════════════════════════════════════
-# 一、视频倒放引擎测试
-# ═══════════════════════════════════════════════════════════════════════════════
-
-
-class TestReverseConfig:
-    """ReverseConfig 配置解析测试."""
-
-    def test_default_disabled(self):
-        """默认配置为关闭."""
-        config = ReverseConfig.from_dict(None)
-        assert config.enabled is False
-        assert config.reverse_video is True
-        assert config.reverse_audio is True
-
-    def test_empty_dict(self):
-        """空字典视为关闭."""
-        config = ReverseConfig.from_dict({})
-        assert config.enabled is False
-
-    def test_enabled(self):
-        """启用倒放."""
-        config = ReverseConfig.from_dict({"enabled": True})
-        assert config.enabled is True
-        assert config.reverse_video is True
-        assert config.reverse_audio is True
-
-    def test_video_only(self):
-        """只倒放视频."""
-        config = ReverseConfig.from_dict(
-            {
-                "enabled": True,
-                "reverse_video": True,
-                "reverse_audio": False,
-            }
-        )
-        assert config.enabled is True
-        assert config.reverse_video is True
-        assert config.reverse_audio is False
-
-    def test_audio_only(self):
-        """只倒放音频."""
-        config = ReverseConfig.from_dict(
-            {
-                "enabled": True,
-                "reverse_video": False,
-                "reverse_audio": True,
-            }
-        )
-        assert config.reverse_video is False
-        assert config.reverse_audio is True
-
-    def test_invalid_config_fallback(self):
-        """无效配置降级为默认."""
-        config = ReverseConfig.from_dict("invalid")  # type: ignore
-        assert config.enabled is False
-
-    def test_none_config(self):
-        """None 配置."""
-        config = ReverseConfig.from_dict(None)
-        assert config.enabled is False
-
-
-class TestReverseEngine:
-    """ReverseEngine 滤镜生成测试."""
-
-    def test_video_reverse_filter(self):
-        """视频倒放滤镜生成."""
-        config = ReverseConfig(enabled=True, reverse_video=True)
-        f = ReverseEngine.build_video_filter(config, duration=10.0)
-        assert f == "reverse"
-
-    def test_video_disabled(self):
-        """视频倒放关闭时返回空."""
-        config = ReverseConfig(enabled=False)
-        f = ReverseEngine.build_video_filter(config, duration=10.0)
-        assert f == ""
-
-    def test_video_disabled_flag(self):
-        """启用但 reverse_video=False."""
-        config = ReverseConfig(enabled=True, reverse_video=False)
-        f = ReverseEngine.build_video_filter(config, duration=10.0)
-        assert f == ""
-
-    def test_audio_reverse_filter(self):
-        """音频倒放滤镜生成."""
-        config = ReverseConfig(enabled=True, reverse_audio=True)
-        f = ReverseEngine.build_audio_filter(config, duration=10.0)
-        assert f == "areverse"
-
-    def test_audio_disabled(self):
-        """音频倒放关闭."""
-        config = ReverseConfig(enabled=False)
-        f = ReverseEngine.build_audio_filter(config, duration=10.0)
-        assert f == ""
-
-    def test_long_video_safety_limit(self):
-        """超长视频安全限制：跳过倒放."""
-        config = ReverseConfig(enabled=True)
-        f = ReverseEngine.build_video_filter(config, duration=200.0)
-        assert f == ""  # 超过 MAX_SAFE_DURATION
-
-    def test_long_audio_safety_limit(self):
-        """超长音频安全限制."""
-        config = ReverseConfig(enabled=True)
-        f = ReverseEngine.build_audio_filter(config, duration=200.0)
-        assert f == ""
-
-    def test_duration_zero(self):
-        """时长为0时正常返回."""
-        config = ReverseConfig(enabled=True)
-        f = ReverseEngine.build_video_filter(config, duration=0.0)
-        assert f == "reverse"
-
-
-# ═══════════════════════════════════════════════════════════════════════════════
-# 二、贴纸引擎测试
-# ═══════════════════════════════════════════════════════════════════════════════
-
-
-class TestStickerPosition:
-    """贴纸位置计算测试."""
-
-    def test_presets_exist(self):
-        """9宫格预设存在."""
-        assert "top_left" in POSITION_PRESETS
-        assert "center" in POSITION_PRESETS
-        assert "bottom_right" in POSITION_PRESETS
-        assert len(POSITION_PRESETS) == 9
-
-    def test_resolve_position_center(self):
-        """居中位置计算."""
-        sticker = ImageStickerConfig(position="center")
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 200, 200)
-        assert abs(x - 400) < 1  # (1000-200)/2 = 400
-        assert abs(y - 400) < 1
-
-    def test_resolve_position_top_left(self):
-        """左上角位置."""
-        sticker = ImageStickerConfig(position="top_left")
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100)
-        assert x == 0  # 0.05*1000 - 50 = 0 (clamped)
-        assert y == 0
-
-    def test_custom_position_percent(self):
-        """自定义百分比位置."""
-        sticker = ImageStickerConfig(
-            position="center",
-            x=30.0,
-            y=70.0,
-            x_unit="percent",
-            y_unit="percent",
-        )
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100)
-        assert abs(x - 250) < 1  # 300 - 50 = 250
-        assert abs(y - 650) < 1  # 700 - 50 = 650
-
-    def test_custom_position_pixel(self):
-        """自定义像素位置."""
-        sticker = ImageStickerConfig(
-            position="center",
-            x=100.0,
-            y=200.0,
-            x_unit="pixel",
-            y_unit="pixel",
-        )
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50)
-        assert abs(x - 75) < 1  # 100 - 25 = 75
-        assert abs(y - 175) < 1  # 200 - 25 = 175
-
-    def test_position_clamped(self):
-        """位置钳制在画布内."""
-        sticker = ImageStickerConfig(
-            position="center",
-            x=-10.0,
-            y=-10.0,
-            x_unit="pixel",
-            y_unit="pixel",
-        )
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50)
-        assert x >= 0
-        assert y >= 0
-
-
-class TestTextSticker:
-    """文字贴纸测试."""
-
-    def test_drawtext_filter_basic(self):
-        """基础文字贴纸滤镜生成."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Hello World",
-            font_size=36,
-            font_color="#FFFFFF",
-            position="center",
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "drawtext" in f
-        assert "Hello World" in f
-        assert "fontsize=36" in f
-        assert "[in]" in f
-        assert "[out]" in f
-
-    def test_drawtext_with_stroke(self):
-        """带描边的文字贴纸."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Test",
-            stroke_width=3,
-            stroke_color="#FF0000",
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "borderw=3" in f
-        assert "bordercolor=#FF0000" in f
-
-    def test_drawtext_with_shadow(self):
-        """带阴影的文字贴纸."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Shadow",
-            shadow_x=4,
-            shadow_y=4,
-            shadow_alpha=0.5,
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "shadowx=4" in f
-        assert "shadowy=4" in f
-
-    def test_drawtext_time_range(self):
-        """带时间范围的文字贴纸."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Timed",
-            start_time=2.0,
-            duration=3.0,
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "enable='between(t,2.0,5.0)'" in f
-
-    def test_drawtext_empty_text(self):
-        """空文字直通."""
-        sticker = TextStickerConfig(enabled=True, text="")
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "[in]copy[out]" in f
-
-    def test_drawtext_with_fade(self):
-        """带淡入淡出的文字贴纸."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Fade",
-            start_time=1.0,
-            duration=5.0,
-            fade_in=0.5,
-            fade_out=0.5,
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "alpha=" in f
-
-
-class TestImageSticker:
-    """图片贴纸测试."""
-
-    def test_image_sticker_overlay(self, sample_image):
-        """图片贴纸 overlay 滤镜生成."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[
-                {
-                    "type": "image",
-                    "image_path": str(sample_image),
-                    "position": "top_right",
-                    "scale": 0.5,
-                    "opacity": 0.8,
-                    "z_index": 10,
-                }
-            ],
-            input_label="[base]",
-            output_label="[final]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        assert result.filter_str != ""
-        assert "overlay" in result.filter_str
-        assert len(result.extra_inputs) == 1
-        assert result.extra_inputs[0] == str(sample_image)
-
-    def test_image_sticker_missing_file(self):
-        """图片贴纸素材不存在时跳过."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[
-                {
-                    "type": "image",
-                    "image_path": "/nonexistent/image.png",
-                    "position": "center",
-                }
-            ],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        # 素材不存在，跳过，返回直通
-        assert "[in]copy[out]" in result.filter_str
-        assert len(result.extra_inputs) == 0
-
-    def test_mixed_stickers(self, sample_image):
-        """混合贴纸：图片 + 文字."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[
-                {
-                    "type": "image",
-                    "image_path": str(sample_image),
-                    "position": "top_left",
-                    "z_index": 5,
-                },
-                {
-                    "type": "text",
-                    "text": "Hello",
-                    "position": "bottom_center",
-                    "z_index": 10,
-                },
-            ],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        assert "overlay" in result.filter_str
-        assert "drawtext" in result.filter_str
-        assert len(result.extra_inputs) == 1
-
-    def test_sticker_z_index_order(self, sample_image):
-        """贴纸按 z_index 排序."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[
-                {"type": "text", "text": "Top", "z_index": 20, "position": "center"},
-                {"type": "text", "text": "Bottom", "z_index": 5, "position": "center"},
-            ],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        # z_index 小的先叠加，大的后叠加（在上面）
-        assert result.filter_str.count("drawtext") == 2
-
-    def test_empty_stickers(self):
-        """空贴纸列表."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        assert "[in]copy[out]" in result.filter_str
-        assert result.extra_inputs == []
-
-    def test_invalid_sticker_skipped(self):
-        """无效贴纸配置跳过."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[{"invalid": "data"}],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        # 解析失败，跳过，直通
-        assert "[in]copy[out]" in result.filter_str
-
-
-class TestStickerHelpers:
-    """贴纸辅助函数测试."""
-
-    def test_parse_stickers_empty(self):
-        """空配置解析."""
-        assert parse_stickers_from_config(None) == []
-        assert parse_stickers_from_config({}) == []
-
-    def test_parse_stickers_list(self):
-        """正常贴纸列表解析."""
-        config = {"stickers": [{"type": "text", "text": "A"}, {"type": "text", "text": "B"}]}
-        result = parse_stickers_from_config(config)
-        assert len(result) == 2
-
-    def test_parse_stickers_not_list(self):
-        """非列表类型返回空."""
-        config = {"stickers": "not a list"}
-        assert parse_stickers_from_config(config) == []
-
-    def test_get_categories(self):
-        """贴纸分类列表."""
-        cats = get_sticker_categories()
-        assert len(cats) == len(STICKER_CATEGORIES)
-        assert cats[0][0] == "emoji"
-
-
-# ═══════════════════════════════════════════════════════════════════════════════
-# 三、封面生成器测试
-# ═══════════════════════════════════════════════════════════════════════════════
-
-
-class TestCoverGenerator:
-    """CoverGenerator 测试."""
-
-    def test_default_dimensions(self):
-        """默认封面尺寸."""
-        assert DEFAULT_COVER_WIDTH == 1080
-        assert DEFAULT_COVER_HEIGHT == 1920
-
-    @patch("video_processing.cover_generator.run_ffmpeg")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_extract_frame_basic(self, mock_probe, mock_run, sample_video, tmp_path):
-        """基础抽帧测试."""
-        mock_probe.return_value = {"duration": 30.0}
-
-        # mock run_ffmpeg 实际创建输出文件
-        def fake_run_ffmpeg(cmd):
-            # 找到输出路径并创建文件
-            output_path = Path(cmd[-1])
-            output_path.parent.mkdir(parents=True, exist_ok=True)
-            output_path.write_bytes(b"fake jpeg data")
-
-        mock_run.side_effect = fake_run_ffmpeg
-
-        output = tmp_path / "cover.jpg"
-        result = CoverGenerator.extract_frame(
-            sample_video,
-            output,
-            time_sec=2.0,
-        )
-
-        assert result == output
-        mock_run.assert_called_once()
-        # 检查命令参数
-        cmd = mock_run.call_args[0][0]
-        assert "-ss" in cmd
-        assert "2.000" in cmd
-        assert "-vframes" in cmd
-        assert "1" in cmd
-
-    @patch("video_processing.cover_generator.run_ffmpeg")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_extract_frame_time_clamped(self, mock_probe, mock_run, sample_video, tmp_path):
-        """抽帧时间超过视频长度时钳制."""
-        mock_probe.return_value = {"duration": 10.0}
-
-        def fake_run_ffmpeg(cmd):
-            output_path = Path(cmd[-1])
-            output_path.parent.mkdir(parents=True, exist_ok=True)
-            output_path.write_bytes(b"fake jpeg data")
-
-        mock_run.side_effect = fake_run_ffmpeg
-
-        output = tmp_path / "cover.jpg"
-        CoverGenerator.extract_frame(
-            sample_video,
-            output,
-            time_sec=100.0,  # 超过视频时长
-        )
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        time_val = float(cmd[ss_idx + 1])
-        # 应该被钳制到中间帧（5秒左右）
-        assert time_val <= 10.0
-
-    @patch("video_processing.cover_generator.run_ffmpeg")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_extract_frame_negative_time(self, mock_probe, mock_run, sample_video, tmp_path):
-        """负时间钳制到0."""
-        mock_probe.return_value = {"duration": 30.0}
-
-        def fake_run_ffmpeg(cmd):
-            output_path = Path(cmd[-1])
-            output_path.parent.mkdir(parents=True, exist_ok=True)
-            output_path.write_bytes(b"fake jpeg data")
-
-        mock_run.side_effect = fake_run_ffmpeg
-
-        output = tmp_path / "cover.jpg"
-        CoverGenerator.extract_frame(
-            sample_video,
-            output,
-            time_sec=-5.0,
-        )
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        time_val = float(cmd[ss_idx + 1])
-        assert time_val >= 0
-
-    def test_extract_frame_file_not_found(self, tmp_path):
-        """视频文件不存在抛异常."""
-        with pytest.raises(FileNotFoundError):
-            CoverGenerator.extract_frame(
-                "/nonexistent/video.mp4",
-                tmp_path / "cover.jpg",
-            )
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_frame")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_smart_cover_3_frames(self, mock_probe, mock_extract, sample_video, tmp_path):
-        """智能封面抽取3帧选最佳."""
-        mock_probe.return_value = {"duration": 30.0}
-
-        # 创建三个大小不同的临时文件（模拟清晰度不同）
-        def create_frame(video_path, output_path, **kwargs):
-            # 第二帧最大（最清晰）
-            p = Path(output_path)
-            p.parent.mkdir(parents=True, exist_ok=True)
-            if "candidate_1" in str(p):
-                p.write_bytes(b"x" * 10000)  # 最大 = 最清晰
-            elif "candidate_0" in str(p):
-                p.write_bytes(b"x" * 1000)
-            else:
-                p.write_bytes(b"x" * 5000)
-            return p
-
-        mock_extract.side_effect = create_frame
-
-        output = tmp_path / "smart_cover.jpg"
-        result = CoverGenerator.extract_smart_cover(
-            sample_video,
-            output,
-            frame_count=3,
-        )
-
-        assert result == output
-        assert output.exists()
-        # 应该选最大的那个文件（candidate_1）
-        assert output.stat().st_size == 10000
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_frame")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_smart_cover_fallback(self, mock_probe, mock_extract, sample_video, tmp_path):
-        """智能封面全部失败时降级."""
-        mock_probe.return_value = {"duration": 0.0}  # 时长为0
-
-        output = tmp_path / "cover.jpg"
-        output.write_bytes(b"x" * 100)
-        mock_extract.return_value = output
-
-        result = CoverGenerator.extract_smart_cover(sample_video, output, frame_count=3)
-        assert result == output
-
-    @patch("video_processing.cover_generator.run_ffmpeg")
-    def test_custom_cover(self, mock_run, sample_image, tmp_path):
-        """自定义封面处理."""
-        output = tmp_path / "custom_cover.jpg"
-
-        result = CoverGenerator.process_custom_cover(
-            sample_image,
-            output,
-        )
-
-        assert result == output
-        mock_run.assert_called_once()
-        cmd = mock_run.call_args[0][0]
-        assert str(sample_image) in cmd
-
-    def test_custom_cover_not_found(self, tmp_path):
-        """自定义封面文件不存在."""
-        with pytest.raises(FileNotFoundError):
-            CoverGenerator.process_custom_cover(
-                "/nonexistent/img.png",
-                tmp_path / "cover.jpg",
-            )
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_frame")
-    def test_generate_cover_time_mode(self, mock_extract, sample_video, tmp_path):
-        """统一入口 - time 模式."""
-        output = tmp_path / "cover.jpg"
-        mock_extract.return_value = output
-
-        result = CoverGenerator.generate_cover(
-            sample_video,
-            output,
-            mode="time",
-            time_sec=3.0,
-        )
-
-        assert result == output
-        mock_extract.assert_called_once()
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover")
-    def test_generate_cover_smart_mode(self, mock_smart, sample_video, tmp_path):
-        """统一入口 - smart 模式."""
-        output = tmp_path / "cover.jpg"
-        mock_smart.return_value = output
-
-        result = CoverGenerator.generate_cover(
-            sample_video,
-            output,
-            mode="smart",
-        )
-
-        assert result == output
-        mock_smart.assert_called_once()
-
-    @patch("video_processing.cover_generator.CoverGenerator.process_custom_cover")
-    def test_generate_cover_custom_mode(self, mock_custom, sample_video, sample_image, tmp_path):
-        """统一入口 - custom 模式."""
-        output = tmp_path / "cover.jpg"
-        mock_custom.return_value = output
-
-        result = CoverGenerator.generate_cover(
-            sample_video,
-            output,
-            mode="custom",
-            custom_image=sample_image,
-        )
-
-        assert result == output
-        mock_custom.assert_called_once()
-
-
-class TestGenerateCoverFromPlan:
-    """从 plan 配置生成封面测试."""
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover")
-    def test_smart_mode_from_plan(self, mock_smart, sample_video, tmp_path):
-        """plan 配置 smart 模式."""
-        plan = FakePlan(id="plan_001", config={"cover_config": {"mode": "smart"}})
-        mock_smart.return_value = tmp_path / "cover.jpg"
-        (tmp_path / "cover.jpg").write_bytes(b"test")
-
-        result = generate_cover_from_plan(plan, sample_video, tmp_path)
-        assert result is not None
-
-    def test_no_cover_config(self, sample_video, tmp_path):
-        """没有封面配置时返回 None."""
-        plan = FakePlan(id="plan_001", config={})
-        result = generate_cover_from_plan(plan, sample_video, tmp_path)
-        assert result is None
-
-    def test_none_config(self, sample_video, tmp_path):
-        """config 为 None."""
-        plan = FakePlan(id="plan_001", config=None)  # type: ignore
-        result = generate_cover_from_plan(plan, sample_video, tmp_path)
-        assert result is None
-
-
-# ═══════════════════════════════════════════════════════════════════════════════
-# 四、UnifiedRenderService 集成测试
-# ═══════════════════════════════════════════════════════════════════════════════
-
-
-def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), clip_type="main", config=None):
-    """创建测试用 ResolvedClip."""
-    return ResolvedClip(
-        clip_id=clip_id,
-        asset_id=asset_id,
-        local_path=path,
-        clip_type=clip_type,
-        order=0,
-        start_time=0.0,
-        duration=0.0,
-        transition_effect="cut",
-        config=config or {},
-        actual_duration=10.0,
-    )
-
-
-def _make_service(plan, clips, asset_path_map=None, work_dir=None, tmp_path=None):
-    """创建测试用 UnifiedRenderService."""
-    from pathlib import Path as P
-
-    work_dir = work_dir or (tmp_path or P("/tmp")) / "render_test"
-    work_dir.mkdir(exist_ok=True, parents=True)
-    return UnifiedRenderService(
-        plan=plan,
-        clips=clips,
-        asset_path_map=asset_path_map or {},
-        work_dir=work_dir,
-        output_width=1080,
-        output_height=1920,
-        output_fps=30,
-        transition_duration=0.5,
-    )
-
-
-class TestReverseIntegration:
-    """倒放功能集成测试."""
-
-    @patch("video_processing.unified_render_service.probe_video_info")
-    @patch("video_processing.unified_render_service.run_ffmpeg")
-    def test_reverse_in_filter_complex(self, mock_run, mock_probe, tmp_path):
-        """filter_complex 路径中包含倒放滤镜."""
-        mock_probe.return_value = {"duration": 10.0, "has_audio": True, "width": 1920, "height": 1080}
-        mock_run.return_value = None
-
-        plan = FakePlan(id="p1")
-        clip = _make_clip(config={"reverse": {"enabled": True}})
-        clip.actual_duration = 5.0
-        # 两个 clip 触发 filter_complex 路径
-        clip2 = _make_clip(clip_id="c2", config={})
-        clip2.actual_duration = 5.0
-        clip2.order = 1
-
-        service = _make_service(plan, [clip, clip2], tmp_path=tmp_path)
-
-        # 直接测 _build_filter_complex
-        from video_processing.unified_render_service import RenderLayer
-
-        layer = RenderLayer(role="main", clips=[clip, clip2])
-        filter_str, inputs = service._build_filter_complex([layer])
-
-        assert "reverse" in filter_str
-
-    def test_can_use_pass_through_with_reverse(self, tmp_path):
-        """倒放不影响直通模式判断（只有贴纸才禁用）."""
-        plan = FakePlan(id="p1")
-        clip = _make_clip(config={"reverse": {"enabled": True}})
-        clip.actual_duration = 5.0
-
-        service = _make_service(plan, [clip], tmp_path=tmp_path)
-        from video_processing.unified_render_service import RenderLayer
-
-        layer = RenderLayer(role="main", clips=[clip])
-        layers = [layer]
-
-        assert service._can_use_pass_through(layers) is True
-
-
-class TestStickerIntegration:
-    """贴纸功能集成测试."""
-
-    def test_can_use_pass_through_with_stickers(self, tmp_path):
-        """有贴纸时禁用直通模式."""
-        plan = FakePlan(id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "center"}]})
-        clip = _make_clip()
-        clip.actual_duration = 5.0
-
-        service = _make_service(plan, [clip], tmp_path=tmp_path)
-        from video_processing.unified_render_service import RenderLayer
-
-        layer = RenderLayer(role="main", clips=[clip])
-        layers = [layer]
-
-        assert service._can_use_pass_through(layers) is False
-
-    def test_can_use_pass_through_no_stickers(self, tmp_path):
-        """无贴纸时直通模式正常."""
-        plan = FakePlan(id="p1", config={})
-        clip = _make_clip()
-        clip.actual_duration = 5.0
-
-        service = _make_service(plan, [clip], tmp_path=tmp_path)
-        from video_processing.unified_render_service import RenderLayer
-
-        layer = RenderLayer(role="main", clips=[clip])
-        layers = [layer]
-
-        assert service._can_use_pass_through(layers) is True
-
-    def test_build_sticker_filters_text(self, tmp_path):
-        """文字贴纸滤镜构建."""
-        plan = FakePlan(
-            id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "top_center", "z_index": 10}]}
-        )
-        service = _make_service(plan, [], tmp_path=tmp_path)
-
-        filter_str, extra_inputs = service._build_sticker_filters("in", "out")
-
-        assert "drawtext" in filter_str
-        assert len(extra_inputs) == 0
-
-    def test_build_sticker_filters_empty(self, tmp_path):
-        """无贴纸返回空."""
-        plan = FakePlan(id="p1", config={})
-        service = _make_service(plan, [], tmp_path=tmp_path)
-
-        filter_str, extra_inputs = service._build_sticker_filters("in", "out")
-
-        assert filter_str == ""
-        assert extra_inputs == []
-
-    def test_build_sticker_filters_image(self, sample_image, tmp_path):
-        """图片贴纸滤镜构建 + 额外输入."""
-        plan = FakePlan(
-            id="p1",
-            config={
-                "stickers": [
-                    {
-                        "type": "image",
-                        "image_path": str(sample_image),
-                        "position": "bottom_right",
-                        "z_index": 5,
-                    }
-                ]
-            },
-        )
-        service = _make_service(plan, [], tmp_path=tmp_path)
-
-        filter_str, extra_inputs = service._build_sticker_filters("in", "out")
-
-        assert "overlay" in filter_str
-        assert len(extra_inputs) == 1

tests/unit/test_cover_service.py

@@ -1,400 +0,0 @@
-"""
-封面管理服务单元测试
-"""
-
-import subprocess
-from pathlib import Path
-from unittest.mock import Mock, patch
-
-import pytest
-
-from apps.api.app.services.cover_service import (
-    COVER_STORAGE_PREFIX,
-    DEFAULT_COVER_HEIGHT,
-    DEFAULT_COVER_QUALITY,
-    DEFAULT_COVER_WIDTH,
-    CoverService,
-)
-
-
-class TestGetCoverConfig:
-    """get_cover_config 静态方法测试"""
-
-    def test_get_cover_config_default(self):
-        """测试默认封面配置"""
-        config = {}
-        result = CoverService.get_cover_config(config)
-        assert result["type"] == "ai_frame"
-        assert result["image_url"] == ""
-        assert result["frame_time"] is None
-
-    def test_get_cover_config_with_custom_values(self):
-        """测试自定义封面配置"""
-        config = {
-            "cover": {
-                "type": "manual",
-                "image_url": "https://example.com/cover.jpg",
-                "frame_time": 5.5,
-            }
-        }
-        result = CoverService.get_cover_config(config)
-        assert result["type"] == "manual"
-        assert result["image_url"] == "https://example.com/cover.jpg"
-        assert result["frame_time"] == 5.5
-
-    def test_get_cover_config_cover_not_dict(self):
-        """测试 cover 不是 dict 时返回默认值"""
-        config = {"cover": "not-a-dict"}
-        result = CoverService.get_cover_config(config)
-        assert result["type"] == "ai_frame"
-        assert result["image_url"] == ""
-        assert result["frame_time"] is None
-
-    def test_get_cover_config_partial_fields(self):
-        """测试部分字段存在时，其余字段用默认值"""
-        config = {"cover": {"type": "custom"}}
-        result = CoverService.get_cover_config(config)
-        assert result["type"] == "custom"
-        assert result["image_url"] == ""
-        assert result["frame_time"] is None
-
-    def test_get_cover_config_empty_cover_dict(self):
-        """测试空的 cover dict"""
-        config = {"cover": {}}
-        result = CoverService.get_cover_config(config)
-        assert result["type"] == "ai_frame"
-        assert result["image_url"] == ""
-
-
-class TestExtractCoverFromClip:
-    """extract_cover_from_clip 测试"""
-
-    @pytest.fixture
-    def mock_storage(self):
-        storage = Mock()
-        storage.download_file = Mock()
-        storage.upload_file = Mock()
-        storage.get_url = Mock(return_value="https://oss.example.com/covers/plan1/cover_1000.jpg")
-        return storage
-
-    @pytest.fixture
-    def mock_asset_repo(self):
-        repo = Mock()
-        repo.get = Mock(return_value=None)
-        return repo
-
-    @pytest.fixture
-    def video_asset(self):
-        asset = Mock()
-        asset.storage_key = "videos/test-video.mp4"
-        asset.mime_type = "video/mp4"
-        return asset
-
-    @pytest.fixture
-    def service(self, mock_storage, mock_asset_repo):
-        return CoverService(storage_service=mock_storage, asset_repository=mock_asset_repo)
-
-    def test_extract_cover_asset_not_found(self, service, mock_asset_repo):
-        """测试素材不存在时报错"""
-        mock_asset_repo.get.return_value = None
-
-        with pytest.raises(ValueError, match="素材不存在"):
-            service.extract_cover_from_clip(plan_id="plan-1", asset_id="nonexistent")
-
-    def test_extract_cover_asset_no_storage_key(self, service, mock_asset_repo):
-        """测试素材没有文件时报错"""
-        asset = Mock()
-        asset.storage_key = ""
-        asset.mime_type = "video/mp4"
-        mock_asset_repo.get.return_value = asset
-
-        with pytest.raises(ValueError, match="素材没有文件"):
-            service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-no-file")
-
-    def test_extract_cover_asset_not_video(self, service, mock_asset_repo):
-        """测试非视频素材报错"""
-        asset = Mock()
-        asset.storage_key = "images/photo.jpg"
-        asset.mime_type = "image/jpeg"
-        mock_asset_repo.get.return_value = asset
-
-        with pytest.raises(ValueError, match="素材不是视频类型"):
-            service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-img")
-
-    def test_extract_cover_download_failure(self, service, mock_asset_repo, mock_storage, video_asset):
-        """测试下载素材失败"""
-        mock_asset_repo.get.return_value = video_asset
-        mock_storage.download_file.side_effect = Exception("网络错误")
-
-        with pytest.raises(RuntimeError, match="下载素材失败"):
-            service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-1")
-
-    def test_extract_cover_upload_failure(self, service, mock_asset_repo, mock_storage, video_asset):
-        """测试上传封面失败"""
-        mock_asset_repo.get.return_value = video_asset
-
-        def fake_download(storage_key, local_path):
-            # 创建一个假的视频文件
-            Path(local_path).parent.mkdir(parents=True, exist_ok=True)
-            with open(local_path, "wb") as f:
-                f.write(b"fake video data")
-
-        mock_storage.download_file.side_effect = fake_download
-        mock_storage.upload_file.side_effect = Exception("上传失败")
-
-        # mock _extract_frame 避免真的调 ffmpeg
-        with patch.object(CoverService, "_extract_frame") as mock_extract:
-
-            def fake_extract(video_path, output_path, **kwargs):
-                # 创建假的封面文件
-                with open(output_path, "wb") as f:
-                    f.write(b"\xff\xd8\xff\xe0fake jpeg data")
-
-            mock_extract.side_effect = fake_extract
-
-            with pytest.raises(RuntimeError, match="上传封面失败"):
-                service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-1")
-
-    def test_extract_cover_get_url_falls_back_to_key(self, service, mock_asset_repo, mock_storage, video_asset):
-        """测试获取 URL 失败时降级为 storage_key"""
-        mock_asset_repo.get.return_value = video_asset
-
-        def fake_download(storage_key, local_path):
-            Path(local_path).parent.mkdir(parents=True, exist_ok=True)
-            with open(local_path, "wb") as f:
-                f.write(b"fake video data")
-
-        mock_storage.download_file.side_effect = fake_download
-        mock_storage.get_url.side_effect = Exception("URL服务不可用")
-
-        with patch.object(CoverService, "_extract_frame") as mock_extract:
-
-            def fake_extract(video_path, output_path, **kwargs):
-                with open(output_path, "wb") as f:
-                    f.write(b"\xff\xd8\xff\xe0fake jpeg")
-
-            mock_extract.side_effect = fake_extract
-
-            result = service.extract_cover_from_clip(plan_id="plan-abc", asset_id="asset-xyz", frame_time=2.5)
-
-        assert result["type"] == "manual"
-        assert result["frame_time"] == 2.5
-        # URL 失败时返回 storage_key
-        assert COVER_STORAGE_PREFIX in result["image_url"]
-        assert "plan-abc" in result["image_url"]
-
-    def test_extract_cover_success(self, service, mock_asset_repo, mock_storage, video_asset):
-        """测试抽帧成功完整流程"""
-        mock_asset_repo.get.return_value = video_asset
-
-        def fake_download(storage_key, local_path):
-            Path(local_path).parent.mkdir(parents=True, exist_ok=True)
-            with open(local_path, "wb") as f:
-                f.write(b"fake video data for testing")
-
-        mock_storage.download_file.side_effect = fake_download
-
-        with patch.object(CoverService, "_extract_frame") as mock_extract:
-
-            def fake_extract(video_path, output_path, **kwargs):
-                with open(output_path, "wb") as f:
-                    f.write(b"\xff\xd8\xff\xe0fake jpeg image data")
-
-            mock_extract.side_effect = fake_extract
-
-            result = service.extract_cover_from_clip(
-                plan_id="plan-123",
-                asset_id="asset-456",
-                frame_time=3.0,
-                width=720,
-                height=1280,
-                quality=3,
-            )
-
-        assert result["type"] == "manual"
-        assert result["image_url"] == "https://oss.example.com/covers/plan1/cover_1000.jpg"
-        assert result["frame_time"] == 3.0
-
-        # 验证上传被调用
-        mock_storage.upload_file.assert_called_once()
-        upload_args = mock_storage.upload_file.call_args[1]
-        assert upload_args["content_type"] == "image/jpeg"
-        assert "plan-123" in upload_args["storage_key"]
-        assert "3000" in upload_args["storage_key"]  # frame_time * 1000
-
-        # 验证 _extract_frame 被调用且参数正确
-        mock_extract.assert_called_once()
-        extract_kwargs = mock_extract.call_args[1]
-        assert extract_kwargs["time_sec"] == 3.0
-        assert extract_kwargs["width"] == 720
-        assert extract_kwargs["height"] == 1280
-        assert extract_kwargs["quality"] == 3
-
-
-class TestGenerateSmartCover:
-    """generate_smart_cover 测试"""
-
-    @pytest.fixture
-    def service(self):
-        return CoverService(storage_service=Mock(), asset_repository=Mock())
-
-    def test_generate_smart_cover_calls_extract_with_default_time(self, service):
-        """测试智能封面调用 extract_cover_from_clip 并设置 type 为 ai_frame"""
-        fake_result = {"type": "manual", "image_url": "test.jpg", "frame_time": 3.0}
-
-        with patch.object(service, "extract_cover_from_clip", return_value=fake_result) as mock_extract:
-            result = service.generate_smart_cover(plan_id="plan-1", asset_id="asset-1")
-
-            mock_extract.assert_called_once()
-            call_kwargs = mock_extract.call_args[1]
-            assert call_kwargs["plan_id"] == "plan-1"
-            assert call_kwargs["asset_id"] == "asset-1"
-            assert call_kwargs["frame_time"] == 3.0  # 默认第3秒
-
-        assert result["type"] == "ai_frame"
-        assert result["image_url"] == "test.jpg"
-
-    def test_generate_smart_cover_passes_dimensions(self, service):
-        """测试智能封面传递尺寸和质量参数"""
-        fake_result = {"type": "manual", "image_url": "test.jpg", "frame_time": 3.0}
-
-        with patch.object(service, "extract_cover_from_clip", return_value=fake_result) as mock_extract:
-            service.generate_smart_cover(
-                plan_id="plan-1",
-                asset_id="asset-1",
-                width=1080,
-                height=1920,
-                quality=5,
-            )
-
-            call_kwargs = mock_extract.call_args[1]
-            assert call_kwargs["width"] == 1080
-            assert call_kwargs["height"] == 1920
-            assert call_kwargs["quality"] == 5
-
-
-class TestExtractFrame:
-    """_extract_frame 静态方法测试（mock subprocess）"""
-
-    @pytest.fixture
-    def video_path(self, tmp_path):
-        path = tmp_path / "test_video.mp4"
-        path.write_bytes(b"fake video")
-        return path
-
-    @pytest.fixture
-    def output_path(self, tmp_path):
-        return tmp_path / "cover.jpg"
-
-    def test_extract_frame_success(self, video_path, output_path):
-        """测试 FFmpeg 抽帧成功"""
-        fake_result = Mock()
-        fake_result.returncode = 0
-
-        with patch("subprocess.run", return_value=fake_result) as mock_run:
-            CoverService._extract_frame(
-                video_path=video_path,
-                output_path=output_path,
-                time_sec=2.5,
-                width=1080,
-                height=1920,
-                quality=5,
-            )
-
-            assert mock_run.call_count == 1
-            cmd = mock_run.call_args[0][0]
-            assert cmd[0] == "ffmpeg"
-            assert "-ss" in cmd
-            assert "2.500" in cmd
-            assert "-vframes" in cmd
-            # 验证 scale+crop 滤镜存在
-            vf_index = cmd.index("-vf") + 1
-            assert "scale=" in cmd[vf_index]
-            assert "crop=" in cmd[vf_index]
-
-    def test_extract_frame_fallback_to_simple_command(self, video_path, output_path):
-        """测试主命令失败时回退到简化命令"""
-        fail_result = Mock()
-        fail_result.returncode = 1
-        fail_result.stderr = "Filter graph error"
-
-        success_result = Mock()
-        success_result.returncode = 0
-
-        call_count = 0
-
-        def fake_run(*args, **kwargs):
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                return fail_result
-            return success_result
-
-        with patch("subprocess.run", side_effect=fake_run) as mock_run:
-            CoverService._extract_frame(
-                video_path=video_path,
-                output_path=output_path,
-                time_sec=1.0,
-                width=1080,
-                height=1920,
-                quality=5,
-            )
-
-            assert mock_run.call_count == 2
-            # 第二次是简化命令（没有 -vf 参数）
-            second_cmd = mock_run.call_args_list[1][0][0]
-            assert "-vf" not in second_cmd
-
-    def test_extract_frame_both_commands_fail(self, video_path, output_path):
-        """测试两个命令都失败时报错"""
-        fail_result = Mock()
-        fail_result.returncode = 1
-        fail_result.stderr = "Invalid data found when processing input"
-
-        with patch("subprocess.run", return_value=fail_result):
-            with pytest.raises(RuntimeError, match="FFmpeg 抽帧失败"):
-                CoverService._extract_frame(
-                    video_path=video_path,
-                    output_path=output_path,
-                    time_sec=1.0,
-                    width=1080,
-                    height=1920,
-                    quality=5,
-                )
-
-    def test_extract_frame_timeout(self, video_path, output_path):
-        """测试 FFmpeg 抽帧超时"""
-        with patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="ffmpeg", timeout=60)):
-            with pytest.raises(RuntimeError, match="FFmpeg 抽帧超时"):
-                CoverService._extract_frame(
-                    video_path=video_path,
-                    output_path=output_path,
-                    time_sec=1.0,
-                    width=1080,
-                    height=1920,
-                    quality=5,
-                )
-
-    def test_extract_frame_ffmpeg_not_found(self, video_path, output_path):
-        """测试 FFmpeg 不可用"""
-        with patch("subprocess.run", side_effect=FileNotFoundError("ffmpeg not found")):
-            with pytest.raises(RuntimeError, match="FFmpeg 不可用"):
-                CoverService._extract_frame(
-                    video_path=video_path,
-                    output_path=output_path,
-                    time_sec=1.0,
-                    width=1080,
-                    height=1920,
-                    quality=5,
-                )
-
-
-class TestDefaults:
-    """默认常量测试"""
-
-    def test_default_dimensions(self):
-        """测试默认尺寸常量"""
-        assert DEFAULT_COVER_WIDTH == 1080
-        assert DEFAULT_COVER_HEIGHT == 1920
-        assert DEFAULT_COVER_QUALITY == 5
-        assert COVER_STORAGE_PREFIX == "covers"

tests/unit/test_generation_cover.py

@@ -0,0 +1,94 @@
+"""Tests for generation cover route — schema validation and import checks."""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+
+def test_generation_cover_router_importable():
+    """新路由模块可以正确导入"""
+    from app.api.routes.generation_cover import router
+
+    assert router is not None
+    # tags 应该是 Generation
+    assert "Generation" in router.tags
+
+
+def test_generation_cover_route_path():
+    """路由路径应为 /generate-cover"""
+    from app.api.routes.generation_cover import router
+
+    paths = [route.path for route in router.routes]
+    assert "/generate-cover" in paths
+
+
+def test_generation_cover_schemas_importable():
+    """Schema 可以从新模块导入"""
+    from app.api.routes.generation_cover import GenerateCoverRequest, GenerateCoverResponse
+
+    # 验证请求 schema 默认值
+    req = GenerateCoverRequest()
+    assert req.asset_ids == []
+    assert req.cover_type == "ai_frame"
+    assert req.frame_time is None
+
+    # 验证响应 schema
+    resp = GenerateCoverResponse(plan_id="p1", cover={"image_url": "http://x"})
+    assert resp.plan_id == "p1"
+    assert resp.cover["image_url"] == "http://x"
+
+
+def test_generation_cover_schemas_not_in_templates_editor():
+    """旧的 templates_editor/schemas.py 不再包含封面 schema"""
+    from app.api.routes.templates_editor import schemas as te_schemas
+
+    assert not hasattr(te_schemas, "GenerateCoverRequest")
+    assert not hasattr(te_schemas, "GenerateCoverResponse")
+
+
+def test_templates_editor_no_cover_router():
+    """templates_editor 不再包含 cover_router"""
+    from app.api.routes.templates_editor import _sub_routers
+
+    # cover_router 应该已被移除
+    for sub in _sub_routers:
+        for route in sub.routes:
+            assert "generate-cover" not in getattr(route, "path", ""), "templates_editor 不应再有 generate-cover 路由"
+
+
+def test_api_router_has_generation_cover():
+    """api_router 应该包含 /api/v1/generation/generate-cover 路径"""
+    from app.api.router import api_router
+
+    all_paths = []
+    for route in api_router.routes:
+        if hasattr(route, "path"):
+            all_paths.append(route.path)
+        # 嵌套 router
+        if hasattr(route, "routes"):
+            for sub_route in route.routes:
+                if hasattr(sub_route, "path"):
+                    all_paths.append(sub_route.path)
+
+    # 应该能找到 generate-cover 路径
+    cover_paths = [p for p in all_paths if "generate-cover" in p]
+    assert len(cover_paths) > 0, f"未找到 generate-cover 路由, 所有路径: {all_paths[:20]}"
+
+
+def test_generation_cover_request_validation():
+    """验证请求 schema 的字段约束"""
+    from app.api.routes.generation_cover import GenerateCoverRequest
+
+    # frame_time 不允许负数
+    with pytest.raises(ValidationError):
+        GenerateCoverRequest(frame_time=-1.0)
+
+    # 合法的 frame_time
+    req = GenerateCoverRequest(frame_time=5.5)
+    assert req.frame_time == 5.5
+
+    # 自定义 cover_type
+    req2 = GenerateCoverRequest(cover_type="upload", asset_ids=["a1", "a2"])
+    assert req2.cover_type == "upload"
+    assert req2.asset_ids == ["a1", "a2"]

tests/unit/test_generation_preview.py

@@ -630,7 +630,7 @@ class TestToPreviewResponse:
         assert resp.file_size == 0
 
     def test_completed_task_with_videos(self):
-        """已完成任务，带视频结果（URL签名后返回）"""
+        """已完成任务，带视频结果（裸URL直接返回，rendered/*已公开读）"""
         task = _make_task(
             status=GenerationTaskStatus.COMPLETED,
             progress=100.0,
@@ -639,12 +639,8 @@ class TestToPreviewResponse:
         video.file_url = "https://cdn.example.com/preview.mp4"
         video.duration = 30.5
         video.file_size = 1024000
-        # Mock storage service to return a signed URL
-        mock_storage = MagicMock()
-        mock_storage.get_download_url.return_value = "https://cdn.example.com/preview.mp4?sig=test123"
-        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
-            resp = _to_preview_response(task, generated_videos=[video])
-        assert resp.video_url == "https://cdn.example.com/preview.mp4?sig=test123"
+        resp = _to_preview_response(task, generated_videos=[video])
+        assert resp.video_url == "https://cdn.example.com/preview.mp4"
         assert resp.duration == 30.5
         assert resp.file_size == 1024000
 
@@ -986,7 +982,7 @@ class TestGetPreviewRoute:
 
         # Mock URL 签名（返回带签名的 URL）
         mock_storage = MagicMock()
-        mock_storage.get_download_url.return_value = "https://cdn.example.com/preview_final.mp4?sig=abc123"
+        mock_storage.get_download_url.return_value = "https://cdn.example.com/preview_final.mp4"
 
         with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockGet:
             MockGet.return_value.execute.return_value = task
@@ -999,7 +995,7 @@ class TestGetPreviewRoute:
                         generation_task_repository=repo,
                         generated_video_repository=vid_repo,
                     )
-        assert resp.video_url == "https://cdn.example.com/preview_final.mp4?sig=abc123"
+        assert resp.video_url == "https://cdn.example.com/preview_final.mp4"
         assert resp.duration == 25.0
 
 
@@ -1088,50 +1084,6 @@ class TestWorkerPreviewResolution:
 # ═══════════════════════════════════════════════════════════════════════════════
 
 
-class TestSignVideoUrl:
-    """_sign_video_url 预签名 URL 测试。"""
-
-    def test_empty_url_returns_empty(self):
-        """空 URL 直接返回空字符串。"""
-        from app.api.routes.generation_preview import _sign_video_url
-
-        assert _sign_video_url("") == ""
-
-    def test_signs_oss_url(self):
-        """OSS URL 应被签名。"""
-        from app.api.routes.generation_preview import _sign_video_url
-
-        mock_storage = MagicMock()
-        mock_storage.get_download_url.return_value = "https://signed.example.com/video.mp4?sig=abc"
-
-        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
-            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
-
-        assert result == "https://signed.example.com/video.mp4?sig=abc"
-        mock_storage.get_download_url.assert_called_once()
-
-    def test_fallback_on_sign_failure(self):
-        """签名失败时降级返回原始 URL。"""
-        from app.api.routes.generation_preview import _sign_video_url
-
-        with patch("app.api.routes.generation_preview.get_storage_service", side_effect=RuntimeError("no storage")):
-            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
-
-        assert result == "https://bucket.oss.example.com/video.mp4"
-
-    def test_sign_returns_none_fallback(self):
-        """get_download_url 返回 None 时降级返回原始 URL。"""
-        from app.api.routes.generation_preview import _sign_video_url
-
-        mock_storage = MagicMock()
-        mock_storage.get_download_url.return_value = None
-
-        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
-            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
-
-        assert result == "https://bucket.oss.example.com/video.mp4"
-
-
 class TestInferVideoRatioFromTemplate:
     """_infer_video_ratio_from_template 单元测试。"""
 

tests/unit/test_mediakit_cover.py

@@ -3,6 +3,7 @@
 测试 #1208: AI封面接入MediaKit视频截帧
 """
 
+from pathlib import Path
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
@@ -140,66 +141,111 @@ class TestMediaKitClient:
 
 
 class TestAICoverService:
-    """AI 封面服务测试."""
+    """AI 封面服务测试（已迁移到 FFmpeg 本地抽帧）。"""
 
-    @patch("packages.shared.mediakit_client.get_mediakit_client")
-    def test_call_ai_cover_with_mediakit_success(self, mock_get_client):
-        """MediaKit 抽帧成功."""
-        mock_client = Mock()
-        mock_client.is_available = True
-        mock_client.extract_frames.return_value = [{"image_url": "https://example.com/frame.jpg", "timestamp": 3.5}]
-        mock_get_client.return_value = mock_client
+    @patch("packages.shared.ai_service.http_requests.head")
+    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
+    def test_call_ai_cover_with_ffmpeg_success(self, mock_ffmpeg, mock_head):
+        """FFmpeg 本地抽帧成功."""
+        import tempfile
+
+        mock_head.return_value.status_code = 200
+
+        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+        tmp.write(b"\xff\xd8" + b"\x00" * 50)
+        tmp.close()
+
+        mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 3.5}]
+
+        with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn:
+            mock_storage = Mock()
+            mock_storage.upload_file = Mock()
+            mock_storage.get_url.return_value = "https://example.com/frame.jpg"
+            mock_storage_fn.return_value = mock_storage
+
+            from packages.shared.ai_service import _call_ai_cover_service
+
+            result = _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_frame",
+                primary_video_url="https://example.com/video.mp4",
+            )
+
+            assert result["type"] == "ai_frame"
+            assert result["image_url"] == "https://example.com/frame.jpg"
+            assert result["frame_time"] == 3.5
+            assert result["confidence"] == 0.85
+
+        Path(tmp.name).unlink(missing_ok=True)
+
+    @patch("packages.shared.ai_service.http_requests.head")
+    def test_call_ai_cover_video_url_unreachable(self, mock_head):
+        """视频 URL 不可访问时抛出 RuntimeError."""
+        mock_head.return_value.status_code = 404
 
         from packages.shared.ai_service import _call_ai_cover_service
 
-        result = _call_ai_cover_service(
-            plan_id="plan-123",
-            asset_ids=["asset-1"],
-            cover_type="ai_frame",
-            primary_video_url="https://example.com/video.mp4",
-        )
+        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
+            _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_frame",
+                primary_video_url="https://example.com/nonexistent.mp4",
+            )
 
-        assert result["type"] == "ai_frame"
-        assert result["image_url"] == "https://example.com/frame.jpg"
-        assert result["frame_time"] == 3.5
-        assert result["confidence"] == 0.85
+    @patch("packages.shared.ai_service.http_requests.head")
+    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
+    def test_call_ai_cover_url_double_slash_normalized(self, mock_ffmpeg, mock_head):
+        """URL 路径中的双斜杠应被规范化."""
+        dirty_url = "https://oss.example.com/generated/projects//tasks/abc123/rendered.mp4"
+        clean_url = "https://oss.example.com/generated/projects/tasks/abc123/rendered.mp4"
 
-        mock_client.extract_frames.assert_called_once()
-
-    @patch("packages.shared.mediakit_client.get_mediakit_client")
-    def test_call_ai_cover_with_mediakit_failure_fallback(self, mock_get_client):
-        """MediaKit 失败时降级到 stub."""
-        mock_client = Mock()
-        mock_client.is_available = True
-        mock_client.extract_frames.side_effect = Exception("API error")
-        mock_get_client.return_value = mock_client
+        mock_head.return_value.status_code = 200
+        mock_ffmpeg.return_value = []
 
         from packages.shared.ai_service import _call_ai_cover_service
 
-        result = _call_ai_cover_service(
-            plan_id="plan-123",
-            asset_ids=["asset-1"],
-            cover_type="ai_frame",
-            primary_video_url="https://example.com/video.mp4",
-        )
+        with pytest.raises(RuntimeError):
+            _call_ai_cover_service(
+                plan_id="plan-1",
+                asset_ids=["a1"],
+                cover_type="ai_frame",
+                primary_video_url=dirty_url,
+            )
 
-        # 应该降级到 stub
-        assert result["type"] == "ai_frame"
-        assert result["image_url"].startswith("data:image/svg+xml,")
+        # HEAD 请求使用规范化后的 URL
+        mock_head.assert_called_once()
+        assert mock_head.call_args[0][0] == clean_url
+
+    @patch("packages.shared.ai_service.http_requests.head")
+    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
+    def test_call_ai_cover_ffmpeg_failure_raises(self, mock_ffmpeg, mock_head):
+        """FFmpeg 抽帧失败时抛出 RuntimeError."""
+        mock_head.return_value.status_code = 200
+        mock_ffmpeg.side_effect = Exception("ffmpeg error")
 
-    def test_call_ai_cover_without_video_url_fallback(self):
-        """没有视频 URL 时使用 stub."""
         from packages.shared.ai_service import _call_ai_cover_service
 
-        result = _call_ai_cover_service(
-            plan_id="plan-123",
-            asset_ids=["asset-1"],
-            cover_type="ai_frame",
-            primary_video_url=None,
-        )
+        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+            _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_frame",
+                primary_video_url="https://example.com/video.mp4",
+            )
 
-        assert result["type"] == "ai_frame"
-        assert result["image_url"].startswith("data:image/svg+xml,")
+    def test_call_ai_cover_without_video_url_raises(self):
+        """没有视频 URL 时抛出 RuntimeError."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+            _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_frame",
+                primary_video_url=None,
+            )
 
     def test_call_ai_cover_upload_type(self):
         """upload 类型直接返回."""
@@ -230,44 +276,22 @@ class TestAICoverService:
         assert result["type"] == "manual"
         assert result["frame_time"] == 5.0
 
-    @patch("packages.shared.mediakit_client.get_mediakit_client")
-    def test_call_ai_cover_mediakit_not_available(self, mock_get_client):
-        """MediaKit 未配置时使用 stub."""
-        mock_client = Mock()
-        mock_client.is_available = False
-        mock_get_client.return_value = mock_client
+    @patch("packages.shared.ai_service.http_requests.head")
+    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
+    def test_call_ai_cover_empty_frames_raises(self, mock_ffmpeg, mock_head):
+        """FFmpeg 返回空帧列表时抛出 RuntimeError."""
+        mock_head.return_value.status_code = 200
+        mock_ffmpeg.return_value = []
 
         from packages.shared.ai_service import _call_ai_cover_service
 
-        result = _call_ai_cover_service(
-            plan_id="plan-123",
-            asset_ids=["asset-1"],
-            cover_type="ai_frame",
-            primary_video_url="https://example.com/video.mp4",
-        )
-
-        assert result["type"] == "ai_frame"
-        assert result["image_url"].startswith("data:image/svg+xml,")
-
-    @patch("packages.shared.mediakit_client.get_mediakit_client")
-    def test_call_ai_cover_empty_frames_fallback(self, mock_get_client):
-        """MediaKit 返回空帧列表时降级."""
-        mock_client = Mock()
-        mock_client.is_available = True
-        mock_client.extract_frames.return_value = []
-        mock_get_client.return_value = mock_client
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        result = _call_ai_cover_service(
-            plan_id="plan-123",
-            asset_ids=["asset-1"],
-            cover_type="ai_frame",
-            primary_video_url="https://example.com/video.mp4",
-        )
-
-        assert result["type"] == "ai_frame"
-        assert result["image_url"].startswith("data:image/svg+xml,")
+        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+            _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_frame",
+                primary_video_url="https://example.com/video.mp4",
+            )
 
 
 class TestGenerateCover:

tests/unit/test_shared_ai_service.py

@@ -434,34 +434,22 @@ class TestAiCoverService:
         assert result["image_url"].startswith("data:image/svg+xml,")
         assert "手动选帧" in result["image_url"]
 
-    def test_cover_type_ai_frame(self):
-        with patch("shared.ai_service.time.sleep"):
-            with patch("shared.ai_service.random.uniform", side_effect=[5.0, 0.9]):
-                result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
-        assert result["type"] == "ai_frame"
-        assert result["frame_time"] == 5.0
-        assert result["confidence"] == 0.9
-        assert result["image_url"].startswith("data:image/svg+xml,")
-        assert "封面生成中" in result["image_url"]
+    def test_cover_type_ai_frame_raises_without_mediakit(self):
+        """ai_frame mode raises RuntimeError when MediaKit is unavailable."""
+        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+            _call_ai_cover_service("plan1", ["a1"], "ai_frame")
 
-    def test_cover_type_ai_regenerate(self):
-        with patch("shared.ai_service.time.sleep"):
-            result = _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
-        assert result["type"] == "ai_frame"
+    def test_cover_type_ai_regenerate_raises_without_mediakit(self):
+        """ai_regenerate mode raises RuntimeError when MediaKit is unavailable."""
+        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+            _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
 
-    def test_cover_frame_time_in_range(self):
+    def test_cover_type_manual_still_works(self):
+        """manual mode does not require MediaKit and still returns stub."""
         with patch("shared.ai_service.time.sleep"):
-            result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
-        assert 1.0 <= result["frame_time"] <= 10.0
-
-    def test_stub_returns_svg_data_uri(self):
-        """stub 降级返回 SVG data URI，不含任何后端 API 路径."""
-        with patch("shared.ai_service.time.sleep"):
-            result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
-        assert result["image_url"].startswith("data:image/svg+xml,")
-        assert "/api/v1/" not in result["image_url"]
-        assert "1080" in result["image_url"]
-        assert "1920" in result["image_url"]
+            result = _call_ai_cover_service("plan1", ["a1"], "manual", frame_time=5.5)
+        assert result["type"] == "manual"
+        assert result["frame_time"] == 5.5
 
     def test_manual_stub_returns_svg_data_uri(self):
         """manual 模式返回 SVG data URI 占位图."""