xiaoxia-saas/packages/domain/bgm_pool.py

"""BGM 池差异化分配 — 打破变体间音频指纹一致性 (Issue #1767).

三层递进策略：
1. **BGM 池分配（核心）**：维护风格匹配的 BGM 池，每个变体基于 variant_seed
   随机分配一首不同 BGM，保证变体间音频指纹不同。
2. **段落差异化（池不够时的补充）**：同一首 BGM 做差异化裁剪，不同变体使用
   不同起始点/段落，进一步降低音频相似度。
3. **音量微调**：不同变体 BGM 音量 ±3dB 微调，混音比例有微小差异。

约束：
- 不破坏现有单视频 BGM 选择逻辑（单视频不走池分配）
- BGM 情绪/风格与视频内容匹配（基于源 plan 的 BGM style 做风格筛选）
- 分配可复现（同 seed 同结果）
"""

from __future__ import annotations

import logging
import random
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)


# ── BGM 池条目 ──────────────────────────────────────────────────────────────


@dataclass(frozen=True)
class BGMPoolEntry:
    """BGM 池条目"""

    id: str
    preset_id: str  # 关联 PRESET_BGM_LIBRARY 中的 ID（用于渲染侧解析音频路径）
    mood: str  # 情绪/风格：upbeat / relax / tech / commerce / emotional / cinematic
    duration: float  # 时长（秒）
    audio_url: str = ""  # CDN/OSS 直链（优先级高于 preset_id）
    tags: list[str] = field(default_factory=list)


# ── BGM 池（10 首，覆盖 6 种风格） ──────────────────────────────────────────

BGM_POOL: list[BGMPoolEntry] = [
    # upbeat (轻快)
    BGMPoolEntry(
        id="pool_upbeat_001", preset_id="bgm_upbeat_001", mood="upbeat", duration=120.0, tags=["轻快", "阳光", "vlog"]
    ),
    BGMPoolEntry(
        id="pool_upbeat_002", preset_id="bgm_upbeat_002", mood="upbeat", duration=95.0, tags=["轻快", "电子", "运动"]
    ),
    BGMPoolEntry(
        id="pool_upbeat_003", preset_id="bgm_upbeat_003", mood="upbeat", duration=110.0, tags=["轻快", "夏日", "旅行"]
    ),
    # relax (治愈)
    BGMPoolEntry(
        id="pool_relax_001", preset_id="bgm_relax_001", mood="relax", duration=180.0, tags=["治愈", "钢琴", "冥想"]
    ),
    BGMPoolEntry(
        id="pool_relax_002", preset_id="bgm_relax_002", mood="relax", duration=150.0, tags=["治愈", "自然", "放松"]
    ),
    BGMPoolEntry(
        id="pool_relax_003", preset_id="bgm_relax_003", mood="relax", duration=200.0, tags=["治愈", "古典", "钢琴"]
    ),
    # tech (科技)
    BGMPoolEntry(
        id="pool_tech_001", preset_id="bgm_tech_001", mood="tech", duration=85.0, tags=["科技", "电子", "数码"]
    ),
    BGMPoolEntry(
        id="pool_tech_002", preset_id="bgm_tech_002", mood="tech", duration=100.0, tags=["科技", "极简", "AI"]
    ),
    # commerce (电商)
    BGMPoolEntry(
        id="pool_commerce_001",
        preset_id="bgm_commerce_001",
        mood="commerce",
        duration=75.0,
        tags=["电商", "时尚", "带货"],
    ),
    BGMPoolEntry(
        id="pool_commerce_002",
        preset_id="bgm_commerce_002",
        mood="commerce",
        duration=90.0,
        tags=["电商", "品牌", "品质"],
    ),
]


# ── 风格 → 情绪映射 ────────────────────────────────────────────────────────
# preset_bgm.py 中 style 字段 → bgm_pool.py 中 mood 字段

STYLE_TO_MOOD: dict[str, str] = {
    "upbeat": "upbeat",
    "relax": "relax",
    "tech": "tech",
    "commerce": "commerce",
    "emotional": "emotional",
    "cinematic": "cinematic",
}


# ── 策略一：BGM 池分配 ─────────────────────────────────────────────────────


def get_bgm_pool_candidates(source_mood: str | None = None) -> list[BGMPoolEntry]:
    """获取 BGM 池候选列表。

    如果指定了 source_mood，优先返回同 mood 的条目；
    如果同 mood 条目不足 2 个，降级返回全池（保证有足够候选）。

    Args:
        source_mood: 源 BGM 的情绪/风格（来自 preset_bgm.py 的 style 字段）

    Returns:
        候选 BGM 列表（至少 2 个条目）
    """
    if not source_mood:
        return list(BGM_POOL)

    mood = STYLE_TO_MOOD.get(source_mood, source_mood)
    matched = [e for e in BGM_POOL if e.mood == mood]

    # 同 mood 至少要有 2 首，否则无法"差异化"，降级全池
    if len(matched) >= 2:
        return matched
    return list(BGM_POOL)


def select_bgm_from_pool(
    variant_seed: int,
    candidates: list[BGMPoolEntry] | None = None,
) -> BGMPoolEntry:
    """基于 variant_seed 从候选池中选一首 BGM（可复现）。

    Args:
        variant_seed: 变体随机种子
        candidates: 候选池（None 时使用全池）

    Returns:
        选中的 BGM 条目
    """
    pool = candidates if candidates is not None else list(BGM_POOL)
    if not pool:
        pool = list(BGM_POOL)
    rng = random.Random(variant_seed)
    return rng.choice(pool)


