diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index cfc395d0b..e982bb248 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -30,6 +30,7 @@ from app.schemas.generation_task import ( GenerationTaskResponse, ListGenerationTasksResponse, ) +from app.services.smart_asset_selector import SmartAssetSelector from fastapi import APIRouter, Depends, HTTPException from packages.application import ( @@ -105,7 +106,7 @@ def _select_assets_from_library( Args: assets: 素材库中所有素材(Asset 实体列表) - mode: 选取模式 — all=全部, random=随机, smart=按质量评分 + mode: 选取模式 — all=全部, random=随机, smart=智能匹配(多维度评分+多样性) count: 选取数量,0 表示全部(仅 random/smart 模式有效) Returns: @@ -123,17 +124,10 @@ def _select_assets_from_library( return [a.id for a in selected] if mode == "smart": - # 按质量分降序排列(质量分高的优先),质量分相同时按时长降序 - sorted_assets = sorted( - ready_video_assets, - key=lambda a: ( - a.quality_score if a.quality_score is not None else 0.0, - a.duration if a.duration is not None else 0.0, - ), - reverse=True, - ) - selected = sorted_assets if count <= 0 else sorted_assets[:count] - return [a.id for a in selected] + # 智能匹配:多维度综合评分 + 时长多样性保证 + selector = SmartAssetSelector() + result = selector.select(ready_video_assets, count=count, ensure_diversity=True) + return result.selected_ids # 默认 all 模式:返回全部 ready 视频素材 return [a.id for a in ready_video_assets] @@ -224,6 +218,20 @@ def create_generation_task( mode=request.asset_select_mode, count=request.asset_select_count, ) + elif project_id and not resolved_asset_ids and request.asset_select_mode in ("random", "smart"): + # 项目级模式:未指定 asset_ids 且选择了 random/smart 模式时,也自动选取 + assets = asset_repository.find_by_project(project_id) + if assets: + resolved_asset_ids = _select_assets_from_library( + assets, + mode=request.asset_select_mode, + count=request.asset_select_count, + ) + if not resolved_asset_ids: + raise HTTPException( + status_code=422, + detail="当前项目没有符合条件的视频素材,请先上传并等待导入完成后再生成。", + ) use_case = CreateGenerationTaskUseCase(generation_task_repository) count = request.count diff --git a/apps/api/app/services/smart_asset_selector.py b/apps/api/app/services/smart_asset_selector.py new file mode 100755 index 000000000..8e781f508 --- /dev/null +++ b/apps/api/app/services/smart_asset_selector.py @@ -0,0 +1,329 @@ +"""SmartAssetSelector — 智能素材选择服务. + +根据多维度评分从素材库中自动选择最优视频素材, +用于一键生成等需要自动选取素材的场景。 + +评分维度(加权求和,总分 0-1): + - 质量分(quality_score):权重 0.5 — 来自人工或AI的质量评分 + - 分辨率适配:权重 0.2 — 分辨率越接近 1080p 得分越高 + - 时长合理性:权重 0.2 — 3-30 秒区间最佳,过短/过长扣分 + - 码率质量:权重 0.1 — 用文件大小/时长估算,码率适中得分高 + +特性: + - 最低质量分门槛:自动过滤低质量素材 + - 时长多样性:保证选出的素材时长分布均匀(短/中/长各占一定比例) + - 兼容全部模式:素材库模式和项目模式都可用 +""" + +from __future__ import annotations + +import logging +import math +from dataclasses import dataclass + +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: + """智能素材选择器. + + 从一组素材中按综合评分选择最优的 N 个, + 同时保证时长分布的多样性。 + """ + + def __init__( + self, + min_quality_score: float = _MIN_QUALITY_SCORE, + target_width: int = _TARGET_WIDTH, + target_height: int = _TARGET_HEIGHT, + ): + 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 = 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) + + 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 = self._score_asset(asset) + 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 = self._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 + + # ── 内部方法 ────────────────────────────────────────────────────────────── + + 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( + 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, + ) + + 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 # 未知分辨率给中评分 + + 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)) + + def _score_duration(self, duration: float | None) -> float: + """时长评分:3-30秒最佳,过短或过长都扣分.""" + 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: + # 太短:线性衰减,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) + + def _score_bitrate(self, file_size: int, duration: float | None) -> float: + """码率评分:根据文件大小和时长估算码率,适中得分高.""" + 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 + + # 码率太高(文件太大):适度扣分 + excess = bitrate / optimal_high - 1.0 + penalty = min(0.5, excess * 0.2) + return max(0.5, 1.0 - penalty) + + 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): + 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] diff --git a/tests/unit/test_smart_asset_selector.py b/tests/unit/test_smart_asset_selector.py new file mode 100755 index 000000000..ed90981e6 --- /dev/null +++ b/tests/unit/test_smart_asset_selector.py @@ -0,0 +1,413 @@ +"""SmartAssetSelector 智能素材选择服务单元测试.""" + +from __future__ import annotations + +import unittest +from dataclasses import dataclass + +from app.services.smart_asset_selector import ( + SmartAssetSelector, + _MEDIUM_BUCKET_MAX, + _SHORT_BUCKET_MAX, +) + + +@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()