xiaoxia-saas/apps/worker/video_processing/bgm_mixer.py

"""BGM 混音模块 — 背景音乐与主音频混合.

基于 FFmpeg 实现：
- BGM 音量调节
- 淡入淡出（afade）
- 循环播放（aloop，短 BGM 铺长视频）
- 人声闪避（sidechaincompress，有人声时BGM自动降低音量）
- amix 混音

作为 render_audio.py 的增强模块，在 mix_audio 后处理阶段被调用。
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING

from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg

if TYPE_CHECKING:
    from video_processing.render_audio import RenderContext

logger = logging.getLogger(__name__)


@dataclass
class BGMConfig:
    """BGM 混音配置（内部使用，从 plan.config.bgm 转换而来）"""

    bgm_path: str  # BGM 本地文件路径
    volume: float = 0.3  # 0.0 ~ 1.0
    fade_in: float = 0.0  # 淡入时长（秒）
    fade_out: float = 0.0  # 淡出时长（秒）
    loop_enabled: bool = True  # 是否循环铺满
    sidechain_enabled: bool = False  # 人声闪避
    sidechain_ratio: float = 0.3  # 闪避时音量降低比例
    sidechain_attack: float = 0.02  # 攻击时间
    sidechain_release: float = 0.5  # 释放时间
    sidechain_threshold: float = -25.0  # 触发阈值（dB）

    @classmethod
    def from_config_dict(cls, bgm_path: str, config: dict) -> "BGMConfig":
        """从 plan.config.bgm 字典创建 BGMConfig。"""
        return cls(
            bgm_path=bgm_path,
            volume=float(config.get("volume", 0.3)),
            fade_in=float(config.get("fade_in", 0.0)),
            fade_out=float(config.get("fade_out", 0.0)),
            loop_enabled=bool(config.get("loop_enabled", True)),
            sidechain_enabled=bool(config.get("sidechain_enabled", False)),
            sidechain_ratio=float(config.get("sidechain_ratio", 0.3)),
            sidechain_attack=float(config.get("sidechain_attack", 0.02)),
            sidechain_release=float(config.get("sidechain_release", 0.5)),
            sidechain_threshold=float(config.get("sidechain_threshold", -25.0)),
        )


# ── BGM 预处理 ────────────────────────────────────────────────────────────────


def prepare_bgm_track(
    ctx: "RenderContext",
    bgm: BGMConfig,
    target_duration: float,
) -> Path:
    """预处理 BGM 轨道：循环/截断 + 音量 + 淡入淡出.

    生成一个时长精确等于 target_duration 的 BGM 音频文件。
    后续再与主音频混音。

    Args:
        ctx: 渲染上下文
        bgm: BGM 配置
        target_duration: 目标时长（秒），通常等于视频总时长

    Returns:
        处理后的 BGM 音频文件路径
    """
    output_path = ctx.work_dir / f"bgm_processed_{ctx.plan_id}.aac"

    if target_duration <= 0:
        target_duration = 5.0  # 兜底

    bgm_dur = probe_duration(bgm.bgm_path)
    needs_loop = bgm.loop_enabled and bgm_dur > 0 and bgm_dur < target_duration * 0.9

    # 构建滤镜链
    filter_parts: list[str] = []

    if needs_loop:
        # 计算需要循环多少次才能铺满
        loop_count = max(1, int(target_duration / bgm_dur) + 2)
        # aloop 滤镜：循环指定次数
        filter_parts.append(f"aloop=loop={loop_count}:size=0")

    # 音量调节
    volume = max(0.0, min(1.0, bgm.volume))
    if abs(volume - 1.0) > 0.001:
        filter_parts.append(f"volume={volume:.3f}")

    # 淡入
    if bgm.fade_in > 0:
        filter_parts.append(f"afade=t=in:st=0:d={bgm.fade_in:.3f}")

    # 淡出（从 target_duration - fade_out 开始）
    if bgm.fade_out > 0 and target_duration > bgm.fade_out:
        fade_start = target_duration - bgm.fade_out
        filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={bgm.fade_out:.3f}")

    # 最终截断到目标时长
    filter_parts.append(f"atrim=0:{target_duration:.3f}")
    filter_parts.append("asetpts=N/SR/TB")  # 重置时间戳

    filter_str = ",".join(filter_parts)

    command = [
        FFMPEG_BIN,
        "-y",
        "-i",
        bgm.bgm_path,
        "-filter:a",
        filter_str,
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        str(output_path),
    ]

    logger.info(
        "[bgm] prepare BGM track: path=%s dur=%.2f target=%.2f loop=%s fade_in=%.2f fade_out=%.2f",
        bgm.bgm_path[-40:],
        bgm_dur,
        target_duration,
        needs_loop,
        bgm.fade_in,
        bgm.fade_out,
    )

    run_ffmpeg(command)
    return output_path


# ── BGM + 主音频混音 ──────────────────────────────────────────────────────────


def mix_bgm_with_main(
    ctx: "RenderContext",
    main_audio_path: Path,
    bgm: BGMConfig,
    target_duration: float,
) -> Path:
    """将 BGM 与主音频混合.

    两种模式：
    1. 普通混音（sidechain 关闭）：amix 两路音频
    2. 人声闪避（sidechain 开启）：用 sidechaincompress 让 BGM 跟随主音频音量自动调整

    Args:
        ctx: 渲染上下文
        main_audio_path: 主音频文件路径（人声/原始音频）
        bgm: BGM 配置
        target_duration: 目标时长

    Returns:
        混音后的音频文件路径
    """
    output_path = ctx.work_dir / f"audio_with_bgm_{ctx.plan_id}.aac"

    # 先预处理 BGM 轨道（循环/音量/淡入淡出/截断）
    bgm_processed = prepare_bgm_track(ctx, bgm, target_duration)

    if not bgm.sidechain_enabled:
        # 普通 amix 混音
        _mix_simple(main_audio_path, bgm_processed, output_path)
    else:
        # sidechain 人声闪避混音
        _mix_sidechain(main_audio_path, bgm_processed, output_path, bgm)

    return output_path


def _mix_simple(main_path: Path, bgm_path: Path, output_path: Path) -> None:
    """简单 amix 混音：主音频 + BGM = 输出.

    主音频权重 1.0，BGM 已经在预处理阶段调好了音量。
    amix 会自动归一化，需要用 volume 补偿。
    """
    # 使用 amix：inputs=2，duration=first（以主音频时长为准）
    # 然后用 volume=2 补偿 amix 的衰减（2路输入每路平均乘0.5）
    filter_complex = "[0:a][1:a]amix=inputs=2:duration=first:dropout_transition=0[outa];" "[outa]volume=2[final]"

    command = [
        FFMPEG_BIN,
        "-y",
        "-i",
        str(main_path),
        "-i",
        str(bgm_path),
        "-filter_complex",
        filter_complex,
        "-map",
        "[final]",
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        str(output_path),
    ]

    logger.info("[bgm] simple amix mix")
    run_ffmpeg(command)


def _mix_sidechain(
    main_path: Path,
    bgm_path: Path,
    output_path: Path,
    bgm: BGMConfig,
) -> None:
    """sidechain 人声闪避混音.

    原理：
    - 主音频作为 sidechain 信号源
    - BGM 轨道经过 sidechaincompress，根据主音频音量动态调整 BGM 音量
    - 最后 amix 混音

    FFmpeg sidechaincompress 参数：
    - threshold: 触发阈值（dB），主音频超过此值时开始压缩
    - ratio: 压缩比，越高压缩越狠
    - attack: 攻击时间（秒）
    - release: 释放时间（秒）
    """
    # sidechain_ratio 表示闪避时 BGM 音量降低比例
    # ratio = 1 / (1 - sidechain_ratio)，但实际压缩比需要更精细调整
    # 简化处理：把 ratio 映射到 2:1 ~ 10:1 范围
    ratio = max(2.0, min(10.0, 1.0 / (1.0 - bgm.sidechain_ratio)))

    filter_complex = (
        # BGM 经过 sidechain 压缩，用主音频做触发
        f"[1:a][0:a]sidechaincompress="
        f"threshold={bgm.sidechain_threshold}dB:"
        f"ratio={ratio:.1f}:"
        f"attack={bgm.sidechain_attack:.3f}:"
        f"release={bgm.sidechain_release:.3f}:"
        f"knee=6[bgm_comp];"
        # 主音频 + 压缩后的 BGM 混音
        f"[0:a][bgm_comp]amix=inputs=2:duration=first:dropout_transition=0[outa];"
        f"[outa]volume=1.5[final]"  # 轻微补偿
    )

    command = [
        FFMPEG_BIN,
        "-y",
        "-i",
        str(main_path),
        "-i",
        str(bgm_path),
        "-filter_complex",
        filter_complex,
        "-map",
        "[final]",
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        str(output_path),
    ]

    logger.info(
        "[bgm] sidechain mix: threshold=%.1fdB ratio=%.1f attack=%.3f release=%.3f",
        bgm.sidechain_threshold,
        ratio,
        bgm.sidechain_attack,
        bgm.sidechain_release,
    )
    run_ffmpeg(command)


# ── 纯 BGM 模式（无主音频） ──────────────────────────────────────────────────


def build_bgm_only(
    ctx: "RenderContext",
    bgm: BGMConfig,
    target_duration: float,
) -> Path:
    """只有 BGM、没有主音频时，直接生成 BGM 音频.

    Args:
        ctx: 渲染上下文
        bgm: BGM 配置
        target_duration: 目标时长

    Returns:
        BGM 音频文件路径
    """
    output_path = ctx.work_dir / f"bgm_only_{ctx.plan_id}.aac"

    if target_duration <= 0:
        target_duration = 5.0

    bgm_processed = prepare_bgm_track(ctx, bgm, target_duration)

    # 直接复制
    import shutil

    shutil.copy2(bgm_processed, output_path)
    return output_path