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

@@ -75,6 +75,86 @@ class GenerateCoverResponse(BaseModel):
 # ── Route ────────────────────────────────────────────────────────────────
 
 
+
+def _select_best_frame_from_snapshots(
+    snapshots: list[dict], plan_id: str
+) -> str:
+    """从 MediaKit 抽帧结果中，通过质量评分选出最佳帧。
+
+    降级策略：cv2 不可用或评分失败时，返回第一帧。
+
+    Args:
+        snapshots: MediaKit 返回的帧列表 [{"image_url": str, ...}, ...]
+        plan_id: 计划 ID（日志用）
+
+    Returns:
+        最佳帧的 image_url，或空字符串
+    """
+    if not snapshots:
+        return ""
+    if len(snapshots) == 1:
+        return snapshots[0].get("image_url") or snapshots[0].get("url") or ""
+
+    try:
+        import tempfile
+
+        import httpx
+
+        from packages.shared.cover_frame_scorer import score_frames
+
+        scored_candidates = []
+        for snap in snapshots:
+            url = snap.get("image_url") or snap.get("url") or ""
+            if not url:
+                continue
+            # 下载帧到临时文件进行评分
+            try:
+                resp = httpx.get(url, timeout=15, follow_redirects=True)
+                resp.raise_for_status()
+                with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
+                    tmp.write(resp.content)
+                    tmp_path = tmp.name
+                scored_candidates.append({"image_path": tmp_path, "url": url})
+            except Exception:
+                # 下载失败的帧跳过，给默认低分
+                scored_candidates.append({"image_path": None, "url": url, "score": 0.0})
+
+        if not scored_candidates:
+            return snapshots[0].get("image_url") or snapshots[0].get("url") or ""
+
+        scored = score_frames(scored_candidates)
+        best = scored[0] if scored else None
+        best_url = best.get("url", "") if best else ""
+        best_score = best.get("score", 0.0) if best else 0.0
+        logger.info(
+            "[封面生成] 帧质量评分完成: plan_id=%s candidates=%d best_score=%.1f",
+            plan_id,
+            len(scored_candidates),
+            best_score,
+        )
+
+        # 清理临时文件
+        for c in scored_candidates:
+            path = c.get("image_path")
+            if path:
+                try:
+                    from pathlib import Path
+
+                    Path(path).unlink(missing_ok=True)
+                except Exception:
+                    pass
+
+        return best_url
+
+    except Exception:
+        logger.warning(
+            "[封面生成] 帧质量评分失败，使用第一帧: plan_id=%s",
+            plan_id,
+            exc_info=True,
+        )
+        return snapshots[0].get("image_url") or snapshots[0].get("url") or ""
+
+
 def _persist_cover_frame(
     frame_url: str,
     plan_id: str,
@@ -652,13 +732,13 @@ def generate_cover(
                     snapshots = mk_client.extract_frames(
                         video_url=primary_video_url,
                         strategy="SpecifiedFrames",
-                        max_frames=1,
+                        max_frames=5,  # 抽 5 帧，通过质量评分选最佳
                         poll_interval=2.0,
                         max_poll_attempts=5,
                         max_retries=0,
                     )
                     if snapshots:
-                        raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
+                        raw = _select_best_frame_from_snapshots(snapshots, plan_id)
                         if raw:
                             cover_url_from_task = _persist_cover_frame(raw, plan_id)
                             logger.info(
@@ -715,13 +795,13 @@ def generate_cover(
                         snapshots = mk_client.extract_frames(
                             video_url=src_url,
                             strategy="SpecifiedFrames",
-                            max_frames=1,
+                            max_frames=5,  # 抽 5 帧，通过质量评分选最佳
                             poll_interval=2.0,
                             max_poll_attempts=5,
                             max_retries=0,
                         )
                         if snapshots:
-                            raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
+                            raw = _select_best_frame_from_snapshots(snapshots, plan_id)
                             if raw:
                                 cover_url_from_task = _persist_cover_frame(
                                     raw,

apps/worker/video_processing/render_adapter.py

@@ -599,7 +599,7 @@ class RenderAdapter:
             # 抽帧天然带标题，因此这里传空字符串，避免 Pillow 二次叠加导致重影。
             # Pillow 叠加仅用于 API 从源素材抽帧（源素材本身无标题）的兜底场景。
             cover_candidates = extract_and_upload_cover_frames(
-                str(result.output_path), plan_id, task_id=job_id, num_frames=3, title_text=""
+                str(result.output_path), plan_id, task_id=job_id, num_frames=5, title_text=""
             )
             if cover_candidates:
                 logger.info(

apps/worker/video_processing/thumbnail_generator.py

@@ -279,21 +279,25 @@ def extract_and_upload_cover_frames(
     plan_id: str,
     *,
     task_id: str = "",
-    num_frames: int = 3,
+    num_frames: int = 5,  # 抽 5 帧候选，通过质量评分选出最佳帧
     title_text: str = "",
     title_color: str = "#ffffff",
     title_position: str = "bottom",
     title_font_size: int | None = None,
 ) -> list[dict]:
-    """从视频中抽取多帧作为封面候选，上传到 OSS。
+    """从视频中抽取多帧作为封面候选，通过质量评分选出最佳帧，上传到 OSS。
 
-    优先使用 MediaKit 智能抽帧，失败时降级到 ffmpeg 直接抽帧。
+    流程：
+    1. 优先使用 MediaKit 智能抽帧（多抽一些供选择）
+    2. MediaKit 不足时降级到 ffmpeg 均匀抽帧
+    3. 对所有候选帧进行质量评分（清晰度/亮度/色彩丰富度）
+    4. 按分数从高到低排序返回
 
     Args:
         video_path: 视频文件路径
         plan_id: 编辑计划 ID（用于生成 storage key）
         task_id: 任务 ID（用于生成独立的 storage key，避免标题变更时封面冲突）
-        num_frames: 抽取帧数（默认 3）
+        num_frames: 抽取候选帧数（默认 5，通过质量评分选出最佳帧）
         title_text: 标题文字；非空时用 Pillow 叠加到每帧。
             从已渲染视频抽帧时通常传空（标题已烧录）；从源素材抽帧时传标题。
         title_color: 标题字体颜色（#RRGGBB）
@@ -301,7 +305,7 @@ def extract_and_upload_cover_frames(
         title_font_size: 标题字号，None 时自动计算
 
     Returns:
-        封面候选列表，每项包含 {"url": str, "position": float}
+        封面候选列表（按质量分数降序），每项包含 {"url": str, "position": float, "score": float}
     """
     import httpx
     from video_processing.ffmpeg_utils import probe_duration
@@ -313,80 +317,125 @@ def extract_and_upload_cover_frames(
         duration = 0.0
 
     candidates: list[dict] = []
+    _temp_paths: list[str] = []  # 收集所有临时文件路径，最后统一清理
 
-    # 优先尝试 MediaKit 智能抽帧
-    mediakit_frames = _extract_frames_via_mediakit(video_path, plan_id, num_frames)
-    if mediakit_frames:
-        for i, frame in enumerate(mediakit_frames):
-            frame_url = frame.get("image_url")
-            if not frame_url:
-                continue
-            tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-            tmp.close()
-            try:
-                # 下载 MediaKit 返回的帧图
-                resp = httpx.get(frame_url, timeout=30, follow_redirects=True)
-                resp.raise_for_status()
-                with open(tmp.name, "wb") as f:
-                    f.write(resp.content)
+    try:
+        # ── 阶段 1：抽帧 ──────────────────────────────────────────────
+        # 优先尝试 MediaKit 智能抽帧
+        mediakit_frames = _extract_frames_via_mediakit(video_path, plan_id, num_frames)
+        if mediakit_frames:
+            for i, frame in enumerate(mediakit_frames):
+                frame_url = frame.get("image_url")
+                if not frame_url:
+                    continue
+                tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+                tmp.close()
+                _temp_paths.append(tmp.name)
+                try:
+                    # 下载 MediaKit 返回的帧图
+                    resp = httpx.get(frame_url, timeout=30, follow_redirects=True)
+                    resp.raise_for_status()
+                    with open(tmp.name, "wb") as f:
+                        f.write(resp.content)
 
-                # 叠加标题文字（如需要）
-                if title_text and title_text.strip():
-                    apply_title_overlay(
-                        tmp.name,
-                        title_text,
-                        color=title_color,
-                        position=title_position,
-                        font_size=title_font_size,
+                    # 叠加标题文字（如需要）
+                    if title_text and title_text.strip():
+                        apply_title_overlay(
+                            tmp.name,
+                            title_text,
+                            color=title_color,
+                            position=title_position,
+                            font_size=title_font_size,
+                        )
+
+                    storage_key = f"covers/{plan_id}/{task_id}/mediakit_frame_{i}.jpg"
+                    url = upload_to_oss(tmp.name, storage_key)
+                    if url:
+                        seek_time = frame.get("timestamp", 0.0)
+                        candidates.append(
+                            {
+                                "url": url,
+                                "position": round(seek_time, 2),
+                                "image_path": tmp.name,
+                            }
+                        )
+                except Exception as e:
+                    logger.warning("[thumbnail] MediaKit 帧 %d 处理失败: %s", i, e)
+
+            if len(candidates) >= num_frames:
+                logger.info("[thumbnail] MediaKit 智能抽帧完成: %d 帧", len(candidates))
+            else:
+                logger.warning("[thumbnail] MediaKit 抽帧不足 %d 帧，降级到 ffmpeg", num_frames)
+
+        # Fallback: ffmpeg 直接抽帧（仅当 MediaKit 不足时）
+        if len(candidates) < num_frames:
+            logger.info("[thumbnail] 使用 ffmpeg 抽帧补充")
+            # 均匀分布抽帧点：从 10% 到 90%
+            for i in range(num_frames):
+                ratio = 0.1 + 0.8 * i / max(num_frames - 1, 1)
+                tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+                tmp.close()
+                _temp_paths.append(tmp.name)
+                try:
+                    frame_path = extract_first_frame(
+                        video_path,
+                        output_path=tmp.name,
+                        seek_ratio=ratio,
+                        min_seek_seconds=0.5,
                     )
+                    # 从源素材抽帧时叠加标题文字；已渲染视频标题已烧录时传空字符串跳过
+                    if title_text and title_text.strip():
+                        apply_title_overlay(
+                            frame_path,
+                            title_text,
+                            color=title_color,
+                            position=title_position,
+                            font_size=title_font_size,
+                        )
+                    storage_key = f"covers/{plan_id}/{task_id}/frame_{i}.jpg"
+                    url = upload_to_oss(frame_path, storage_key)
+                    if url:
+                        seek_time = max(0.5, duration * ratio) if duration > 0 else 0.0
+                        candidates.append(
+                            {
+                                "url": url,
+                                "position": round(seek_time, 2),
+                                "image_path": tmp.name,
+                            }
+                        )
+                except Exception as e:
+                    logger.warning("[thumbnail] 封面候选帧 %d 提取失败: %s", i, e)
 
-                storage_key = f"covers/{plan_id}/{task_id}/mediakit_frame_{i}.jpg"
-                url = upload_to_oss(tmp.name, storage_key)
-                if url:
-                    seek_time = frame.get("timestamp", 0.0)
-                    candidates.append({"url": url, "position": round(seek_time, 2)})
-            except Exception as e:
-                logger.warning("[thumbnail] MediaKit 帧 %d 处理失败: %s", i, e)
-            finally:
-                Path(tmp.name).unlink(missing_ok=True)
+        # ── 阶段 2：质量评分 ────────────────────────────────────────────
+        if len(candidates) > 1:
+            try:
+                from packages.shared.cover_frame_scorer import score_frames
 
-        if len(candidates) >= num_frames:
-            logger.info("[thumbnail] MediaKit 智能抽帧完成: %d 帧", len(candidates))
-            return candidates[:num_frames]
-
-        logger.warning("[thumbnail] MediaKit 抽帧不足 %d 帧，降级到 ffmpeg", num_frames)
-
-    # Fallback: ffmpeg 直接抽帧
-    logger.info("[thumbnail] 使用 ffmpeg 抽帧")
-    # 均匀分布抽帧点：从 10% 到 90%
-    for i in range(num_frames):
-        ratio = 0.1 + 0.8 * i / max(num_frames - 1, 1)
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.close()
-        try:
-            frame_path = extract_first_frame(
-                video_path,
-                output_path=tmp.name,
-                seek_ratio=ratio,
-                min_seek_seconds=0.5,
-            )
-            # 从源素材抽帧时叠加标题文字；已渲染视频标题已烧录时传空字符串跳过
-            if title_text and title_text.strip():
-                apply_title_overlay(
-                    frame_path,
-                    title_text,
-                    color=title_color,
-                    position=title_position,
-                    font_size=title_font_size,
+                candidates = score_frames(candidates)
+                logger.info(
+                    "[thumbnail] 封面帧质量评分完成: plan_id=%s count=%d best_score=%.1f",
+                    plan_id,
+                    len(candidates),
+                    candidates[0].get("score", 0.0) if candidates else 0.0,
+                )
+            except Exception:
+                logger.warning(
+                    "[thumbnail] 封面帧质量评分失败，保持原始顺序: plan_id=%s",
+                    plan_id,
+                    exc_info=True,
                 )
-            storage_key = f"covers/{plan_id}/{task_id}/frame_{i}.jpg"
-            url = upload_to_oss(frame_path, storage_key)
-            if url:
-                seek_time = max(0.5, duration * ratio) if duration > 0 else 0.0
-                candidates.append({"url": url, "position": round(seek_time, 2)})
-        except Exception as e:
-            logger.warning("[thumbnail] 封面候选帧 %d 提取失败: %s", i, e)
-        finally:
-            Path(tmp.name).unlink(missing_ok=True)
 
-    return candidates
+        # ── 阶段 3：清理临时文件 ────────────────────────────────────────
+        # 移除 image_path（不再需要），但临时文件统一清理
+        for c in candidates:
+            c.pop("image_path", None)
+
+        return candidates
+
+    finally:
+        # 统一清理所有临时文件
+        for path in _temp_paths:
+            try:
+                Path(path).unlink(missing_ok=True)
+            except Exception:
+                pass

packages/shared/cover_frame_scorer.py

@@ -0,0 +1,141 @@
+"""封面帧质量评分器 — 评估视频帧的视觉质量，用于智能选帧。
+
+评分维度（总分 0~100）：
+1. 清晰度（0~40）：拉普拉斯方差，越高越清晰
+2. 亮度（0~30）：均值亮度，80~180 区间满分
+3. 色彩丰富度（0~30）：RGB 三维直方图非零 bin 数
+
+设计原则：
+- 纯函数，输入 numpy array，输出 float
+- 无副作用、无 IO、无网络
+- 不依赖 OpenCV 的 GUI 模块，仅用 cv2/numpy 的计算函数
+- 降级策略：cv2 不可用时返回 50.0（中间值）
+"""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+def score_frame(frame: "np.ndarray") -> float:
+    """评估单帧图像质量，返回 0~100 分（越高越好）。
+
+    Args:
+        frame: BGR 格式的 numpy array（OpenCV  imread 或 video capture 读取）
+
+    Returns:
+        float: 质量评分 0~100
+
+    评分维度：
+        - 清晰度 (0~40): 拉普拉斯方差，越高越清晰
+        - 亮度 (0~30): 均值亮度，80~180 区间满分
+        - 色彩丰富度 (0~30): RGB 三维直方图非零 bin 数
+    """
+    try:
+        import cv2
+        import numpy as np
+
+        # 确保 cv2 和 numpy 是真实的模块，不是 mock 对象
+        # MagicMock 会有所有属性，但调用结果不是真实数值类型
+        if not hasattr(cv2, "Laplacian") or not hasattr(np, "ndarray"):
+            raise ImportError("cv2/numpy 缺少核心方法")
+        # 验证 cv2 的核心函数是真实的 C/Python 函数，不是 MagicMock
+        if not inspect.isroutine(cv2.Laplacian) or "mock" in str(type(cv2.Laplacian)).lower():
+            raise ImportError("cv2 是 mock 对象")
+    except (ImportError, TypeError, AttributeError):
+        logger.warning("[cover_scorer] cv2/numpy 不可用或是 mock，返回默认分 50.0")
+        return 50.0
+
+    if frame is None or frame.size == 0:
+        return 0.0
+
+    # ── 1. 清晰度（拉普拉斯方差）────────────────────────────────────
+    # 拉普拉斯算子检测边缘，方差越大说明高频细节越多，图像越清晰
+    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+    laplacian_var = cv2.Laplacian(gray, cv2.CV_64F).var()
+    # 归一化：一般视频拉普拉斯方差在 0~500 之间，映射到 0~40 分
+    clarity = min(laplacian_var / 500.0 * 40.0, 40.0)
+
+    # ── 2. 亮度（适中最好）──────────────────────────────────────────
+    # 均值亮度在 80~180 区间视觉舒适；过暗（<80）或过亮（>180）扣分
+    mean_brightness = float(np.mean(gray))
+    if 80.0 <= mean_brightness <= 180.0:
+        brightness_score = 30.0
+    else:
+        # 偏离舒适区间越远扣分越多，最多扣 30 分
+        deviation = abs(mean_brightness - 130.0) - 50.0  # 130 是舒适区中心
+        brightness_score = max(0.0, 30.0 - deviation * 0.3)
+
+    # ── 3. 色彩丰富度（RGB 三维直方图）──────────────────────────────
+    # 将 RGB 各通道量化为 8 个 bin，共 8x8x8=512 个 bin
+    # 非零 bin 越多说明色彩越丰富
+    hist = cv2.calcHist([frame], [0, 1, 2], None, [8, 8, 8], [0, 256] * 3)
+    nonzero_bins = float(np.count_nonzero(hist))
+    # 归一化：一般帧非零 bin 在 0~200 之间，映射到 0~30 分
+    color_richness = min(nonzero_bins / 200.0 * 30.0, 30.0)
+
+    total = clarity + brightness_score + color_richness
+    return round(total, 2)
+
+
+def score_frames(frames: list[dict]) -> list[dict]:
+    """批量评估帧质量，在每帧 dict 中增加 score 字段并排序。
+
+    Args:
+        frames: 帧列表，每项需包含 image_path 或 image_array。
+            支持格式：
+            - {"image_path": "/path/to/frame.jpg", ...}
+            - {"image_array": np.ndarray, ...}
+
+    Returns:
+        排序后的帧列表（分数从高到低），每项增加 "score" 字段
+    """
+    try:
+        import cv2
+    except ImportError:
+        logger.warning("[cover_scorer] cv2 不可用，跳过评分")
+        for f in frames:
+            f["score"] = 50.0
+        return frames
+
+    scored: list[dict] = []
+    for f in frames:
+        arr = f.get("image_array")
+        if arr is None:
+            path = f.get("image_path", "")
+            if path:
+                try:
+                    arr = cv2.imread(path)
+                except Exception:
+                    logger.warning("[cover_scorer] 读取帧图片失败: %s", path)
+        if arr is not None and arr.size > 0:
+            f["score"] = score_frame(arr)
+        else:
+            f["score"] = 0.0
+        scored.append(f)
+
+    # 按分数从高到低排序
+    scored.sort(key=lambda x: x.get("score", 0.0), reverse=True)
+    return scored
+
+
+def select_best_frame(frames: list[dict]) -> dict | None:
+    """从候选帧中选择质量最高的一帧。
+
+    Args:
+        frames: 帧列表，每项需包含 image_path 或 image_array
+
+    Returns:
+        评分最高的帧 dict（包含 score 字段），空列表返回 None
+    """
+    if not frames:
+        return None
+    scored = score_frames(frames)
+    return scored[0] if scored else None

tests/unit/test_cover_frame_scorer.py

@@ -0,0 +1,226 @@
+"""封面帧质量评分器测试 — cover_frame_scorer.
+
+测试维度：
+1. score_frame: 清晰度/亮度/色彩丰富度各维度评分
+2. score_frames: 批量评分和排序
+3. select_best_frame: 选出最佳帧
+4. 降级策略：cv2 不可用时返回默认分
+5. 边界条件：空帧、None、损坏数据
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import numpy as np
+import pytest
+
+
+def _cv2_available() -> bool:
+    """检测 cv2 是否真实可用（非 mock）."""
+    try:
+        import cv2
+        import numpy as np
+
+        if not hasattr(cv2, "Laplacian") or not inspect.isroutine(cv2.Laplacian):
+            return False
+        if "mock" in str(type(cv2.Laplacian)).lower():
+            return False
+        # 实际调用测试
+        _test = np.zeros((2, 2, 3), dtype=np.uint8)
+        _result = cv2.cvtColor(_test, cv2.COLOR_BGR2GRAY)
+        return isinstance(_result, np.ndarray)
+    except Exception:
+        return False
+
+
+HAS_CV2 = _cv2_available()
+requires_cv2 = pytest.mark.skipif(not HAS_CV2, reason="cv2 不可用或是 mock 对象")
+
+
+class TestScoreFrame:
+    """score_frame 单元测试."""
+
+    @requires_cv2
+    def test_clear_image_high_score(self):
+        """清晰、亮度适中、色彩丰富的图像应得高分."""
+        # 创建一个清晰的渐变图像（色彩丰富、亮度适中）
+        img = np.zeros((100, 100, 3), dtype=np.uint8)
+        for i in range(100):
+            for j in range(100):
+                img[i, j] = [i * 2, j * 2, (i + j) % 256]
+
+        from packages.shared.cover_frame_scorer import score_frame
+
+        score = score_frame(img)
+        assert 50.0 <= score <= 100.0, f"清晰图像应得高分，实际: {score}"
+
+    @requires_cv2
+    def test_blurry_image_low_clarity(self):
+        """模糊图像的清晰度分数应较低."""
+        # 纯色图像（无高频细节）
+        img = np.full((100, 100, 3), 128, dtype=np.uint8)
+
+        from packages.shared.cover_frame_scorer import score_frame
+
+        score = score_frame(img)
+        # 纯色图像清晰度为 0，亮度满分 30，色彩为 0
+        assert score <= 35.0, f"模糊图像应低分，实际: {score}"
+
+    @requires_cv2
+    def test_dark_image_low_brightness(self):
+        """过暗图像应扣分."""
+        # 全黑图像
+        img = np.zeros((100, 100, 3), dtype=np.uint8)
+
+        from packages.shared.cover_frame_scorer import score_frame
+
+        score = score_frame(img)
+        # 全黑：清晰度 0，亮度 0，色彩 0
+        assert score <= 5.0, f"全黑图像应接近 0 分，实际: {score}"
+
+    @requires_cv2
+    def test_bright_image_low_brightness(self):
+        """过亮图像应扣分."""
+        # 全白图像
+        img = np.full((100, 100, 3), 255, dtype=np.uint8)
+
+        from packages.shared.cover_frame_scorer import score_frame
+
+        score = score_frame(img)
+        # 全白：清晰度 0（无边缘），亮度偏低（偏离 130），色彩 0
+        assert score <= 20.0, f"全白图像应较低分，实际: {score}"
+
+    @requires_cv2
+    def test_medium_brightness_full_score(self):
+        """亮度在 80~180 区间应得亮度满分."""
+        # 中等亮度灰色
+        img = np.full((100, 100, 3), 130, dtype=np.uint8)
+        # 加一些纹理增加清晰度
+        for i in range(0, 100, 10):
+            img[i : i + 5, :] = 180
+
+        from packages.shared.cover_frame_scorer import score_frame
+
+        score = score_frame(img)
+        # 亮度应在舒适区间
+        assert score >= 25.0, f"中等亮度图像应有一定分数，实际: {score}"
+
+    @requires_cv2
+    def test_colorful_image_high_color_score(self):
+        """色彩丰富的图像应得高色彩分."""
+        # 彩虹渐变
+        img = np.zeros((100, 100, 3), dtype=np.uint8)
+        for i in range(100):
+            img[i, :, 0] = int(i * 2.55)  # R
+            img[i, :, 1] = int((100 - i) * 2.55)  # G
+            img[:, i, 2] = int(i * 2.55)  # B
+
+        from packages.shared.cover_frame_scorer import score_frame
+
+        score = score_frame(img)
+        assert score >= 40.0, f"彩色图像应得高分，实际: {score}"
+
+    @requires_cv2
+    def test_empty_frame_returns_zero(self):
+        """空帧返回 0 分（有 cv2）或默认分（无 cv2）."""
+        from packages.shared.cover_frame_scorer import score_frame
+
+        result1 = score_frame(np.array([]))
+        result2 = score_frame(None)
+        if HAS_CV2:
+            assert result1 == 0.0
+            assert result2 == 0.0
+        else:
+            assert result1 == 50.0
+            assert result2 == 50.0
+
+    @requires_cv2
+    def test_score_range(self):
+        """评分必须在 0~100 范围内."""
+        from packages.shared.cover_frame_scorer import score_frame
+
+        # 各种极端情况
+        for val in [0, 50, 128, 200, 255]:
+            img = np.full((50, 50, 3), val, dtype=np.uint8)
+            score = score_frame(img)
+            assert 0.0 <= score <= 100.0, f"评分 {score} 超出范围 [0, 100]"
+
+
+class TestScoreFrames:
+    """score_frames 批量评分测试."""
+
+    @requires_cv2
+    def test_returns_sorted_by_score(self):
+        """返回结果应按分数从高到低排序."""
+        from packages.shared.cover_frame_scorer import score_frames
+
+        frames = [
+            {"image_array": np.full((50, 50, 3), 128, dtype=np.uint8), "id": "mid"},
+            {"image_array": np.zeros((50, 50, 3), dtype=np.uint8), "id": "dark"},
+        ]
+        # 添加一个清晰帧
+        clear = np.zeros((50, 50, 3), dtype=np.uint8)
+        for i in range(50):
+            clear[i, :, :] = i * 5
+        frames.insert(0, {"image_array": clear, "id": "clear"})
+
+        scored = score_frames(frames)
+        assert len(scored) == 3
+        # 第一个应该是分数最高的
+        assert scored[0]["score"] >= scored[1]["score"]
+        assert scored[1]["score"] >= scored[2]["score"]
+
+    @requires_cv2
+    def test_all_have_score_field(self):
+        """每个帧都应该有 score 字段."""
+        from packages.shared.cover_frame_scorer import score_frames
+
+        frames = [
+            {"image_array": np.full((30, 30, 3), 100, dtype=np.uint8)},
+            {"image_array": np.full((30, 30, 3), 200, dtype=np.uint8)},
+        ]
+        scored = score_frames(frames)
+        for f in scored:
+            assert "score" in f
+            assert isinstance(f["score"], float)
+
+    def test_empty_list(self):
+        """空列表返回空列表."""
+        from packages.shared.cover_frame_scorer import score_frames
+
+        assert score_frames([]) == []
+
+
+class TestSelectBestFrame:
+    """select_best_frame 测试."""
+
+    @requires_cv2
+    def test_returns_highest_score(self):
+        """返回分数最高的帧."""
+        from packages.shared.cover_frame_scorer import select_best_frame
+
+        frames = [
+            {"image_array": np.zeros((30, 30, 3), dtype=np.uint8), "id": "dark"},
+            {"image_array": np.full((30, 30, 3), 128, dtype=np.uint8), "id": "mid"},
+        ]
+        best = select_best_frame(frames)
+        assert best is not None
+        # 中等亮度帧应该得分更高
+        assert best["id"] == "mid"
+
+    def test_empty_returns_none(self):
+        """空列表返回 None."""
+        from packages.shared.cover_frame_scorer import select_best_frame
+
+        assert select_best_frame([]) is None
+
+    @requires_cv2
+    def test_single_frame(self):
+        """单帧直接返回."""
+        from packages.shared.cover_frame_scorer import select_best_frame
+
+        frames = [{"image_array": np.full((30, 30, 3), 128, dtype=np.uint8), "id": "only"}]
+        best = select_best_frame(frames)
+        assert best is not None
+        assert best["id"] == "only"

tests/unit/test_generation_cover.py

@@ -1318,7 +1318,7 @@ class TestCoverFromFinalVideo:
             call_kwargs = mock_mk.extract_frames.call_args.kwargs
             assert "rendered/final/video.mp4" in call_kwargs["video_url"]
             assert call_kwargs["strategy"] == "SpecifiedFrames"
-            assert call_kwargs["max_frames"] == 1
+            assert call_kwargs["max_frames"] == 5
             assert call_kwargs["max_retries"] == 0
             mock_persist.assert_called_once()