xiaoxia-saas/packages/domain/asset_scoring.py

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