xiaoxia-saas/packages/domain/atom_clip_selector.py

"""原子片段级选片核心 — #1970 智能剪辑流程重构 P1.

选片单元从"整条素材 + 随机起点"升级为"原子片段（atom clip）"：

- 每个 EditPlanClip 指向一个 atom_clip_id（含 asset_id + start/end）；
- 同一素材的不同原子片段可被同一视频多次选用；
- 同一原子片段在一个视频内只用一次；
- 跨变体/跨任务的避让升级为原子片段级（同 asset 的不同片段天然不重叠）；
- atom_clips 未就绪（老素材/切片失败）时由调用方走内存兜底切片，
  再不行回退到现有的整条素材随机起点逻辑。

本模块是纯函数：原子片段数据由调用方从 repository 读取后注入，不直接碰 DB，
便于单元测试。评分维度与 smart_match 保持一致（质量分、时长适配、新鲜度、
未使用加分），只是评分对象从素材变为原子片段。
"""

from __future__ import annotations

import random
from dataclasses import dataclass
from typing import Any

from packages.domain.asset_atom_clip import AssetAtomClip


@dataclass(slots=True)
class ScoredAtomClip:
    """带评分的候选原子片段。"""

    clip: AssetAtomClip
    score: float

    @property
    def atom_clip_id(self) -> str:
        return self.clip.id

    @property
    def asset_id(self) -> str:
        return self.clip.asset_id

    @property
    def start_time(self) -> float:
        return self.clip.start_time

    @property
    def end_time(self) -> float:
        return self.clip.end_time

    @property
    def duration(self) -> float:
        return self.clip.duration


# 评分权重（与 smart_match.score_asset 的维度对齐）
W_QUALITY = 0.35
W_DURATION_FIT = 0.30
W_FRESHNESS = 0.15
W_UNUSED_BONUS = 0.10
W_ASSET_BALANCE = 0.10

# 评分随机噪声上限（与 SCORE_RANDOM_NOISE_MAX 同量级，避免反复选同一组合）
SCORE_NOISE_MAX = 0.05


def score_atom_clip(
    clip: AssetAtomClip,
    *,
    target_duration: float,
    asset_quality: dict[str, float] | None = None,
    asset_freshness: dict[str, float] | None = None,
    used_in_video: set[str] | None = None,
    asset_usage_counts: dict[str, int] | None = None,
    recently_used: set[str] | None = None,
    required_count: int = 1,
    total_candidates: int = 1,
) -> float:
    """评估单个原子片段对某个目标槽位的适配分（越高越优先）.

    评分维度：
    - 质量分（继承素材质量，缺省中性 0.6）；
    - 时长适配（片段时长越接近目标越好，覆盖不满显著扣分）；
    - 新鲜度（缺省中性 0.5）；
    - 未使用加分（本视频内未用过 +1，已用 0）；
    - 素材均衡（同一素材在本视频用得越多，其剩余片段扣分越多，鼓励分散到多素材）；
    - 跨视频/历史使用降权（recently_used 中的片段扣分，不硬禁）。
    """
    asset_quality = asset_quality or {}
    asset_freshness = asset_freshness or {}
    used_in_video = used_in_video or set()
    asset_usage_counts = asset_usage_counts or {}
    recently_used = recently_used or set()

    quality = asset_quality.get(clip.asset_id, 0.6)

    if target_duration > 0:
        coverage = min(1.0, clip.duration / target_duration)
        overshoot = max(0.0, (clip.duration - target_duration) / target_duration)
        duration_fit = max(0.0, coverage - 0.15 * overshoot)
    else:
        duration_fit = 0.5

    freshness = asset_freshness.get(clip.asset_id, 0.5)
    unused_bonus = 0.0 if clip.id in used_in_video else 1.0

    # 素材均衡：该素材已被本视频选用 k 次，其片段逐次扣分
    times_used = asset_usage_counts.get(clip.asset_id, 0)
    balance = 1.0 / (1.0 + times_used)

    # 跨视频/历史使用降权（不硬禁）
    history_penalty = 0.35 if clip.id in recently_used else 0.0

    score = (
        W_QUALITY * quality
        + W_DURATION_FIT * duration_fit
        + W_FRESHNESS * freshness
        + W_UNUSED_BONUS * unused_bonus
        + W_ASSET_BALANCE * balance
        - history_penalty
    )
    return score


