test(wave92): extract asset_scoring domain module + 89 unit tests (#955)

apps/api/app/services/smart_asset_selector.py

@@ -13,57 +13,31 @@
   - 最低质量分门槛：自动过滤低质量素材
   - 时长多样性：保证选出的素材时长分布均匀（短/中/长各占一定比例）
   - 兼容全部模式：素材库模式和项目模式都可用
+
+纯逻辑部分已抽离到 packages.domain.asset_scoring。
 """
 
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass
+
+from packages.domain.asset_scoring import (
+    MEDIUM_BUCKET_MAX as _MEDIUM_BUCKET_MAX,
+    MIN_QUALITY_SCORE as _MIN_QUALITY_SCORE,
+    OPTIMAL_DURATION_MAX as _OPTIMAL_DURATION_MAX,
+    OPTIMAL_DURATION_MIN as _OPTIMAL_DURATION_MIN,
+    SHORT_BUCKET_MAX as _SHORT_BUCKET_MAX,
+    TARGET_HEIGHT as _TARGET_HEIGHT,
+    TARGET_WIDTH as _TARGET_WIDTH,
+    AssetScoreDetail,
+    SmartSelectResult,
+    diverse_selection,
+    filter_candidates,
+    score_asset_detail,
+)
 
 logger = logging.getLogger(__name__)
 
-# ── 评分权重 ──────────────────────────────────────────────────────────────────
-_WEIGHT_QUALITY = 0.5
-_WEIGHT_RESOLUTION = 0.2
-_WEIGHT_DURATION = 0.2
-_WEIGHT_BITRATE = 0.1
-
-# ── 评分参数 ──────────────────────────────────────────────────────────────────
-_TARGET_WIDTH = 1920  # 目标分辨率宽度基准
-_TARGET_HEIGHT = 1080  # 目标分辨率高度基准
-_MIN_QUALITY_SCORE = 30.0  # 最低质量分门槛（低于此值的素材直接排除）
-_OPTIMAL_DURATION_MIN = 3.0  # 最佳时长区间（秒）
-_OPTIMAL_DURATION_MAX = 30.0
-
-# ── 多样性分桶 ───────────────────────────────────────────────────────────────
-_SHORT_BUCKET_MAX = 5.0  # 短素材：< 5s
-_MEDIUM_BUCKET_MAX = 15.0  # 中素材：5-15s
-# 长素材：> 15s
-
-
-@dataclass
-class SmartSelectResult:
-    """智能选择结果."""
-
-    selected_ids: list[str]
-    total_candidates: int
-    filtered_out: int  # 被质量门槛过滤的数量
-    avg_score: float
-    details: list[AssetScoreDetail]
-
-
-@dataclass
-class AssetScoreDetail:
-    """单个素材的评分详情."""
-
-    asset_id: str
-    total_score: float
-    quality_score: float
-    resolution_score: float
-    duration_score: float
-    bitrate_score: float
-    duration: float | None
-
 
 class SmartAssetSelector:
     """智能素材选择器.
@@ -74,9 +48,9 @@ class SmartAssetSelector:
 
     def __init__(
         self,
-        min_quality_score: float = _MIN_QUALITY_SCORE,
-        target_width: int = _TARGET_WIDTH,
-        target_height: int = _TARGET_HEIGHT,
+        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
@@ -102,21 +76,7 @@ class SmartAssetSelector:
             SmartSelectResult 选择结果
         """
         # 1. 过滤：只保留 ready 状态的视频素材 + 最低质量分门槛
-        candidates = []
-        filtered_out = 0
-        for asset in assets:
-            status = getattr(asset, "status", None)
-            status_val = status.value if hasattr(status, "value") else str(status)
-            if status_val != "ready":
-                continue
-            mime_type = getattr(asset, "mime_type", "") or ""
-            if not mime_type.startswith("video"):
-                continue
-            quality = getattr(asset, "quality_score", None)
-            if quality is not None and quality < self.min_quality_score:
-                filtered_out += 1
-                continue
-            candidates.append(asset)
+        candidates, filtered_out = filter_candidates(assets, self.min_quality_score)
 
         if not candidates:
             return SmartSelectResult(
@@ -130,7 +90,16 @@ class SmartAssetSelector:
         # 2. 对每个候选素材评分
         scored: list[AssetScoreDetail] = []
         for asset in candidates:
-            detail = self._score_asset(asset)
+            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. 按总分降序排列
@@ -138,7 +107,7 @@ class SmartAssetSelector:
 
         # 4. 多样性选择（如果需要且数量有限制）
         if ensure_diversity and count > 0 and len(scored) > count:
-            selected = self._diverse_selection(scored, count)
+            selected = diverse_selection(scored, count)
         else:
             # 无数量限制或不要求多样性，直接按排名取
             selected = scored if count <= 0 else scored[:count]
@@ -162,165 +131,39 @@ class SmartAssetSelector:
         )
         return result
 
-    # ── 内部方法 ──────────────────────────────────────────────────────────────
+    # ── 向后兼容：私有方法别名（委托给 asset_scoring 纯函数） ────────────────
 
     def _score_asset(self, asset) -> AssetScoreDetail:
-        """对单个素材进行多维度评分."""
-        # 质量分
-        quality = getattr(asset, "quality_score", None)
-        quality_score = (quality / 100.0) if quality is not None else 0.5
-
-        # 分辨率评分：越接近目标分辨率得分越高
-        width = getattr(asset, "width", None)
-        height = getattr(asset, "height", None)
-        resolution_score = self._score_resolution(width, height)
-
-        # 时长评分：在最佳区间内得分高，过短过长扣分
-        duration = getattr(asset, "duration", None)
-        duration_score = self._score_duration(duration)
-
-        # 码率评分：用 file_size/duration 估算，适中得分高
-        file_size = getattr(asset, "file_size", 0) or 0
-        bitrate_score = self._score_bitrate(file_size, duration)
-
-        # 加权总分
-        total = (
-            _WEIGHT_QUALITY * quality_score
-            + _WEIGHT_RESOLUTION * resolution_score
-            + _WEIGHT_DURATION * duration_score
-            + _WEIGHT_BITRATE * bitrate_score
-        )
-
-        return AssetScoreDetail(
+        """对单个素材进行多维度评分（向后兼容）."""
+        return score_asset_detail(
             asset_id=asset.id,
-            total_score=round(total, 4),
-            quality_score=round(quality_score, 4),
-            resolution_score=round(resolution_score, 4),
-            duration_score=round(duration_score, 4),
-            bitrate_score=round(bitrate_score, 4),
-            duration=duration,
+            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:
-        """分辨率评分：越接近目标分辨率得分越高，低于480p扣分严重."""
-        if width is None or height is None or width <= 0 or height <= 0:
-            return 0.5  # 未知分辨率给中评分
+        """分辨率评分（向后兼容）."""
+        from packages.domain.asset_scoring import score_resolution
 
-        target_pixels = self.target_width * self.target_height
-        actual_pixels = width * height
-
-        # 计算像素数比例
-        ratio = actual_pixels / target_pixels
-
-        if ratio >= 1.0:
-            # 高于或等于目标分辨率：满分，略高不扣分（4K也给满分）
-            return 1.0
-        else:
-            # 低于目标分辨率：线性衰减，但最低不低于 0.1
-            # 例如：720p (921600) / 1080p (2073600) = 0.44 → 得分 0.6
-            score = 0.3 + 0.7 * ratio
-            return max(0.1, min(1.0, score))
+        return score_resolution(width, height, self.target_width, self.target_height)
 
     def _score_duration(self, duration: float | None) -> float:
-        """时长评分：3-30秒最佳，过短或过长都扣分."""
-        if duration is None or duration <= 0:
-            return 0.5  # 未知时长给中评分
+        """时长评分（向后兼容）."""
+        from packages.domain.asset_scoring import score_duration
 
-        if _OPTIMAL_DURATION_MIN <= duration <= _OPTIMAL_DURATION_MAX:
-            # 最佳区间：满分
-            return 1.0
-
-        if duration < _OPTIMAL_DURATION_MIN:
-            # 太短：线性衰减，1秒以下给 0.3
-            ratio = duration / _OPTIMAL_DURATION_MIN
-            return 0.3 + 0.7 * ratio
-
-        # 太长：每超过最佳区间上限10秒扣 0.1 分，最低 0.2
-        excess = duration - _OPTIMAL_DURATION_MAX
-        penalty = min(0.8, excess / 10.0 * 0.1)
-        return max(0.2, 1.0 - penalty)
+        return score_duration(duration)
 
     def _score_bitrate(self, file_size: int, duration: float | None) -> float:
-        """码率评分：根据文件大小和时长估算码率，适中得分高."""
-        if not file_size or not duration or duration <= 0:
-            return 0.5  # 未知给中评分
+        """码率评分（向后兼容）."""
+        from packages.domain.asset_scoring import score_bitrate
 
-        # 估算码率（bps）
-        bitrate = (file_size * 8) / duration
-
-        # 最佳码率范围：2-8 Mbps
-        optimal_low = 2_000_000  # 2 Mbps
-        optimal_high = 8_000_000  # 8 Mbps
-
-        if optimal_low <= bitrate <= optimal_high:
-            return 1.0
-
-        if bitrate < optimal_low:
-            # 码率太低：线性衰减
-            ratio = bitrate / optimal_low
-            return 0.3 + 0.7 * ratio
-
-        # 码率太高（文件太大）：适度扣分
-        excess = bitrate / optimal_high - 1.0
-        penalty = min(0.5, excess * 0.2)
-        return max(0.5, 1.0 - penalty)
+        return score_bitrate(file_size, duration)
 
     def _diverse_selection(self, scored: list[AssetScoreDetail], count: int) -> list[AssetScoreDetail]:
-        """多样性选择：按时长分桶，保证每个桶都有素材.
-
-        策略：
-        1. 按时长分为三桶：短(<5s)、中(5-15s)、长(>15s)
-        2. 每个桶配额 = max(1, count / 3)
-        3. 先从每桶按配额取最高分的
-        4. 剩余名额从全局最高分中取（不重复）
-        """
-        # 分桶
-        short_bucket = [d for d in scored if d.duration is not None and d.duration < _SHORT_BUCKET_MAX]
-        medium_bucket = [
-            d for d in scored if d.duration is not None and _SHORT_BUCKET_MAX <= d.duration < _MEDIUM_BUCKET_MAX
-        ]
-        long_bucket = [d for d in scored if d.duration is not None and d.duration >= _MEDIUM_BUCKET_MAX]
-        unknown_bucket = [d for d in scored if d.duration is None]
-
-        buckets = [short_bucket, medium_bucket, long_bucket]
-        bucket_names = ["short", "medium", "long"]
-
-        # 每个桶基础配额（至少1个，如果桶非空且需要的话）
-        base_quota = max(1, count // 3)
-
-        selected: list[AssetScoreDetail] = []
-        selected_ids: set[str] = set()
-
-        # 先按配额从每个桶取
-        for bucket, _name in zip(buckets, bucket_names, strict=False):
-            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]
+        """多样性选择（向后兼容）."""
+        return diverse_selection(scored, count)

packages/domain/asset_scoring.py

@@ -0,0 +1,372 @@
+"""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/test_asset_scoring.py

@@ -0,0 +1,755 @@
+"""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 (
+    AssetScoreDetail,
+    MEDIUM_BUCKET_MAX,
+    MIN_QUALITY_SCORE,
+    OPTIMAL_DURATION_MAX,
+    OPTIMAL_DURATION_MIN,
+    SHORT_BUCKET_MAX,
+    SmartSelectResult,
+    TARGET_HEIGHT,
+    TARGET_WIDTH,
+    WEIGHT_BITRATE,
+    WEIGHT_DURATION,
+    WEIGHT_QUALITY,
+    WEIGHT_RESOLUTION,
+    _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