test(wave129): asset_scoring资产评分+64单测

- 覆盖8大模块：权重/分辨率评分/时长评分/码率评分/加权总分/单素材评分/分桶/多样性选择/候选过滤/数据类
- 边界值全覆盖：0/None/负数/临界值/最佳区间
- 多样性选择验证：多桶分布/去重/配额/不足时回退

tests/unit/test_asset_scoring_domain.py

@@ -0,0 +1,515 @@
+"""资产评分纯逻辑单元测试 — wave129."""
+
+from __future__ import annotations
+
+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,
+    WEIGHT_QUALITY,
+    WEIGHT_RESOLUTION,
+    WEIGHT_DURATION,
+    WEIGHT_BITRATE,
+)
+
+# ── 常量与权重 ──────────────────────────────────────────────────────────────
+
+
+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 == []  # 默认空列表