refactor: 统一智能选素材逻辑，迁移 generation_tasks smart 模式到 smart_match

- 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

@@ -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]

apps/api/app/services/smart_asset_selector.py

@@ -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)

packages/domain/asset_scoring.py

@@ -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

tests/unit/domain/test_asset_scoring.py

@@ -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

tests/unit/test_asset_scoring.py

@@ -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

tests/unit/test_asset_scoring_domain.py

@@ -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 == []  # 默认空列表

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"]

tests/unit/test_smart_asset_selector.py

@@ -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()