style: auto-format with black + isort + prettier [skip ci-format-check]

根因：封面图片生成有3条路径，只有路径B(extract_and_upload_cover_frames)
有标题叠加，但其产物(render_result.cover_candidates)被_render_video丢弃，
导致封面API永远拿不到带标题的候选帧，fallback到路径C远程抽帧(无标题)。

修改内容：
1. _render_video 返回值从 tuple[Path,float] 改为 tuple[Path,float,list[dict]]
   将 render_result.cover_candidates 传递给调用方
2. generate_video 渲染完成后将 cover_candidates 持久化到
   GenerationTask.metadata(JSON列)，无需DB migration
3. generation_cover API 从 GenerationTask.metadata 读取 cover_candidates
   替代原来从 plan.config 读取(从来没人写入过)
4. 删除路径A: 移除 RenderAdapter 中 generate_and_upload_thumbnail 调用
   (无标题且产物从未被使用)
5. 清理死代码: 删除 extract_first_frame、generate_and_upload_thumbnail
   函数及 RenderAdapterResult.thumbnail_url 字段
6. 路径C(_extract_frames_with_ffmpeg)保留作为兜底不变

标题传递链路验证:
custom_title → task_info → _render_video → virtual_plan.config[title]
→ render_adapter plan_config → extract_and_upload_cover_frames(title_text)
→ _overlay_title_on_image → 最终封面候选帧带标题

影响文件:
- apps/worker/worker_app/tasks/generation.py
- apps/worker/video_processing/render_adapter.py
- apps/worker/video_processing/thumbnail_generator.py
- apps/api/app/api/routes/generation_cover.py

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

@@ -78,8 +78,26 @@ def generate_cover(
     _, plan_svc = services
     plan = plan_svc.get_plan_or_raise(plan_id)
 
-    # ── 3 步查找预览视频 URL ──────────────────────────────────────────
-    # 第一步：从 plan.config 读取
+    # ── 查找预览视频 URL + 封面候选帧 ──────────────────────────────────
+    # 封面候选帧从 GenerationTask.metadata 读取（渲染时预抽帧，已叠加标题）
+    cover_candidates: list[dict] = []
+
+    def _read_cover_candidates_from_task(task_id: str) -> list[dict]:
+        """从 GenerationTask.metadata 列读取封面候选帧。"""
+        if not task_id:
+            return []
+        try:
+            from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+            task_model = db.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
+            if task_model:
+                meta = dict(task_model.metadata or {})
+                return meta.get("cover_candidates", [])
+        except Exception:
+            logger.warning("读取 GenerationTask 封面候选帧失败: task_id=%s", task_id, exc_info=True)
+        return []
+
+    # 第一步：从 plan.config 读取 rendered_storage_key
     logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id)
     rendered_storage_key = (plan.config or {}).get("rendered_storage_key", "")
 
@@ -105,10 +123,8 @@ def generate_cover(
                             generation_task_id,
                             rendered_storage_key[:80],
                         )
