From 012d97ae4edc803f1ad2127e89702991de711cba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=89=8D=E7=AB=AFAgent?= Date: Wed, 5 Aug 2026 11:13:19 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E6=99=BA?= =?UTF-8?q?=E8=83=BD=E9=80=89=E7=B4=A0=E6=9D=90=E9=80=BB=E8=BE=91=EF=BC=8C?= =?UTF-8?q?=E8=BF=81=E7=A7=BB=20generation=5Ftasks=20smart=20=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=E5=88=B0=20smart=5Fmatch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - generation_tasks.py 的 smart 模式从内联 quality+duration 排序改为调用 packages/domain/smart_match.py 的 smart_select_assets()，复用统一的多维 评分（质量40%+时长30%+新鲜度20%+未使用10%）和时长分桶多样性算法 - 删除已无引用的旧代码： - packages/domain/asset_scoring.py（旧4维评分：质量0.5+分辨率0.2+时长0.2+码率0.1） - apps/api/app/services/smart_asset_selector.py（SmartAssetSelector 类，无生产调用方） - 对应测试文件：test_asset_scoring.py、test_asset_scoring_domain.py、 tests/unit/domain/test_asset_scoring.py、test_smart_asset_selector.py - 更新 test_asset_select_mode.py 的 smart 模式测试用例，适配新评分体系 - 全量单测 13587 passed 通过 --- apps/api/app/api/routes/generation_tasks.py | 19 +- apps/api/app/services/smart_asset_selector.py | 164 ---- packages/domain/asset_scoring.py | 372 --------- tests/unit/domain/test_asset_scoring.py | 530 ------------ tests/unit/test_asset_scoring.py | 755 ------------------ tests/unit/test_asset_scoring_domain.py | 515 ------------ tests/unit/test_asset_select_mode.py | 40 +- tests/unit/test_smart_asset_selector.py | 413 ---------- 8 files changed, 29 insertions(+), 2779 deletions(-) delete mode 100755 apps/api/app/services/smart_asset_selector.py delete mode 100755 packages/domain/asset_scoring.py delete mode 100755 tests/unit/domain/test_asset_scoring.py delete mode 100755 tests/unit/test_asset_scoring.py delete mode 100755 tests/unit/test_asset_scoring_domain.py delete mode 100755 tests/unit/test_smart_asset_selector.py diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 795514544..d42be2e44 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -38,6 +38,7 @@ from packages.application import ( GetGenerationTaskUseCase, ListGeneratedVideosByTaskUseCase, ) +from packages.domain.smart_match import smart_select_assets logger = logging.getLogger(__name__) @@ -124,19 +125,11 @@ def _select_assets_from_library( return [a.id for a in selected] if mode == "smart": - # 智能匹配：按质量分降序 + 时长降序作为tiebreaker - # 注意：这里使用简单的 quality_score 排序保持向后兼容 - # 更复杂的4维评分+多样性策略由 SmartAssetSelector 服务提供（用于 AI 精选等场景） - scored_assets = sorted( - ready_video_assets, - key=lambda a: ( - -(a.quality_score if a.quality_score is not None else 0.0), - -(getattr(a, "duration", 0.0) or 0.0), - ), - ) - if count > 0: - scored_assets = scored_assets[:count] - return [a.id for a in scored_assets] + # 智能匹配：统一使用 packages/domain/smart_match.py 的多维评分+多样性选取 + # 评分维度：质量分(40%) + 时长适配(30%) + 新鲜度(20%) + 未使用加分(10%) + limit = count if count > 0 else None + results = smart_select_assets(ready_video_assets, limit=limit, kind="video") + return [r.asset.id for r in results] # 默认 all 模式：返回全部 ready 视频素材 return [a.id for a in ready_video_assets] diff --git a/apps/api/app/services/smart_asset_selector.py b/apps/api/app/services/smart_asset_selector.py deleted file mode 100755 index 6b2c17856..000000000 --- a/apps/api/app/services/smart_asset_selector.py +++ /dev/null @@ -1,164 +0,0 @@ -"""SmartAssetSelector — 智能素材选择服务. - -根据多维度评分从素材库中自动选择最优视频素材， -用于一键生成等需要自动选取素材的场景。 - -评分维度（加权求和，总分 0-1）： - - 质量分（quality_score）：权重 0.5 — 来自人工或AI的质量评分 - - 分辨率适配：权重 0.2 — 分辨率越接近 1080p 得分越高 - - 时长合理性：权重 0.2 — 3-30 秒区间最佳，过短/过长扣分 - - 码率质量：权重 0.1 — 用文件大小/时长估算，码率适中得分高 - -特性： - - 最低质量分门槛：自动过滤低质量素材 - - 时长多样性：保证选出的素材时长分布均匀（短/中/长各占一定比例） - - 兼容全部模式：素材库模式和项目模式都可用 - -纯逻辑部分已抽离到 packages.domain.asset_scoring。 -""" - -from __future__ import annotations - -import logging - -from packages.domain.asset_scoring import MEDIUM_BUCKET_MAX as _MEDIUM_BUCKET_MAX # noqa: F401 - re-export for tests -from packages.domain.asset_scoring import SHORT_BUCKET_MAX as _SHORT_BUCKET_MAX # noqa: F401 - re-export for tests -from packages.domain.asset_scoring import ( - AssetScoreDetail, - SmartSelectResult, - diverse_selection, - filter_candidates, - score_asset_detail, -) - -logger = logging.getLogger(__name__) - - -class SmartAssetSelector: - """智能素材选择器. - - 从一组素材中按综合评分选择最优的 N 个， - 同时保证时长分布的多样性。 - """ - - def __init__( - self, - 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 - self.target_height = target_height - - # ── 公开方法 ────────────────────────────────────────────────────────────── - - def select( - self, - assets: list, - count: int = 0, - *, - ensure_diversity: bool = True, - ) -> SmartSelectResult: - """从素材列表中智能选择最优素材. - - Args: - assets: Asset 实体列表（需要有 id/quality_score/width/height/duration/file_size 属性） - count: 选取数量，0 表示全部符合条件的 - ensure_diversity: 是否保证时长多样性（默认开启） - - Returns: - SmartSelectResult 选择结果 - """ - # 1. 过滤：只保留 ready 状态的视频素材 + 最低质量分门槛 - candidates, filtered_out = filter_candidates(assets, self.min_quality_score) - - if not candidates: - return SmartSelectResult( - selected_ids=[], - total_candidates=0, - filtered_out=filtered_out, - avg_score=0.0, - details=[], - ) - - # 2. 对每个候选素材评分 - scored: list[AssetScoreDetail] = [] - for asset in candidates: - 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. 按总分降序排列 - scored.sort(key=lambda d: d.total_score, reverse=True) - - # 4. 多样性选择（如果需要且数量有限制） - if ensure_diversity and count > 0 and len(scored) > count: - selected = diverse_selection(scored, count) - else: - # 无数量限制或不要求多样性，直接按排名取 - selected = scored if count <= 0 else scored[:count] - - avg_score = sum(d.total_score for d in selected) / len(selected) if selected else 0.0 - - result = SmartSelectResult( - selected_ids=[d.asset_id for d in selected], - total_candidates=len(candidates), - filtered_out=filtered_out, - avg_score=avg_score, - details=selected, - ) - - logger.info( - "智能素材选择完成: 候选=%d, 过滤=%d, 选中=%d, 平均分=%.3f", - result.total_candidates, - result.filtered_out, - len(result.selected_ids), - result.avg_score, - ) - return result - - # ── 向后兼容：私有方法别名（委托给 asset_scoring 纯函数） ──────────────── - - def _score_asset(self, asset) -> AssetScoreDetail: - """对单个素材进行多维度评分（向后兼容）.""" - return 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, - ) - - def _score_resolution(self, width: int | None, height: int | None) -> float: - """分辨率评分（向后兼容）.""" - from packages.domain.asset_scoring import score_resolution - - return score_resolution(width, height, self.target_width, self.target_height) - - def _score_duration(self, duration: float | None) -> float: - """时长评分（向后兼容）.""" - from packages.domain.asset_scoring import score_duration - - return score_duration(duration) - - def _score_bitrate(self, file_size: int, duration: float | None) -> float: - """码率评分（向后兼容）.""" - from packages.domain.asset_scoring import score_bitrate - - return score_bitrate(file_size, duration) - - def _diverse_selection(self, scored: list[AssetScoreDetail], count: int) -> list[AssetScoreDetail]: - """多样性选择（向后兼容）.""" - return diverse_selection(scored, count) diff --git a/packages/domain/asset_scoring.py b/packages/domain/asset_scoring.py deleted file mode 100755 index dfb61ebae..000000000 --- a/packages/domain/asset_scoring.py +++ /dev/null @@ -1,372 +0,0 @@ -"""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/domain/test_asset_scoring.py b/tests/unit/domain/test_asset_scoring.py deleted file mode 100755 index 983a3f21d..000000000 --- a/tests/unit/domain/test_asset_scoring.py +++ /dev/null @@ -1,530 +0,0 @@ -"""asset_scoring 单元测试 - wave162 - -覆盖： -- 分辨率评分 score_resolution -- 时长评分 score_duration -- 码率评分 score_bitrate -- 加权总分 calculate_total_score -- 单个素材评分 score_asset_detail -- 时长分桶 _bucket_by_duration -- 多样性选择 diverse_selection -- 候选过滤 filter_candidates -""" - -from dataclasses import dataclass - -import pytest - -from packages.domain.asset_scoring import ( - AssetScoreDetail, - SmartSelectResult, - _bucket_by_duration, - calculate_total_score, - diverse_selection, - filter_candidates, - score_asset_detail, - score_bitrate, - score_duration, - score_resolution, -) - -# ============================================================ -# score_resolution -# ============================================================ - - -class TestScoreResolution: - def test_none_width_returns_mid(self): - assert score_resolution(None, 1080) == 0.5 - - def test_none_height_returns_mid(self): - assert score_resolution(1920, None) == 0.5 - - def test_zero_dimension_returns_mid(self): - assert score_resolution(0, 1080) == 0.5 - assert score_resolution(1920, 0) == 0.5 - assert score_resolution(-1, 1080) == 0.5 - - def test_exact_target_returns_1(self): - assert score_resolution(1920, 1080) == 1.0 - - def test_higher_than_target_returns_1(self): - assert score_resolution(3840, 2160) == 1.0 # 4K - assert score_resolution(2560, 1440) == 1.0 # 2K - - def test_lower_than_target_linear_decay(self): - # 720p = 1280*720 / 1920*1080 = 0.444 ratio - # score = 0.3 + 0.7 * 0.444 = 0.611 - score = score_resolution(1280, 720) - assert 0.55 < score < 0.7 - - def test_very_low_has_floor(self): - # 最低不低于 0.1 - score = score_resolution(100, 100) - assert score >= 0.1 - - def test_480p_still_reasonable(self): - score = score_resolution(640, 480) - assert 0.3 < score < 0.5 - - def test_custom_target(self): - score = score_resolution(1280, 720, target_width=1280, target_height=720) - assert score == 1.0 - - def test_between_0_and_1(self): - for w, h in [(1920, 1080), (1280, 720), (640, 480), (3840, 2160)]: - s = score_resolution(w, h) - assert 0.0 <= s <= 1.0 - - -# ============================================================ -# score_duration -# ============================================================ - - -class TestScoreDuration: - def test_none_returns_mid(self): - assert score_duration(None) == 0.5 - - def test_zero_or_negative_returns_mid(self): - assert score_duration(0) == 0.5 - assert score_duration(-1) == 0.5 - - def test_optimal_range_returns_1(self): - assert score_duration(3.0) == 1.0 - assert score_duration(10.0) == 1.0 - assert score_duration(30.0) == 1.0 - assert score_duration(15.0) == 1.0 - - def test_short_duration_linear_decay(self): - # 1.5s: ratio = 1.5/3 = 0.5, score = 0.3 + 0.7*0.5 = 0.65 - score = score_duration(1.5) - assert score == pytest.approx(0.65) - - def test_very_short_above_floor(self): - score = score_duration(0.1) - assert 0.3 <= score < 0.5 - - def test_long_duration_penalty(self): - # 40s: excess=10, penalty=10/10*0.1=0.1, score=0.9 - score = score_duration(40.0) - assert score == pytest.approx(0.9) - - def test_very_long_minimum_floor(self): - # 超过很多，最低 0.2 - score = score_duration(1000.0) - assert score >= 0.2 - assert score < 0.5 - - def test_just_below_optimal(self): - score = score_duration(2.9) - assert 0.9 < score < 1.0 - - def test_just_above_optimal(self): - score = score_duration(30.1) - assert 0.9 < score < 1.0 - - -# ============================================================ -# score_bitrate -# ============================================================ - - -class TestScoreBitrate: - def test_no_file_size_returns_mid(self): - assert score_bitrate(0, 10.0) == 0.5 - - def test_no_duration_returns_mid(self): - assert score_bitrate(1000000, None) == 0.5 - assert score_bitrate(1000000, 0) == 0.5 - assert score_bitrate(1000000, -1) == 0.5 - - def test_optimal_range_returns_1(self): - # 5 Mbps for 10s = 5*10^6 * 10 / 8 = 6,250,000 bytes - size_5mbps_10s = int(5_000_000 * 10 / 8) - assert score_bitrate(size_5mbps_10s, 10.0) == 1.0 - - def test_low_bitrate_decay(self): - # 500 Kbps for 10s - size_500kbps = int(500_000 * 10 / 8) - score = score_bitrate(size_500kbps, 10.0) - assert 0.3 < score < 0.7 - - def test_high_bitrate_moderate_penalty(self): - # 16 Mbps (2x optimal high), excess=1.0, penalty=min(0.5, 1.0*0.2)=0.2 - # score = 0.8 - size_16mbps = int(16_000_000 * 10 / 8) - score = score_bitrate(size_16mbps, 10.0) - assert 0.7 < score < 0.9 - - def test_very_high_bitrate_floor(self): - # 极高码率，最低 0.5 - huge_size = 10**9 # 1GB for 1s = 8Gbps - score = score_bitrate(huge_size, 1.0) - assert score >= 0.5 - - def test_between_0_and_1(self): - for size, dur in [(1000, 1), (1000000, 10), (100000000, 5)]: - s = score_bitrate(size, dur) - assert 0.0 <= s <= 1.0 - - -# ============================================================ -# calculate_total_score -# ============================================================ - - -class TestCalculateTotalScore: - def test_all_perfect_equals_1(self): - assert calculate_total_score(1.0, 1.0, 1.0, 1.0) == 1.0 - - def test_all_zero_equals_0(self): - assert calculate_total_score(0.0, 0.0, 0.0, 0.0) == 0.0 - - def test_weighted_sum(self): - # 0.5*0.5 + 0.2*0.5 + 0.2*0.5 + 0.1*0.5 = 0.25+0.1+0.1+0.05 = 0.5 - assert calculate_total_score(0.5, 0.5, 0.5, 0.5) == pytest.approx(0.5) - - def test_quality_has_highest_weight(self): - # 只提高质量分，对比只提高其他 - q_high = calculate_total_score(1.0, 0.0, 0.0, 0.0) - r_high = calculate_total_score(0.0, 1.0, 0.0, 0.0) - assert q_high > r_high # 0.5 > 0.2 - - def test_bitrate_has_lowest_weight(self): - b_high = calculate_total_score(0.0, 0.0, 0.0, 1.0) - q_high = calculate_total_score(1.0, 0.0, 0.0, 0.0) - assert b_high < q_high # 0.1 < 0.5 - - def test_rounded_to_4_decimals(self): - result = calculate_total_score(0.3333, 0.3333, 0.3333, 0.3333) - assert round(result, 4) == result - - -# ============================================================ -# score_asset_detail -# ============================================================ - - -class TestScoreAssetDetail: - def test_returns_detail_object(self): - detail = score_asset_detail( - asset_id="a1", - quality=80.0, - width=1920, - height=1080, - duration=10.0, - file_size=5_000_000, - ) - assert isinstance(detail, AssetScoreDetail) - assert detail.asset_id == "a1" - assert 0.0 <= detail.total_score <= 1.0 - - def test_perfect_asset_high_score(self): - detail = score_asset_detail( - asset_id="perfect", - quality=100.0, - width=1920, - height=1080, - duration=10.0, - file_size=6_250_000, # 5Mbps for 10s - ) - assert detail.total_score > 0.9 - - def test_quality_none_defaults_mid(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_normalized(self): - detail = score_asset_detail( - asset_id="a1", - quality=50.0, - width=1920, - height=1080, - duration=10.0, - file_size=5_000_000, - ) - assert detail.quality_score == pytest.approx(0.5) - - 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_total_score_matches_components(self): - detail = score_asset_detail( - asset_id="a1", - quality=80.0, - width=1920, - height=1080, - duration=10.0, - file_size=5_000_000, - ) - expected = calculate_total_score( - detail.quality_score, - detail.resolution_score, - detail.duration_score, - detail.bitrate_score, - ) - assert detail.total_score == pytest.approx(expected, abs=0.001) - - -# ============================================================ -# _bucket_by_duration -# ============================================================ - - -class TestBucketByDuration: - def test_none_is_unknown(self): - item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, None) - assert _bucket_by_duration(item) == "unknown" - - def test_short(self): - item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 3.0) - assert _bucket_by_duration(item) == "short" - - def test_short_boundary(self): - item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 4.9) - assert _bucket_by_duration(item) == "short" - - def test_medium(self): - item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 10.0) - assert _bucket_by_duration(item) == "medium" - - def test_medium_boundary(self): - item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 5.0) - assert _bucket_by_duration(item) == "medium" - - def test_medium_upper_boundary(self): - item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 14.9) - assert _bucket_by_duration(item) == "medium" - - def test_long(self): - item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 20.0) - assert _bucket_by_duration(item) == "long" - - def test_long_boundary(self): - item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 15.0) - assert _bucket_by_duration(item) == "long" - - -# ============================================================ -# diverse_selection -# ============================================================ - - -def _make_detail(asset_id: str, score: float, duration: float) -> AssetScoreDetail: - return AssetScoreDetail( - asset_id=asset_id, - total_score=score, - quality_score=score, - resolution_score=score, - duration_score=score, - bitrate_score=score, - duration=duration, - ) - - -class TestDiverseSelection: - def test_empty_input_returns_empty(self): - assert diverse_selection([], 5) == [] - - def test_zero_count_returns_empty(self): - items = [_make_detail("a1", 0.9, 10.0)] - assert diverse_selection(items, 0) == [] - - def test_negative_count_returns_empty(self): - items = [_make_detail("a1", 0.9, 10.0)] - assert diverse_selection(items, -1) == [] - - def test_fewer_items_than_count(self): - items = [_make_detail("a1", 0.9, 10.0), _make_detail("a2", 0.8, 3.0)] - result = diverse_selection(items, 10) - assert len(result) == 2 - - def test_picks_top_from_each_bucket(self): - # 3个桶各有3个素材，选3个 - items = [ - _make_detail("s1", 0.95, 2.0), - _make_detail("m1", 0.9, 10.0), - _make_detail("l1", 0.85, 20.0), - _make_detail("s2", 0.8, 3.0), - _make_detail("m2", 0.75, 8.0), - _make_detail("l2", 0.7, 25.0), - ] - result = diverse_selection(items, 3) - assert len(result) == 3 - ids = [d.asset_id for d in result] - assert "s1" in ids - assert "m1" in ids - assert "l1" in ids - - def test_base_quota_when_count_large(self): - # count=6, base_quota=max(1, 6//3)=2 - items = [ - _make_detail("s1", 1.0, 2.0), - _make_detail("s2", 0.9, 3.0), - _make_detail("s3", 0.8, 4.0), - _make_detail("m1", 0.95, 10.0), - _make_detail("m2", 0.85, 12.0), - _make_detail("l1", 0.92, 20.0), - _make_detail("l2", 0.82, 30.0), - ] - result = diverse_selection(items, 6) - assert len(result) == 6 - ids = [d.asset_id for d in result] - # 每桶至少2个 - short_count = sum(1 for d in result if d.duration and d.duration < 5) - assert short_count >= 2 - - def test_remaining_filled_by_global_score(self): - # 只有2个桶有内容，count=5，配额用完后剩余从全局取 - items = [ - _make_detail("s1", 1.0, 2.0), - _make_detail("s2", 0.9, 3.0), - _make_detail("m1", 0.95, 10.0), - _make_detail("m2", 0.8, 12.0), - _make_detail("s3", 0.7, 4.0), - _make_detail("s4", 0.6, 1.0), - _make_detail("m3", 0.5, 8.0), - ] - result = diverse_selection(items, 5) - assert len(result) == 5 - # 最高分的都应该在 - ids = [d.asset_id for d in result] - assert "s1" in ids - assert "m1" in ids - - def test_single_bucket(self): - items = [ - _make_detail("s1", 1.0, 2.0), - _make_detail("s2", 0.9, 3.0), - _make_detail("s3", 0.8, 4.0), - ] - result = diverse_selection(items, 2) - assert len(result) == 2 - assert result[0].asset_id == "s1" - assert result[1].asset_id == "s2" - - def test_unknown_duration_fallback(self): - # 已知素材不够时用未知时长的补充 - items = [ - _make_detail("s1", 1.0, 2.0), - _make_detail("u1", 0.95, None), - _make_detail("u2", 0.9, None), - ] - result = diverse_selection(items, 3) - assert len(result) == 3 - ids = [d.asset_id for d in result] - assert "s1" in ids - assert "u1" in ids - - def test_no_duplicates(self): - items = [ - _make_detail("s1", 1.0, 2.0), - _make_detail("m1", 0.9, 10.0), - ] - result = diverse_selection(items, 5) - ids = [d.asset_id for d in result] - assert len(ids) == len(set(ids)) - - -# ============================================================ -# filter_candidates -# ============================================================ - - -@dataclass -class FakeAsset: - status: str = "ready" - mime_type: str = "video/mp4" - quality_score: float | None = 50.0 - - -class TestFilterCandidates: - def test_ready_video_passes(self): - assets = [FakeAsset()] - candidates, filtered = filter_candidates(assets) - assert len(candidates) == 1 - assert filtered == 0 - - def test_non_ready_filtered(self): - assets = [FakeAsset(status="uploading"), FakeAsset(status="processing")] - candidates, filtered = filter_candidates(assets) - assert len(candidates) == 0 - assert filtered == 0 # 被状态过滤的不计入质量门槛 - - def test_non_video_filtered(self): - assets = [FakeAsset(mime_type="image/jpeg"), FakeAsset(mime_type="audio/mp3")] - candidates, filtered = filter_candidates(assets) - assert len(candidates) == 0 - - def test_low_quality_filtered(self): - assets = [FakeAsset(quality_score=10.0), FakeAsset(quality_score=80.0)] - candidates, filtered = filter_candidates(assets, min_quality_score=30.0) - assert len(candidates) == 1 - assert filtered == 1 - - def test_quality_none_passes(self): - assets = [FakeAsset(quality_score=None)] - candidates, filtered = filter_candidates(assets) - assert len(candidates) == 1 - assert filtered == 0 - - def test_exactly_min_quality_passes(self): - assets = [FakeAsset(quality_score=30.0)] - candidates, filtered = filter_candidates(assets, min_quality_score=30.0) - assert len(candidates) == 1 - - def test_custom_min_quality(self): - assets = [ - FakeAsset(quality_score=40.0), - FakeAsset(quality_score=60.0), - FakeAsset(quality_score=80.0), - ] - candidates, filtered = filter_candidates(assets, min_quality_score=50.0) - assert len(candidates) == 2 - assert filtered == 1 - - def test_empty_input(self): - candidates, filtered = filter_candidates([]) - assert candidates == [] - assert filtered == 0 - - def test_mime_type_none(self): - # None 的 mime_type 也应该被过滤掉（不是video开头） - asset = FakeAsset(mime_type="") - candidates, _ = filter_candidates([asset]) - assert len(candidates) == 0 - - def test_with_enum_status(self): - from enum import Enum - - class StatusEnum(Enum): - READY = "ready" - UPLOADING = "uploading" - - @dataclass - class EnumAsset: - status: StatusEnum = StatusEnum.READY - mime_type: str = "video/mp4" - quality_score: float = 50.0 - - assets = [EnumAsset()] - candidates, filtered = filter_candidates(assets) - assert len(candidates) == 1 diff --git a/tests/unit/test_asset_scoring.py b/tests/unit/test_asset_scoring.py deleted file mode 100755 index 4a58a1313..000000000 --- a/tests/unit/test_asset_scoring.py +++ /dev/null @@ -1,755 +0,0 @@ -"""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 ( - MEDIUM_BUCKET_MAX, - MIN_QUALITY_SCORE, - OPTIMAL_DURATION_MAX, - OPTIMAL_DURATION_MIN, - SHORT_BUCKET_MAX, - TARGET_HEIGHT, - TARGET_WIDTH, - WEIGHT_BITRATE, - WEIGHT_DURATION, - WEIGHT_QUALITY, - WEIGHT_RESOLUTION, - AssetScoreDetail, - SmartSelectResult, - _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 diff --git a/tests/unit/test_asset_scoring_domain.py b/tests/unit/test_asset_scoring_domain.py deleted file mode 100755 index 2a1cccd9a..000000000 --- a/tests/unit/test_asset_scoring_domain.py +++ /dev/null @@ -1,515 +0,0 @@ -"""资产评分纯逻辑单元测试 — wave129.""" - -from __future__ import annotations - -from dataclasses import dataclass - -import pytest - -from packages.domain.asset_scoring import ( - WEIGHT_BITRATE, - WEIGHT_DURATION, - WEIGHT_QUALITY, - WEIGHT_RESOLUTION, - AssetScoreDetail, - SmartSelectResult, - _bucket_by_duration, - calculate_total_score, - diverse_selection, - filter_candidates, - score_asset_detail, - score_bitrate, - score_duration, - score_resolution, -) - -# ── 常量与权重 ────────────────────────────────────────────────────────────── - - -class TestWeights: - 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_quality_is_highest_weight(self): - assert WEIGHT_QUALITY > WEIGHT_RESOLUTION - assert WEIGHT_QUALITY > WEIGHT_DURATION - assert WEIGHT_QUALITY > WEIGHT_BITRATE - - -# ── 分辨率评分 ────────────────────────────────────────────────────────────── - - -class TestScoreResolution: - def test_exact_target_full_score(self): - assert score_resolution(1920, 1080) == 1.0 - - def test_higher_than_target_full_score(self): - """4K 等高于目标分辨率也给满分.""" - assert score_resolution(3840, 2160) == 1.0 - assert score_resolution(2560, 1440) == 1.0 - - def test_seventytwop_less_than_one(self): - score = score_resolution(1280, 720) - assert 0.5 < score < 1.0 - - def test_fourheightyp_even_lower(self): - score_480 = score_resolution(854, 480) - score_720 = score_resolution(1280, 720) - assert score_480 < score_720 - - def test_none_returns_medium(self): - assert score_resolution(None, None) == 0.5 - assert score_resolution(None, 1080) == 0.5 - assert score_resolution(1920, None) == 0.5 - - def test_zero_or_negative_returns_medium(self): - assert score_resolution(0, 1080) == 0.5 - assert score_resolution(-100, 1080) == 0.5 - assert score_resolution(1920, 0) == 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_score_between_zero_one(self): - score = score_resolution(320, 240) - assert 0.0 < score < 1.0 - - def test_very_low_resolution_not_zero(self): - """低分也不会低于 0.1.""" - score = score_resolution(160, 120) - assert score >= 0.1 - - -# ── 时长评分 ──────────────────────────────────────────────────────────────── - - -class TestScoreDuration: - def test_optimal_range_full_score(self): - """3-30秒最佳区间满分.""" - assert score_duration(3.0) == 1.0 - assert score_duration(10.0) == 1.0 - assert score_duration(30.0) == 1.0 - - def test_very_short_lower_score(self): - score_1s = score_duration(1.0) - assert 0.3 <= score_1s < 1.0 - - def test_shorter_than_optimal_lower(self): - """越短分越低.""" - score_1 = score_duration(1.0) - score_2 = score_duration(2.0) - assert score_1 < score_2 - - def test_just_below_optimal(self): - score = score_duration(2.9) - assert score < 1.0 - assert score > 0.8 # 接近满分 - - def test_too_long_penalty(self): - score_30 = score_duration(30.0) - score_60 = score_duration(60.0) - assert score_60 < score_30 - - def test_very_long_minimum_floor(self): - """超长素材最低 0.2 分.""" - score = score_duration(1000.0) - assert score >= 0.2 - - def test_none_returns_medium(self): - assert score_duration(None) == 0.5 - - def test_zero_returns_medium(self): - assert score_duration(0) == 0.5 - assert score_duration(0.0) == 0.5 - - def test_negative_returns_medium(self): - assert score_duration(-5.0) == 0.5 - - -# ── 码率评分 ──────────────────────────────────────────────────────────────── - - -class TestScoreBitrate: - def test_optimal_bitrate_full_score(self): - """2-8 Mbps 区间满分.""" - # 5 Mbps, 10秒 = 50Mbit = 6.25MB = 6,250,000 字节 - size_5mbps_10s = int(5_000_000 * 10 / 8) - assert score_bitrate(size_5mbps_10s, 10.0) == 1.0 - - # 3 Mbps, 5秒 - size_3mbps_5s = int(3_000_000 * 5 / 8) - assert score_bitrate(size_3mbps_5s, 5.0) == 1.0 - - def test_low_bitrate_lower_score(self): - """码率低得分低.""" - # 500 Kbps - size_low = int(500_000 * 10 / 8) - score = score_bitrate(size_low, 10.0) - assert 0.3 <= score < 1.0 - - def test_high_bitrate_moderate_penalty(self): - """码率过高适度扣分，最低0.5.""" - # 50 Mbps，远超 8Mbps - size_high = int(50_000_000 * 10 / 8) - score = score_bitrate(size_high, 10.0) - assert 0.5 <= score < 1.0 - - def test_none_duration_returns_medium(self): - assert score_bitrate(1_000_000, None) == 0.5 - - def test_zero_file_size_returns_medium(self): - assert score_bitrate(0, 10.0) == 0.5 - - def test_zero_duration_returns_medium(self): - assert score_bitrate(1_000_000, 0) == 0.5 - assert score_bitrate(1_000_000, -5.0) == 0.5 - - -# ── 加权总分 ──────────────────────────────────────────────────────────────── - - -class TestCalculateTotalScore: - def test_all_perfect(self): - assert calculate_total_score(1.0, 1.0, 1.0, 1.0) == 1.0 - - def test_all_zero(self): - assert calculate_total_score(0.0, 0.0, 0.0, 0.0) == 0.0 - - def test_weighted_calculation(self): - """手动验证加权计算.""" - q, r, d, b = 0.8, 0.6, 0.4, 0.2 - expected = WEIGHT_QUALITY * q + WEIGHT_RESOLUTION * r + WEIGHT_DURATION * d + WEIGHT_BITRATE * b - assert calculate_total_score(q, r, d, b) == pytest.approx(expected, rel=1e-3) - - def test_quality_dominates(self): - """质量分权重最高，质量分变化影响最大.""" - score_high_quality = calculate_total_score(1.0, 0.5, 0.5, 0.5) - score_low_quality = calculate_total_score(0.0, 0.5, 0.5, 0.5) - diff_quality = score_high_quality - score_low_quality - - score_high_res = calculate_total_score(0.5, 1.0, 0.5, 0.5) - score_low_res = calculate_total_score(0.5, 0.0, 0.5, 0.5) - diff_res = score_high_res - score_low_res - - assert diff_quality > diff_res - - def test_rounded_to_4_decimals(self): - result = calculate_total_score(0.3333, 0.3333, 0.3333, 0.3333) - assert result == round(result, 4) - - -# ── 单个素材评分详情 ──────────────────────────────────────────────────────── - - -class TestScoreAssetDetail: - def test_normal_asset(self): - detail = score_asset_detail( - asset_id="asset_001", - quality=80.0, - width=1920, - height=1080, - duration=10.0, - file_size=5_000_000, - ) - assert detail.asset_id == "asset_001" - assert 0.0 <= detail.total_score <= 1.0 - assert detail.quality_score == pytest.approx(0.8) - assert detail.resolution_score == 1.0 - assert detail.duration_score == 1.0 - assert detail.duration == 10.0 - - def test_unknown_quality_defaults(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_hundred_is_one(self): - detail = score_asset_detail( - asset_id="a1", - quality=100.0, - width=1920, - height=1080, - duration=10.0, - file_size=1_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=1_000_000, - ) - assert detail.quality_score == 0.0 - - def test_custom_target_resolution(self): - detail = score_asset_detail( - asset_id="a1", - quality=50.0, - width=1280, - height=720, - duration=10.0, - file_size=1_000_000, - target_width=1280, - target_height=720, - ) - assert detail.resolution_score == 1.0 - - -# ── 时长分桶 ──────────────────────────────────────────────────────────────── - - -def _make_detail(asset_id: str, duration: float | None, score: float = 0.8) -> AssetScoreDetail: - return AssetScoreDetail( - asset_id=asset_id, - total_score=score, - quality_score=score, - resolution_score=score, - duration_score=score, - bitrate_score=score, - duration=duration, - ) - - -class TestBucketByDuration: - def test_short_bucket(self): - item = _make_detail("s1", duration=3.0) - assert _bucket_by_duration(item) == "short" - - def test_short_boundary(self): - item = _make_detail("s1", duration=4.9) - assert _bucket_by_duration(item) == "short" - - def test_medium_bucket(self): - item = _make_detail("m1", duration=10.0) - assert _bucket_by_duration(item) == "medium" - - def test_medium_boundary(self): - item = _make_detail("m1", duration=14.9) - assert _bucket_by_duration(item) == "medium" - - def test_long_bucket(self): - item = _make_detail("l1", duration=20.0) - assert _bucket_by_duration(item) == "long" - - def test_long_at_boundary(self): - """>=15s 为长素材.""" - item = _make_detail("l1", duration=15.0) - assert _bucket_by_duration(item) == "long" - - def test_unknown_bucket(self): - item = _make_detail("u1", duration=None) - assert _bucket_by_duration(item) == "unknown" - - -# ── 多样性选择 ────────────────────────────────────────────────────────────── - - -class TestDiverseSelection: - def _make_scored_list(self) -> list[AssetScoreDetail]: - """构造一个包含各时长桶的测试列表，按分数降序.""" - items = [ - _make_detail("high_short", duration=3.0, score=0.95), - _make_detail("high_medium", duration=8.0, score=0.9), - _make_detail("high_long", duration=30.0, score=0.85), - _make_detail("mid_short", duration=2.0, score=0.8), - _make_detail("mid_medium", duration=10.0, score=0.75), - _make_detail("mid_long", duration=20.0, score=0.7), - _make_detail("low_short", duration=4.0, score=0.6), - _make_detail("low_medium", duration=12.0, score=0.5), - _make_detail("low_long", duration=60.0, score=0.4), - ] - items.sort(key=lambda x: x.total_score, reverse=True) - return items - - def test_empty_list_returns_empty(self): - assert diverse_selection([], 5) == [] - - def test_zero_count_returns_empty(self): - items = self._make_scored_list() - assert diverse_selection(items, 0) == [] - - def test_negative_count_returns_empty(self): - items = self._make_scored_list() - assert diverse_selection(items, -1) == [] - - def test_selects_from_multiple_buckets(self): - items = self._make_scored_list() - result = diverse_selection(items, 6) - assert len(result) == 6 - # 应该包含来自不同桶的素材 - durations = [r.duration for r in result] - has_short = any(d and d < 5.0 for d in durations) - has_medium = any(d and 5.0 <= d < 15.0 for d in durations) - has_long = any(d and d >= 15.0 for d in durations) - assert has_short and has_medium and has_long - - def test_no_duplicate_ids(self): - items = self._make_scored_list() - result = diverse_selection(items, 9) - ids = [r.asset_id for r in result] - assert len(ids) == len(set(ids)) - - def test_not_more_than_count(self): - items = self._make_scored_list() - result = diverse_selection(items, 3) - assert len(result) <= 3 - - def test_fewer_assets_than_count(self): - items = [_make_detail("a1", duration=3.0, score=0.9)] - result = diverse_selection(items, 10) - assert len(result) == 1 - - def test_only_short_bucket(self): - items = [ - _make_detail("s1", duration=1.0, score=0.9), - _make_detail("s2", duration=2.0, score=0.8), - _make_detail("s3", duration=3.0, score=0.7), - ] - result = diverse_selection(items, 3) - assert len(result) == 3 - # 只有短素材，应该都返回 - assert all(r.duration and r.duration < 5.0 for r in result) - - def test_highest_scores_priority(self): - """分数最高的素材应该优先被选中.""" - items = self._make_scored_list() - result = diverse_selection(items, 3) - # 最高分的那个应该在结果里 - assert result[0].asset_id == "high_short" - - def test_contains_top_scoring_items(self): - """结果中应该包含全局最高分的素材.""" - items = self._make_scored_list() - result = diverse_selection(items, 6) - result_ids = {r.asset_id for r in result} - # 全局最高分的应该在结果中 - assert "high_short" in result_ids - assert "high_medium" in result_ids - - -# ── 候选过滤 ──────────────────────────────────────────────────────────────── - - -@dataclass -class MockAsset: - asset_id: str - status: str = "ready" - mime_type: str = "video/mp4" - quality_score: float | None = None - - -class TestFilterCandidates: - def test_ready_video_passes(self): - assets = [MockAsset("a1", status="ready", mime_type="video/mp4")] - candidates, filtered = filter_candidates(assets) - assert len(candidates) == 1 - assert filtered == 0 - - def test_not_ready_filtered(self): - assets = [ - MockAsset("a1", status="processing"), - MockAsset("a2", status="failed"), - MockAsset("a3", status="ready"), - ] - candidates, filtered = filter_candidates(assets) - assert len(candidates) == 1 - assert candidates[0].asset_id == "a3" - assert filtered == 0 # 状态不对的不算质量过滤 - - def test_non_video_filtered(self): - assets = [ - MockAsset("a1", mime_type="image/jpeg"), - MockAsset("a2", mime_type="audio/mp3"), - MockAsset("a3", mime_type="video/mp4"), - ] - candidates, filtered = filter_candidates(assets) - assert len(candidates) == 1 - assert filtered == 0 - - def test_video_mime_prefix(self): - """所有以 video 开头的 MIME 都通过.""" - assets = [ - MockAsset("a1", mime_type="video/mp4"), - MockAsset("a2", mime_type="video/webm"), - MockAsset("a3", mime_type="video/x-matroska"), - ] - candidates, _ = filter_candidates(assets) - assert len(candidates) == 3 - - def test_low_quality_filtered_counted(self): - """质量分低于门槛的计入 filtered_out.""" - assets = [ - MockAsset("good", quality_score=80.0), - MockAsset("bad", quality_score=20.0), - ] - candidates, filtered = filter_candidates(assets, min_quality_score=30.0) - assert len(candidates) == 1 - assert candidates[0].asset_id == "good" - assert filtered == 1 - - def test_none_quality_passes(self): - """quality_score 为 None 的不做质量检查，通过.""" - assets = [MockAsset("a1", quality_score=None)] - candidates, filtered = filter_candidates(assets, min_quality_score=30.0) - assert len(candidates) == 1 - assert filtered == 0 - - def test_quality_at_threshold_passes(self): - """刚好等于门槛值的通过.""" - assets = [MockAsset("a1", quality_score=30.0)] - candidates, filtered = filter_candidates(assets, min_quality_score=30.0) - assert len(candidates) == 1 - assert filtered == 0 - - def test_empty_list(self): - candidates, filtered = filter_candidates([]) - assert candidates == [] - assert filtered == 0 - - def test_none_mime_type_treated_as_empty(self): - assets = [MockAsset("a1", mime_type=None)] # type: ignore - candidates, _ = filter_candidates(assets) - assert len(candidates) == 0 - - -# ── 数据类 ────────────────────────────────────────────────────────────────── - - -class TestDataClasses: - def test_score_detail_defaults(self): - detail = AssetScoreDetail( - asset_id="test", - total_score=0.5, - quality_score=0.5, - resolution_score=0.5, - duration_score=0.5, - bitrate_score=0.5, - duration=None, - ) - assert detail.asset_id == "test" - assert detail.total_score == 0.5 - assert detail.duration is None - - def test_smart_select_result_defaults(self): - result = SmartSelectResult( - selected_ids=["a1", "a2"], - total_candidates=10, - filtered_out=3, - avg_score=0.75, - ) - assert len(result.selected_ids) == 2 - assert result.total_candidates == 10 - assert result.filtered_out == 3 - assert result.details == [] # 默认空列表 diff --git a/tests/unit/test_asset_select_mode.py b/tests/unit/test_asset_select_mode.py index 98469e7fa..86e3662b2 100644 --- a/tests/unit/test_asset_select_mode.py +++ b/tests/unit/test_asset_select_mode.py @@ -4,7 +4,7 @@ 覆盖： - all 模式：返回全部 ready 视频素材 ID - random 模式：随机选取 N 个 -- smart 模式：按质量分/时长评分降序选取 +- smart 模式：使用 smart_match 多维评分+多样性选取 - 无 ready 视频素材时返回空列表 - count=0 时返回全部（random/smart 模式） - 非视频素材和非 ready 状态素材被过滤 @@ -118,47 +118,53 @@ class TestSelectAssetsRandomMode: class TestSelectAssetsSmartMode: - """smart 模式：按质量分/时长评分降序选取。""" + """smart 模式：使用 smart_match 多维评分（质量40%+时长30%+新鲜度20%+未使用10%）。""" def test_smart_sorts_by_quality_score_desc(self): assets = [ - _asset("low", "low.mp4", quality_score=0.3), - _asset("high", "high.mp4", quality_score=0.9), - _asset("mid", "mid.mp4", quality_score=0.6), + _asset("low", "low.mp4", quality_score=30), + _asset("high", "high.mp4", quality_score=90), + _asset("mid", "mid.mp4", quality_score=60), ] result = _select_assets_from_library(assets, mode="smart", count=0) assert result == ["high", "mid", "low"] - def test_smart_tiebreak_by_duration_desc(self): + def test_smart_duration_optimal_beats_too_short(self): + """最优时长区间(5-30s)的素材得分高于过短素材。""" assets = [ - _asset("short", "short.mp4", quality_score=0.8, duration=10.0), - _asset("long", "long.mp4", quality_score=0.8, duration=60.0), + _asset("too_short", "short.mp4", quality_score=80, duration=1.0), + _asset("optimal", "optimal.mp4", quality_score=80, duration=15.0), ] result = _select_assets_from_library(assets, mode="smart", count=0) - assert result == ["long", "short"] + # Both: quality=80*0.4=32, recency/unused equal + # optimal(15s): duration_fitness=30 → total=62+ + # too_short(1s): duration_fitness=20+(1/5)*80=36 → 36*0.3=10.8 → total=42.8+ + assert result == ["optimal", "too_short"] def test_smart_with_count_limits_results(self): assets = [ - _asset("a1", "v1.mp4", quality_score=0.9), - _asset("a2", "v2.mp4", quality_score=0.7), - _asset("a3", "v3.mp4", quality_score=0.5), + _asset("a1", "v1.mp4", quality_score=90), + _asset("a2", "v2.mp4", quality_score=70), + _asset("a3", "v3.mp4", quality_score=50), ] result = _select_assets_from_library(assets, mode="smart", count=2) assert result == ["a1", "a2"] - def test_smart_null_quality_treated_as_zero(self): + def test_smart_null_quality_treated_as_default(self): + """无质量分的素材按50分计算（0-100标度）。""" assets = [ - _asset("scored", "scored.mp4", quality_score=0.5), + _asset("scored", "scored.mp4", quality_score=80), _asset("unscored", "unscored.mp4", quality_score=None), ] result = _select_assets_from_library(assets, mode="smart", count=0) + # scored(80): quality=80*0.4=32; unscored(None→50): quality=50*0.4=20 assert result == ["scored", "unscored"] def test_smart_count_zero_returns_all_sorted(self): assets = [ - _asset("a1", "v1.mp4", quality_score=0.1), - _asset("a2", "v2.mp4", quality_score=0.9), - _asset("a3", "v3.mp4", quality_score=0.5), + _asset("a1", "v1.mp4", quality_score=10), + _asset("a2", "v2.mp4", quality_score=90), + _asset("a3", "v3.mp4", quality_score=50), ] result = _select_assets_from_library(assets, mode="smart", count=0) assert result == ["a2", "a3", "a1"] diff --git a/tests/unit/test_smart_asset_selector.py b/tests/unit/test_smart_asset_selector.py deleted file mode 100755 index 258209d50..000000000 --- a/tests/unit/test_smart_asset_selector.py +++ /dev/null @@ -1,413 +0,0 @@ -"""SmartAssetSelector 智能素材选择服务单元测试.""" - -from __future__ import annotations - -import unittest -from dataclasses import dataclass - -from app.services.smart_asset_selector import ( - _MEDIUM_BUCKET_MAX, - _SHORT_BUCKET_MAX, - SmartAssetSelector, -) - - -@dataclass -class MockAsset: - """模拟 Asset 实体.""" - - id: str - quality_score: float | None = None - width: int | None = None - height: int | None = None - duration: float | None = None - file_size: int = 0 - mime_type: str = "video/mp4" - status: str = "ready" - - @property - def status_value(self) -> str: - return self.status - - -class TestSmartAssetSelectorScoring(unittest.TestCase): - """评分维度测试.""" - - def setUp(self): - self.selector = SmartAssetSelector() - - def test_quality_score_normalization(self): - """质量分正确归一化到 0-1.""" - asset_high = MockAsset(id="1", quality_score=90.0) - asset_low = MockAsset(id="2", quality_score=30.0) - asset_none = MockAsset(id="3", quality_score=None) - - detail_high = self.selector._score_asset(asset_high) - detail_low = self.selector._score_asset(asset_low) - detail_none = self.selector._score_asset(asset_none) - - # 90分 → 0.9 × 0.5权重 = 0.45 基础贡献 - self.assertAlmostEqual(detail_high.quality_score, 0.9, delta=0.01) - # 30分 → 0.3 × 0.5权重 = 0.15 基础贡献 - self.assertAlmostEqual(detail_low.quality_score, 0.3, delta=0.01) - # 无质量分给默认 0.5 - self.assertAlmostEqual(detail_none.quality_score, 0.5, delta=0.01) - - def test_resolution_score_1080p_full(self): - """1080p 分辨率得满分.""" - asset = MockAsset(id="1", width=1920, height=1080) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.resolution_score, 1.0, delta=0.01) - - def test_resolution_score_4k_full(self): - """4K 也得满分（高于目标分辨率不扣分）.""" - asset = MockAsset(id="1", width=3840, height=2160) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.resolution_score, 1.0, delta=0.01) - - def test_resolution_score_720p_lower(self): - """720p 低于 1080p，得分低于 1.""" - asset = MockAsset(id="1", width=1280, height=720) - detail = self.selector._score_asset(asset) - self.assertLess(detail.resolution_score, 1.0) - self.assertGreater(detail.resolution_score, 0.3) - - def test_resolution_score_none(self): - """分辨率未知给中评分.""" - asset = MockAsset(id="1", width=None, height=None) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.resolution_score, 0.5, delta=0.01) - - def test_duration_score_optimal(self): - """最佳时长区间内得满分.""" - asset = MockAsset(id="1", duration=10.0) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.duration_score, 1.0, delta=0.01) - - def test_duration_score_too_short(self): - """时长过短扣分.""" - asset = MockAsset(id="1", duration=1.0) - detail = self.selector._score_asset(asset) - self.assertLess(detail.duration_score, 1.0) - - def test_duration_score_too_long(self): - """时长过长扣分.""" - asset = MockAsset(id="1", duration=120.0) - detail = self.selector._score_asset(asset) - self.assertLess(detail.duration_score, 1.0) - - def test_duration_score_none(self): - """时长未知给中评分.""" - asset = MockAsset(id="1", duration=None) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.duration_score, 0.5, delta=0.01) - - def test_total_score_weighted_sum(self): - """总分是各维度的加权和.""" - asset = MockAsset( - id="1", - quality_score=100.0, # 1.0 × 0.5 = 0.5 - width=1920, # 1.0 × 0.2 = 0.2 - height=1080, - duration=10.0, # 1.0 × 0.2 = 0.2 - file_size=10_000_000, # ~8Mbps，10秒 → 约 1.0 × 0.1 = 0.1 - ) - detail = self.selector._score_asset(asset) - # 理论上接近 1.0 - self.assertGreater(detail.total_score, 0.85) - self.assertLessEqual(detail.total_score, 1.0) - - -class TestSmartAssetSelectorSelection(unittest.TestCase): - """选择逻辑测试.""" - - def setUp(self): - self.selector = SmartAssetSelector(min_quality_score=0) # 测试时关闭质量门槛 - - def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: - assets = [] - for i in range(count): - assets.append( - MockAsset( - id=f"asset_{i}", - quality_score=base_quality - i * 5, # 质量递减 - width=1920, - height=1080, - duration=10.0 + i, - file_size=5_000_000 + i * 100_000, - ) - ) - return assets - - def test_select_all_when_count_zero(self): - """count=0 时返回全部符合条件的.""" - assets = self._make_assets(10) - result = self.selector.select(assets, count=0) - self.assertEqual(len(result.selected_ids), 10) - self.assertEqual(result.total_candidates, 10) - - def test_select_top_n(self): - """返回指定数量的 top N.""" - assets = self._make_assets(10) - result = self.selector.select(assets, count=3) - self.assertEqual(len(result.selected_ids), 3) - # 最高分的应该是 asset_0（质量分最高） - self.assertEqual(result.selected_ids[0], "asset_0") - - def test_select_more_than_available(self): - """请求数量超过候选数量时返回全部.""" - assets = self._make_assets(5) - result = self.selector.select(assets, count=10) - self.assertEqual(len(result.selected_ids), 5) - - def test_filter_non_ready(self): - """非 ready 状态的素材被过滤.""" - assets = [ - MockAsset(id="1", quality_score=90.0, status="ready"), - MockAsset(id="2", quality_score=80.0, status="processing"), - MockAsset(id="3", quality_score=70.0, status="ready"), - ] - result = self.selector.select(assets, count=0) - self.assertEqual(len(result.selected_ids), 2) - self.assertIn("1", result.selected_ids) - self.assertIn("3", result.selected_ids) - self.assertNotIn("2", result.selected_ids) - - def test_filter_non_video(self): - """非视频素材被过滤.""" - assets = [ - MockAsset(id="1", quality_score=90.0, mime_type="video/mp4"), - MockAsset(id="2", quality_score=80.0, mime_type="image/jpeg"), - MockAsset(id="3", quality_score=70.0, mime_type="video/quicktime"), - ] - result = self.selector.select(assets, count=0) - self.assertEqual(len(result.selected_ids), 2) - - def test_min_quality_filter(self): - """最低质量分门槛过滤.""" - selector = SmartAssetSelector(min_quality_score=60.0) - assets = [ - MockAsset(id="1", quality_score=90.0), - MockAsset(id="2", quality_score=50.0), # 低于门槛 - MockAsset(id="3", quality_score=70.0), - MockAsset(id="4", quality_score=30.0), # 低于门槛 - ] - result = selector.select(assets, count=0) - self.assertEqual(len(result.selected_ids), 2) - self.assertEqual(result.filtered_out, 2) - self.assertIn("1", result.selected_ids) - self.assertIn("3", result.selected_ids) - - def test_empty_input(self): - """空输入返回空结果.""" - result = self.selector.select([], count=5) - self.assertEqual(result.selected_ids, []) - self.assertEqual(result.total_candidates, 0) - self.assertEqual(result.avg_score, 0.0) - - def test_sorted_by_score_descending(self): - """结果按总分降序排列.""" - assets = self._make_assets(5) - result = self.selector.select(assets, count=0, ensure_diversity=False) - scores = [d.total_score for d in result.details] - # 应该是降序 - self.assertEqual(scores, sorted(scores, reverse=True)) - - -class TestSmartAssetSelectorDiversity(unittest.TestCase): - """多样性选择测试.""" - - def setUp(self): - self.selector = SmartAssetSelector(min_quality_score=0) - - def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: - assets = [] - for i in range(count): - assets.append( - MockAsset( - id=f"asset_{i}", - quality_score=base_quality - i * 5, - width=1920, - height=1080, - duration=10.0 + i, - file_size=5_000_000 + i * 100_000, - ) - ) - return assets - - def test_diversity_all_short(self): - """全是短素材时不报错，正常返回.""" - assets = [] - for i in range(10): - assets.append( - MockAsset( - id=f"short_{i}", - quality_score=80.0 + i, - width=1920, - height=1080, - duration=2.0 + i * 0.1, # 都 < 5s - file_size=1_000_000, - ) - ) - result = self.selector.select(assets, count=5, ensure_diversity=True) - self.assertEqual(len(result.selected_ids), 5) - - def test_diversity_mixed_buckets(self): - """混合时长素材时，各桶都有代表.""" - assets = [] - # 短素材（质量分高） - for i in range(5): - assets.append( - MockAsset( - id=f"short_{i}", - quality_score=95.0 - i, - width=1920, - height=1080, - duration=3.0, - file_size=2_000_000, - ) - ) - # 中素材（质量分中等） - for i in range(5): - assets.append( - MockAsset( - id=f"medium_{i}", - quality_score=85.0 - i, - width=1920, - height=1080, - duration=10.0, - file_size=5_000_000, - ) - ) - # 长素材（质量分低） - for i in range(5): - assets.append( - MockAsset( - id=f"long_{i}", - quality_score=75.0 - i, - width=1920, - height=1080, - duration=60.0, - file_size=20_000_000, - ) - ) - - result = self.selector.select(assets, count=6, ensure_diversity=True) - selected = result.selected_ids - - # 6个素材，每个桶至少有1个（基础配额 max(1, 6//3)=2） - short_count = sum(1 for sid in selected if sid.startswith("short_")) - medium_count = sum(1 for sid in selected if sid.startswith("medium_")) - long_count = sum(1 for sid in selected if sid.startswith("long_")) - - # 每个桶至少1个 - self.assertGreaterEqual(short_count, 1) - self.assertGreaterEqual(medium_count, 1) - self.assertGreaterEqual(long_count, 1) - self.assertEqual(len(selected), 6) - - def test_diversity_disabled_returns_top(self): - """关闭多样性时，直接返回 top N（可能全是短素材）.""" - assets = [] - # 短素材（质量分最高） - for i in range(10): - assets.append( - MockAsset( - id=f"short_{i}", - quality_score=95.0 - i, - width=1920, - height=1080, - duration=3.0, - file_size=2_000_000, - ) - ) - # 长素材（质量分低） - for i in range(5): - assets.append( - MockAsset( - id=f"long_{i}", - quality_score=70.0, - width=1920, - height=1080, - duration=60.0, - file_size=20_000_000, - ) - ) - - result = self.selector.select(assets, count=5, ensure_diversity=False) - selected = result.selected_ids - # 全是短素材（因为质量分高） - self.assertTrue(all(s.startswith("short_") for s in selected)) - - def test_avg_score_calculated(self): - """平均分正确计算.""" - assets = self._make_assets(3) - result = self.selector.select(assets, count=3, ensure_diversity=False) - expected_avg = sum(d.total_score for d in result.details) / 3 - self.assertAlmostEqual(result.avg_score, expected_avg, delta=0.001) - - -class TestSmartAssetSelectorEdgeCases(unittest.TestCase): - """边界情况测试.""" - - def setUp(self): - self.selector = SmartAssetSelector(min_quality_score=0) - - def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: - assets = [] - for i in range(count): - assets.append( - MockAsset( - id=f"asset_{i}", - quality_score=base_quality - i * 5, - width=1920, - height=1080, - duration=10.0 + i, - file_size=5_000_000 + i * 100_000, - ) - ) - return assets - - def test_single_asset(self): - """单个素材正常返回.""" - assets = [MockAsset(id="1", quality_score=80.0, width=1920, height=1080, duration=10.0)] - result = self.selector.select(assets, count=1) - self.assertEqual(len(result.selected_ids), 1) - self.assertEqual(result.selected_ids[0], "1") - - def test_zero_width_height(self): - """宽高为0时按未知处理.""" - asset = MockAsset(id="1", width=0, height=0) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.resolution_score, 0.5, delta=0.01) - - def test_negative_duration(self): - """负时长按未知处理.""" - asset = MockAsset(id="1", duration=-5.0) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.duration_score, 0.5, delta=0.01) - - def test_zero_file_size_with_duration(self): - """文件大小为0时码率评分中等.""" - asset = MockAsset(id="1", file_size=0, duration=10.0) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.bitrate_score, 0.5, delta=0.01) - - def test_bitrate_score_optimal(self): - """最佳码率范围得满分.""" - # 5 Mbps × 10秒 = 6.25 MB → file_size = 6,250,000 bytes - asset = MockAsset(id="1", file_size=6_250_000, duration=10.0) - detail = self.selector._score_asset(asset) - self.assertAlmostEqual(detail.bitrate_score, 1.0, delta=0.01) - - def test_details_match_selected_ids(self): - """details 列表和 selected_ids 顺序一致.""" - assets = self._make_assets(5) - result = self.selector.select(assets, count=3, ensure_diversity=False) - self.assertEqual(len(result.details), 3) - for i, aid in enumerate(result.selected_ids): - self.assertEqual(result.details[i].asset_id, aid) - - -if __name__ == "__main__": - unittest.main() -- 2.54.0