From b896873ecec97a9ebb64db41aad08f8b1e07d787 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 26 Jul 2026 18:15:59 +0800 Subject: [PATCH] test(wave92): extract asset_scoring domain module + 89 unit tests (#955) --- apps/api/app/services/smart_asset_selector.py | 263 ++---- packages/domain/asset_scoring.py | 372 +++++++++ tests/unit/test_asset_scoring.py | 755 ++++++++++++++++++ 3 files changed, 1180 insertions(+), 210 deletions(-) create mode 100755 packages/domain/asset_scoring.py create mode 100755 tests/unit/test_asset_scoring.py diff --git a/apps/api/app/services/smart_asset_selector.py b/apps/api/app/services/smart_asset_selector.py index 7878b1bd5..65f2aa8c1 100755 --- a/apps/api/app/services/smart_asset_selector.py +++ b/apps/api/app/services/smart_asset_selector.py @@ -13,57 +13,31 @@ - 最低质量分门槛：自动过滤低质量素材 - 时长多样性：保证选出的素材时长分布均匀（短/中/长各占一定比例） - 兼容全部模式：素材库模式和项目模式都可用 + +纯逻辑部分已抽离到 packages.domain.asset_scoring。 """ from __future__ import annotations import logging -from dataclasses import dataclass + +from packages.domain.asset_scoring import ( + MEDIUM_BUCKET_MAX as _MEDIUM_BUCKET_MAX, + MIN_QUALITY_SCORE as _MIN_QUALITY_SCORE, + OPTIMAL_DURATION_MAX as _OPTIMAL_DURATION_MAX, + OPTIMAL_DURATION_MIN as _OPTIMAL_DURATION_MIN, + SHORT_BUCKET_MAX as _SHORT_BUCKET_MAX, + TARGET_HEIGHT as _TARGET_HEIGHT, + TARGET_WIDTH as _TARGET_WIDTH, + AssetScoreDetail, + SmartSelectResult, + diverse_selection, + filter_candidates, + score_asset_detail, +) logger = logging.getLogger(__name__) -# ── 评分权重 ────────────────────────────────────────────────────────────────── -_WEIGHT_QUALITY = 0.5 -_WEIGHT_RESOLUTION = 0.2 -_WEIGHT_DURATION = 0.2 -_WEIGHT_BITRATE = 0.1 - -# ── 评分参数 ────────────────────────────────────────────────────────────────── -_TARGET_WIDTH = 1920 # 目标分辨率宽度基准 -_TARGET_HEIGHT = 1080 # 目标分辨率高度基准 -_MIN_QUALITY_SCORE = 30.0 # 最低质量分门槛（低于此值的素材直接排除） -_OPTIMAL_DURATION_MIN = 3.0 # 最佳时长区间（秒） -_OPTIMAL_DURATION_MAX = 30.0 - -# ── 多样性分桶 ─────────────────────────────────────────────────────────────── -_SHORT_BUCKET_MAX = 5.0 # 短素材：< 5s -_MEDIUM_BUCKET_MAX = 15.0 # 中素材：5-15s -# 长素材：> 15s - - -@dataclass -class SmartSelectResult: - """智能选择结果.""" - - selected_ids: list[str] - total_candidates: int - filtered_out: int # 被质量门槛过滤的数量 - avg_score: float - details: list[AssetScoreDetail] - - -@dataclass -class AssetScoreDetail: - """单个素材的评分详情.""" - - asset_id: str - total_score: float - quality_score: float - resolution_score: float - duration_score: float - bitrate_score: float - duration: float | None - class SmartAssetSelector: """智能素材选择器. @@ -74,9 +48,9 @@ class SmartAssetSelector: def __init__( self, - min_quality_score: float = _MIN_QUALITY_SCORE, - target_width: int = _TARGET_WIDTH, - target_height: int = _TARGET_HEIGHT, + min_quality_score: float = 30.0, + target_width: int = 1920, + target_height: int = 1080, ): self.min_quality_score = min_quality_score self.target_width = target_width @@ -102,21 +76,7 @@ class SmartAssetSelector: SmartSelectResult 选择结果 """ # 1. 过滤：只保留 ready 状态的视频素材 + 最低质量分门槛 - candidates = [] - filtered_out = 0 - for asset in assets: - status = getattr(asset, "status", None) - status_val = status.value if hasattr(status, "value") else str(status) - if status_val != "ready": - continue - mime_type = getattr(asset, "mime_type", "") or "" - if not mime_type.startswith("video"): - continue - quality = getattr(asset, "quality_score", None) - if quality is not None and quality < self.min_quality_score: - filtered_out += 1 - continue - candidates.append(asset) + candidates, filtered_out = filter_candidates(assets, self.min_quality_score) if not candidates: return SmartSelectResult( @@ -130,7 +90,16 @@ class SmartAssetSelector: # 2. 对每个候选素材评分 scored: list[AssetScoreDetail] = [] for asset in candidates: - detail = self._score_asset(asset) + detail = score_asset_detail( + asset_id=asset.id, + quality=getattr(asset, "quality_score", None), + width=getattr(asset, "width", None), + height=getattr(asset, "height", None), + duration=getattr(asset, "duration", None), + file_size=getattr(asset, "file_size", 0) or 0, + target_width=self.target_width, + target_height=self.target_height, + ) scored.append(detail) # 3. 按总分降序排列 @@ -138,7 +107,7 @@ class SmartAssetSelector: # 4. 多样性选择（如果需要且数量有限制） if ensure_diversity and count > 0 and len(scored) > count: - selected = self._diverse_selection(scored, count) + selected = diverse_selection(scored, count) else: # 无数量限制或不要求多样性，直接按排名取 selected = scored if count <= 0 else scored[:count] @@ -162,165 +131,39 @@ class SmartAssetSelector: ) return result - # ── 内部方法 ────────────────────────────────────────────────────────────── + # ── 向后兼容：私有方法别名（委托给 asset_scoring 纯函数） ──────────────── def _score_asset(self, asset) -> AssetScoreDetail: - """对单个素材进行多维度评分.""" - # 质量分 - quality = getattr(asset, "quality_score", None) - quality_score = (quality / 100.0) if quality is not None else 0.5 - - # 分辨率评分：越接近目标分辨率得分越高 - width = getattr(asset, "width", None) - height = getattr(asset, "height", None) - resolution_score = self._score_resolution(width, height) - - # 时长评分：在最佳区间内得分高，过短过长扣分 - duration = getattr(asset, "duration", None) - duration_score = self._score_duration(duration) - - # 码率评分：用 file_size/duration 估算，适中得分高 - file_size = getattr(asset, "file_size", 0) or 0 - bitrate_score = self._score_bitrate(file_size, duration) - - # 加权总分 - total = ( - _WEIGHT_QUALITY * quality_score - + _WEIGHT_RESOLUTION * resolution_score - + _WEIGHT_DURATION * duration_score - + _WEIGHT_BITRATE * bitrate_score - ) - - return AssetScoreDetail( + """对单个素材进行多维度评分（向后兼容）.""" + return score_asset_detail( asset_id=asset.id, - total_score=round(total, 4), - quality_score=round(quality_score, 4), - resolution_score=round(resolution_score, 4), - duration_score=round(duration_score, 4), - bitrate_score=round(bitrate_score, 4), - duration=duration, + quality=getattr(asset, "quality_score", None), + width=getattr(asset, "width", None), + height=getattr(asset, "height", None), + duration=getattr(asset, "duration", None), + file_size=getattr(asset, "file_size", 0) or 0, + target_width=self.target_width, + target_height=self.target_height, ) def _score_resolution(self, width: int | None, height: int | None) -> float: - """分辨率评分：越接近目标分辨率得分越高，低于480p扣分严重.""" - if width is None or height is None or width <= 0 or height <= 0: - return 0.5 # 未知分辨率给中评分 + """分辨率评分（向后兼容）.""" + from packages.domain.asset_scoring import score_resolution - target_pixels = self.target_width * self.target_height - actual_pixels = width * height - - # 计算像素数比例 - ratio = actual_pixels / target_pixels - - if ratio >= 1.0: - # 高于或等于目标分辨率：满分，略高不扣分（4K也给满分） - return 1.0 - else: - # 低于目标分辨率：线性衰减，但最低不低于 0.1 - # 例如：720p (921600) / 1080p (2073600) = 0.44 → 得分 0.6 - score = 0.3 + 0.7 * ratio - return max(0.1, min(1.0, score)) + return score_resolution(width, height, self.target_width, self.target_height) def _score_duration(self, duration: float | None) -> float: - """时长评分：3-30秒最佳，过短或过长都扣分.""" - if duration is None or duration <= 0: - return 0.5 # 未知时长给中评分 + """时长评分（向后兼容）.""" + from packages.domain.asset_scoring import score_duration - if _OPTIMAL_DURATION_MIN <= duration <= _OPTIMAL_DURATION_MAX: - # 最佳区间：满分 - return 1.0 - - if duration < _OPTIMAL_DURATION_MIN: - # 太短：线性衰减，1秒以下给 0.3 - ratio = duration / _OPTIMAL_DURATION_MIN - return 0.3 + 0.7 * ratio - - # 太长：每超过最佳区间上限10秒扣 0.1 分，最低 0.2 - excess = duration - _OPTIMAL_DURATION_MAX - penalty = min(0.8, excess / 10.0 * 0.1) - return max(0.2, 1.0 - penalty) + return score_duration(duration) def _score_bitrate(self, file_size: int, duration: float | None) -> float: - """码率评分：根据文件大小和时长估算码率，适中得分高.""" - if not file_size or not duration or duration <= 0: - return 0.5 # 未知给中评分 + """码率评分（向后兼容）.""" + from packages.domain.asset_scoring import score_bitrate - # 估算码率（bps） - bitrate = (file_size * 8) / duration - - # 最佳码率范围：2-8 Mbps - optimal_low = 2_000_000 # 2 Mbps - optimal_high = 8_000_000 # 8 Mbps - - if optimal_low <= bitrate <= optimal_high: - return 1.0 - - if bitrate < optimal_low: - # 码率太低：线性衰减 - ratio = bitrate / optimal_low - return 0.3 + 0.7 * ratio - - # 码率太高（文件太大）：适度扣分 - excess = bitrate / optimal_high - 1.0 - penalty = min(0.5, excess * 0.2) - return max(0.5, 1.0 - penalty) + return score_bitrate(file_size, duration) def _diverse_selection(self, scored: list[AssetScoreDetail], count: int) -> list[AssetScoreDetail]: - """多样性选择：按时长分桶，保证每个桶都有素材. - - 策略： - 1. 按时长分为三桶：短(<5s)、中(5-15s)、长(>15s) - 2. 每个桶配额 = max(1, count / 3) - 3. 先从每桶按配额取最高分的 - 4. 剩余名额从全局最高分中取（不重复） - """ - # 分桶 - short_bucket = [d for d in scored if d.duration is not None and d.duration < _SHORT_BUCKET_MAX] - medium_bucket = [ - d for d in scored if d.duration is not None and _SHORT_BUCKET_MAX <= d.duration < _MEDIUM_BUCKET_MAX - ] - long_bucket = [d for d in scored if d.duration is not None and d.duration >= _MEDIUM_BUCKET_MAX] - unknown_bucket = [d for d in scored if d.duration is None] - - buckets = [short_bucket, medium_bucket, long_bucket] - bucket_names = ["short", "medium", "long"] - - # 每个桶基础配额（至少1个，如果桶非空且需要的话） - base_quota = max(1, count // 3) - - selected: list[AssetScoreDetail] = [] - selected_ids: set[str] = set() - - # 先按配额从每个桶取 - for bucket, _name in zip(buckets, bucket_names, strict=False): - quota = min(base_quota, len(bucket)) - if quota <= 0: - continue - # 桶内已经按分数排好序了，直接取前 quota 个 - for item in bucket[:quota]: - if item.asset_id not in selected_ids: - selected.append(item) - selected_ids.add(item.asset_id) - if len(selected) >= count: - return selected - - # 剩余名额：从全局（未被选中的）中按分数高低取 - remaining_needed = count - len(selected) - if remaining_needed > 0: - for item in scored: - if item.asset_id not in selected_ids: - selected.append(item) - selected_ids.add(item.asset_id) - if len(selected) >= count: - break - - # 如果还不够（不应该发生），加上未知时长的 - if len(selected) < count and unknown_bucket: - for item in unknown_bucket: - if item.asset_id not in selected_ids: - selected.append(item) - selected_ids.add(item.asset_id) - if len(selected) >= count: - break - - return selected[:count] + """多样性选择（向后兼容）.""" + return diverse_selection(scored, count) diff --git a/packages/domain/asset_scoring.py b/packages/domain/asset_scoring.py new file mode 100755 index 000000000..dfb61ebae --- /dev/null +++ b/packages/domain/asset_scoring.py @@ -0,0 +1,372 @@ +"""Asset scoring pure logic — multi-dimensional scoring + diverse selection. + +从 smart_asset_selector.py 抽出来的纯逻辑模块： +- 评分维度：质量分、分辨率、时长、码率（加权求和，总分 0-1） +- 多样性选择：按时长分桶（短/中/长）保证分布均匀 +- 数据类：AssetScoreDetail, SmartSelectResult +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +# ── 评分权重（总和 = 1.0） ──────────────────────────────────────────────────── + +WEIGHT_QUALITY = 0.5 +WEIGHT_RESOLUTION = 0.2 +WEIGHT_DURATION = 0.2 +WEIGHT_BITRATE = 0.1 + +# ── 评分参数 ────────────────────────────────────────────────────────────────── + +TARGET_WIDTH = 1920 # 目标分辨率宽度基准 +TARGET_HEIGHT = 1080 # 目标分辨率高度基准 +MIN_QUALITY_SCORE = 30.0 # 最低质量分门槛（低于此值的素材直接排除） +OPTIMAL_DURATION_MIN = 3.0 # 最佳时长区间（秒） +OPTIMAL_DURATION_MAX = 30.0 + +# ── 多样性分桶阈值 ─────────────────────────────────────────────────────────── + +SHORT_BUCKET_MAX = 5.0 # 短素材：< 5s +MEDIUM_BUCKET_MAX = 15.0 # 中素材：5-15s +# 长素材：>= 15s + + +# ── 数据类 ─────────────────────────────────────────────────────────────────── + + +@dataclass +class AssetScoreDetail: + """单个素材的评分详情.""" + + asset_id: str + total_score: float + quality_score: float + resolution_score: float + duration_score: float + bitrate_score: float + duration: float | None + + +@dataclass +class SmartSelectResult: + """智能选择结果.""" + + selected_ids: list[str] + total_candidates: int + filtered_out: int # 被质量门槛过滤的数量 + avg_score: float + details: list[AssetScoreDetail] = field(default_factory=list) + + +# ── 评分函数 ───────────────────────────────────────────────────────────────── + + +def score_resolution( + width: int | None, + height: int | None, + target_width: int = TARGET_WIDTH, + target_height: int = TARGET_HEIGHT, +) -> float: + """分辨率评分：越接近目标分辨率得分越高，低于480p扣分严重. + + Args: + width: 素材宽度（像素） + height: 素材高度（像素） + target_width: 目标宽度 + target_height: 目标高度 + + Returns: + 0.0 - 1.0 之间的评分 + """ + if width is None or height is None or width <= 0 or height <= 0: + return 0.5 # 未知分辨率给中评分 + + target_pixels = target_width * target_height + actual_pixels = width * height + + # 计算像素数比例 + ratio = actual_pixels / target_pixels + + if ratio >= 1.0: + # 高于或等于目标分辨率：满分，略高不扣分（4K也给满分） + return 1.0 + else: + # 低于目标分辨率：线性衰减，但最低不低于 0.1 + score = 0.3 + 0.7 * ratio + return max(0.1, min(1.0, score)) + + +def score_duration(duration: float | None) -> float: + """时长评分：3-30秒最佳，过短或过长都扣分. + + Args: + duration: 时长（秒） + + Returns: + 0.0 - 1.0 之间的评分 + """ + if duration is None or duration <= 0: + return 0.5 # 未知时长给中评分 + + if OPTIMAL_DURATION_MIN <= duration <= OPTIMAL_DURATION_MAX: + # 最佳区间：满分 + return 1.0 + + if duration < OPTIMAL_DURATION_MIN: + # 太短：线性衰减，趋近于 0.3 + ratio = duration / OPTIMAL_DURATION_MIN + return 0.3 + 0.7 * ratio + + # 太长：每超过最佳区间上限10秒扣 0.1 分，最低 0.2 + excess = duration - OPTIMAL_DURATION_MAX + penalty = min(0.8, excess / 10.0 * 0.1) + return max(0.2, 1.0 - penalty) + + +def score_bitrate(file_size: int, duration: float | None) -> float: + """码率评分：根据文件大小和时长估算码率，适中得分高. + + Args: + file_size: 文件大小（字节） + duration: 时长（秒） + + Returns: + 0.0 - 1.0 之间的评分 + """ + if not file_size or not duration or duration <= 0: + return 0.5 # 未知给中评分 + + # 估算码率（bps） + bitrate = (file_size * 8) / duration + + # 最佳码率范围：2-8 Mbps + optimal_low = 2_000_000 # 2 Mbps + optimal_high = 8_000_000 # 8 Mbps + + if optimal_low <= bitrate <= optimal_high: + return 1.0 + + if bitrate < optimal_low: + # 码率太低：线性衰减 + ratio = bitrate / optimal_low + return 0.3 + 0.7 * ratio + + # 码率太高（文件太大）：适度扣分，最低 0.5 + excess = bitrate / optimal_high - 1.0 + penalty = min(0.5, excess * 0.2) + return max(0.5, 1.0 - penalty) + + +def calculate_total_score( + quality_score: float, + resolution_score: float, + duration_score: float, + bitrate_score: float, +) -> float: + """计算加权总分. + + Args: + quality_score: 质量分（0-1） + resolution_score: 分辨率分（0-1） + duration_score: 时长分（0-1） + bitrate_score: 码率分（0-1） + + Returns: + 加权总分（0-1） + """ + total = ( + WEIGHT_QUALITY * quality_score + + WEIGHT_RESOLUTION * resolution_score + + WEIGHT_DURATION * duration_score + + WEIGHT_BITRATE * bitrate_score + ) + return round(total, 4) + + +def score_asset_detail( + asset_id: str, + quality: float | None, + width: int | None, + height: int | None, + duration: float | None, + file_size: int, + target_width: int = TARGET_WIDTH, + target_height: int = TARGET_HEIGHT, +) -> AssetScoreDetail: + """对单个素材进行多维度评分，返回详细评分结果. + + Args: + asset_id: 素材ID + quality: 质量分（0-100，None表示未知） + width: 宽度 + height: 高度 + duration: 时长 + file_size: 文件大小 + target_width: 目标宽度 + target_height: 目标高度 + + Returns: + AssetScoreDetail 评分详情 + """ + # 质量分归一化到 0-1 + quality_score = (quality / 100.0) if quality is not None else 0.5 + + resolution_score = score_resolution(width, height, target_width, target_height) + duration_score = score_duration(duration) + bitrate_score = score_bitrate(file_size, duration) + + total_score = calculate_total_score( + quality_score, + resolution_score, + duration_score, + bitrate_score, + ) + + return AssetScoreDetail( + asset_id=asset_id, + total_score=total_score, + quality_score=round(quality_score, 4), + resolution_score=round(resolution_score, 4), + duration_score=round(duration_score, 4), + bitrate_score=round(bitrate_score, 4), + duration=duration, + ) + + +# ── 多样性选择 ─────────────────────────────────────────────────────────────── + + +def _bucket_by_duration(item: AssetScoreDetail) -> str: + """根据时长判断所属桶. + + Returns: + 'short' / 'medium' / 'long' / 'unknown' + """ + if item.duration is None: + return "unknown" + if item.duration < SHORT_BUCKET_MAX: + return "short" + if item.duration < MEDIUM_BUCKET_MAX: + return "medium" + return "long" + + +def diverse_selection( + scored: list[AssetScoreDetail], + count: int, +) -> list[AssetScoreDetail]: + """多样性选择：按时长分桶，保证每个桶都有素材. + + 策略： + 1. 按时长分为三桶：短(<5s)、中(5-15s)、长(>=15s) + 2. 每个桶配额 = max(1, count // 3) + 3. 先从每桶按配额取最高分的 + 4. 剩余名额从全局最高分中取（不重复） + 5. 如果还不够，加上未知时长的 + + Args: + scored: 已按总分降序排列的评分列表 + count: 需要选取的数量 + + Returns: + 选中的评分列表（不超过 count 个） + """ + if count <= 0 or not scored: + return [] + + # 分桶 + short_bucket = [d for d in scored if _bucket_by_duration(d) == "short"] + medium_bucket = [d for d in scored if _bucket_by_duration(d) == "medium"] + long_bucket = [d for d in scored if _bucket_by_duration(d) == "long"] + unknown_bucket = [d for d in scored if _bucket_by_duration(d) == "unknown"] + + buckets = [short_bucket, medium_bucket, long_bucket] + + # 每个桶基础配额（至少1个，如果桶非空且需要的话） + base_quota = max(1, count // 3) + + selected: list[AssetScoreDetail] = [] + selected_ids: set[str] = set() + + # 先按配额从每个桶取 + for bucket in buckets: + quota = min(base_quota, len(bucket)) + if quota <= 0: + continue + # 桶内已经按分数排好序了，直接取前 quota 个 + for item in bucket[:quota]: + if item.asset_id not in selected_ids: + selected.append(item) + selected_ids.add(item.asset_id) + if len(selected) >= count: + return selected + + # 剩余名额：从全局（未被选中的）中按分数高低取 + remaining_needed = count - len(selected) + if remaining_needed > 0: + for item in scored: + if item.asset_id not in selected_ids: + selected.append(item) + selected_ids.add(item.asset_id) + if len(selected) >= count: + break + + # 如果还不够（不应该发生），加上未知时长的 + if len(selected) < count and unknown_bucket: + for item in unknown_bucket: + if item.asset_id not in selected_ids: + selected.append(item) + selected_ids.add(item.asset_id) + if len(selected) >= count: + break + + return selected[:count] + + +# ── 候选过滤 ───────────────────────────────────────────────────────────────── + + +def filter_candidates( + assets: list[Any], + min_quality_score: float = MIN_QUALITY_SCORE, +) -> tuple[list[Any], int]: + """从素材列表中筛选出合格的候选素材. + + 筛选条件： + - status == 'ready' + - mime_type 以 'video' 开头 + - quality_score >= min_quality_score（如果quality不为None） + + Args: + assets: 素材列表 + min_quality_score: 最低质量分门槛 + + Returns: + (合格素材列表, 被质量门槛过滤的数量) + """ + candidates = [] + filtered_out = 0 + + for asset in assets: + # 状态检查 + status = getattr(asset, "status", None) + status_val = status.value if hasattr(status, "value") else str(status) + if status_val != "ready": + continue + + # 类型检查 + mime_type = getattr(asset, "mime_type", "") or "" + if not mime_type.startswith("video"): + continue + + # 质量分门槛 + quality = getattr(asset, "quality_score", None) + if quality is not None and quality < min_quality_score: + filtered_out += 1 + continue + + candidates.append(asset) + + return candidates, filtered_out diff --git a/tests/unit/test_asset_scoring.py b/tests/unit/test_asset_scoring.py new file mode 100755 index 000000000..3b1e47619 --- /dev/null +++ b/tests/unit/test_asset_scoring.py @@ -0,0 +1,755 @@ +"""Deep unit tests for asset_scoring.py — multi-dimensional scoring + diverse selection. + +深度覆盖： +- score_resolution: 10+ 边界情况 +- score_duration: 10+ 边界情况 +- score_bitrate: 10+ 边界情况 +- calculate_total_score: 加权验证 +- score_asset_detail: 完整评分流程 +- diverse_selection: 各种分桶场景 +- filter_candidates: 各种过滤条件 +- 数据类 + 常量 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + +import pytest + +from packages.domain.asset_scoring import ( + AssetScoreDetail, + MEDIUM_BUCKET_MAX, + MIN_QUALITY_SCORE, + OPTIMAL_DURATION_MAX, + OPTIMAL_DURATION_MIN, + SHORT_BUCKET_MAX, + SmartSelectResult, + TARGET_HEIGHT, + TARGET_WIDTH, + WEIGHT_BITRATE, + WEIGHT_DURATION, + WEIGHT_QUALITY, + WEIGHT_RESOLUTION, + _bucket_by_duration, + calculate_total_score, + diverse_selection, + filter_candidates, + score_asset_detail, + score_bitrate, + score_duration, + score_resolution, +) + +# ── 辅助：模拟 asset 对象 ──────────────────────────────────────────────────── + + +class MockStatus: + def __init__(self, value: str): + self.value = value + + +@dataclass +class MockAsset: + id: str = "asset_001" + status: Any = None + mime_type: str = "video/mp4" + quality_score: Optional[float] = None + width: Optional[int] = None + height: Optional[int] = None + duration: Optional[float] = None + file_size: int = 0 + + def __post_init__(self): + if self.status is None: + self.status = MockStatus("ready") + + +# ── 常量测试 ──────────────────────────────────────────────────────────────── + + +class TestConstants: + def test_weights_sum_to_one(self): + total = WEIGHT_QUALITY + WEIGHT_RESOLUTION + WEIGHT_DURATION + WEIGHT_BITRATE + assert abs(total - 1.0) < 0.001 + + def test_target_resolution_1080p(self): + assert TARGET_WIDTH == 1920 + assert TARGET_HEIGHT == 1080 + + def test_bucket_thresholds(self): + assert SHORT_BUCKET_MAX == 5.0 + assert MEDIUM_BUCKET_MAX == 15.0 + assert SHORT_BUCKET_MAX < MEDIUM_BUCKET_MAX + + def test_optimal_duration_range(self): + assert OPTIMAL_DURATION_MIN == 3.0 + assert OPTIMAL_DURATION_MAX == 30.0 + assert OPTIMAL_DURATION_MIN < OPTIMAL_DURATION_MAX + + def test_min_quality_score(self): + assert MIN_QUALITY_SCORE == 30.0 + + +# ── 数据类测试 ────────────────────────────────────────────────────────────── + + +class TestDataClasses: + def test_asset_score_detail_defaults(self): + detail = AssetScoreDetail( + asset_id="a1", + total_score=0.8, + quality_score=0.7, + resolution_score=0.9, + duration_score=0.85, + bitrate_score=0.75, + duration=10.0, + ) + assert detail.asset_id == "a1" + assert detail.total_score == 0.8 + assert detail.duration == 10.0 + + def test_smart_select_result_defaults(self): + result = SmartSelectResult( + selected_ids=["a1", "a2"], + total_candidates=10, + filtered_out=3, + avg_score=0.75, + ) + assert result.selected_ids == ["a1", "a2"] + assert result.details == [] + assert result.total_candidates == 10 + + def test_smart_select_result_with_details(self): + detail = AssetScoreDetail( + asset_id="a1", + total_score=0.9, + quality_score=0.8, + resolution_score=0.95, + duration_score=0.9, + bitrate_score=0.85, + duration=5.0, + ) + result = SmartSelectResult( + selected_ids=["a1"], + total_candidates=5, + filtered_out=0, + avg_score=0.9, + details=[detail], + ) + assert len(result.details) == 1 + assert result.details[0].asset_id == "a1" + + +# ── score_resolution ──────────────────────────────────────────────────────── + + +class TestScoreResolution: + def test_exact_target_1080p(self): + score = score_resolution(1920, 1080) + assert score == 1.0 + + def test_4k_full_score(self): + score = score_resolution(3840, 2160) + assert score == 1.0 + + def test_higher_than_target_full_score(self): + score = score_resolution(2560, 1440) + assert score == 1.0 + + def test_720p_lower(self): + score = score_resolution(1280, 720) + # 720p 像素 = 921600, 1080p = 2073600 + # ratio = 0.444, score = 0.3 + 0.7 * 0.444 = 0.611 + assert 0.5 < score < 0.75 + + def test_480p_much_lower(self): + score = score_resolution(854, 480) + # 480p = 409,920 pixels, ratio = 0.197 + # score = 0.3 + 0.7 * 0.197 = 0.438 + assert 0.3 < score < 0.5 + + def test_none_width(self): + score = score_resolution(None, 1080) + assert score == 0.5 + + def test_none_height(self): + score = score_resolution(1920, None) + assert score == 0.5 + + def test_both_none(self): + score = score_resolution(None, None) + assert score == 0.5 + + def test_zero_width(self): + score = score_resolution(0, 1080) + assert score == 0.5 + + def test_zero_height(self): + score = score_resolution(1920, 0) + assert score == 0.5 + + def test_negative_width(self): + score = score_resolution(-100, 1080) + assert score == 0.5 + + def test_very_low_res_floor(self): + score = score_resolution(100, 100) + # 10000 pixels, ratio = 0.0048, score = 0.3 + 0.7*0.0048 = 0.303 + # 但最低不低于 0.1 + assert score >= 0.1 + assert score < 0.5 + + def test_custom_target_resolution(self): + score = score_resolution(1280, 720, target_width=1280, target_height=720) + assert score == 1.0 + + def test_sd_resolution(self): + score = score_resolution(640, 480) + # VGA = 307,200, ratio = 0.148 + assert score > 0.1 + + +# ── score_duration ────────────────────────────────────────────────────────── + + +class TestScoreDuration: + def test_none_duration(self): + assert score_duration(None) == 0.5 + + def test_zero_duration(self): + assert score_duration(0.0) == 0.5 + + def test_negative_duration(self): + assert score_duration(-5.0) == 0.5 + + def test_optimal_lower_bound(self): + assert score_duration(OPTIMAL_DURATION_MIN) == 1.0 + + def test_optimal_upper_bound(self): + assert score_duration(OPTIMAL_DURATION_MAX) == 1.0 + + def test_optimal_middle(self): + assert score_duration(10.0) == 1.0 + + def test_below_optimal_short(self): + score = score_duration(1.5) + # ratio = 1.5/3 = 0.5, score = 0.3 + 0.7*0.5 = 0.65 + assert score == pytest.approx(0.65, rel=1e-3) + + def test_very_short_approaches_03(self): + score = score_duration(0.1) + # ratio = 0.1/3 = 0.033, score = 0.3 + 0.7*0.033 = 0.323 + assert 0.3 < score < 0.4 + + def test_just_below_optimal(self): + score = score_duration(2.9) + assert score < 1.0 + assert score > 0.9 + + def test_above_optimal_slightly(self): + score = score_duration(35.0) + # excess = 5, penalty = 5/10 * 0.1 = 0.05, score = 0.95 + assert score == pytest.approx(0.95, rel=1e-3) + + def test_above_optimal_moderate(self): + score = score_duration(60.0) + # excess = 30, penalty = 30/10 * 0.1 = 0.3, score = 0.7 + assert score == pytest.approx(0.7, rel=1e-3) + + def test_very_long_floor(self): + score = score_duration(1000.0) + # excess = 970, penalty = 970/10 * 0.1 = 9.7, capped at 0.8 + # score = max(0.2, 1.0 - 0.8) = 0.2 + assert score == 0.2 + + def test_1_second(self): + score = score_duration(1.0) + # ratio = 1/3 = 0.333, score = 0.3 + 0.7*0.333 = 0.533 + assert score == pytest.approx(0.3 + 0.7 * (1.0 / 3.0), rel=1e-3) + + +# ── score_bitrate ─────────────────────────────────────────────────────────── + + +class TestScoreBitrate: + def test_no_file_size(self): + assert score_bitrate(0, 10.0) == 0.5 + + def test_no_duration(self): + assert score_bitrate(1_000_000, None) == 0.5 + + def test_zero_duration(self): + assert score_bitrate(1_000_000, 0.0) == 0.5 + + def test_negative_duration(self): + assert score_bitrate(1_000_000, -5.0) == 0.5 + + def test_optimal_low_end(self): + # 2 Mbps for 10s = 2.5 MB + file_size = int(2_000_000 * 10 / 8) + score = score_bitrate(file_size, 10.0) + assert score == 1.0 + + def test_optimal_high_end(self): + # 8 Mbps for 10s = 10 MB + file_size = int(8_000_000 * 10 / 8) + score = score_bitrate(file_size, 10.0) + assert score == 1.0 + + def test_optimal_middle(self): + # 5 Mbps for 10s = 6.25 MB + file_size = int(5_000_000 * 10 / 8) + score = score_bitrate(file_size, 10.0) + assert score == 1.0 + + def test_low_bitrate(self): + # 1 Mbps for 10s = 1.25 MB + file_size = int(1_000_000 * 10 / 8) + score = score_bitrate(file_size, 10.0) + # ratio = 1/2 = 0.5, score = 0.3 + 0.7*0.5 = 0.65 + assert score == pytest.approx(0.65, rel=1e-2) + + def test_very_low_bitrate(self): + # 100 kbps for 10s = 125 KB + file_size = int(100_000 * 10 / 8) + score = score_bitrate(file_size, 10.0) + # ratio = 0.05, score = 0.3 + 0.7*0.05 = 0.335 + assert 0.3 < score < 0.5 + + def test_high_bitrate_slightly(self): + # 10 Mbps (just above 8Mbps) + file_size = int(10_000_000 * 10 / 8) + score = score_bitrate(file_size, 10.0) + # excess ratio = 10/8 - 1 = 0.25, penalty = min(0.5, 0.25*0.2) = 0.05 + # score = max(0.5, 1.0 - 0.05) = 0.95 + assert score == pytest.approx(0.95, rel=1e-2) + + def test_very_high_bitrate_floor(self): + # 100 Mbps + file_size = int(100_000_000 * 10 / 8) + score = score_bitrate(file_size, 10.0) + # excess ratio = 100/8 - 1 = 11.5, penalty = min(0.5, 11.5*0.2) = 0.5 + # score = max(0.5, 1.0 - 0.5) = 0.5 + assert score == 0.5 + + def test_1mbps_file_10s(self): + file_size = 1_000_000 # 1 MB + score = score_bitrate(file_size, 10.0) + # bitrate = 8*1M/10 = 0.8 Mbps + assert 0.3 < score < 0.7 + + +# ── calculate_total_score ─────────────────────────────────────────────────── + + +class TestCalculateTotalScore: + def test_perfect_score(self): + total = calculate_total_score(1.0, 1.0, 1.0, 1.0) + assert total == 1.0 + + def test_zero_score(self): + total = calculate_total_score(0.0, 0.0, 0.0, 0.0) + assert total == 0.0 + + def test_weighted_sum(self): + # 各维度不同分数 + q, r, d, b = 0.8, 0.6, 0.9, 0.7 + expected = WEIGHT_QUALITY * q + WEIGHT_RESOLUTION * r + WEIGHT_DURATION * d + WEIGHT_BITRATE * b + total = calculate_total_score(q, r, d, b) + assert total == pytest.approx(expected, rel=1e-4) + + def test_quality_dominates(self): + # 质量分权重最高(0.5)，变化影响最大 + base = calculate_total_score(0.5, 0.5, 0.5, 0.5) + quality_up = calculate_total_score(1.0, 0.5, 0.5, 0.5) + resolution_up = calculate_total_score(0.5, 1.0, 0.5, 0.5) + # 质量分变化带来的差异最大 + assert (quality_up - base) > (resolution_up - base) + + def test_rounded_to_4_decimals(self): + # 1/3 这样的无限小数应该被截断 + total = calculate_total_score(1 / 3, 1 / 3, 1 / 3, 1 / 3) + assert len(str(total).split(".")[-1]) <= 4 + + +# ── score_asset_detail ────────────────────────────────────────────────────── + + +class TestScoreAssetDetail: + def test_full_asset(self): + detail = score_asset_detail( + asset_id="test_001", + quality=80.0, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + assert detail.asset_id == "test_001" + assert detail.quality_score == pytest.approx(0.8, rel=1e-3) + assert detail.resolution_score == 1.0 + assert detail.duration_score == 1.0 + assert 0.0 < detail.total_score <= 1.0 + assert detail.duration == 10.0 + + def test_no_quality_default_05(self): + detail = score_asset_detail( + asset_id="a1", + quality=None, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + assert detail.quality_score == 0.5 + + def test_quality_100_is_1_0(self): + detail = score_asset_detail( + asset_id="a1", + quality=100.0, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + assert detail.quality_score == 1.0 + + def test_quality_zero_is_zero(self): + detail = score_asset_detail( + asset_id="a1", + quality=0.0, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + assert detail.quality_score == 0.0 + + def test_all_unknown_medium_score(self): + detail = score_asset_detail( + asset_id="a1", + quality=None, + width=None, + height=None, + duration=None, + file_size=0, + ) + # 全部未知：质量0.5，分辨率0.5，时长0.5，码率0.5 + assert detail.total_score == pytest.approx(0.5, rel=1e-3) + + def test_custom_target_resolution(self): + detail = score_asset_detail( + asset_id="a1", + quality=100.0, + width=1280, + height=720, + duration=10.0, + file_size=5_000_000, + target_width=1280, + target_height=720, + ) + assert detail.resolution_score == 1.0 + + def test_scores_are_rounded(self): + detail = score_asset_detail( + asset_id="a1", + quality=33.3, + width=854, + height=480, + duration=1.5, + file_size=1_000_000, + ) + # 所有分数字符串长度不超过 0.xxxx 格式 + for attr in ["quality_score", "resolution_score", "duration_score", "bitrate_score", "total_score"]: + val = getattr(detail, attr) + assert isinstance(val, float) + + +# ── _bucket_by_duration ───────────────────────────────────────────────────── + + +class TestBucketByDuration: + def test_short_bucket(self): + d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 3.0) + assert _bucket_by_duration(d) == "short" + + def test_short_bucket_boundary(self): + d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 4.9) + assert _bucket_by_duration(d) == "short" + + def test_medium_bucket(self): + d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 10.0) + assert _bucket_by_duration(d) == "medium" + + def test_medium_lower_boundary(self): + d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, SHORT_BUCKET_MAX) + assert _bucket_by_duration(d) == "medium" + + def test_medium_upper_boundary(self): + d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 14.9) + assert _bucket_by_duration(d) == "medium" + + def test_long_bucket(self): + d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 20.0) + assert _bucket_by_duration(d) == "long" + + def test_long_lower_boundary(self): + d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, MEDIUM_BUCKET_MAX) + assert _bucket_by_duration(d) == "long" + + def test_none_duration(self): + d = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, None) + assert _bucket_by_duration(d) == "unknown" + + +# ── diverse_selection ─────────────────────────────────────────────────────── + + +def _make_scored(items: list[tuple[str, float, float]]) -> list[AssetScoreDetail]: + """构造评分列表: (asset_id, total_score, duration)""" + return [ + AssetScoreDetail( + asset_id=aid, + total_score=score, + quality_score=score, + resolution_score=score, + duration_score=score, + bitrate_score=score, + duration=dur, + ) + for aid, score, dur in items + ] + + +class TestDiverseSelection: + def test_empty_input(self): + result = diverse_selection([], 5) + assert result == [] + + def test_zero_count(self): + scored = _make_scored([("a1", 0.9, 10.0)]) + result = diverse_selection(scored, 0) + assert result == [] + + def test_negative_count(self): + scored = _make_scored([("a1", 0.9, 10.0)]) + result = diverse_selection(scored, -1) + assert result == [] + + def test_fewer_than_count(self): + scored = _make_scored([("a1", 0.9, 10.0)]) + result = diverse_selection(scored, 5) + assert len(result) == 1 + + def test_mixed_buckets_diversity(self): + # 3短 + 3中 + 3长，取6个 + items = [ + ("s1", 0.9, 2.0), + ("s2", 0.8, 3.0), + ("s3", 0.7, 4.0), + ("m1", 0.95, 8.0), + ("m2", 0.85, 10.0), + ("m3", 0.75, 12.0), + ("l1", 0.92, 20.0), + ("l2", 0.82, 25.0), + ("l3", 0.72, 30.0), + ] + scored = _make_scored(items) + scored.sort(key=lambda d: d.total_score, reverse=True) + result = diverse_selection(scored, 6) + assert len(result) == 6 + # 每个桶至少1个（base_quota = max(1, 6//3) = 2） + ids = [r.asset_id for r in result] + short_count = sum(1 for r in result if r.duration and r.duration < SHORT_BUCKET_MAX) + medium_count = sum(1 for r in result if r.duration and SHORT_BUCKET_MAX <= r.duration < MEDIUM_BUCKET_MAX) + long_count = sum(1 for r in result if r.duration and r.duration >= MEDIUM_BUCKET_MAX) + assert short_count >= 1 + assert medium_count >= 1 + assert long_count >= 1 + + def test_all_short_fallback_to_global(self): + items = [("s1", 0.9, 2.0), ("s2", 0.8, 3.0), ("s3", 0.7, 4.0)] + scored = _make_scored(items) + result = diverse_selection(scored, 3) + assert len(result) == 3 + # 都是短素材，只能取短的 + assert all(r.duration and r.duration < SHORT_BUCKET_MAX for r in result) + + def test_sorted_by_score_descending(self): + items = [("a1", 0.5, 10.0), ("a2", 0.9, 10.0), ("a3", 0.7, 10.0)] + scored = _make_scored(items) + scored.sort(key=lambda d: d.total_score, reverse=True) + result = diverse_selection(scored, 3) + assert len(result) == 3 + assert result[0].total_score >= result[1].total_score >= result[2].total_score + + def test_count_one_each_bucket(self): + # count=3, base_quota=max(1,1)=1，每桶1个共3个 + items = [ + ("s1", 0.9, 2.0), + ("m1", 0.95, 8.0), + ("l1", 0.92, 20.0), + ] + scored = _make_scored(items) + scored.sort(key=lambda d: d.total_score, reverse=True) + result = diverse_selection(scored, 3) + assert len(result) == 3 + # 每桶1个 + assert any(r.duration and r.duration < SHORT_BUCKET_MAX for r in result) + assert any(r.duration and SHORT_BUCKET_MAX <= r.duration < MEDIUM_BUCKET_MAX for r in result) + assert any(r.duration and r.duration >= MEDIUM_BUCKET_MAX for r in result) + + def test_unknown_duration_used_last(self): + items = [ + ("u1", 0.99, None), # 分最高但未知 + ("s1", 0.9, 2.0), + ("m1", 0.8, 10.0), + ("l1", 0.7, 20.0), + ] + scored = _make_scored(items) + scored.sort(key=lambda d: d.total_score, reverse=True) + result = diverse_selection(scored, 3) + # 前3个应该是三个已知桶各一个 + ids = [r.asset_id for r in result] + # u1 不应该在前3（因为 unknown 桶最后才用） + assert "s1" in ids + assert "m1" in ids + assert "l1" in ids + + def test_no_duplicates(self): + items = [("s1", 0.9, 2.0), ("s2", 0.8, 3.0)] + scored = _make_scored(items) + result = diverse_selection(scored, 5) + ids = [r.asset_id for r in result] + assert len(ids) == len(set(ids)) + + def test_many_more_than_count(self): + # 30个素材，取6个 + items = [] + for i in range(10): + items.append((f"s{i}", 0.9 - i * 0.05, 2.0 + i * 0.2)) + items.append((f"m{i}", 0.9 - i * 0.03, 6.0 + i * 0.8)) + items.append((f"l{i}", 0.9 - i * 0.04, 16.0 + i * 1.5)) + scored = _make_scored(items) + scored.sort(key=lambda d: d.total_score, reverse=True) + result = diverse_selection(scored, 6) + assert len(result) == 6 + # 有多样性 + durations = [r.duration for r in result] + short = sum(1 for d in durations if d and d < SHORT_BUCKET_MAX) + medium = sum(1 for d in durations if d and SHORT_BUCKET_MAX <= d < MEDIUM_BUCKET_MAX) + long_ = sum(1 for d in durations if d and d >= MEDIUM_BUCKET_MAX) + assert short >= 1 + assert medium >= 1 + assert long_ >= 1 + + +# ── filter_candidates ─────────────────────────────────────────────────────── + + +class TestFilterCandidates: + def test_empty_list(self): + candidates, filtered = filter_candidates([]) + assert candidates == [] + assert filtered == 0 + + def test_ready_video_passes(self): + assets = [MockAsset(id="a1", status=MockStatus("ready"), mime_type="video/mp4")] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert filtered == 0 + + def test_non_ready_filtered(self): + assets = [ + MockAsset(id="a1", status=MockStatus("processing"), mime_type="video/mp4"), + MockAsset(id="a2", status=MockStatus("ready"), mime_type="video/mp4"), + ] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert candidates[0].id == "a2" + assert filtered == 0 # 非ready不算filtered_out（filtered_out只算质量分过滤的） + + def test_non_video_filtered(self): + assets = [ + MockAsset(id="a1", mime_type="image/jpeg"), + MockAsset(id="a2", mime_type="video/mp4"), + ] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert candidates[0].id == "a2" + + def test_low_quality_filtered(self): + assets = [ + MockAsset(id="low", quality_score=20.0), + MockAsset(id="high", quality_score=80.0), + ] + candidates, filtered = filter_candidates(assets, min_quality_score=30.0) + assert len(candidates) == 1 + assert candidates[0].id == "high" + assert filtered == 1 + + def test_quality_none_passes(self): + assets = [MockAsset(id="a1", quality_score=None)] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert filtered == 0 + + def test_quality_exact_min_passes(self): + assets = [MockAsset(id="a1", quality_score=30.0)] + candidates, filtered = filter_candidates(assets, min_quality_score=30.0) + assert len(candidates) == 1 + assert filtered == 0 + + def test_string_status(self): + # status 是字符串不是 Enum + @dataclass + class StrAsset: + id: str = "a1" + status: str = "ready" + mime_type: str = "video/mp4" + quality_score: float = 80.0 + width: int = 1920 + height: int = 1080 + duration: float = 10.0 + file_size: int = 5_000_000 + + assets = [StrAsset()] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + + def test_empty_mime_type(self): + assets = [MockAsset(id="a1", mime_type="")] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 0 + + def test_none_mime_type(self): + # mime_type 是 None + @dataclass + class NoneMimeAsset: + id: str = "a1" + status: Any = None + mime_type: str | None = None + quality_score: float = 80.0 + width: int = 1920 + height: int = 1080 + duration: float = 10.0 + file_size: int = 5_000_000 + + def __post_init__(self): + if self.status is None: + self.status = MockStatus("ready") + + assets = [NoneMimeAsset()] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 0 + + def test_custom_min_quality(self): + assets = [ + MockAsset(id="low", quality_score=40.0), + MockAsset(id="high", quality_score=60.0), + ] + candidates, filtered = filter_candidates(assets, min_quality_score=50.0) + assert len(candidates) == 1 + assert filtered == 1