From 3be0aca709ef01e3d494b441903d7ea1a4edafb6 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 26 Jul 2026 16:29:39 +0800 Subject: [PATCH 1/3] =?UTF-8?q?refactor+test(asset):=20=E7=AC=AC89?= =?UTF-8?q?=E6=B3=A2=20-=20=E6=8B=86=E5=88=86asset=5Fanalyzer=E8=AF=84?= =?UTF-8?q?=E5=88=86=E7=BA=AF=E9=80=BB=E8=BE=91=20+=2064=E5=8D=95=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 从799行asset_analyzer.py拆出asset_quality_scoring.py纯逻辑模块 - 质量评分：分辨率/帧率/码率/清晰度/稳定性 5个维度纯函数 - 分类评分：9个分类（风景/产品/人物/动物/美食/科技/运动/音乐/其他） - 64个单测全覆盖：边界值、numpy数组、分类组合、数据类 - 原文件从799行瘦身到467行（-332行） - 向后兼容：classify_asset_real / calculate_quality_score_real 入口不变 --- .../worker/worker_app/tasks/asset_analyzer.py | 376 +--------- .../worker_app/tasks/asset_quality_scoring.py | 459 ++++++++++++ tests/unit/test_asset_quality_scoring.py | 667 ++++++++++++++++++ 3 files changed, 1148 insertions(+), 354 deletions(-) create mode 100755 apps/worker/worker_app/tasks/asset_quality_scoring.py create mode 100755 tests/unit/test_asset_quality_scoring.py 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 -- 2.54.0 From 5354093417f0863dc0b74244fcdfed9438d9d078 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 26 Jul 2026 16:48:25 +0800 Subject: [PATCH 2/3] test(wave90): add 52 unit tests for plan_generator_utils + extract from plan_generator_service - Extract pure logic from plan_generator_service.py to packages/domain/plan_generator_utils.py - distribute_assets: 4-mode asset distribution (ONE_TAKE/PIP/VOICE_OVER/VOICE_PIP) - map_clip_types_for_mode: clip type mapping based on generation mode - generate_default_clips: default clip structure generation per mode - create_clips_from_configs: clip creation from config list with duration calculation - Constants: DEFAULT_DURATION, mode clip configs - Original file: 437 lines -> 209 lines (-228, -52%) - 52 unit tests covering all functions and edge cases - black formatted --- .../app/services/plan_generator_service.py | 268 +------ packages/domain/plan_generator_utils.py | 347 +++++++++ tests/unit/test_plan_generator_utils.py | 670 ++++++++++++++++++ 3 files changed, 1037 insertions(+), 248 deletions(-) create mode 100755 packages/domain/plan_generator_utils.py create mode 100755 tests/unit/test_plan_generator_utils.py diff --git a/apps/api/app/services/plan_generator_service.py b/apps/api/app/services/plan_generator_service.py index 57d986f7d..0bf79f168 100755 --- a/apps/api/app/services/plan_generator_service.py +++ b/apps/api/app/services/plan_generator_service.py @@ -26,6 +26,13 @@ from packages.domain.edit_plan import EditPlan from packages.domain.edit_plan_clip import EditPlanClip from packages.domain.edit_template import EditTemplate from packages.domain.editing_mode import EditingMode +from packages.domain.plan_generator_utils import ( + DEFAULT_CLIP_DURATION, + create_clips_from_configs, + distribute_assets, + generate_default_clips, + map_clip_types_for_mode, +) from packages.domain.template_clip_config import ClipType, TemplateClipConfig logger = logging.getLogger(__name__) @@ -163,83 +170,18 @@ class PlanGeneratorService: plan_id: str, clip_configs: List[TemplateClipConfig], ) -> List[EditPlanClip]: - """从 TemplateClipConfig 列表创建 EditPlanClip 列表（未持久化）""" - clips: List[EditPlanClip] = [] - # 按 order 排序 - sorted_configs = sorted(clip_configs, key=lambda c: c.order) + """从 TemplateClipConfig 列表创建 EditPlanClip 列表（未持久化）. - for cfg in sorted_configs: - # 计算时长：取 min_duration 和 max_duration 的中间值 - if cfg.min_duration > 0 and cfg.max_duration > 0: - duration = (cfg.min_duration + cfg.max_duration) / 2 - elif cfg.min_duration > 0: - duration = cfg.min_duration - elif cfg.max_duration > 0: - duration = cfg.max_duration - else: - duration = _DEFAULT_CLIP_DURATION - - # clip_type 可能是枚举或字符串 - clip_type = cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type - - # transition_effect 可能是枚举或字符串 - transition = ( - cfg.transition_effect.value if hasattr(cfg.transition_effect, "value") else cfg.transition_effect - ) - - # 从 clip config 中解析 playback_speed（兼容 speed_ratio 字段名） - clip_cfg = cfg.config or {} - playback_speed = clip_cfg.get("playback_speed", clip_cfg.get("speed_ratio", 1.0)) or 1.0 - - clip = EditPlanClip.create( - plan_id=plan_id, - clip_type=clip_type, - order=cfg.order, - template_clip_config_id=cfg.id, - text_content=getattr(cfg, "text_template", "") or "", - duration=duration, - transition_effect=transition or "cut", - playback_speed=playback_speed, - config=clip_cfg, - ) - clips.append(clip) - - return clips + 委托给 plan_generator_utils.create_clips_from_configs 纯函数。 + """ + return create_clips_from_configs(plan_id, clip_configs) def _map_clip_types_for_mode(self, clips: List[EditPlanClip], editing_mode: str) -> None: - """将模板 clip_config 生成的 MAIN 类型片段，按 editing_mode 映射为对应角色类型。 + """将 MAIN 类型片段按 editing_mode 映射为对应角色类型. - 模板的 clip_config 使用 ClipType 枚举（main/intro/outro 等）， - 但 PIP / VOICE_PIP 模式的素材分配和渲染分层依赖特定的 clip_type 命名 - （overlay / background / corner_voice / b_roll）。 - - 映射规则（仅修改 MAIN 类型片段，非 MAIN 片段保持原类型）： - - PIP: 第1个 MAIN → main（背景），其余 MAIN → overlay（画中画） - - VOICE_PIP: 第1个 → background，第2个 → corner_voice，第3+个 → b_roll - - ONE_TAKE / VOICE_OVER: 保持 main 不变 + 委托给 plan_generator_utils.map_clip_types_for_mode 纯函数。 """ - from packages.domain.template_clip_config import ClipType - - main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] - if not main_clips: - return - - if editing_mode == EditingMode.PIP.value: - # 第1个 main 保持（背景层），其余改为 overlay（画中画层） - for i, clip in enumerate(main_clips): - if i > 0: - clip.clip_type = "overlay" - - elif editing_mode == EditingMode.VOICE_PIP.value: - for i, clip in enumerate(main_clips): - if i == 0: - clip.clip_type = "background" - elif i == 1: - clip.clip_type = "corner_voice" - else: - clip.clip_type = "b_roll" - - # ONE_TAKE / VOICE_OVER: 保持 main 不变，无需处理 + map_clip_types_for_mode(clips, editing_mode) def _generate_default_clips( self, @@ -247,101 +189,11 @@ class PlanGeneratorService: editing_mode: str, asset_count: int, ) -> List[EditPlanClip]: - """无 clip_configs 时，根据 editing_mode 生成默认 clip 结构 + """无 clip_configs 时，根据 editing_mode 生成默认 clip 结构. - - ONE_TAKE: N 个 main clips（N = asset_count，至少1个） - - PIP: 1 个 main + (N-1) 个 overlay（N = asset_count） - - VOICE_OVER: N 个 main clips + 标记需要配音 - - VOICE_PIP: 1 个 background + 1 个 corner_voice + (N-2) 个 b_roll + 委托给 plan_generator_utils.generate_default_clips 纯函数。 """ - n = max(asset_count, 1) - clips: List[EditPlanClip] = [] - order = 0 - - if editing_mode == EditingMode.PIP.value: - # 1 个 main（全屏背景） - clips.append( - EditPlanClip.create( - plan_id=plan_id, - clip_type=ClipType.MAIN.value, - order=order, - duration=_DEFAULT_CLIP_DURATION, - ) - ) - order += 1 - # 剩余为 overlay - for _ in range(1, n): - clips.append( - EditPlanClip.create( - plan_id=plan_id, - clip_type="overlay", - order=order, - duration=_DEFAULT_CLIP_DURATION, - ) - ) - order += 1 - - elif editing_mode == EditingMode.VOICE_OVER.value: - # N 个 main clips（B-roll） - for _ in range(n): - clips.append( - EditPlanClip.create( - plan_id=plan_id, - clip_type=ClipType.MAIN.value, - order=order, - duration=_DEFAULT_CLIP_DURATION, - config={"role": "b_roll"}, - ) - ) - order += 1 - - elif editing_mode == EditingMode.VOICE_PIP.value: - # 1 个 background - clips.append( - EditPlanClip.create( - plan_id=plan_id, - clip_type="background", - order=order, - duration=_DEFAULT_CLIP_DURATION, - ) - ) - order += 1 - # 1 个 corner_voice - clips.append( - EditPlanClip.create( - plan_id=plan_id, - clip_type="corner_voice", - order=order, - duration=_DEFAULT_CLIP_DURATION, - ) - ) - order += 1 - # 剩余为 b_roll - for _ in range(2, n): - clips.append( - EditPlanClip.create( - plan_id=plan_id, - clip_type="b_roll", - order=order, - duration=_DEFAULT_CLIP_DURATION, - ) - ) - order += 1 - - else: - # ONE_TAKE: N 个 main clips - for _ in range(n): - clips.append( - EditPlanClip.create( - plan_id=plan_id, - clip_type=ClipType.MAIN.value, - order=order, - duration=_DEFAULT_CLIP_DURATION, - ) - ) - order += 1 - - return clips + return generate_default_clips(plan_id, editing_mode, asset_count) def _distribute_assets( self, @@ -349,89 +201,9 @@ class PlanGeneratorService: asset_ids: List[str], editing_mode: str, ) -> None: - """按 editing_mode 将素材分配到 clips（就地修改，未持久化） + """按 editing_mode 将素材分配到 clips（就地修改，未持久化）. - 分配策略: - - ONE_TAKE: 素材按顺序依次分配给 main 类型 clips - - PIP: 第1个素材→main(全屏背景)，其余→交替分配给 overlay clips - - VOICE_OVER: 素材→main clips (B-roll) - - VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll + 委托给 plan_generator_utils.distribute_assets 纯函数。 """ - if not asset_ids or not clips: - return + distribute_assets(clips, asset_ids, editing_mode) - if editing_mode == EditingMode.ONE_TAKE.value: - self._distribute_one_take(clips, asset_ids) - elif editing_mode == EditingMode.PIP.value: - self._distribute_pip(clips, asset_ids) - elif editing_mode == EditingMode.VOICE_OVER.value: - self._distribute_voice_over(clips, asset_ids) - elif editing_mode == EditingMode.VOICE_PIP.value: - self._distribute_voice_pip(clips, asset_ids) - else: - # 未知模式，退化为 one_take - self._distribute_one_take(clips, asset_ids) - - def _distribute_one_take( - self, - clips: List[EditPlanClip], - asset_ids: List[str], - ) -> None: - """ONE_TAKE: 素材按顺序依次分配给 main 类型 clips""" - main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] - for i, clip in enumerate(main_clips): - if i < len(asset_ids): - clip.assign_asset(asset_ids[i]) - - def _distribute_pip( - self, - clips: List[EditPlanClip], - asset_ids: List[str], - ) -> None: - """PIP: 第1个素材→main(全屏背景)，其余→overlay clips""" - # 第1个素材 → main clip - main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] - if main_clips and asset_ids: - main_clips[0].assign_asset(asset_ids[0]) - - # 其余素材 → overlay clips - overlay_clips = [c for c in clips if c.clip_type == "overlay"] - remaining = asset_ids[1:] - for i, clip in enumerate(overlay_clips): - if i < len(remaining): - clip.assign_asset(remaining[i]) - - def _distribute_voice_over( - self, - clips: List[EditPlanClip], - asset_ids: List[str], - ) -> None: - """VOICE_OVER: 素材→main clips (B-roll)""" - main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] - for i, clip in enumerate(main_clips): - if i < len(asset_ids): - clip.assign_asset(asset_ids[i]) - - def _distribute_voice_pip( - self, - clips: List[EditPlanClip], - asset_ids: List[str], - ) -> None: - """VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll""" - bg_clips = [c for c in clips if c.clip_type == "background"] - corner_clips = [c for c in clips if c.clip_type == "corner_voice"] - broll_clips = [c for c in clips if c.clip_type == "b_roll"] - - # 第1个素材 → background - if bg_clips and len(asset_ids) > 0: - bg_clips[0].assign_asset(asset_ids[0]) - - # 第2个素材 → corner_voice - if corner_clips and len(asset_ids) > 1: - corner_clips[0].assign_asset(asset_ids[1]) - - # 其余素材 → b_roll - remaining = asset_ids[2:] - for i, clip in enumerate(broll_clips): - if i < len(remaining): - clip.assign_asset(remaining[i]) diff --git a/packages/domain/plan_generator_utils.py b/packages/domain/plan_generator_utils.py new file mode 100755 index 000000000..4bca02417 --- /dev/null +++ b/packages/domain/plan_generator_utils.py @@ -0,0 +1,347 @@ +"""剪辑计划生成 — 纯逻辑工具函数. + +从 PlanGeneratorService 提取的纯业务逻辑： +- 素材分配策略（4 种 editing_mode） +- 默认 clip 结构生成 +- clip_type 按模式映射 +- 从 TemplateClipConfig 创建 EditPlanClip + +纯函数，无副作用，不依赖 DB/外部服务。 +""" + +from __future__ import annotations + +from typing import List + +from packages.domain.edit_plan_clip import EditPlanClip +from packages.domain.editing_mode import EditingMode +from packages.domain.template_clip_config import ClipType, TemplateClipConfig + +# ── 默认片段时长（秒） ──────────────────────────────────────────────────────── +DEFAULT_CLIP_DURATION = 5.0 +DEFAULT_INTRO_DURATION = 3.0 +DEFAULT_OUTRO_DURATION = 3.0 + + +# ── 素材分配 ──────────────────────────────────────────────────────────────── + + +def distribute_assets( + clips: List[EditPlanClip], + asset_ids: List[str], + editing_mode: str, +) -> None: + """按 editing_mode 将素材分配到 clips（就地修改）. + + 分配策略: + - ONE_TAKE: 素材按顺序依次分配给 main 类型 clips + - PIP: 第1个素材→main(全屏背景)，其余→overlay clips + - VOICE_OVER: 素材→main clips (B-roll) + - VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll + + Args: + clips: 剪辑片段列表（就地修改 asset_id） + asset_ids: 素材 ID 列表 + editing_mode: 剪辑模式字符串 + """ + if not asset_ids or not clips: + return + + if editing_mode == EditingMode.ONE_TAKE.value: + _distribute_one_take(clips, asset_ids) + elif editing_mode == EditingMode.PIP.value: + _distribute_pip(clips, asset_ids) + elif editing_mode == EditingMode.VOICE_OVER.value: + _distribute_voice_over(clips, asset_ids) + elif editing_mode == EditingMode.VOICE_PIP.value: + _distribute_voice_pip(clips, asset_ids) + else: + # 未知模式，退化为 one_take + _distribute_one_take(clips, asset_ids) + + +def _distribute_one_take( + clips: List[EditPlanClip], + asset_ids: List[str], +) -> None: + """ONE_TAKE: 素材按顺序依次分配给 main 类型 clips.""" + main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] + for i, clip in enumerate(main_clips): + if i < len(asset_ids): + clip.assign_asset(asset_ids[i]) + + +def _distribute_pip( + clips: List[EditPlanClip], + asset_ids: List[str], +) -> None: + """PIP: 第1个素材→main(全屏背景)，其余→overlay clips.""" + # 第1个素材 → main clip + main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] + if main_clips and asset_ids: + main_clips[0].assign_asset(asset_ids[0]) + + # 其余素材 → overlay clips + overlay_clips = [c for c in clips if c.clip_type == "overlay"] + remaining = asset_ids[1:] + for i, clip in enumerate(overlay_clips): + if i < len(remaining): + clip.assign_asset(remaining[i]) + + +def _distribute_voice_over( + clips: List[EditPlanClip], + asset_ids: List[str], +) -> None: + """VOICE_OVER: 素材→main clips (B-roll).""" + main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] + for i, clip in enumerate(main_clips): + if i < len(asset_ids): + clip.assign_asset(asset_ids[i]) + + +def _distribute_voice_pip( + clips: List[EditPlanClip], + asset_ids: List[str], +) -> None: + """VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll.""" + bg_clips = [c for c in clips if c.clip_type == "background"] + voice_clips = [c for c in clips if c.clip_type == "corner_voice"] + broll_clips = [c for c in clips if c.clip_type == "b_roll"] + + idx = 0 + + # 第1个 → background + if idx < len(asset_ids) and bg_clips: + bg_clips[0].assign_asset(asset_ids[idx]) + idx += 1 + + # 第2个 → corner_voice + if idx < len(asset_ids) and voice_clips: + voice_clips[0].assign_asset(asset_ids[idx]) + idx += 1 + + # 剩余 → b_roll clips + remaining = asset_ids[idx:] + for i, clip in enumerate(broll_clips): + if i < len(remaining): + clip.assign_asset(remaining[i]) + + +# ── clip_type 映射 ──────────────────────────────────────────────────────── + + +def map_clip_types_for_mode( + clips: List[EditPlanClip], + editing_mode: str, +) -> None: + """将 MAIN 类型片段按 editing_mode 映射为对应角色类型. + + 模板的 clip_config 使用 ClipType 枚举（main/intro/outro 等）， + 但 PIP / VOICE_PIP 模式的素材分配和渲染分层依赖特定的 clip_type 命名 + （overlay / background / corner_voice / b_roll）。 + + 映射规则（仅修改 MAIN 类型片段，非 MAIN 片段保持原类型）： + - PIP: 第1个 MAIN → main（背景），其余 MAIN → overlay（画中画） + - VOICE_PIP: 第1个 → background，第2个 → corner_voice，第3+个 → b_roll + - ONE_TAKE / VOICE_OVER: 保持 main 不变 + + Args: + clips: 剪辑片段列表（就地修改 clip_type） + editing_mode: 剪辑模式字符串 + """ + main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] + if not main_clips: + return + + if editing_mode == EditingMode.PIP.value: + # 第1个 main 保持（背景层），其余改为 overlay（画中画层） + for i, clip in enumerate(main_clips): + if i > 0: + clip.clip_type = "overlay" + + elif editing_mode == EditingMode.VOICE_PIP.value: + for i, clip in enumerate(main_clips): + if i == 0: + clip.clip_type = "background" + elif i == 1: + clip.clip_type = "corner_voice" + else: + clip.clip_type = "b_roll" + + # ONE_TAKE / VOICE_OVER: 保持 main 不变，无需处理 + + +# ── 默认 clip 生成 ──────────────────────────────────────────────────────── + + +def generate_default_clips( + plan_id: str, + editing_mode: str, + asset_count: int, +) -> List[EditPlanClip]: + """无 clip_configs 时，根据 editing_mode 生成默认 clip 结构. + + - ONE_TAKE: N 个 main clips（N = asset_count，至少1个） + - PIP: 1 个 main + (N-1) 个 overlay（N = asset_count） + - VOICE_OVER: N 个 main clips + 标记需要配音 + - VOICE_PIP: 1 个 background + 1 个 corner_voice + (N-2) 个 b_roll + + Args: + plan_id: 剪辑计划 ID + editing_mode: 剪辑模式字符串 + asset_count: 素材数量 + + Returns: + List[EditPlanClip]: 生成的默认剪辑片段列表 + """ + n = max(asset_count, 1) + clips: List[EditPlanClip] = [] + order = 0 + + if editing_mode == EditingMode.PIP.value: + # 1 个 main（全屏背景） + clips.append( + EditPlanClip.create( + plan_id=plan_id, + clip_type=ClipType.MAIN.value, + order=order, + duration=DEFAULT_CLIP_DURATION, + ) + ) + order += 1 + # 剩余为 overlay + for _ in range(1, n): + clips.append( + EditPlanClip.create( + plan_id=plan_id, + clip_type="overlay", + order=order, + duration=DEFAULT_CLIP_DURATION, + ) + ) + order += 1 + + elif editing_mode == EditingMode.VOICE_OVER.value: + # N 个 main clips（B-roll） + for _ in range(n): + clips.append( + EditPlanClip.create( + plan_id=plan_id, + clip_type=ClipType.MAIN.value, + order=order, + duration=DEFAULT_CLIP_DURATION, + config={"role": "b_roll"}, + ) + ) + order += 1 + + elif editing_mode == EditingMode.VOICE_PIP.value: + # 1 个 background + clips.append( + EditPlanClip.create( + plan_id=plan_id, + clip_type="background", + order=order, + duration=DEFAULT_CLIP_DURATION, + ) + ) + order += 1 + # 1 个 corner_voice（至少有1个素材就有） + if n >= 2: + clips.append( + EditPlanClip.create( + plan_id=plan_id, + clip_type="corner_voice", + order=order, + duration=DEFAULT_CLIP_DURATION, + ) + ) + order += 1 + # 剩余为 b_roll + for _ in range(2, n): + clips.append( + EditPlanClip.create( + plan_id=plan_id, + clip_type="b_roll", + order=order, + duration=DEFAULT_CLIP_DURATION, + ) + ) + order += 1 + + else: + # ONE_TAKE / 未知模式：N 个 main clips + for _ in range(n): + clips.append( + EditPlanClip.create( + plan_id=plan_id, + clip_type=ClipType.MAIN.value, + order=order, + duration=DEFAULT_CLIP_DURATION, + ) + ) + order += 1 + + return clips + + +# ── 从配置创建 clips ──────────────────────────────────────────────────────── + + +def create_clips_from_configs( + plan_id: str, + clip_configs: List[TemplateClipConfig], +) -> List[EditPlanClip]: + """从 TemplateClipConfig 列表创建 EditPlanClip 列表. + + Args: + plan_id: 剪辑计划 ID + clip_configs: 模板片段配置列表 + + Returns: + List[EditPlanClip]: 创建的剪辑片段列表（按 order 排序） + """ + clips: List[EditPlanClip] = [] + # 按 order 排序 + sorted_configs = sorted(clip_configs, key=lambda c: c.order) + + for cfg in sorted_configs: + # 计算时长：取 min_duration 和 max_duration 的中间值 + if cfg.min_duration > 0 and cfg.max_duration > 0: + duration = (cfg.min_duration + cfg.max_duration) / 2 + elif cfg.min_duration > 0: + duration = cfg.min_duration + elif cfg.max_duration > 0: + duration = cfg.max_duration + else: + duration = DEFAULT_CLIP_DURATION + + # clip_type 可能是枚举或字符串 + clip_type = cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type + + # transition_effect 可能是枚举或字符串 + transition = ( + cfg.transition_effect.value + if hasattr(cfg.transition_effect, "value") + else cfg.transition_effect + ) + + # 从 clip config 中解析 playback_speed（兼容 speed_ratio 字段名） + clip_cfg = cfg.config or {} + playback_speed = clip_cfg.get("playback_speed", clip_cfg.get("speed_ratio", 1.0)) or 1.0 + + clip = EditPlanClip.create( + plan_id=plan_id, + clip_type=clip_type, + order=cfg.order, + template_clip_config_id=cfg.id, + text_content=getattr(cfg, "text_template", "") or "", + duration=duration, + transition_effect=transition or "cut", + playback_speed=playback_speed, + config=clip_cfg, + ) + clips.append(clip) + + return clips diff --git a/tests/unit/test_plan_generator_utils.py b/tests/unit/test_plan_generator_utils.py new file mode 100755 index 000000000..be7a15d27 --- /dev/null +++ b/tests/unit/test_plan_generator_utils.py @@ -0,0 +1,670 @@ +"""plan_generator_utils 纯逻辑单测 — 第90波. + +测试素材分配、clip_type映射、默认clip生成、配置转clip等纯函数。 +不依赖 DB，使用领域对象直接构造。 +""" + +import pytest + +from packages.domain.edit_plan_clip import EditPlanClip +from packages.domain.editing_mode import EditingMode +from packages.domain.plan_generator_utils import ( + DEFAULT_CLIP_DURATION, + create_clips_from_configs, + distribute_assets, + generate_default_clips, + map_clip_types_for_mode, +) +from packages.domain.template_clip_config import ClipType, TemplateClipConfig + + +# ── 辅助函数 ────────────────────────────────────────────────────────── + + +def _make_main_clip(plan_id: str = "plan1", order: int = 0) -> EditPlanClip: + """创建一个 MAIN 类型的 clip.""" + return EditPlanClip.create( + plan_id=plan_id, + clip_type=ClipType.MAIN.value, + order=order, + duration=5.0, + ) + + +def _make_clips(n: int, clip_type: str = "main") -> list[EditPlanClip]: + """创建 n 个指定类型的 clip.""" + return [ + EditPlanClip.create( + plan_id="plan1", + clip_type=clip_type, + order=i, + duration=5.0, + ) + for i in range(n) + ] + + +def _collect_asset_ids(clips: list[EditPlanClip]) -> list[str]: + """按顺序收集 clips 的 asset_id（空的跳过）.""" + return [c.asset_id for c in clips if c.asset_id] + + +# ── distribute_assets: ONE_TAKE ───────────────────────────────────── + + +class TestDistributeOneTake: + """ONE_TAKE 模式素材分配.""" + + def test_equal_count(self): + """素材数 == clip 数：一一对应.""" + clips = _make_clips(3) + distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.ONE_TAKE.value) + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "a2" + assert clips[2].asset_id == "a3" + + def test_more_assets_than_clips(self): + """素材多于 clip：多余的不用.""" + clips = _make_clips(2) + distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.ONE_TAKE.value) + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "a2" + + def test_fewer_assets_than_clips(self): + """素材少于 clip：后面的 clip 没素材.""" + clips = _make_clips(5) + distribute_assets(clips, ["a1", "a2"], EditingMode.ONE_TAKE.value) + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "a2" + assert clips[2].asset_id == "" + assert clips[3].asset_id == "" + assert clips[4].asset_id == "" + + def test_empty_assets(self): + """空素材列表：无分配.""" + clips = _make_clips(3) + distribute_assets(clips, [], EditingMode.ONE_TAKE.value) + for c in clips: + assert c.asset_id == "" + + def test_empty_clips(self): + """空 clip 列表：不报错.""" + distribute_assets([], ["a1"], EditingMode.ONE_TAKE.value) + + def test_only_main_clips_get_assigned(self): + """只分配给 MAIN 类型 clip，其他类型不受影响.""" + clips = _make_clips(2) + _make_clips(2, "intro") + _make_clips(2, "outro") + distribute_assets(clips, ["a1", "a2", "a3", "a4"], EditingMode.ONE_TAKE.value) + mains = [c for c in clips if c.clip_type == "main"] + others = [c for c in clips if c.clip_type != "main"] + assert mains[0].asset_id == "a1" + assert mains[1].asset_id == "a2" + for c in others: + assert c.asset_id == "" + + +# ── distribute_assets: PIP ────────────────────────────────────────── + + +class TestDistributePip: + """PIP 模式素材分配.""" + + def test_basic_pip_distribution(self): + """第1个素材给 main，其余给 overlay.""" + clips = _make_clips(1) + _make_clips(3, "overlay") + distribute_assets(clips, ["a1", "a2", "a3", "a4"], EditingMode.PIP.value) + mains = [c for c in clips if c.clip_type == "main"] + overlays = [c for c in clips if c.clip_type == "overlay"] + assert mains[0].asset_id == "a1" + assert overlays[0].asset_id == "a2" + assert overlays[1].asset_id == "a3" + assert overlays[2].asset_id == "a4" + + def test_single_asset_only_main(self): + """只有1个素材：只分配给 main，overlay 没素材.""" + clips = _make_clips(1) + _make_clips(2, "overlay") + distribute_assets(clips, ["a1"], EditingMode.PIP.value) + mains = [c for c in clips if c.clip_type == "main"] + overlays = [c for c in clips if c.clip_type == "overlay"] + assert mains[0].asset_id == "a1" + assert overlays[0].asset_id == "" + assert overlays[1].asset_id == "" + + def test_more_overlays_than_assets(self): + """overlay 多于剩余素材：后面的 overlay 没素材.""" + clips = _make_clips(1) + _make_clips(5, "overlay") + distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.PIP.value) + overlays = [c for c in clips if c.clip_type == "overlay"] + assert overlays[0].asset_id == "a2" + assert overlays[1].asset_id == "a3" + assert overlays[2].asset_id == "" + assert overlays[3].asset_id == "" + assert overlays[4].asset_id == "" + + def test_no_main_clip(self): + """没有 main clip：第1个素材没人拿，overlay 从第2个素材开始.""" + clips = _make_clips(3, "overlay") + distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.PIP.value) + overlays = [c for c in clips if c.clip_type == "overlay"] + # PIP 逻辑：先给 main 分配第1个素材（没有 main 则跳过）， + # 剩余从第2个开始分配给 overlay + assert overlays[0].asset_id == "a2" + assert overlays[1].asset_id == "a3" + assert overlays[2].asset_id == "" + + +# ── distribute_assets: VOICE_OVER ─────────────────────────────────── + + +class TestDistributeVoiceOver: + """VOICE_OVER 模式素材分配.""" + + def test_voice_over_same_as_one_take(self): + """VOICE_OVER 和 ONE_TAKE 分配策略相同：按顺序给 main.""" + clips = _make_clips(3) + assets = ["a1", "a2", "a3"] + distribute_assets(clips, assets, EditingMode.VOICE_OVER.value) + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "a2" + assert clips[2].asset_id == "a3" + + def test_voice_over_fewer_assets(self): + """素材不足时，后面的 main clip 没素材.""" + clips = _make_clips(5) + distribute_assets(clips, ["a1"], EditingMode.VOICE_OVER.value) + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "" + + +# ── distribute_assets: VOICE_PIP ──────────────────────────────────── + + +class TestDistributeVoicePip: + """VOICE_PIP 模式素材分配.""" + + def test_three_assets_full_distribution(self): + """3个素材：background + corner_voice + b_roll 各一个.""" + clips = ( + _make_clips(1, "background") + + _make_clips(1, "corner_voice") + + _make_clips(1, "b_roll") + ) + distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.VOICE_PIP.value) + bgs = [c for c in clips if c.clip_type == "background"] + voices = [c for c in clips if c.clip_type == "corner_voice"] + brolls = [c for c in clips if c.clip_type == "b_roll"] + assert bgs[0].asset_id == "a1" + assert voices[0].asset_id == "a2" + assert brolls[0].asset_id == "a3" + + def test_single_asset_only_background(self): + """1个素材：只分配给 background.""" + clips = ( + _make_clips(1, "background") + + _make_clips(1, "corner_voice") + + _make_clips(2, "b_roll") + ) + distribute_assets(clips, ["a1"], EditingMode.VOICE_PIP.value) + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "" + assert clips[2].asset_id == "" + assert clips[3].asset_id == "" + + def test_two_assets_bg_and_voice(self): + """2个素材：background + corner_voice.""" + clips = ( + _make_clips(1, "background") + + _make_clips(1, "corner_voice") + + _make_clips(2, "b_roll") + ) + distribute_assets(clips, ["a1", "a2"], EditingMode.VOICE_PIP.value) + bgs = [c for c in clips if c.clip_type == "background"] + voices = [c for c in clips if c.clip_type == "corner_voice"] + brolls = [c for c in clips if c.clip_type == "b_roll"] + assert bgs[0].asset_id == "a1" + assert voices[0].asset_id == "a2" + assert brolls[0].asset_id == "" + + def test_many_broll_clips(self): + """多个 b_roll clip：按顺序分配剩余素材.""" + clips = ( + _make_clips(1, "background") + + _make_clips(1, "corner_voice") + + _make_clips(5, "b_roll") + ) + distribute_assets( + clips, + ["a1", "a2", "a3", "a4", "a5"], + EditingMode.VOICE_PIP.value, + ) + brolls = [c for c in clips if c.clip_type == "b_roll"] + assert brolls[0].asset_id == "a3" + assert brolls[1].asset_id == "a4" + assert brolls[2].asset_id == "a5" + assert brolls[3].asset_id == "" + assert brolls[4].asset_id == "" + + def test_missing_some_layer_clips(self): + """缺少某些层的 clip 不影响其他层.""" + # 没有 corner_voice，素材应该按顺序：bg 拿 a1，b_roll 从 a2 开始 + clips = _make_clips(1, "background") + _make_clips(3, "b_roll") + distribute_assets(clips, ["a1", "a2", "a3"], EditingMode.VOICE_PIP.value) + bgs = [c for c in clips if c.clip_type == "background"] + brolls = [c for c in clips if c.clip_type == "b_roll"] + assert bgs[0].asset_id == "a1" + # 没有 corner_voice，b_roll 从第2个素材开始 + assert brolls[0].asset_id == "a2" + assert brolls[1].asset_id == "a3" + + +# ── distribute_assets: 边缘情况 ───────────────────────────────────── + + +class TestDistributeEdgeCases: + """素材分配边缘情况.""" + + def test_unknown_mode_falls_back_to_one_take(self): + """未知模式退化为 ONE_TAKE.""" + clips = _make_clips(3) + distribute_assets(clips, ["a1", "a2", "a3"], "unknown_mode") + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "a2" + assert clips[2].asset_id == "a3" + + def test_both_empty(self): + """两边都空：不报错.""" + distribute_assets([], [], EditingMode.ONE_TAKE.value) + + +# ── map_clip_types_for_mode ───────────────────────────────────────── + + +class TestMapClipTypesForMode: + """clip_type 按模式映射.""" + + def test_one_take_unchanged(self): + """ONE_TAKE 模式：main 保持 main.""" + clips = _make_clips(5) + map_clip_types_for_mode(clips, EditingMode.ONE_TAKE.value) + for c in clips: + assert c.clip_type == "main" + + def test_voice_over_unchanged(self): + """VOICE_OVER 模式：main 保持 main.""" + clips = _make_clips(5) + map_clip_types_for_mode(clips, EditingMode.VOICE_OVER.value) + for c in clips: + assert c.clip_type == "main" + + def test_pip_first_main_stays_rest_become_overlay(self): + """PIP 模式：第1个 main 保持，其余变 overlay.""" + clips = _make_clips(5) + map_clip_types_for_mode(clips, EditingMode.PIP.value) + assert clips[0].clip_type == "main" + assert clips[1].clip_type == "overlay" + assert clips[2].clip_type == "overlay" + assert clips[3].clip_type == "overlay" + assert clips[4].clip_type == "overlay" + + def test_pip_single_main_unchanged(self): + """PIP 模式只有1个 main：保持 main.""" + clips = _make_clips(1) + map_clip_types_for_mode(clips, EditingMode.PIP.value) + assert clips[0].clip_type == "main" + + def test_voice_pip_three_types(self): + """VOICE_PIP 模式：background + corner_voice + b_roll.""" + clips = _make_clips(5) + map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value) + assert clips[0].clip_type == "background" + assert clips[1].clip_type == "corner_voice" + assert clips[2].clip_type == "b_roll" + assert clips[3].clip_type == "b_roll" + assert clips[4].clip_type == "b_roll" + + def test_voice_pip_one_main(self): + """VOICE_PIP 只有1个 main：变成 background.""" + clips = _make_clips(1) + map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value) + assert clips[0].clip_type == "background" + + def test_voice_pip_two_mains(self): + """VOICE_PIP 2个 main：background + corner_voice.""" + clips = _make_clips(2) + map_clip_types_for_mode(clips, EditingMode.VOICE_PIP.value) + assert clips[0].clip_type == "background" + assert clips[1].clip_type == "corner_voice" + + def test_non_main_clips_unchanged(self): + """非 MAIN 类型 clip 不受影响.""" + clips = ( + _make_clips(1, "intro") + + _make_clips(3) # main + + _make_clips(1, "outro") + ) + map_clip_types_for_mode(clips, EditingMode.PIP.value) + assert clips[0].clip_type == "intro" + assert clips[1].clip_type == "main" # 第1个 main + assert clips[2].clip_type == "overlay" # 第2个 main → overlay + assert clips[3].clip_type == "overlay" # 第3个 main → overlay + assert clips[4].clip_type == "outro" + + def test_no_main_clips_noop(self): + """没有 main clip：什么都不做.""" + clips = _make_clips(3, "intro") + original_types = [c.clip_type for c in clips] + map_clip_types_for_mode(clips, EditingMode.PIP.value) + assert [c.clip_type for c in clips] == original_types + + def test_empty_clips_noop(self): + """空列表：不报错.""" + map_clip_types_for_mode([], EditingMode.PIP.value) + + +# ── generate_default_clips ────────────────────────────────────────── + + +class TestGenerateDefaultClips: + """默认 clip 生成.""" + + def test_one_take_normal(self): + """ONE_TAKE：N 个 main clip.""" + clips = generate_default_clips("plan1", EditingMode.ONE_TAKE.value, 5) + assert len(clips) == 5 + for c in clips: + assert c.clip_type == "main" + assert c.plan_id == "plan1" + assert c.duration == DEFAULT_CLIP_DURATION + # order 递增 + for i in range(5): + assert clips[i].order == i + + def test_pip_structure(self): + """PIP：1个 main + (N-1)个 overlay.""" + clips = generate_default_clips("plan1", EditingMode.PIP.value, 4) + assert len(clips) == 4 + assert clips[0].clip_type == "main" + assert clips[1].clip_type == "overlay" + assert clips[2].clip_type == "overlay" + assert clips[3].clip_type == "overlay" + assert clips[0].order == 0 + assert clips[3].order == 3 + + def test_pip_single_asset(self): + """PIP 只有1个素材：1个 main，没有 overlay.""" + clips = generate_default_clips("plan1", EditingMode.PIP.value, 1) + assert len(clips) == 1 + assert clips[0].clip_type == "main" + + def test_voice_over_structure(self): + """VOICE_OVER：N 个 main clip，带 b_roll 标记.""" + clips = generate_default_clips("plan1", EditingMode.VOICE_OVER.value, 3) + assert len(clips) == 3 + for c in clips: + assert c.clip_type == "main" + assert c.config.get("role") == "b_roll" + + def test_voice_pip_three_layers(self): + """VOICE_PIP：background + corner_voice + b_roll.""" + clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 5) + assert len(clips) == 5 + assert clips[0].clip_type == "background" + assert clips[1].clip_type == "corner_voice" + assert clips[2].clip_type == "b_roll" + assert clips[3].clip_type == "b_roll" + assert clips[4].clip_type == "b_roll" + + def test_voice_pip_single_asset(self): + """VOICE_PIP 1个素材：只有 background.""" + clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 1) + assert len(clips) == 1 + assert clips[0].clip_type == "background" + + def test_voice_pip_two_assets(self): + """VOICE_PIP 2个素材：background + corner_voice.""" + clips = generate_default_clips("plan1", EditingMode.VOICE_PIP.value, 2) + assert len(clips) == 2 + assert clips[0].clip_type == "background" + assert clips[1].clip_type == "corner_voice" + + def test_zero_assets_at_least_one(self): + """0 个素材：至少生成 1 个 clip.""" + for mode in [ + EditingMode.ONE_TAKE.value, + EditingMode.PIP.value, + EditingMode.VOICE_OVER.value, + EditingMode.VOICE_PIP.value, + ]: + clips = generate_default_clips("plan1", mode, 0) + assert len(clips) >= 1 + + def test_unknown_mode_falls_back(self): + """未知模式退化为 ONE_TAKE 风格.""" + clips = generate_default_clips("plan1", "unknown", 3) + assert len(clips) == 3 + for c in clips: + assert c.clip_type == "main" + + def test_order_is_sequential(self): + """所有模式下 order 都是从 0 开始连续递增.""" + for mode in [ + EditingMode.ONE_TAKE.value, + EditingMode.PIP.value, + EditingMode.VOICE_OVER.value, + EditingMode.VOICE_PIP.value, + ]: + clips = generate_default_clips("plan1", mode, 5) + for i, c in enumerate(clips): + assert c.order == i + + +# ── create_clips_from_configs ─────────────────────────────────────── + + +class TestCreateClipsFromConfigs: + """从模板配置创建 clips.""" + + def test_basic_creation(self): + """基本创建：按 order 排序，属性正确传递.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=1, + min_duration=3.0, + max_duration=7.0, + transition_effect="fade", + ), + TemplateClipConfig( + id="cfg2", + template_id="tpl1", + clip_type=ClipType.INTRO, + order=0, + min_duration=2.0, + max_duration=4.0, + transition_effect="cut", + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert len(clips) == 2 + # 按 order 排序：intro(order=0) 在前，main(order=1) 在后 + assert clips[0].clip_type == "intro" + assert clips[1].clip_type == "main" + assert clips[0].order == 0 + assert clips[1].order == 1 + assert clips[0].template_clip_config_id == "cfg2" + assert clips[1].template_clip_config_id == "cfg1" + + def test_duration_average_of_min_max(self): + """min_duration 和 max_duration 都有时，取平均值.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=4.0, + max_duration=6.0, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].duration == 5.0 # (4+6)/2 + + def test_duration_only_min(self): + """只有 min_duration 时，用 min_duration.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=3.5, + max_duration=0, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].duration == 3.5 + + def test_duration_only_max(self): + """只有 max_duration 时，用 max_duration.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=0, + max_duration=8.0, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].duration == 8.0 + + def test_duration_default_when_both_zero(self): + """都为 0 时用默认时长.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=0, + max_duration=0, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].duration == DEFAULT_CLIP_DURATION + + def test_empty_configs_returns_empty(self): + """空配置列表返回空列表.""" + clips = create_clips_from_configs("plan1", []) + assert clips == [] + + def test_plan_id_passed_through(self): + """plan_id 正确传递给所有 clip.""" + configs = [ + TemplateClipConfig( + id=f"cfg{i}", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=i, + min_duration=3, + max_duration=5, + ) + for i in range(3) + ] + clips = create_clips_from_configs("my_plan", configs) + for c in clips: + assert c.plan_id == "my_plan" + + def test_playback_speed_from_config(self): + """playback_speed 从 config.playback_speed 读取.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=3, + max_duration=5, + config={"playback_speed": 1.5}, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].playback_speed == 1.5 + + def test_playback_speed_fallback_to_speed_ratio(self): + """playback_speed 不存在时回退到 speed_ratio.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=3, + max_duration=5, + config={"speed_ratio": 0.8}, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].playback_speed == 0.8 + + def test_playback_speed_default_1(self): + """没有 speed 配置时默认为 1.0.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=3, + max_duration=5, + config={}, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].playback_speed == 1.0 + + def test_playback_speed_none_falls_back(self): + """playback_speed 为 None 时回退到 1.0.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=3, + max_duration=5, + config={"playback_speed": None}, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].playback_speed == 1.0 + + def test_transition_default_cut(self): + """transition_effect 为空时默认为 cut.""" + configs = [ + TemplateClipConfig( + id="cfg1", + template_id="tpl1", + clip_type=ClipType.MAIN, + order=0, + min_duration=3, + max_duration=5, + transition_effect=None, + ), + ] + clips = create_clips_from_configs("plan1", configs) + assert clips[0].transition_effect == "cut" + + +# ── 常量导出 ──────────────────────────────────────────────────────── + + +class TestConstants: + """常量导出验证.""" + + def test_default_duration_value(self): + """默认片段时长应为 5 秒.""" + assert DEFAULT_CLIP_DURATION == 5.0 -- 2.54.0 From 7da2a07a1afccc3e80fbdf7edcba4f156f789c20 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 27 Jul 2026 00:33:05 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix(test):=20=E4=BF=AE=E5=A4=8Dtest=5Fagent?= =?UTF-8?q?=5Fdocs.py=E7=9B=B8=E5=AF=B9=E8=B7=AF=E5=BE=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98=EF=BC=8C=E4=BB=8Eapps/api=E8=BF=90=E8=A1=8C=E4=B9=9F?= =?UTF-8?q?=E8=83=BD=E6=89=BE=E5=88=B0docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 使用 __file__ 定位项目根目录，避免依赖pytest运行目录。 修复后Unit Tests CI从apps/api目录运行也能正常通过。 --- tests/unit/test_agent_docs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_agent_docs.py b/tests/unit/test_agent_docs.py index c23b26a76..581f2a68b 100644 --- a/tests/unit/test_agent_docs.py +++ b/tests/unit/test_agent_docs.py @@ -1,6 +1,6 @@ from pathlib import Path -AGENT_DOCS = Path("docs/agents") +AGENT_DOCS = Path(__file__).parent.parent.parent / "docs" / "agents" EXPECTED_AGENTS = [ "requirement_agent", "arch_agent", -- 2.54.0