xiaoxia-saas/apps/api/app/services/smart_asset_selector.py

"""SmartAssetSelector — 智能素材选择服务.

根据多维度评分从素材库中自动选择最优视频素材，
用于一键生成等需要自动选取素材的场景。

评分维度（加权求和，总分 0-1）：
  - 质量分（quality_score）：权重 0.5 — 来自人工或AI的质量评分
  - 分辨率适配：权重 0.2 — 分辨率越接近 1080p 得分越高
  - 时长合理性：权重 0.2 — 3-30 秒区间最佳，过短/过长扣分
  - 码率质量：权重 0.1 — 用文件大小/时长估算，码率适中得分高

特性：
  - 最低质量分门槛：自动过滤低质量素材
  - 时长多样性：保证选出的素材时长分布均匀（短/中/长各占一定比例）
  - 兼容全部模式：素材库模式和项目模式都可用
"""

from __future__ import annotations

import logging
from dataclasses import dataclass

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:
    """智能素材选择器.

    从一组素材中按综合评分选择最优的 N 个，
    同时保证时长分布的多样性。
    """

    def __init__(
        self,
        min_quality_score: float = _MIN_QUALITY_SCORE,
        target_width: int = _TARGET_WIDTH,
        target_height: int = _TARGET_HEIGHT,
    ):
        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 = 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)

        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 = self._score_asset(asset)
            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 = self._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

    # ── 内部方法 ──────────────────────────────────────────────────────────────

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

    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  # 未知分辨率给中评分

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

    def _score_duration(self, duration: float | None) -> float:
        """时长评分：3-30秒最佳，过短或过长都扣分."""
        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:
            # 太短：线性衰减，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)

    def _score_bitrate(self, file_size: int, duration: float | None) -> float:
        """码率评分：根据文件大小和时长估算码率，适中得分高."""
        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

        # 码率太高（文件太大）：适度扣分
        excess = bitrate / optimal_high - 1.0
        penalty = min(0.5, excess * 0.2)
        return max(0.5, 1.0 - penalty)

    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]