xiaoxia-saas/packages/domain/smart_match.py

"""统一智能选素材算法 — 合并 _helpers / generation_tasks / auto_clip_service 的重叠逻辑。

设计目标：
- 单一入口，替代 3 套分散的选素材代码
- 多维度加权评分：质量分 + 时长均衡 + 新鲜度 + 未使用偏好
- 多样性保障：按时长分桶（短/中/长）均衡选取，避免同质化
- 可扩展：后续接入 AI 模型时只需替换 score_asset()
"""

from __future__ import annotations

import math
import random
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

# 素材选取排序时注入的随机噪声上限（分）。
# score_asset 综合得分范围为 0-100，噪声 0~20 意味着：
# - 素材间得分差距 > 20 分时，排名不受影响（质量差异显著的素材保持稳定优先级）
# - 得分接近（差距 <= 20 分）的素材排名会随机浮动，使每次生成选出的素材组合不同，
#   从素材组合层面降低成片重复率；排名靠后的低分素材也有机会入选。
SCORE_RANDOM_NOISE_MAX = 20.0


@dataclass
class SmartMatchResult:
    """单条素材的匹配结果。"""

    asset: Any  # Asset entity
    score: float  # 综合得分 0-100
    breakdown: dict[str, float] = field(default_factory=dict)  # 各维度得分明细


def _get_enum_value(obj: Any, attr: str) -> str:
    """安全获取属性值，兼容 StrEnum / 普通字符串。"""
    val = getattr(obj, attr, None)
    if val is None:
        return ""
    return val.value if hasattr(val, "value") else str(val)


def _duration_bucket(duration: float | None) -> str:
    """将素材时长分为 3 档：short(<10s) / medium(10-30s) / long(>30s)。"""
    if duration is None or duration <= 0:
        return "unknown"
    if duration < 10:
        return "short"
    if duration <= 30:
        return "medium"
    return "long"


def score_asset(
    asset: Any,
    now: datetime | None = None,
) -> tuple[float, dict[str, float]]:
    """为单个素材计算综合得分（0-100）。

    维度权重：
    - quality_score (40%)：素材质量分（0-100），无质量分按 50 计
    - duration_fitness (30%)：时长适配度，5-30s 为最优区间
    - recency (20%)：新鲜度，30 天内衰减
    - unused_bonus (10%)：未被使用过的素材加分

    Returns:
        (total_score, breakdown_dict)
    """
    if now is None:
        now = datetime.now(timezone.utc)

    breakdown: dict[str, float] = {}

    # 1. 质量分 (0-100) → 权重 40%
    raw_quality = asset.quality_score if asset.quality_score is not None else 50.0
    quality_component = raw_quality * 0.4
    breakdown["quality"] = round(quality_component, 2)

    # 2. 时长适配度 (0-100) → 权重 30%
    # 最优区间 5-30s 得满分，越偏离越低
    duration = getattr(asset, "duration", None) or 0.0
    if duration <= 0:
        duration_fitness = 30.0  # 未知时长给中等分
    elif 5 <= duration <= 30:
        duration_fitness = 100.0
    elif duration < 5:
        # 0-5s: 线性增长 20→100
        duration_fitness = 20.0 + (duration / 5) * 80
    else:
        # >30s: 指数衰减，60s 时约 50 分
        duration_fitness = 100.0 * math.exp(-0.02 * (duration - 30))
        duration_fitness = max(duration_fitness, 10.0)
    duration_component = duration_fitness * 0.3
    breakdown["duration"] = round(duration_component, 2)

    # 3. 新鲜度 (0-100) → 权重 20%
    # 30 天半衰期
    created_at = getattr(asset, "created_at", None)
    if created_at is None:
        recency = 50.0
    else:
        if created_at.tzinfo is None:
            created_at = created_at.replace(tzinfo=timezone.utc)
        age_days = max(0, (now - created_at).total_seconds() / 86400)
        recency = 100.0 * math.exp(-0.05 * age_days)  # ~14天半衰期
    recency_component = recency * 0.2
    breakdown["recency"] = round(recency_component, 2)

    # 4. 未使用偏好 (0-100) → 权重 10%
    metadata = getattr(asset, "metadata", None) or {}
    try:
        use_count = int(metadata.get("generation_use_count") or 0)
    except (ValueError, TypeError):
        use_count = 0  # 脏数据时按未使用处理（保守策略：给未使用加分）
    if use_count == 0:
        unused_score = 100.0
    elif use_count <= 3:
        unused_score = 70.0
    else:
        unused_score = 30.0
    unused_component = unused_score * 0.1
    breakdown["unused"] = round(unused_component, 2)

    total = quality_component + duration_component + recency_component + unused_component
    return round(total, 2), breakdown