def select_atom_clips(
    candidates: list[AssetAtomClip],
    *,
    target_duration: float = 0.0,
    used_atom_clip_ids: set[str] | None = None,
    asset_usage_counts: dict[str, int] | None = None,
    recently_used_atom_ids: set[str] | None = None,
    required_count: int = 1,
    limit: int = 0,
    asset_quality: dict[str, float] | None = None,
    asset_freshness: dict[str, float] | None = None,
    rng: random.Random | None = None,
) -> list[ScoredAtomClip]:
    """为一个目标槽位从候选原子片段中评分选片（纯函数）.

    Args:
        candidates: 候选原子片段（可跨多素材）。
        target_duration: 槽位目标时长（秒）。
        used_atom_clip_ids: 本视频已用过的原子片段 ID（硬排除，同片段不重复）。
        asset_usage_counts: 本视频各素材已选片段数（均衡评分用）。
        recently_used_atom_ids: 跨视频/历史成片用过的片段 ID（降权，不硬禁）。
        required_count: 整个视频需要的片段总数（预留，供覆盖策略判断）。
        limit: 最多返回条数；<=0 表示返回全部排序结果。
        asset_quality / asset_freshness: 评分注入。
        rng: 可选随机源（测试注入）。

    Returns:
        评分降序的 ScoredAtomClip 列表（已排除本视频用过的片段）。
    """
    rng = rng or random.Random()
    used = used_atom_clip_ids or set()
    asset_usage_counts = asset_usage_counts or {}
    recently_used = recently_used_atom_ids or set()

    available = [c for c in candidates if c.id not in used]
    scored: list[ScoredAtomClip] = []
    for clip in available:
        base = score_atom_clip(
            clip,
            target_duration=target_duration,
            asset_quality=asset_quality,
            asset_freshness=asset_freshness,
            used_in_video=used,
            asset_usage_counts=asset_usage_counts,
            recently_used=recently_used,
            required_count=required_count,
            total_candidates=len(candidates),
        )
        noise = rng.uniform(0.0, SCORE_NOISE_MAX)
        scored.append(ScoredAtomClip(clip=clip, score=base + noise))

    scored.sort(key=lambda s: s.score, reverse=True)
    if limit and limit > 0:
        return scored[:limit]
    return scored


def clips_to_segments(clips: list[AssetAtomClip]) -> dict[str, list[tuple[float, float]]]:
    """把选中的原子片段转换为旧的 {asset_id: [(start, end), ...]} 区间结构.

    用于与现有跨变体区间避让（variant_plan_selector / metadata.used_segments）对接。
    原子片段级天然不重叠，同素材多片段直接形成多段不重叠区间。
    """
    segments: dict[str, list[tuple[float, float]]] = {}
    for clip in clips:
        segments.setdefault(clip.asset_id, []).append((clip.start_time, clip.end_time))
    for asset_id in segments:
        segments[asset_id].sort()
    return segments


def estimate_required_clip_count(
    voice_total_duration: float,
    average_clip_duration: float = 4.5,
) -> int:
    """配音总时长 / 平均片段时长 ≈ 需要的片段数（至少 1）。"""
    if voice_total_duration <= 0 or average_clip_duration <= 0:
        return 1
    return max(1, round(voice_total_duration / average_clip_duration))


def reselect_clips_from_atoms(
    source_clips: list[dict[str, Any]],
    candidates: list[AssetAtomClip],
    *,
    historical_atom_ids: set[str] | None = None,
    batch_used_atom_ids: set[str] | None = None,
    rng: random.Random | None = None,
) -> list[dict[str, Any]] | None:
    """#1970 变体重选的原子片段级实现.

    与 variant_plan_selector.reselect_clips_for_variant 对应：保留源 plan 的
    片段骨架（order/clip_type/文案/转场），从候选原子片段中为每个 main 片段
    选取一个原子片段；同变体/批次内同一片段不可重复，历史成片用过的片段降权。

    Returns:
        新 clips_data（dict 列表，含 asset_id/atom_clip_id/start_time/duration），
        候选不足（main 片段多于去重后片段数）时返回 None，由调用方回退整条素材路径。
        非 main 片段（intro/outro 等）原样保留不分配素材。
    """
    if not source_clips or not candidates:
        return None

    rng = rng or random.Random()
    main_indexes = [i for i, c in enumerate(source_clips) if c.get("clip_type", "main") == "main"]
    if len(main_indexes) > len({c.id for c in candidates}):
        return None

    used: set[str] = set(batch_used_atom_ids or ())
    result: list[dict[str, Any]] = [dict(c) for c in source_clips]
    asset_usage: dict[str, int] = {}

    for idx in main_indexes:
        skeleton = source_clips[idx]
        target_duration = float(skeleton.get("duration") or 0.0)
        ranked = select_atom_clips(
            candidates,
            target_duration=target_duration,
            used_atom_clip_ids=used,
            asset_usage_counts=asset_usage,
            recently_used_atom_ids=historical_atom_ids or set(),
            required_count=len(main_indexes),
            limit=1,
            rng=rng,
        )
        if not ranked:
            return None
        picked = ranked[0]
        # 段长：片段短于槽位时取片段全长（渲染末帧冻结铺满），长于槽位时按槽位时长 trim
        new_duration = picked.duration if target_duration <= 0 else min(target_duration, picked.duration)
        result[idx].update(
            {
                "asset_id": picked.asset_id,
                "atom_clip_id": picked.atom_clip_id,
                "start_time": round(picked.start_time, 3),
                "duration": round(new_duration, 3),
            }
        )
        used.add(picked.atom_clip_id)
        asset_usage[picked.asset_id] = asset_usage.get(picked.asset_id, 0) + 1

    return result