# ── 策略二：段落差异化 ─────────────────────────────────────────────────────


def generate_bgm_segment_offset(variant_seed: int, bgm_duration: float) -> float:
    """为变体生成 BGM 段落起始偏移（策略二）。

    不同变体从同一首 BGM 的不同位置开始播放，进一步降低音频指纹相似度。

    偏移范围 [0, max_offset]，max_offset = min(30s, bgm_duration * 0.3)。
    量化到 5 秒整数倍，便于复现和调试。

    Args:
        variant_seed: 变体随机种子
        bgm_duration: BGM 总时长（秒）

    Returns:
        起始偏移（秒），0 ~ max_offset 之间，5s 步长
    """
    rng = random.Random(variant_seed + 7919)  # 加素数偏移，避免与 BGM 选择 seed 序列重合
    max_offset = min(30.0, bgm_duration * 0.3)
    steps = int(max_offset // 5.0)
    if steps <= 0:
        return 0.0
    return float(rng.randint(0, steps) * 5)


# ── 策略三：音量微调 ─────────────────────────────────────────────────────


def generate_bgm_volume_adjust(variant_seed: int) -> float:
    """为变体生成 BGM 音量微调值（策略三）。

    ±3dB 微调，让不同变体的 BGM/配音混音比例有微小差异。
    离散步长：-3, -2, -1, 0, 1, 2, 3 dB。

    Args:
        variant_seed: 变体随机种子

    Returns:
        音量调整值（dB），-3.0 ~ 3.0
    """
    rng = random.Random(variant_seed + 104729)  # 另一个素数偏移
    return float(rng.choice([-3, -2, -1, 0, 1, 2, 3]))


# ── 批量分配入口 ──────────────────────────────────────────────────────────


def allocate_bgm_pool_for_variants(
    source_bgm_config: dict,
    variant_seeds: list[int],
) -> list[dict]:
    """为批量变体分配不同的 BGM 池配置。

    整合三层策略：池分配 + 段落偏移 + 音量微调。
    每个变体得到一个 dict，可直接合并到 plan.config["bgm"] 中。

    Args:
        source_bgm_config: 源 plan 的 BGM 配置（用于风格匹配）
        variant_seeds: 每个变体的随机种子列表

    Returns:
        每个变体的 BGM 池配置 dict 列表（与 variant_seeds 等长），
        每项包含 preset_id / audio_url / audio_offset / volume_adjust_db。
        如果源 BGM 未启用，返回空列表。
    """
    if not source_bgm_config or not source_bgm_config.get("enabled", False):
        return []
    if not variant_seeds:
        return []

    # 从源 BGM 配置中推断风格
    source_mood = _infer_source_mood(source_bgm_config)

    # 策略一：获取候选池
    candidates = get_bgm_pool_candidates(source_mood)

    # 为每个变体分配不同的 BGM（尽量不重复）
    assignments = _assign_unique_bgm(candidates, variant_seeds)

    results = []
    for i, (entry, seed) in enumerate(zip(assignments, variant_seeds, strict=False)):
        # 策略二：段落偏移
        offset = generate_bgm_segment_offset(seed, entry.duration)

        # 策略三：音量微调
        volume_adj = generate_bgm_volume_adjust(seed)

        result = {
            "preset_id": entry.preset_id,
            "audio_url": entry.audio_url,
            "audio_offset": offset,
            "volume_adjust_db": volume_adj,
            "bgm_pool_entry_id": entry.id,
            "bgm_pool_mood": entry.mood,
        }
        results.append(result)
        logger.info(
            "变体 %d BGM 池分配: seed=%d bgm=%s mood=%s offset=%.1fs vol_adj=%+.0fdB",
            i,
            seed,
            entry.id,
            entry.mood,
            offset,
            volume_adj,
        )

    return results


def _infer_source_mood(source_bgm_config: dict) -> str | None:
    """从源 BGM 配置推断风格/情绪。

    优先级：
    1. preset_id → 查 preset_bgm 库获取 style
    2. bgm_pool_mood → 上游已设置过（二次分配场景）
    3. 无法推断 → None（返回全池候选）
    """
    preset_id = source_bgm_config.get("preset_id", "")
    if preset_id:
        from packages.domain.preset_bgm import get_preset_bgm

        preset = get_preset_bgm(preset_id)
        if preset:
            return STYLE_TO_MOOD.get(preset.style, preset.style)

    # 如果之前已经分配过 BGM 池，直接用 mood
    pool_mood = source_bgm_config.get("bgm_pool_mood", "")
    if pool_mood:
        return pool_mood

    return None


def _assign_unique_bgm(
    candidates: list[BGMPoolEntry],
    variant_seeds: list[int],
) -> list[BGMPoolEntry]:
    """尽量让每个变体选到不同的 BGM。

    策略：用 seed 选 BGM，如果与前面变体重复，用递增 seed 重试。
    如果候选池大小 < 变体数，允许重复但不连续。
    """
    if not candidates or not variant_seeds:
        return []

    assignments: list[BGMPoolEntry] = []
    used_ids: set[str] = set()

    for i, seed in enumerate(variant_seeds):
        rng = random.Random(seed)
        # 先尝试选一个没用过的
        chosen = None
        for _attempt in range(len(candidates)):
            candidate = rng.choice(candidates)
            if candidate.id not in used_ids:
                chosen = candidate
                break
        if chosen is None:
            # 候选池已用完，允许重复但取下一个（循环）
            idx = i % len(candidates)
            chosen = candidates[idx]

        assignments.append(chosen)
        used_ids.add(chosen.id)

    return assignments