04463fbdce
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 2s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 28s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 30s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 1m38s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m48s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m46s
CI/CD Pipeline / Validate - Style (pull_request) Successful in 2m10s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m42s
CI/CD Pipeline / Validate - Security (pull_request) Successful in 4m19s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 6m39s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Failing after 4s
AI Code Review / AI Code Review (pull_request) Successful in 6m53s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 3m9s
142 lines
5.3 KiB
Python
142 lines
5.3 KiB
Python
"""封面帧质量评分器 — 评估视频帧的视觉质量,用于智能选帧。
|
||
|
||
评分维度(总分 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
|