diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index bcea7df42..766637849 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/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, diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index c52f3b2d0..ac87ab956 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/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( diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py index 4282c0027..a0d4b9dcc 100755 --- a/apps/worker/video_processing/thumbnail_generator.py +++ b/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 diff --git a/packages/shared/cover_frame_scorer.py b/packages/shared/cover_frame_scorer.py new file mode 100644 index 000000000..80de9c350 --- /dev/null +++ b/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 diff --git a/tests/unit/test_cover_frame_scorer.py b/tests/unit/test_cover_frame_scorer.py new file mode 100644 index 000000000..809d10300 --- /dev/null +++ b/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" diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index ea10796b8..96f5c260f 100644 --- a/tests/unit/test_generation_cover.py +++ b/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()