xiaoxia-saas/packages/domain/narrative_match.py

"""叙事剪辑素材标签匹配 — #1970 PR3 + P2 AI 标签加权.

叙事模式下，选片在现有评分（smart_match / atom_clip_selector）之前先做一层
文案标签匹配：

- 文案 tags 与素材 tag 名归一化后求交集；
- 命中任一标签的素材作为「优先候选池」，未命中的作为普通池；
- 调用方对优先池跑现有 smart_select_assets，数量不足时用普通池补足
  （无任何匹配 → 完全降级为现有随机逻辑，行为与改造前一致）。

P2 AI 标签加权（#1970 fragment-level AI tagging）：
- 片段级 AI 标签（scene/objects/action）与文案标签做交集时权重 2.0
- 素材级标签（tag_ids 映射名）与文案标签交集时权重 1.0
- 综合得分 = sum(命中权重) / max(可能权重)
- 有 AI 标签的片段命中时优先于仅素材标签命中的片段

纯函数模块：标签 id→名称映射由调用方查 TagModel 后注入，不直接碰 DB。
"""

from __future__ import annotations

from typing import Any, Iterable

# 标签归一化后仍短于此长度的标签不参与匹配（避免「的」「是」这类噪声短词）
MIN_TAG_LEN = 2

# 标签匹配权重
AI_TAG_WEIGHT = 2.0  # AI 标签命中权重
ASSET_TAG_WEIGHT = 1.0  # 素材标签命中权重


def normalize_tag(tag: Any) -> str:
    """标签归一化：去空白、小写。数字/英文统一小写，中文不受影响。"""
    if tag is None:
        return ""
    return str(tag).strip().lower()


def _normalize_tags(tags: Iterable[Any]) -> set[str]:
    out: set[str] = set()
    for t in tags or []:
        norm = normalize_tag(t)
        if len(norm) >= MIN_TAG_LEN:
            out.add(norm)
    return out


def build_asset_tag_name_index(tag_names_by_id: dict[str, Any]) -> dict[str, set[str]]:
    """构造 asset_id → 归一化标签名集合 的索引。

    Args:
        tag_names_by_id: {asset_id: [标签名或标签id, ...]}，允许混入 None/空值
    """
    index: dict[str, set[str]] = {}
    for asset_id, names in (tag_names_by_id or {}).items():
        index[asset_id] = _normalize_tags(names)
    return index


def _extract_ai_tag_names(ai_tags: dict) -> set[str]:
    """从 AI 标签 dict 中提取所有标签名（scene + objects + action）.

    Args:
        ai_tags: 片段级 AI 标签 dict，如 {"scene": [...], "objects": [...], "action": [...], ...}

    Returns:
        归一化后的标签名集合。
    """
    names: set[str] = set()
    for key in ("scene", "objects", "action"):
        values = ai_tags.get(key)
        if isinstance(values, list):
            names |= _normalize_tags(values)
    return names


def _compute_ai_score(
    asset_id: str,
    wanted: set[str],
    clip_ai_tags_by_asset: dict[str, list[dict]] | None,
) -> float:
    """计算单个素材的 AI 标签加权得分.

    对该素材的所有片段 AI 标签，求各片段标签名与文案标签交集的加权总和。
    每个片段的命中权重 = 命中数 × AI_TAG_WEIGHT。
    最终取所有片段的最高得分（而非累加，避免片段数多的素材不公平占优）。

    Args:
        asset_id: 素材 ID。
        wanted: 归一化后的文案标签集合。
        clip_ai_tags_by_asset: {asset_id: [ai_tag_dict, ...]} 每个片段一个。

    Returns:
        AI 标签加权得分（≥0）。
    """
    if not clip_ai_tags_by_asset or not wanted:
        return 0.0

    clips = clip_ai_tags_by_asset.get(asset_id)
    if not clips:
        return 0.0

    best_score = 0.0
    for ai_tags in clips:
        if not ai_tags or not isinstance(ai_tags, dict):
            continue
        ai_names = _extract_ai_tag_names(ai_tags)
        hits = ai_names & wanted
        score = len(hits) * AI_TAG_WEIGHT
        if score > best_score:
            best_score = score

    return best_score


