diff --git a/apps/worker/worker_app/tasks/asset_analyzer.py b/apps/worker/worker_app/tasks/asset_analyzer.py index 25f7ffb1d..686e60811 100755 --- a/apps/worker/worker_app/tasks/asset_analyzer.py +++ b/apps/worker/worker_app/tasks/asset_analyzer.py @@ -19,74 +19,21 @@ from PIL import Image from packages.domain.classification import AssetClassification +from .asset_quality_scoring import ( + AudioAnalysis, + ClassificationResult, + ColorAnalysis, + MotionAnalysis, + QualityScore, + VideoInfo, + calculate_category_scores, + calculate_quality_score, + classify_from_analysis, +) + logger = logging.getLogger(__name__) -@dataclass -class VideoInfo: - """视频基本信息""" - - width: int = 0 - height: int = 0 - fps: float = 0.0 - duration: float = 0.0 - bitrate: int = 0 - codec: str = "" - has_audio: bool = False - file_size: int = 0 - - -@dataclass -class ColorAnalysis: - """色彩分析结果""" - - dominant_hue: float = 0.0 # 主色调 (0-360) - green_ratio: float = 0.0 # 绿色占比 - warm_ratio: float = 0.0 # 暖色调占比 - cool_ratio: float = 0.0 # 冷色调占比 - avg_saturation: float = 0.0 - avg_brightness: float = 0.0 - - -@dataclass -class MotionAnalysis: - """运动分析结果""" - - motion_score: float = 0.0 # 运动幅度 (0-1) - scene_changes: int = 0 # 场景切换次数 - - -@dataclass -class AudioAnalysis: - """音频分析结果""" - - has_audio: bool = False - speech_ratio: float = 0.0 # 人声比例 - music_ratio: float = 0.0 # 音乐比例 - ambient_ratio: float = 0.0 # 环境音比例 - - -@dataclass -class ClassificationResult: - """分类结果""" - - category: AssetClassification - confidence: float - scores: dict[str, float] = field(default_factory=dict) - - -@dataclass -class QualityScore: - """质量评分结果""" - - total: float - resolution_score: float = 0.0 - fps_score: float = 0.0 - bitrate_score: float = 0.0 - clarity_score: float = 0.0 - stability_score: float = 0.0 - - class AssetAnalyzer: """ 轻量级视频素材分析器 @@ -449,316 +396,37 @@ class AssetAnalyzer: """ 综合分析得出分类结果 + 评分逻辑在 asset_quality_scoring.calculate_category_scores / classify_from_analysis + 纯函数中，此处只负责采集分析数据后委托计算。 + Returns: ClassificationResult 对象 """ - # 提取分析数据 frames = self.extract_frames() color = self.analyze_color_distribution(frames) motion = self.analyze_motion(frames) audio = self.analyze_audio() - # 计算各类别得分 - scores = self._calculate_category_scores(color, motion, audio) - - # 找最高分 - if not scores: - return ClassificationResult( - category=AssetClassification.OTHER, - confidence=0.3, - scores={}, - ) - - best_category = max(scores.items(), key=lambda x: x[1]) - category = AssetClassification(best_category[0]) - confidence = min(0.95, max(0.3, best_category[1])) - - return ClassificationResult( - category=category, - confidence=confidence, - scores=scores, - ) - - def _calculate_category_scores( - self, - color: ColorAnalysis, - motion: MotionAnalysis, - audio: AudioAnalysis, - ) -> dict[str, float]: - """ - 计算各类别的置信度得分 - - Args: - color: 色彩分析结果 - motion: 运动分析结果 - audio: 音频分析结果 - - Returns: - 各类别得分字典 - """ - scores = {} - - # 1. 风景 (scenic) - 绿色、户外、自然 - scenic_score = 0.0 - if color.green_ratio > 0.3: - scenic_score += 0.4 * color.green_ratio - if color.avg_saturation > 0.3: - scenic_score += 0.2 * color.avg_saturation - if color.avg_brightness > 0.4: - scenic_score += 0.2 - if motion.motion_score > 0.1 and motion.motion_score < 0.5: - scenic_score += 0.2 # 适度运动（如云朵、树叶） - if not audio.has_audio or audio.ambient_ratio > 0.5: - scenic_score += 0.2 # 自然环境音 - scores[AssetClassification.SCENIC.value] = min(1.0, scenic_score) - - # 2. 产品 (product) - 中等亮度、均匀色彩、低运动 - product_score = 0.0 - if 0.3 < color.avg_brightness < 0.7: - product_score += 0.3 - if color.avg_saturation < 0.5: - product_score += 0.2 - if motion.motion_score < 0.15: - product_score += 0.4 # 低运动 = 产品展示 - if color.cool_ratio > 0.3: - product_score += 0.2 # 冷色调 = 科技感 - scores[AssetClassification.PRODUCT.value] = min(1.0, product_score) - - # 3. 人物 (person) - 中等运动、有时有人声 - person_score = 0.0 - if 0.1 < motion.motion_score < 0.4: - person_score += 0.3 # 适度运动 - if audio.has_audio and audio.speech_ratio > 0.3: - person_score += 0.5 # 有人声 - if color.avg_brightness > 0.3: - person_score += 0.2 - scores[AssetClassification.PERSON.value] = min(1.0, person_score) - - # 4. 动物 (animal) - 高运动、有时自然音 - animal_score = 0.0 - if motion.motion_score > 0.3: - animal_score += 0.4 # 高运动 - if motion.scene_changes > 2: - animal_score += 0.2 - if audio.has_audio and (audio.ambient_ratio > 0.3 or audio.speech_ratio > 0.2): - animal_score += 0.3 - scores[AssetClassification.ANIMAL.value] = min(1.0, animal_score) - - # 5. 美食 (food) - 暖色调、高饱和度 - food_score = 0.0 - if color.warm_ratio > 0.4: - food_score += 0.5 - if color.avg_saturation > 0.5: - food_score += 0.3 - if 0.4 < color.avg_brightness < 0.8: - food_score += 0.2 - scores[AssetClassification.FOOD.value] = min(1.0, food_score) - - # 6. 科技 (tech) - 冷色调、低饱和度、低运动 - tech_score = 0.0 - if color.cool_ratio > 0.4: - tech_score += 0.4 - if color.avg_saturation < 0.4: - tech_score += 0.3 - if motion.motion_score < 0.2: - tech_score += 0.3 - scores[AssetClassification.TECH.value] = min(1.0, tech_score) - - # 7. 运动 (sport) - 高运动 - sport_score = 0.0 - if motion.motion_score > 0.4: - sport_score += 0.6 - if motion.scene_changes > 3: - sport_score += 0.2 - if color.avg_brightness > 0.4: - sport_score += 0.2 - scores[AssetClassification.SPORT.value] = min(1.0, sport_score) - - # 8. 音乐 (music) - 有节奏性音乐 - music_score = 0.0 - if audio.has_audio and audio.music_ratio > 0.4: - music_score += 0.6 - # 纯视觉判断：色彩丰富但非自然 - if color.avg_saturation > 0.5 and color.green_ratio < 0.2: - music_score += 0.3 - scores[AssetClassification.MUSIC.value] = min(1.0, music_score) - - # 9. 其他 (other) - 默认最低分 - scores[AssetClassification.OTHER.value] = 0.1 - - return scores + return classify_from_analysis(color, motion, audio) def calculate_quality_score(self) -> QualityScore: """ 计算视频质量综合评分 (0-100) + 评分逻辑在 asset_quality_scoring.calculate_quality_score 纯函数中， + 此处只负责采集数据后委托计算。 + 评分维度: 1. 分辨率得分 (25分) 2. 帧率得分 (20分) 3. 码率得分 (20分) - 4. 清晰度得分 (20分) - Laplacian 方差 - 5. 稳定性得分 (15分) - 帧间位移方差 + 4. 清晰度得分 (20分) + 5. 稳定性得分 (15分) """ info = self.get_video_info() frames = self.extract_frames() - # 1. 分辨率得分 - resolution_score = self._score_resolution(info.width, info.height) - - # 2. 帧率得分 - fps_score = self._score_framerate(info.fps) - - # 3. 码率得分 - bitrate_score = self._score_bitrate(info.bitrate) - - # 4. 清晰度得分 - clarity_score = self._score_clarity(frames) - - # 5. 稳定性得分 - stability_score = self._score_stability(frames) - - total = resolution_score + fps_score + bitrate_score + clarity_score + stability_score - - return QualityScore( - total=round(min(100, max(0, total)), 1), - resolution_score=resolution_score, - fps_score=fps_score, - bitrate_score=bitrate_score, - clarity_score=clarity_score, - stability_score=stability_score, - ) - - def _score_resolution(self, width: int, height: int) -> float: - """分辨率评分 (满分 25)""" - pixels = width * height - - if pixels >= 3840 * 2160: # 4K - return 25.0 - elif pixels >= 2560 * 1440: # 2K - return 22.0 - elif pixels >= 1920 * 1080: # 1080p - return 20.0 - elif pixels >= 1280 * 720: # 720p - return 15.0 - elif pixels >= 854 * 480: # 480p - return 8.0 - else: - return 3.0 - - def _score_framerate(self, fps: float) -> float: - """帧率评分 (满分 20)""" - if fps >= 60: - return 20.0 - elif fps >= 30: - return 15.0 - elif fps >= 24: - return 10.0 - elif fps >= 15: - return 7.0 - else: - return 5.0 - - def _score_bitrate(self, bitrate: int) -> float: - """码率评分 (满分 20)""" - bitrate_mbps = bitrate / 1_000_000 - - if bitrate_mbps > 10: - return 20.0 - elif bitrate_mbps >= 5: - return 15.0 - elif bitrate_mbps >= 2: - return 10.0 - elif bitrate_mbps >= 0.5: - return 5.0 - else: - return 3.0 - - def _score_clarity(self, frames: list[np.ndarray]) -> float: - """ - 清晰度评分 (满分 20) - - 使用 Laplacian 方差评估画面清晰度 - 高方差 = 细节丰富 = 高分 - """ - if not frames: - return 10.0 # 默认中等分 - - try: - variances = [] - - for frame in frames[:5]: # 只分析前 5 帧 - if len(frame.shape) == 3: - # 转灰度 - gray = np.dot(frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8) - else: - gray = frame - - # Laplacian 算子 - laplacian = np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float32) - - # 手动计算卷积 - from scipy import signal - - laplacian_img = signal.convolve2d(gray.astype(float), laplacian, mode="same") - variance = np.var(laplacian_img) - variances.append(variance) - - # 归一化方差到 0-20 分 - avg_variance = np.mean(variances) - # 根据经验值调整 - score = min(20.0, avg_variance / 100) - return float(score) - - except ImportError: - # 如果没有 scipy，使用简化方法 - return 10.0 - except Exception: - return 10.0 - - def _score_stability(self, frames: list[np.ndarray]) -> float: - """ - 稳定性评分 (满分 15) - - 分析帧间位移方差 - 画面稳定 = 高分 - 剧烈抖动 = 低分 - """ - if len(frames) < 2: - return 10.0 # 默认中等分 - - try: - displacements = [] - - for i in range(len(frames) - 1): - # 缩小帧以加速处理 - scale = 0.25 - new_h = int(frames[i].shape[0] * scale) - new_w = int(frames[i].shape[1] * scale) - frame1_small = np.array(Image.fromarray(frames[i]).resize((new_w, new_h))) - new_h2 = int(frames[i + 1].shape[0] * scale) - new_w2 = int(frames[i + 1].shape[1] * scale) - frame2_small = np.array(Image.fromarray(frames[i + 1]).resize((new_w2, new_h2))) - - # 简单位移检测：灰度差 - gray1 = np.mean(frame1_small, axis=2) if len(frame1_small.shape) == 3 else frame1_small - gray2 = np.mean(frame2_small, axis=2) if len(frame2_small.shape) == 3 else frame2_small - - diff = np.abs(gray2.astype(float) - gray1.astype(float)) - displacement = np.mean(diff) / 255.0 - displacements.append(displacement) - - # 高位移方差 = 不稳定 - if displacements: - displacement_variance = np.var(displacements) - # 归一化 - instability = min(1.0, displacement_variance * 10) - score = 15.0 * (1.0 - instability) - return float(max(0.0, score)) - - return 10.0 - - except Exception: - return 10.0 + return calculate_quality_score(info, frames) def classify_asset_real(video_path: str) -> tuple[str, float]: diff --git a/apps/worker/worker_app/tasks/asset_quality_scoring.py b/apps/worker/worker_app/tasks/asset_quality_scoring.py new file mode 100755 index 000000000..9c2d3f5c1 --- /dev/null +++ b/apps/worker/worker_app/tasks/asset_quality_scoring.py @@ -0,0 +1,459 @@ +"""素材质量与分类评分 — 纯逻辑模块. + +从 asset_analyzer.py 提取的评分计算逻辑，纯函数，无副作用。 +输入分析结果对象，输出评分/分类结果。 + +拆分目的： +1. 大文件瘦身（asset_analyzer.py 799行 → 拆出 200+ 行纯逻辑） +2. 评分逻辑可独立单测，不依赖 FFmpeg/视频文件 +3. 评分策略调整时不需要触碰分析主流程 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Optional + +import numpy as np + +from packages.domain.classification import AssetClassification + +# ── 数据类 ──────────────────────────────────────────────────────────── + + +@dataclass +class VideoInfo: + """视频基本信息""" + + width: int = 0 + height: int = 0 + fps: float = 0.0 + duration: float = 0.0 + bitrate: int = 0 + codec: str = "" + has_audio: bool = False + file_size: int = 0 + + +@dataclass +class ColorAnalysis: + """色彩分析结果""" + + dominant_hue: float = 0.0 # 主色调 (0-360) + green_ratio: float = 0.0 # 绿色占比 + warm_ratio: float = 0.0 # 暖色调占比 + cool_ratio: float = 0.0 # 冷色调占比 + avg_saturation: float = 0.0 + avg_brightness: float = 0.0 + + +@dataclass +class MotionAnalysis: + """运动分析结果""" + + motion_score: float = 0.0 # 运动幅度 (0-1) + scene_changes: int = 0 # 场景切换次数 + + +@dataclass +class AudioAnalysis: + """音频分析结果""" + + has_audio: bool = False + speech_ratio: float = 0.0 # 人声比例 + music_ratio: float = 0.0 # 音乐比例 + ambient_ratio: float = 0.0 # 环境音比例 + + +@dataclass +class ClassificationResult: + """分类结果""" + + category: AssetClassification + confidence: float + scores: dict[str, float] = field(default_factory=dict) + + +@dataclass +class QualityScore: + """质量评分结果""" + + total: float + resolution_score: float = 0.0 + fps_score: float = 0.0 + bitrate_score: float = 0.0 + clarity_score: float = 0.0 + stability_score: float = 0.0 + + +# ── 质量评分纯函数 ──────────────────────────────────────────────────── + + +def score_resolution(width: int, height: int) -> float: + """分辨率评分 (满分 25). + + 按像素总数阶梯评分：4K > 2K > 1080p > 720p > 480p > 其他. + + Args: + width: 视频宽度（像素） + height: 视频高度（像素） + + Returns: + float: 0-25 分 + """ + pixels = width * height + + if pixels >= 3840 * 2160: # 4K + return 25.0 + elif pixels >= 2560 * 1440: # 2K + return 22.0 + elif pixels >= 1920 * 1080: # 1080p + return 20.0 + elif pixels >= 1280 * 720: # 720p + return 15.0 + elif pixels >= 854 * 480: # 480p + return 8.0 + else: + return 3.0 + + +def score_framerate(fps: float) -> float: + """帧率评分 (满分 20). + + 60fps 满分，阶梯递减. + + Args: + fps: 帧率（帧/秒） + + Returns: + float: 0-20 分 + """ + if fps >= 60: + return 20.0 + elif fps >= 30: + return 15.0 + elif fps >= 24: + return 10.0 + elif fps >= 15: + return 7.0 + else: + return 5.0 + + +def score_bitrate(bitrate: int) -> float: + """码率评分 (满分 20). + + 按 Mbps 阶梯评分. + + Args: + bitrate: 码率（bps） + + Returns: + float: 0-20 分 + """ + bitrate_mbps = bitrate / 1_000_000 + + if bitrate_mbps > 10: + return 20.0 + elif bitrate_mbps >= 5: + return 15.0 + elif bitrate_mbps >= 2: + return 10.0 + elif bitrate_mbps >= 0.5: + return 5.0 + else: + return 3.0 + + +def score_clarity(frames: list[np.ndarray]) -> float: + """清晰度评分 (满分 20). + + 使用 Laplacian 方差评估画面清晰度。 + 高方差 = 细节丰富 = 高分. + + Args: + frames: 视频帧列表（numpy 数组，RGB 或灰度） + + Returns: + float: 0-20 分 + """ + if not frames: + return 10.0 # 默认中等分 + + try: + variances = [] + + for frame in frames[:5]: # 只分析前 5 帧 + if len(frame.shape) == 3: + # 转灰度 + gray = np.dot(frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8) + else: + gray = frame + + # Laplacian 算子 + laplacian = np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float32) + + # 手动计算卷积 + from scipy import signal + + laplacian_img = signal.convolve2d(gray.astype(float), laplacian, mode="same") + variance = np.var(laplacian_img) + variances.append(variance) + + # 归一化方差到 0-20 分 + avg_variance = np.mean(variances) + score = min(20.0, avg_variance / 100) + return float(score) + + except ImportError: + # 如果没有 scipy，使用简化方法 + return 10.0 + except Exception: + return 10.0 + + +def score_stability(frames: list[np.ndarray]) -> float: + """稳定性评分 (满分 15). + + 分析帧间位移方差。 + 画面稳定 = 高分；剧烈抖动 = 低分. + + Args: + frames: 视频帧列表（numpy 数组，RGB 或灰度） + + Returns: + float: 0-15 分 + """ + if len(frames) < 2: + return 10.0 # 默认中等分 + + try: + from PIL import Image + + displacements = [] + + for i in range(len(frames) - 1): + # 缩小帧以加速处理 + scale = 0.25 + new_h = int(frames[i].shape[0] * scale) + new_w = int(frames[i].shape[1] * scale) + frame1_small = np.array(Image.fromarray(frames[i]).resize((new_w, new_h))) + new_h2 = int(frames[i + 1].shape[0] * scale) + new_w2 = int(frames[i + 1].shape[1] * scale) + frame2_small = np.array(Image.fromarray(frames[i + 1]).resize((new_w2, new_h2))) + + # 简单位移检测：灰度差 + gray1 = np.mean(frame1_small, axis=2) if len(frame1_small.shape) == 3 else frame1_small + gray2 = np.mean(frame2_small, axis=2) if len(frame2_small.shape) == 3 else frame2_small + + diff = np.abs(gray2.astype(float) - gray1.astype(float)) + displacement = np.mean(diff) / 255.0 + displacements.append(displacement) + + # 高位移方差 = 不稳定 + if displacements: + displacement_variance = np.var(displacements) + # 归一化 + instability = min(1.0, displacement_variance * 10) + score = 15.0 * (1.0 - instability) + return float(max(0.0, score)) + + return 10.0 + + except Exception: + return 10.0 + + +def calculate_quality_score( + info: VideoInfo, + frames: Optional[list[np.ndarray]] = None, +) -> QualityScore: + """计算视频质量综合评分 (0-100). + + 评分维度: + 1. 分辨率得分 (25分) + 2. 帧率得分 (20分) + 3. 码率得分 (20分) + 4. 清晰度得分 (20分) - 无帧时默认10分 + 5. 稳定性得分 (15分) - 帧不足时默认10分 + + Args: + info: 视频基本信息 + frames: 采样帧列表（可选，无则清晰度/稳定性给默认分） + + Returns: + QualityScore: 各维度得分 + 总分 + """ + # 1. 分辨率得分 + resolution_score = score_resolution(info.width, info.height) + + # 2. 帧率得分 + fps_score = score_framerate(info.fps) + + # 3. 码率得分 + bitrate_score = score_bitrate(info.bitrate) + + # 4. 清晰度得分 + clarity_score = score_clarity(frames) if frames else 10.0 + + # 5. 稳定性得分 + stability_score = score_stability(frames) if frames and len(frames) >= 2 else 10.0 + + total = resolution_score + fps_score + bitrate_score + clarity_score + stability_score + + return QualityScore( + total=round(min(100.0, max(0.0, total)), 1), + resolution_score=resolution_score, + fps_score=fps_score, + bitrate_score=bitrate_score, + clarity_score=clarity_score, + stability_score=stability_score, + ) + + +# ── 分类评分纯函数 ──────────────────────────────────────────────────── + + +def calculate_category_scores( + color: ColorAnalysis, + motion: MotionAnalysis, + audio: AudioAnalysis, +) -> dict[str, float]: + """计算各类别的置信度得分. + + 9 个分类：风景、产品、人物、动物、美食、科技、运动、音乐、其他. + + Args: + color: 色彩分析结果 + motion: 运动分析结果 + audio: 音频分析结果 + + Returns: + dict[str, float]: 各分类名称 -> 得分 (0-1) + """ + scores: dict[str, float] = {} + + # 1. 风景 (scenic) - 绿色、户外、自然 + scenic_score = 0.0 + if color.green_ratio > 0.3: + scenic_score += 0.4 * color.green_ratio + if color.avg_saturation > 0.3: + scenic_score += 0.2 * color.avg_saturation + if color.avg_brightness > 0.4: + scenic_score += 0.2 + if 0.1 < motion.motion_score < 0.5: + scenic_score += 0.2 # 适度运动（如云朵、树叶） + if not audio.has_audio or audio.ambient_ratio > 0.5: + scenic_score += 0.2 # 自然环境音 + scores[AssetClassification.SCENIC.value] = min(1.0, scenic_score) + + # 2. 产品 (product) - 中等亮度、均匀色彩、低运动 + product_score = 0.0 + if 0.3 < color.avg_brightness < 0.7: + product_score += 0.3 + if color.avg_saturation < 0.5: + product_score += 0.2 + if motion.motion_score < 0.15: + product_score += 0.4 # 低运动 = 产品展示 + if color.cool_ratio > 0.3: + product_score += 0.2 # 冷色调 = 科技感 + scores[AssetClassification.PRODUCT.value] = min(1.0, product_score) + + # 3. 人物 (person) - 中等运动、有时有人声 + person_score = 0.0 + if 0.1 < motion.motion_score < 0.4: + person_score += 0.3 # 适度运动 + if audio.has_audio and audio.speech_ratio > 0.3: + person_score += 0.5 # 有人声 + if color.avg_brightness > 0.3: + person_score += 0.2 + scores[AssetClassification.PERSON.value] = min(1.0, person_score) + + # 4. 动物 (animal) - 高运动、有时自然音 + animal_score = 0.0 + if motion.motion_score > 0.3: + animal_score += 0.4 # 高运动 + if motion.scene_changes > 2: + animal_score += 0.2 + if audio.has_audio and (audio.ambient_ratio > 0.3 or audio.speech_ratio > 0.2): + animal_score += 0.3 + scores[AssetClassification.ANIMAL.value] = min(1.0, animal_score) + + # 5. 美食 (food) - 暖色调、高饱和度 + food_score = 0.0 + if color.warm_ratio > 0.4: + food_score += 0.5 + if color.avg_saturation > 0.5: + food_score += 0.3 + if 0.4 < color.avg_brightness < 0.8: + food_score += 0.2 + scores[AssetClassification.FOOD.value] = min(1.0, food_score) + + # 6. 科技 (tech) - 冷色调、低饱和度、低运动 + tech_score = 0.0 + if color.cool_ratio > 0.4: + tech_score += 0.4 + if color.avg_saturation < 0.4: + tech_score += 0.3 + if motion.motion_score < 0.2: + tech_score += 0.3 + scores[AssetClassification.TECH.value] = min(1.0, tech_score) + + # 7. 运动 (sport) - 高运动 + sport_score = 0.0 + if motion.motion_score > 0.4: + sport_score += 0.6 + if motion.scene_changes > 3: + sport_score += 0.2 + if color.avg_brightness > 0.4: + sport_score += 0.2 + scores[AssetClassification.SPORT.value] = min(1.0, sport_score) + + # 8. 音乐 (music) - 有节奏性音乐 + music_score = 0.0 + if audio.has_audio and audio.music_ratio > 0.4: + music_score += 0.6 + # 纯视觉判断：色彩丰富但非自然 + if color.avg_saturation > 0.5 and color.green_ratio < 0.2: + music_score += 0.3 + scores[AssetClassification.MUSIC.value] = min(1.0, music_score) + + # 9. 其他 (other) - 默认最低分 + scores[AssetClassification.OTHER.value] = 0.1 + + return scores + + +def classify_from_analysis( + color: ColorAnalysis, + motion: MotionAnalysis, + audio: AudioAnalysis, +) -> ClassificationResult: + """综合分析得出分类结果. + + Args: + color: 色彩分析结果 + motion: 运动分析结果 + audio: 音频分析结果 + + Returns: + ClassificationResult: 分类结果（最高分类别 + 置信度 + 全部分数） + """ + scores = calculate_category_scores(color, motion, audio) + + if not scores: + return ClassificationResult( + category=AssetClassification.OTHER, + confidence=0.3, + scores={}, + ) + + best_category = max(scores.items(), key=lambda x: x[1]) + category = AssetClassification(best_category[0]) + confidence = min(0.95, max(0.3, best_category[1])) + + return ClassificationResult( + category=category, + confidence=confidence, + scores=scores, + ) diff --git a/tests/unit/test_asset_quality_scoring.py b/tests/unit/test_asset_quality_scoring.py new file mode 100755 index 000000000..103adf697 --- /dev/null +++ b/tests/unit/test_asset_quality_scoring.py @@ -0,0 +1,667 @@ +"""asset_quality_scoring 纯逻辑单测 — 第89波. + +测试评分纯函数，不依赖 FFmpeg/视频文件。 +覆盖：分辨率评分、帧率评分、码率评分、清晰度评分、稳定性评分、 + 质量总评分、9分类评分、分类结果计算。 +""" + +import numpy as np +import pytest + +from apps.worker.worker_app.tasks.asset_quality_scoring import ( + AudioAnalysis, + ClassificationResult, + ColorAnalysis, + MotionAnalysis, + QualityScore, + VideoInfo, + calculate_category_scores, + calculate_quality_score, + classify_from_analysis, + score_bitrate, + score_clarity, + score_framerate, + score_resolution, + score_stability, +) +from packages.domain.classification import AssetClassification + +# ── 分辨率评分 ──────────────────────────────────────────────────────── + + +class TestScoreResolution: + """分辨率评分边界测试.""" + + def test_4k_full_score(self): + """4K 及以上满分 25.""" + assert score_resolution(3840, 2160) == 25.0 + assert score_resolution(4096, 2160) == 25.0 + assert score_resolution(7680, 4320) == 25.0 # 8K + + def test_2k_score(self): + """2K 档 22 分.""" + assert score_resolution(2560, 1440) == 22.0 + assert score_resolution(3000, 1600) == 22.0 + # 刚好低于 4K + assert score_resolution(3839, 2159) == 22.0 + + def test_1080p_score(self): + """1080p 档 20 分.""" + assert score_resolution(1920, 1080) == 20.0 + assert score_resolution(2000, 1080) == 20.0 + # 刚好低于 2K + assert score_resolution(2559, 1439) == 20.0 + + def test_720p_score(self): + """720p 档 15 分.""" + assert score_resolution(1280, 720) == 15.0 + assert score_resolution(1280, 720) == 15.0 + # 刚好低于 1080p + assert score_resolution(1919, 1079) == 15.0 + # 1080x720 像素数 < 1280x720，掉到 480p 档 + assert score_resolution(1080, 720) == 8.0 + + def test_480p_score(self): + """480p 档 8 分.""" + assert score_resolution(854, 480) == 8.0 + assert score_resolution(854, 480) == 8.0 + # 刚好低于 720p + assert score_resolution(1279, 719) == 8.0 + # 720x480 像素数 < 854x480，掉到最低档 + assert score_resolution(720, 480) == 3.0 + + def test_low_resolution_score(self): + """低于 480p 给 3 分.""" + assert score_resolution(640, 360) == 3.0 + assert score_resolution(320, 240) == 3.0 + assert score_resolution(0, 0) == 3.0 + + def test_non_standard_aspect_ratio(self): + """非标准宽高比按像素总数计算.""" + # 竖屏 1080x1920 像素数 = 1080p + assert score_resolution(1080, 1920) == 20.0 + # 超宽屏 + assert score_resolution(2560, 1080) == 20.0 # 像素≈2.7M < 2K(3.6M) + # 1x1 极低分辨率 + assert score_resolution(1, 1) == 3.0 + + def test_negative_values(self): + """负尺寸：负负得正按像素数算，一正一负 = 负数像素 = 最低档.""" + # 一正一负 → 负像素总数 → < 480p → 3分 + assert score_resolution(1920, -1080) == 3.0 + assert score_resolution(-1920, 1080) == 3.0 + # 都是 0 → 3分 + assert score_resolution(0, 0) == 3.0 + + +# ── 帧率评分 ────────────────────────────────────────────────────────── + + +class TestScoreFramerate: + """帧率评分边界测试.""" + + def test_60fps_full_score(self): + """60fps 及以上满分 20.""" + assert score_framerate(60) == 20.0 + assert score_framerate(120) == 20.0 + assert score_framerate(240) == 20.0 + + def test_30fps_score(self): + """30-59fps 给 15 分.""" + assert score_framerate(30) == 15.0 + assert score_framerate(59.9) == 15.0 + assert score_framerate(59) == 15.0 + + def test_24fps_score(self): + """24-29fps 给 10 分.""" + assert score_framerate(24) == 10.0 + assert score_framerate(29.97) == 10.0 + assert score_framerate(25) == 10.0 + + def test_15fps_score(self): + """15-23fps 给 7 分.""" + assert score_framerate(15) == 7.0 + assert score_framerate(23.9) == 7.0 + assert score_framerate(20) == 7.0 + + def test_low_fps_score(self): + """低于 15fps 给 5 分.""" + assert score_framerate(10) == 5.0 + assert score_framerate(1) == 5.0 + assert score_framerate(0) == 5.0 + + def test_negative_fps(self): + """负帧率按最低档.""" + assert score_framerate(-30) == 5.0 + + def test_float_fps(self): + """浮点帧率正确判断边界.""" + # 29.97 (NTSC) < 30 → 24fps 档 + assert score_framerate(29.97) == 10.0 + assert score_framerate(23.976) == 7.0 + assert score_framerate(59.94) == 15.0 # 59.94 < 60 → 30fps 档 + assert score_framerate(30.0) == 15.0 + + +# ── 码率评分 ────────────────────────────────────────────────────────── + + +class TestScoreBitrate: + """码率评分边界测试.""" + + def test_high_bitrate_full_score(self): + """10Mbps 以上满分 20.""" + assert score_bitrate(10_000_001) == 20.0 + assert score_bitrate(50_000_000) == 20.0 + assert score_bitrate(100_000_000) == 20.0 + + def test_5mbps_score(self): + """5-10Mbps 给 15 分.""" + assert score_bitrate(5_000_000) == 15.0 + assert score_bitrate(8_000_000) == 15.0 + assert score_bitrate(10_000_000) == 15.0 # 刚好 10Mbps = 不 > 10 + + def test_2mbps_score(self): + """2-5Mbps 给 10 分.""" + assert score_bitrate(2_000_000) == 10.0 + assert score_bitrate(3_000_000) == 10.0 + assert score_bitrate(4_999_999) == 10.0 + + def test_05mbps_score(self): + """0.5-2Mbps 给 5 分.""" + assert score_bitrate(500_000) == 5.0 + assert score_bitrate(1_000_000) == 5.0 + assert score_bitrate(1_999_999) == 5.0 + + def test_low_bitrate_score(self): + """低于 0.5Mbps 给 3 分.""" + assert score_bitrate(499_999) == 3.0 + assert score_bitrate(100_000) == 3.0 + assert score_bitrate(0) == 3.0 + + def test_negative_bitrate(self): + """负码率按最低档.""" + assert score_bitrate(-5_000_000) == 3.0 + + def test_zero_bitrate(self): + """0 码率 = 最低档.""" + assert score_bitrate(0) == 3.0 + + +# ── 清晰度评分 ──────────────────────────────────────────────────────── + + +class TestScoreClarity: + """清晰度评分测试.""" + + def test_empty_frames_default_score(self): + """空帧列表给默认 10 分.""" + assert score_clarity([]) == 10.0 + + def test_constant_image_low_clarity(self): + """纯色图像比高细节图像清晰度低很多.""" + # 纯灰色图像 + gray_frame = np.full((100, 100), 128, dtype=np.uint8) + score_constant = score_clarity([gray_frame]) + + # 高细节随机图像 + detail_frame = np.random.randint(0, 256, (100, 100), dtype=np.uint8) + score_detail = score_clarity([detail_frame]) + + # 纯色图应该显著低于高细节图 + assert score_constant < score_detail + assert 0.0 <= score_constant <= 20.0 + + def test_edge_rich_image_high_clarity(self): + """高频边缘图像有较高清晰度得分.""" + # 棋盘格图案，边缘丰富 + frame = np.zeros((100, 100), dtype=np.uint8) + for i in range(0, 100, 10): + for j in range(0, 100, 10): + if (i // 10 + j // 10) % 2 == 0: + frame[i : i + 10, j : j + 10] = 255 + score = score_clarity([frame]) + assert score > 1.0 # 应有一定清晰度 + assert 0.0 <= score <= 20.0 + + def test_rgb_frame_converts_to_gray(self): + """RGB 帧会被转灰度后计算.""" + rgb_frame = np.random.randint(0, 256, (50, 50, 3), dtype=np.uint8) + score_rgb = score_clarity([rgb_frame]) + # 对应灰度图 + gray = np.dot(rgb_frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8) + score_gray = score_clarity([gray]) + # 两者应近似相等 + assert abs(score_rgb - score_gray) < 0.01 + + def test_only_first_five_frames_analyzed(self): + """只分析前 5 帧.""" + # 10 帧：前 5 帧纯色，后 5 帧高对比度 + frames = [] + for _ in range(5): + frames.append(np.full((50, 50), 128, dtype=np.uint8)) + for _ in range(5): + high_freq = np.random.randint(0, 256, (50, 50), dtype=np.uint8) + frames.append(high_freq) + score_10 = score_clarity(frames) + score_5 = score_clarity(frames[:5]) + # 前 5 帧相同，得分应相同 + assert abs(score_10 - score_5) < 0.01 + + def test_score_within_bounds(self): + """得分始终在 0-20 范围内.""" + for _ in range(10): + frame = np.random.randint(0, 256, (30, 30, 3), dtype=np.uint8) + score = score_clarity([frame]) + assert 0.0 <= score <= 20.0 + + def test_multiple_frames_averaged(self): + """多帧取平均方差.""" + # 第 1 帧低细节，第 2 帧高细节 + low_detail = np.full((50, 50), 100, dtype=np.uint8) + high_detail = np.random.randint(0, 256, (50, 50), dtype=np.uint8) + + score_low = score_clarity([low_detail]) + score_high = score_clarity([high_detail]) + score_both = score_clarity([low_detail, high_detail]) + + # 混合得分应在两者之间 + assert score_low <= score_both <= score_high + + +# ── 稳定性评分 ──────────────────────────────────────────────────────── + + +class TestScoreStability: + """稳定性评分测试.""" + + def test_single_frame_default_score(self): + """不足 2 帧给默认 10 分.""" + assert score_stability([]) == 10.0 + assert score_stability([np.zeros((10, 10, 3), dtype=np.uint8)]) == 10.0 + + def test_identical_frames_max_stability(self): + """完全相同的帧 = 高稳定性.""" + frame = np.random.randint(50, 200, (100, 100, 3), dtype=np.uint8) + score = score_stability([frame, frame.copy()]) + assert score > 10.0 # 应该接近满分 15 + + def test_very_different_frames_low_stability(self): + """位移方差大的多帧序列 = 低稳定性.""" + # 构造 4 帧：帧间位移差异大（有的帧相似、有的帧完全不同） + # 位移方差大 → 不稳定 → 低分 + base = np.random.randint(100, 150, (100, 100, 3), dtype=np.uint8) + frames = [ + base, # 帧0 + base, # 帧1 (完全相同 → 位移0) + np.full_like(base, 255), # 帧2 (纯白 → 位移大) + base, # 帧3 (回到基准 → 位移又大) + ] + score = score_stability(frames) + # 位移差异大 → 方差大 → 稳定性低 + assert score < 10.0 + assert 0.0 <= score <= 15.0 + + def test_score_within_bounds(self): + """得分始终在 0-15 范围内.""" + for _ in range(10): + f1 = np.random.randint(0, 256, (40, 40, 3), dtype=np.uint8) + f2 = np.random.randint(0, 256, (40, 40, 3), dtype=np.uint8) + score = score_stability([f1, f2]) + assert 0.0 <= score <= 15.0 + + def test_gray_frames_also_work(self): + """灰度帧也能计算.""" + f1 = np.random.randint(0, 256, (50, 50), dtype=np.uint8) + f2 = np.random.randint(0, 256, (50, 50), dtype=np.uint8) + score = score_stability([f1, f2]) + assert 0.0 <= score <= 15.0 + + def test_multiple_frame_pairs(self): + """多对帧取方差.""" + base = np.random.randint(100, 150, (60, 60, 3), dtype=np.uint8) + # 5 帧相似的 + frames = [] + for i in range(5): + f = base.copy() + # 轻微变化 + f = np.clip(f.astype(int) + np.random.randint(-5, 6, f.shape), 0, 255).astype(np.uint8) + frames.append(f) + score = score_stability(frames) + assert score > 5.0 # 相似帧应该有一定稳定性 + + +# ── 质量总评分 ──────────────────────────────────────────────────────── + + +class TestCalculateQualityScore: + """质量综合评分测试.""" + + def test_perfect_video_near_100(self): + """完美参数的视频接近 100 分.""" + info = VideoInfo( + width=3840, + height=2160, + fps=60, + bitrate=20_000_000, + ) + # 用高细节帧提升清晰度分 + frame = np.random.randint(0, 256, (100, 100, 3), dtype=np.uint8) + result = calculate_quality_score(info, [frame]) + assert isinstance(result, QualityScore) + assert result.resolution_score == 25.0 + assert result.fps_score == 20.0 + assert result.bitrate_score == 20.0 + assert 50.0 <= result.total <= 100.0 + + def test_low_quality_video(self): + """低质量视频得分低.""" + info = VideoInfo( + width=320, + height=240, + fps=10, + bitrate=100_000, + ) + result = calculate_quality_score(info, []) + assert isinstance(result, QualityScore) + assert result.resolution_score == 3.0 + assert result.fps_score == 5.0 + assert result.bitrate_score == 3.0 + # 无帧时清晰度和稳定性各给 10 分默认 + assert result.clarity_score == 10.0 + assert result.stability_score == 10.0 + assert result.total == 31.0 # 3+5+3+10+10 + + def test_no_frames_uses_defaults(self): + """不传 frames 时清晰度/稳定性给默认分.""" + info = VideoInfo(width=1920, height=1080, fps=30, bitrate=5_000_000) + result = calculate_quality_score(info) + assert result.clarity_score == 10.0 + assert result.stability_score == 10.0 + assert result.resolution_score == 20.0 + assert result.fps_score == 15.0 + assert result.bitrate_score == 15.0 + assert result.total == 70.0 + + def test_total_capped_at_100(self): + """总分不超过 100.""" + info = VideoInfo( + width=7680, + height=4320, + fps=240, + bitrate=100_000_000, + ) + # 即使所有维度都满，总分不超 100 + result = calculate_quality_score(info, []) + assert result.total <= 100.0 + + def test_total_minimum_zero(self): + """总分不低于 0.""" + info = VideoInfo(width=0, height=0, fps=0, bitrate=0) + result = calculate_quality_score(info, []) + assert result.total >= 0.0 + + def test_total_is_rounded(self): + """总分保留 1 位小数.""" + info = VideoInfo(width=1920, height=1080, fps=30, bitrate=5_000_000) + result = calculate_quality_score(info, []) + # 检查是 1 位小数 + assert round(result.total, 1) == result.total + + +# ── 分类评分 ────────────────────────────────────────────────────────── + + +class TestCalculateCategoryScores: + """分类评分计算测试.""" + + def test_scenic_high_green_and_motion(self): + """绿色+适度运动+自然音 → 风景高分.""" + color = ColorAnalysis( + green_ratio=0.5, + avg_saturation=0.5, + avg_brightness=0.6, + warm_ratio=0.2, + cool_ratio=0.3, + ) + motion = MotionAnalysis(motion_score=0.3, scene_changes=1) + audio = AudioAnalysis(has_audio=True, ambient_ratio=0.7) + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.SCENIC.value] > 0.5 + assert scores[AssetClassification.SCENIC.value] <= 1.0 + + def test_product_low_motion_cool_tone(self): + """低运动+冷色调 → 产品高分.""" + color = ColorAnalysis( + avg_brightness=0.5, + avg_saturation=0.3, + cool_ratio=0.5, + green_ratio=0.1, + warm_ratio=0.2, + ) + motion = MotionAnalysis(motion_score=0.1, scene_changes=0) + audio = AudioAnalysis(has_audio=False) + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.PRODUCT.value] > 0.5 + + def test_person_with_speech(self): + """有人声+适度运动 → 人物高分.""" + color = ColorAnalysis(avg_brightness=0.5) + motion = MotionAnalysis(motion_score=0.25, scene_changes=1) + audio = AudioAnalysis(has_audio=True, speech_ratio=0.6) + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.PERSON.value] > 0.5 + + def test_animal_high_motion(self): + """高运动+多场景切换 → 动物高分.""" + color = ColorAnalysis() + motion = MotionAnalysis(motion_score=0.6, scene_changes=5) + audio = AudioAnalysis(has_audio=True, ambient_ratio=0.5) + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.ANIMAL.value] > 0.5 + + def test_food_warm_saturated(self): + """暖色调+高饱和 → 美食高分.""" + color = ColorAnalysis( + warm_ratio=0.6, + avg_saturation=0.7, + avg_brightness=0.6, + green_ratio=0.1, + cool_ratio=0.2, + ) + motion = MotionAnalysis(motion_score=0.1, scene_changes=0) + audio = AudioAnalysis(has_audio=False) + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.FOOD.value] > 0.5 + + def test_tech_cool_low_saturation(self): + """冷色调+低饱和+低运动 → 科技高分.""" + color = ColorAnalysis( + cool_ratio=0.6, + avg_saturation=0.3, + avg_brightness=0.5, + green_ratio=0.1, + warm_ratio=0.2, + ) + motion = MotionAnalysis(motion_score=0.1, scene_changes=0) + audio = AudioAnalysis(has_audio=False) + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.TECH.value] > 0.5 + + def test_sport_high_motion(self): + """高运动+多场景 → 运动高分.""" + color = ColorAnalysis(avg_brightness=0.6) + motion = MotionAnalysis(motion_score=0.7, scene_changes=5) + audio = AudioAnalysis(has_audio=False) + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.SPORT.value] > 0.5 + + def test_music_high_music_ratio(self): + """高音乐比例 → 音乐高分.""" + color = ColorAnalysis(avg_saturation=0.6, green_ratio=0.1) + motion = MotionAnalysis(motion_score=0.2) + audio = AudioAnalysis(has_audio=True, music_ratio=0.7) + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.MUSIC.value] > 0.5 + + def test_other_has_base_score(self): + """其他分类有基础分 0.1.""" + color = ColorAnalysis() + motion = MotionAnalysis() + audio = AudioAnalysis() + scores = calculate_category_scores(color, motion, audio) + assert scores[AssetClassification.OTHER.value] == 0.1 + + def test_all_scores_within_bounds(self): + """所有分类得分都在 0-1 范围内.""" + color = ColorAnalysis( + green_ratio=0.9, + warm_ratio=0.9, + cool_ratio=0.9, + avg_saturation=0.99, + avg_brightness=0.99, + ) + motion = MotionAnalysis(motion_score=0.9, scene_changes=100) + audio = AudioAnalysis( + has_audio=True, + speech_ratio=0.99, + music_ratio=0.99, + ambient_ratio=0.99, + ) + scores = calculate_category_scores(color, motion, audio) + for cat, score in scores.items(): + assert 0.0 <= score <= 1.0, f"{cat} score {score} out of bounds" + + def test_all_nine_categories_present(self): + """返回 9 个分类的得分.""" + color = ColorAnalysis() + motion = MotionAnalysis() + audio = AudioAnalysis() + scores = calculate_category_scores(color, motion, audio) + assert len(scores) == 9 + + +# ── 分类结果计算 ────────────────────────────────────────────────────── + + +class TestClassifyFromAnalysis: + """分类结果计算测试.""" + + def test_returns_classification_result(self): + """返回 ClassificationResult 对象.""" + color = ColorAnalysis() + motion = MotionAnalysis() + audio = AudioAnalysis() + result = classify_from_analysis(color, motion, audio) + assert isinstance(result, ClassificationResult) + assert isinstance(result.category, AssetClassification) + assert isinstance(result.confidence, float) + assert isinstance(result.scores, dict) + + def test_highest_score_wins(self): + """得分最高的分类胜出.""" + # 构造明显偏向风景的特征 + color = ColorAnalysis( + green_ratio=0.8, + avg_saturation=0.6, + avg_brightness=0.7, + ) + motion = MotionAnalysis(motion_score=0.3) + audio = AudioAnalysis(has_audio=True, ambient_ratio=0.8) + result = classify_from_analysis(color, motion, audio) + assert result.category == AssetClassification.SCENIC + + def test_confidence_within_bounds(self): + """置信度在 0.3-0.95 范围内.""" + color = ColorAnalysis() + motion = MotionAnalysis() + audio = AudioAnalysis() + result = classify_from_analysis(color, motion, audio) + assert 0.3 <= result.confidence <= 0.95 + + def test_confidence_capped_at_095(self): + """极高得分也被限制在 0.95.""" + color = ColorAnalysis( + green_ratio=0.9, + avg_saturation=0.9, + avg_brightness=0.9, + warm_ratio=0.9, + ) + motion = MotionAnalysis(motion_score=0.9, scene_changes=10) + audio = AudioAnalysis( + has_audio=True, + speech_ratio=0.9, + music_ratio=0.9, + ambient_ratio=0.9, + ) + result = classify_from_analysis(color, motion, audio) + assert result.confidence <= 0.95 + + def test_confidence_floored_at_03(self): + """极低得分也有 0.3 最低置信度.""" + color = ColorAnalysis() + motion = MotionAnalysis() + audio = AudioAnalysis() + result = classify_from_analysis(color, motion, audio) + assert result.confidence >= 0.3 + + def test_scores_dict_included(self): + """结果中包含完整分数字典.""" + color = ColorAnalysis() + motion = MotionAnalysis() + audio = AudioAnalysis() + result = classify_from_analysis(color, motion, audio) + assert len(result.scores) == 9 + assert AssetClassification.OTHER.value in result.scores + + def test_food_category_wins_on_warm_colors(self): + """暖色调+高饱和 → 美食分类胜出.""" + color = ColorAnalysis( + warm_ratio=0.7, + avg_saturation=0.8, + avg_brightness=0.6, + green_ratio=0.05, + ) + motion = MotionAnalysis(motion_score=0.05) + audio = AudioAnalysis(has_audio=False) + result = classify_from_analysis(color, motion, audio) + assert result.category == AssetClassification.FOOD + + +# ── 数据类默认值 ────────────────────────────────────────────────────── + + +class TestDataclassDefaults: + """数据类默认值测试.""" + + def test_video_info_defaults(self): + info = VideoInfo() + assert info.width == 0 + assert info.height == 0 + assert info.fps == 0.0 + assert info.bitrate == 0 + assert info.has_audio is False + + def test_color_analysis_defaults(self): + color = ColorAnalysis() + assert color.green_ratio == 0.0 + assert color.avg_brightness == 0.0 + assert color.dominant_hue == 0.0 + + def test_motion_analysis_defaults(self): + motion = MotionAnalysis() + assert motion.motion_score == 0.0 + assert motion.scene_changes == 0 + + def test_audio_analysis_defaults(self): + audio = AudioAnalysis() + assert audio.has_audio is False + assert audio.speech_ratio == 0.0 + assert audio.music_ratio == 0.0 + + def test_quality_score_requires_total(self): + with pytest.raises(TypeError): + QualityScore() + qs = QualityScore(total=50.0) + assert qs.total == 50.0 + assert qs.resolution_score == 0.0