def smart_select_assets(
    assets: list[Any],
    *,
    limit: int | None = None,
    kind: str | None = None,
    now: datetime | None = None,
    rng: random.Random | None = None,
) -> list[SmartMatchResult]:
    """从素材列表中智能选取素材。

    Args:
        assets: 候选素材列表（Asset 实体）
        limit: 最大返回数量，None 表示不限制
        kind: 按文件类型过滤（video/image/audio），None 表示不过滤
        now: 当前时间（用于测试注入）
        rng: 随机数生成器（用于测试注入，控制排序噪声可复现）

    Returns:
        按有效得分（综合得分 + 随机噪声）降序排列的 SmartMatchResult 列表。
        r.score 始终为无噪声的原始综合得分；噪声仅用于排序/分桶顺序。
    """
    # Step 1: 过滤 ready 状态
    ready_assets = [a for a in assets if _get_enum_value(a, "status") == "ready"]

    # Step 2: 按 kind 过滤
    if kind:
        ready_assets = [a for a in ready_assets if a.file_type == kind]

    if not ready_assets:
        return []

    # Step 3: 评分
    scored: list[SmartMatchResult] = []
    for a in ready_assets:
        total, breakdown = score_asset(a, now=now)
        scored.append(SmartMatchResult(asset=a, score=total, breakdown=breakdown))

    # Step 4: 按「得分 + 随机噪声」降序排序
    # 同分/近分素材（分差 <= SCORE_RANDOM_NOISE_MAX）每次选出的顺序与组合不同，
    # 从素材组合层面降低成片重复率；分差显著（>20）的高质量素材排名不受影响。
    # 噪声以 asset.id 为 key 缓存，保证同一次调用内排序与分桶轮询顺序一致。
    rng = rng or random.Random()
    noise_by_asset: dict[str, float] = {
        getattr(a, "id", ""): rng.uniform(0.0, SCORE_RANDOM_NOISE_MAX) for a in ready_assets
    }

    def _effective(r: SmartMatchResult) -> float:
        return r.score + noise_by_asset.get(getattr(r.asset, "id", ""), 0.0)

    scored.sort(key=_effective, reverse=True)

    # Step 5: 多样性保障 — 时长分桶均衡选取（桶内同样按含噪声顺序）
    if limit and limit > 0 and len(scored) > limit:
        scored = _diversity_select(scored, limit, effective_key=_effective)
    elif limit and limit > 0:
        scored = scored[:limit]

    return scored


def _diversity_select(
    scored: list[SmartMatchResult],
    limit: int,
    effective_key: Any | None = None,
) -> list[SmartMatchResult]:
    """从已排序的候选中按分桶均衡选取，避免全选中同一时长档。

    策略：轮流从 short/medium/long 桶中按顺序取，直到凑满 limit。
    scored 已按含噪声的有效得分排序，桶内直接继承该顺序；
    effective_key 给出时最终输出也按有效得分排序（同一次调用内噪声一致）。
    """
    buckets: dict[str, list[SmartMatchResult]] = {
        "short": [],
        "medium": [],
        "long": [],
        "unknown": [],
    }
    for r in scored:
        bucket = _duration_bucket(getattr(r.asset, "duration", None))
        buckets.setdefault(bucket, []).append(r)

    selected: list[SmartMatchResult] = []
    selected_ids: set[str] = set()
    bucket_order = ["medium", "short", "long", "unknown"]  # medium 优先
    bucket_idx = {b: 0 for b in bucket_order}

    while len(selected) < limit:
        added = False
        for b in bucket_order:
            if len(selected) >= limit:
                break
            items = buckets.get(b, [])
            idx = bucket_idx[b]
            while idx < len(items):
                candidate = items[idx]
                idx += 1
                if candidate.asset.id not in selected_ids:
                    selected.append(candidate)
                    selected_ids.add(candidate.asset.id)
                    added = True
                    break
            bucket_idx[b] = idx
        if not added:
            break

    # 按有效得分（含噪声）降序输出；未传 effective_key 时退回原始得分
    selected.sort(key=effective_key or (lambda r: r.score), reverse=True)
    return selected