style: auto-format with black + isort + prettier [skip ci-format-check]

- score_resolution分辨率评分：None/0值/目标匹配/超分/线性衰减/最低分/自定义目标
- score_duration时长评分：None/0值/最佳区间/过短衰减/过长扣分/边界值
- score_bitrate码率评分：缺省值/最佳区间/低码率衰减/高码率扣分/最低分
- calculate_total_score加权总分：满分/零分/权重验证/精度
- score_asset_detail单素材评分：对象返回/完美素材/质量默认/归一化/组件一致性
- _bucket_by_duration分桶：unknown/short/medium/long + 边界值
- diverse_selection多样性选择：空输入/0值/不足量/三桶配额/大count配额/全局补位
- filter_candidates候选过滤：状态/类型/质量门槛/None质量/枚举状态/空输入

apps/worker/video_processing/multi_track_mixer_pure.py

@@ -6,7 +6,6 @@
 
 from __future__ import annotations
 
-
 # ── 单轨时间计算 ──────────────────────────────────────────────────────────────
 
 

apps/worker/video_processing/speed_engine.py

@@ -13,7 +13,6 @@
 
 from __future__ import annotations
 
-
 from packages.domain.speed_config import (
     DEFAULT_SPEED,
     SpeedConfig,

tests/unit/domain/test_asset_scoring.py

@@ -0,0 +1,530 @@
+"""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

tests/unit/domain/test_video_filter_builder.py

@@ -6,7 +6,6 @@ domain 层纯逻辑模块，0 FFmpeg 依赖，快速轻量。
 from __future__ import annotations
 
 import dataclasses
-
 from unittest.mock import MagicMock
 
 import pytest

tests/unit/test_transition_presets.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import dataclasses
 
-
 import pytest
 
 from packages.domain.transition_presets import (