def match_assets_by_script_tags(
    assets: list[Any],
    *,
    script_tags: Iterable[Any],
    tag_names_by_id: dict[str, Any] | None = None,
    clip_ai_tags_by_asset: dict[str, list[dict]] | None = None,
) -> tuple[list[Any], list[Any]]:
    """按文案标签把素材拆成「命中池 / 未命中池」，保持输入相对顺序。

    P2 加权逻辑：
    - AI 标签命中（scene/objects/action ∩ 文案标签）权重 2.0
    - 素材标签命中（tag_ids 映射名 ∩ 文案标签）权重 1.0
    - 任一权重 > 0 → 命中池，否则 → 未命中池

    Args:
        assets: 候选素材（domain Asset，需有 id 与 tag_ids）。
        script_tags: 文案 tags（字符串数组，名称语义）。
        tag_names_by_id: asset_id → 素材标签名列表。
        clip_ai_tags_by_asset: #1970 P2 — {asset_id: [ai_tag_dict, ...]}。

    Returns:
        (matched, unmatched)：命中任一文案标签的素材 / 其余素材。
        文案无有效标签时 matched 为空（调用方直接走随机逻辑）。
    """
    wanted = _normalize_tags(script_tags)
    if not wanted:
        return [], list(assets)

    name_index = build_asset_tag_name_index(tag_names_by_id or {})
    matched: list[Any] = []
    unmatched: list[Any] = []
    for asset in assets:
        asset_id = str(getattr(asset, "id", "") or "")

        # P2: AI 标签加权得分
        ai_score = _compute_ai_score(asset_id, wanted, clip_ai_tags_by_asset)

        # 素材标签得分
        names = set(name_index.get(asset_id, set()))
        raw_tags = getattr(asset, "tags", None)
        if raw_tags:
            names |= _normalize_tags(raw_tags)
        asset_score = len(names & wanted) * ASSET_TAG_WEIGHT

        # 综合得分 > 0 → 命中池
        if ai_score > 0 or asset_score > 0:
            matched.append(asset)
        else:
            unmatched.append(asset)
    return matched, unmatched


def compute_tag_match_score(
    asset_id: str,
    *,
    script_tags: Iterable[Any],
    tag_names_by_id: dict[str, Any] | None = None,
    clip_ai_tags_by_asset: dict[str, list[dict]] | None = None,
) -> float:
    """计算单个素材的标签匹配综合得分（0.0 ~ 1.0）.

    综合得分 = sum(命中权重) / max(可能权重)
    - AI 标签每命中一个 +2.0
    - 素材标签每命中一个 +1.0
    - max_possible = len(wanted) * (AI_TAG_WEIGHT + ASSET_TAG_WEIGHT)

    Args:
        asset_id: 素材 ID。
        script_tags: 文案标签。
        tag_names_by_id: 素材标签名索引。
        clip_ai_tags_by_asset: AI 标签索引。

    Returns:
        归一化得分 0.0~1.0。
    """
    wanted = _normalize_tags(script_tags)
    if not wanted:
        return 0.0

    # AI 得分
    ai_score = _compute_ai_score(asset_id, wanted, clip_ai_tags_by_asset)

    # 素材标签得分
    name_index = build_asset_tag_name_index(tag_names_by_id or {})
    names = name_index.get(asset_id, set())
    asset_score = len(names & wanted) * ASSET_TAG_WEIGHT

    # 归一化：最大可能得分 = 文案标签数 × (AI权重 + 素材权重)
    max_possible = len(wanted) * (AI_TAG_WEIGHT + ASSET_TAG_WEIGHT)
    if max_possible <= 0:
        return 0.0

    return min((ai_score + asset_score) / max_possible, 1.0)


def pick_narrative_assets(
    assets: list[Any],
    *,
    script_tags: Iterable[Any],
    tag_names_by_id: dict[str, Any] | None = None,
    clip_ai_tags_by_asset: dict[str, list[dict]] | None = None,
    limit: int | None = None,
    rng: Any = None,
) -> list[Any]:
    """叙事模式选片：标签命中池优先，不足部分从未命中池按现有评分补齐。

    本函数只负责「标签优先 + 兜底降级」的顺序编排；评分仍复用
    smart_match.smart_select_assets（质量/时长/新鲜度/未使用 + 随机噪声），
    不重写评分维度。

    P2 增强：有 AI 标签的片段命中时权重更高（2.0 vs 1.0），
    命中池内部按综合标签得分排序（AI 标签命中多的排前面）。

    Args:
        assets: ready 视频素材候选（调用方负责状态/类型过滤）。
        script_tags / tag_names_by_id: 见 match_assets_by_script_tags。
        clip_ai_tags_by_asset: #1970 P2 — {asset_id: [ai_tag_dict, ...]}。
        limit: 需要的素材数量；None 表示全部（命中池 + 全部未命中池）。
        rng: 注入 smart_select_assets 的随机源（可复现）。

    Returns:
        选中的素材列表。无任何标签命中时等价于对全量跑 smart_select_assets。
    """
    from packages.domain.smart_match import smart_select_assets

    matched, unmatched = match_assets_by_script_tags(
        assets,
        script_tags=script_tags,
        tag_names_by_id=tag_names_by_id,
        clip_ai_tags_by_asset=clip_ai_tags_by_asset,
    )

    need = limit if (limit is not None and limit > 0) else None

    if not matched:
        # 完全降级：与改造前随机混剪同一逻辑
        return [r.asset for r in smart_select_assets(assets, kind="video", limit=need, rng=rng)]

    picked = [r.asset for r in smart_select_assets(matched, kind="video", limit=need, rng=rng)]
    if need is not None and len(picked) < need and unmatched:
        rest_need = need - len(picked)
        picked.extend(r.asset for r in smart_select_assets(unmatched, kind="video", limit=rest_need, rng=rng))
    elif need is None:
        picked.extend(r.asset for r in smart_select_assets(unmatched, kind="video", rng=rng))
    return picked