-                        logger.info(
-                            plan_id,
-                            generation_task_id,
-                        )
+                    # 同时读取封面候选帧
+                    cover_candidates = _read_cover_candidates_from_task(generation_task_id)
             except Exception:
                 logger.warning(
                     "封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s",
@@ -138,6 +154,8 @@ def generate_cover(
                         template_id,
                         completed_preview.id,
                     )
+                # 同时读取封面候选帧
+                cover_candidates = _read_cover_candidates_from_task(completed_preview.id)
         except Exception:
             logger.warning(
                 "封面警告: user+template 查找预览任务失败: plan_id=%s template_id=%s",
@@ -185,8 +203,7 @@ def generate_cover(
             detail=f"获取预览视频URL失败: {e}",
         ) from e
 
-    # 优先使用渲染时预抽的封面候选帧（跳过 MediaKit，秒级返回）
-    cover_candidates = (plan.config or {}).get("cover_candidates", [])
+    # 优先使用渲染时预抽的封面候选帧（已从 GenerationTask.metadata 读取，跳过 MediaKit，秒级返回）
     if cover_candidates and body.cover_type in ("ai_frame", "ai_regenerate"):
         logger.info(
             "[封面生成] 使用预存封面候选帧: plan_id=%s count=%d",

apps/worker/video_processing/render_adapter.py

@@ -71,7 +71,6 @@ class RenderAdapterResult:
     success: bool
     output_url: str = ""
     output_path: Path | None = None
-    thumbnail_url: str = ""
     duration: float = 0.0
     file_size: int = 0
     width: int = 0
@@ -562,23 +561,9 @@ class RenderAdapter:
         storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4"
         output_url = upload_to_oss(result.output_path, storage_key)
 
-        self._report_progress(progress_cb, 90.0, "生成封面缩略图")
+        self._report_progress(progress_cb, 90.0, "生成封面候选帧")
 
-        # 6. 生成封面缩略图
-        thumbnail_url = ""
-        try:
-            from video_processing.thumbnail_generator import generate_and_upload_thumbnail
-
-            thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg"
-            thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key)
-        except Exception as thumb_err:
-            logger.warning(
-                "[render-adapter] 缩略图生成失败（不影响主流程）: plan_id=%s error=%s",
-                plan_id,
-                thumb_err,
-            )
-
-        # 7. 抽取封面候选帧并上传 OSS（失败不阻断主流程）
+        # 6. 抽取封面候选帧并上传 OSS（失败不阻断主流程）
         cover_candidates = None
         try:
             from video_processing.thumbnail_generator import extract_and_upload_cover_frames
@@ -626,7 +611,6 @@ class RenderAdapter:
             success=True,
             output_url=output_url or "",
             output_path=result.output_path,
-            thumbnail_url=thumbnail_url,
             duration=result.duration,
             file_size=result.file_size,
             width=result.width,

apps/worker/video_processing/thumbnail_generator.py

@@ -1,4 +1,4 @@
-"""视频缩略图生成工具 — 抽取首帧上传到 OSS。"""
+"""视频封面候选帧生成工具 — 抽帧叠加标题上传到 OSS。"""
 
 from __future__ import annotations
 
@@ -9,109 +9,6 @@ from pathlib import Path
 logger = logging.getLogger(__name__)
 
 
-def extract_first_frame(
-    video_path: str,
-    output_path: str | None = None,
-    *,
-    width: int = 640,
-    height: int = -1,
-    timeout: int = 30,
-    seek_ratio: float = 0.15,
-    min_seek_seconds: float = 1.0,
-) -> str:
-    """抽取视频封面图（默认取视频时长 15% 处的帧，避开片头纯色画面）。
-
-    Args:
-        video_path: 视频文件路径
-        output_path: 输出图片路径，不传则用临时文件
-        width: 输出宽度（默认 640，-1 表示按比例缩放）
-        height: 输出高度（默认 -1，按比例缩放）
-        timeout: 超时时间（秒）
-        seek_ratio: 抽帧位置占视频时长的比例（默认 0.15，即 15% 处）
-        min_seek_seconds: 最小抽帧时间（秒），避免极短视频 seek 到 0
-
-    Returns:
-        生成的缩略图文件路径
-
-    Raises:
-        subprocess.CalledProcessError: ffmpeg 执行失败
-    """
-    from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
-
-    _is_temp_output = False
-    if output_path is None:
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.close()
-        output_path = tmp.name
-        _is_temp_output = True
-
-    try:
-        # 计算抽帧时间点：取视频时长 * seek_ratio，最少 min_seek_seconds 秒
-        try:
-            duration = probe_duration(video_path)
-            seek_time = max(min_seek_seconds, duration * seek_ratio)
-        except Exception:
-            # probe 失败时 fallback 到第1秒
-            seek_time = min_seek_seconds
-
-        # 格式化为 HH:MM:SS.xx
-        seek_str = _format_seek_time(seek_time)
-
-        # -ss 放在 -i 前面（input seeking，更快但精度稍低，缩略图够用）
-        # -vframes 1 只取一帧
-        # -q:v 2 jpeg 高质量
-        scale_filter = f"scale={width}:{height}: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,
-        ]
-
-        try:
-            run_ffmpeg(cmd, capture_output=True, timeout=timeout)
-        except Exception:
-            # 失败时退回到第0帧兜底
-            cmd2 = [
-                FFMPEG_BIN,
-                "-y",
-                "-i",
-                video_path,
-                "-ss",
-                "00:00:00",
-                "-vframes",
-                "1",
-                "-vf",
-                scale_filter,
-                "-q:v",
-                "2",
-                output_path,
-            ]
-            run_ffmpeg(cmd2, capture_output=True, timeout=timeout)
-
-        if not Path(output_path).exists() or Path(output_path).stat().st_size == 0:
-            raise RuntimeError(f"Thumbnail generation failed: {output_path}")
-
-        return output_path
-    except Exception:
-        # 失败时清理自己创建的临时文件
-        if _is_temp_output and output_path:
-            try:
-                Path(output_path).unlink(missing_ok=True)
-            except Exception:
-                pass
-        raise
-
-
 def _format_seek_time(seconds: float) -> str:
     """将秒数格式化为 HH:MM:SS.xx 格式。"""
     h = int(seconds // 3600)
@@ -206,43 +103,6 @@ def _overlay_title_on_image(
     return image_path
 
 
-def generate_and_upload_thumbnail(
-    video_path: str,
-    storage_key: str,
-) -> str | None:
-    """生成缩略图并上传到 OSS，返回 URL。
-
-    Args:
-        video_path: 本地视频路径
-        storage_key: OSS 存储 key（如 generated/projects/xxx/thumbnails/yyy.jpg）
-
-    Returns:
-        上传成功返回 URL，失败返回 None
-    """
-    thumbnail_path = None
-    try:
-        thumbnail_path = extract_first_frame(video_path)
-    except Exception as e:
-        logger.warning("Failed to extract thumbnail from %s: %s", video_path, e)
-        return None
-
-    try:
-        from video_processing.oss_helpers import upload_to_oss
-
-        url = upload_to_oss(thumbnail_path, storage_key)
-        return url
-    except Exception as e:
-        logger.warning("Failed to upload thumbnail to OSS: %s", e)
-        return None
-    finally:
-        # 清理临时文件
-        if thumbnail_path:
-            try:
-                Path(thumbnail_path).unlink(missing_ok=True)
-            except Exception:
-                pass
-
-
 def extract_cover_candidates(
     video_path: str,
     num_frames: int = 3,

apps/worker/worker_app/tasks/generation.py

@@ -1124,14 +1124,14 @@ def _render_video(
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
     custom_title: str = "",
-) -> tuple[Path, float]:
+) -> tuple[Path, float, list[dict]]:
     """渲染视频（含配音混音）。
 
-    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
+    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/封面候选帧逻辑。
 
     Args:
     Returns:
-        (output_path, render_duration)
+        (output_path, render_duration, cover_candidates)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1259,8 +1259,9 @@ def _render_video(
 
     # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
     output_path = render_output_path
+    cover_candidates = render_result.cover_candidates or []
 
-    return output_path, render_duration
+    return output_path, render_duration, cover_candidates
 
 
 def _upload_and_record(
@@ -1531,7 +1532,7 @@ def generate_video(self, task_id: str) -> dict:
             else:
                 _resolved_resolution = task_info.get("resolution", "")
 
-            output_path, render_duration = _render_video(
+            output_path, render_duration, cover_candidates = _render_video(
                 task_id=task_id,
                 downloaded_videos=downloaded_videos,
                 voice_path=audio_path,
@@ -1547,6 +1548,35 @@ def generate_video(self, task_id: str) -> dict:
                 custom_title=task_info.get("custom_title", ""),
             )
 
+            # 持久化封面候选帧到 GenerationTask.metadata，供封面 API 直接读取
+            if cover_candidates:
+                try:
+                    _cover_session = SessionLocal()
+                    try:
+                        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+                        _cover_model = (
+                            _cover_session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
+                        )
+                        if _cover_model:
+                            existing_meta = dict(_cover_model.metadata or {})
+                            existing_meta["cover_candidates"] = cover_candidates
+                            _cover_model.metadata = existing_meta
+                            _cover_session.commit()
+                            logger.info(
+                                "[task_id=%s] 封面候选帧已持久化: count=%d",
+                                task_id,
+                                len(cover_candidates),
+                            )
+                    finally:
+                        _cover_session.close()
+                except Exception:
+                    logger.warning(
+                        "[task_id=%s] 封面候选帧持久化失败（不影响主流程）",
+                        task_id,
+                        exc_info=True,
+                    )
+
             if gen_task:
                 gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
                 _flush_logs(task_id, gen_task)