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

"""多轨道混音引擎 — 支持多路音频独立音量调节与混合.

基于 FFmpeg amix / amerge 实现：
- 支持任意数量音频轨道（原音、BGM、配音、音效等）
- 每轨独立音量调节
- 每轨独立淡入淡出
- 每轨独立时间偏移（delay）
- 总输出音量归一化补偿

作为 render_audio.py 的增强模块，在 mix_audio 后处理阶段被调用。
与 bgm_mixer.py 的关系：
- bgm_mixer 专注 BGM 单轨道的复杂处理（循环、人声闪避）
- 本模块专注多路轨道的统一音量调节与混合
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import TYPE_CHECKING

from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path

from packages.domain.audio_track_config import (  # noqa: F401 — 向后兼容
    ALLOWED_AUDIO_EXTENSIONS,
    DEFAULT_VOLUMES,
    MAX_AUDIO_TRACKS,
    TRACK_TYPE_AMBIENT,
    TRACK_TYPE_BGM,
    TRACK_TYPE_MAIN,
    TRACK_TYPE_SFX,
    TRACK_TYPE_VOICEOVER,
    AudioTrack,
    MultiTrackMixConfig,
)

if TYPE_CHECKING:
    from video_processing.render_audio import RenderContext

logger = logging.getLogger(__name__)


# ── 路径安全校验 ────────────────────────────────────────────────────────────


def _validate_audio_path(audio_path: str, work_dir: Path) -> None:
    """校验音频文件路径安全性.

    规则：
    - local:// schema → 必须在 work_dir 内
    - 相对路径 → 必须在 work_dir 内
    - 绝对路径 → 必须在允许目录白名单内
    - 扩展名必须是音频格式

    Raises:
        PathSecurityError: 路径不安全
    """
    if not audio_path or not isinstance(audio_path, str):
        raise PathSecurityError("音频路径不能为空")

    # 本地路径（local:// 或相对路径）
    if audio_path.startswith("local://") or not audio_path.startswith(("http://", "https://", "oss://")):
        is_abs = audio_path.startswith("/") and not audio_path.startswith("local://")
        resolved_path = safe_resolve_path(
            audio_path,
            work_dir,
            allow_outside=is_abs,
            allowed_extensions=ALLOWED_AUDIO_EXTENSIONS,
        )
        # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过）
        if is_abs:
            resolved_work_dir = work_dir.resolve()
            try:
                resolved_path.relative_to(resolved_work_dir)
            except ValueError as _e:
                if not is_in_allowed_dirs(resolved_path):
                    raise PathSecurityError(f"音频路径不在允许目录内: {audio_path[:80]}") from _e
    # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责）
    # 但检查扩展名
    else:
        # URL路径，检查扩展名白名单（取 ? 之前的部分）
        path_part = audio_path.split("?")[0].split("#")[0]
        from pathlib import Path as _P

        ext = _P(path_part).suffix.lower()
        if ext and ext not in ALLOWED_AUDIO_EXTENSIONS:
            raise PathSecurityError(f"不允许的音频文件类型: {ext}")


# ── 单轨道预处理 ────────────────────────────────────────────────────────────


def _prepare_single_track(
    ctx: "RenderContext",
    track: AudioTrack,
    target_duration: float,
    output_path: Path,
) -> bool:
    """预处理单条轨道：音量 + 淡入淡出 + 时间偏移 + 截断.

    生成一个精确对齐时间轴的音频文件，后续统一 amix 混音。

    Returns:
        True 表示处理成功，False 表示失败（跳过）
    """
    try:
        audio_dur = probe_duration(track.audio_path)
    except Exception:
        logger.warning("[multi-track] probe failed, skip track: %s", track.track_id)
        return False

    if audio_dur <= 0:
        return False

    # 计算实际有效时长
    effective_start = track.start_time
    if track.duration > 0:
        effective_dur = min(track.duration, audio_dur)
    else:
        effective_dur = audio_dur

    # 如果轨道完全在视频时长之外，跳过
    if effective_start >= target_duration:
        return False
    if effective_start + effective_dur <= 0:
        return False

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

    # 1. 先截断到有效范围
    trim_start = 0.0  # 从源文件的哪个位置开始取
    if effective_start < 0:
        trim_start = -effective_start
        effective_start = 0.0

    # 实际需要的源时长
    need_dur = min(effective_dur, target_duration - effective_start)
    if need_dur <= 0:
        return False

    filter_parts.append(f"atrim={trim_start:.3f}:{trim_start + need_dur:.3f}")
    filter_parts.append("asetpts=N/SR/TB")

    # 2. 音量调节
    if abs(track.volume - 1.0) > 0.001:
        filter_parts.append(f"volume={track.volume:.3f}")

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

    # 4. 淡出
    if track.fade_out > 0 and track.fade_out < need_dur:
        fade_start = need_dur - track.fade_out
        if fade_start > 0:
            filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={track.fade_out:.3f}")

    # 5. 时间偏移（用 adelay 实现开头静音填充）
    if effective_start > 0.01:
        delay_ms = int(effective_start * 1000)
        filter_parts.append(f"adelay={delay_ms}|{delay_ms}")

    # 6. 最终截断到目标总时长
    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",
        track.audio_path,
        "-filter:a",
        filter_str,
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        str(output_path),
    ]

    logger.info(
        "[multi-track] prepare track: id=%s type=%s vol=%.2f start=%.2f dur=%.2f",
        track.track_id,
        track.track_type,
        track.volume,
        effective_start,
        need_dur,
    )

    try:
        run_ffmpeg(command)
        return True
    except Exception as e:
        logger.warning("[multi-track] track prepare failed: %s, error=%s", track.track_id, e)
        return False


# ── 多轨道混音主入口 ─────────────────────────────────────────────────────────


def mix_multi_track(
    ctx: "RenderContext",
    main_audio_path: Path,
    config: MultiTrackMixConfig,
    target_duration: float,
) -> Path:
    """多轨道混音：主音频 + 多条附加轨道.

    Args:
        ctx: 渲染上下文
        main_audio_path: 主音频文件路径（原音）
        config: 多轨道混音配置
        target_duration: 目标总时长

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

    if target_duration <= 0:
        target_duration = 5.0

    # ── 安全校验：轨道数量上限 ──
    enabled_tracks = [t for t in config.tracks if t.enabled and t.audio_path]
    if len(enabled_tracks) > MAX_AUDIO_TRACKS:
        logger.warning(
            "[multi-track] too many tracks: %d > %d, truncating to max",
            len(enabled_tracks),
            MAX_AUDIO_TRACKS,
        )
        enabled_tracks = enabled_tracks[:MAX_AUDIO_TRACKS]
        # 更新 config.tracks 为截断后的列表
        config.tracks = enabled_tracks

    # ── 安全校验：所有音频路径白名单校验 ──
    # 主音频路径
    try:
        _validate_audio_path(str(main_audio_path), ctx.work_dir)
    except PathSecurityError as e:
        logger.error("[multi-track] main audio path security check failed: %s", e)
        raise

    # 各轨道音频路径
    valid_tracks = []
    for track in enabled_tracks:
        try:
            _validate_audio_path(track.audio_path, ctx.work_dir)
            valid_tracks.append(track)
        except PathSecurityError as e:
            logger.warning("[multi-track] skip track %s: path security check failed: %s", track.track_id, e)

    if len(valid_tracks) != len(enabled_tracks):
        config.tracks = valid_tracks
        logger.info("[multi-track] %d tracks passed security check", len(valid_tracks))

    # 收集所有有效轨道（已预处理好的）
    prepared_tracks: list[Path] = []

    # 主音频作为第0轨
    prepared_tracks.append(main_audio_path)

    # 预处理每条附加轨道
    for i, track in enumerate(config.tracks):
        if not track.enabled or not track.audio_path:
            continue

        track_out = ctx.work_dir / f"track_{i}_{ctx.plan_id}.aac"
        if _prepare_single_track(ctx, track, target_duration, track_out):
            prepared_tracks.append(track_out)

    # 如果只有主音频，直接返回（无需混音）
    if len(prepared_tracks) <= 1:
        import shutil

        shutil.copy2(main_audio_path, output_path)
        return output_path

    # 使用 amix 混音
    num_inputs = len(prepared_tracks)

    # 构建输入参数
    input_args: list[str] = []
    for tp in prepared_tracks:
        input_args.extend(["-i", str(tp)])

    # amix 的 duration=first 以第一个输入（主音频）时长为准
    # normalize 补偿：amix 会把每路音量除以 N，需要乘回来
    # 但如果所有轨道都同时有声，可能会爆音，所以用 master_volume 控制
    if config.normalize:
        # 经验值：不是所有轨道都同时有声，补偿系数取 N * 0.7
        compensate = num_inputs * 0.7
    else:
        compensate = 1.0

    final_volume = compensate * config.master_volume
    final_volume = min(final_volume, config.max_output_volume)

    # 构建 filter_complex
    inputs_label = "".join(f"[{i}:a]" for i in range(num_inputs))
    filter_complex = (
        f"{inputs_label}amix=inputs={num_inputs}:duration=first:dropout_transition=0[outa];"
        f"[outa]volume={final_volume:.3f}[final]"
    )

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

    logger.info(
        "[multi-track] mix %d tracks, master_vol=%.2f compensate=%.2f final_vol=%.2f",
        num_inputs,
        config.master_volume,
        compensate,
        final_volume,
    )

    try:
        run_ffmpeg(command)
    except Exception as e:
        logger.error("[multi-track] mix failed, fallback to main audio only: %s", e)
        import shutil

        shutil.copy2(main_audio_path, output_path)

    return output_path


# ── 便捷函数：从 plan.config 快速混音 ───────────────────────────────────────


def mix_audio_tracks_from_config(
    ctx: "RenderContext",
    main_audio_path: Path,
    audio_tracks_config: dict | None,
    target_duration: float,
) -> Path:
    """从 plan.config.audio_tracks 配置执行多轨道混音.

    降级策略：配置无效或混音失败时返回主音频。
    """
    config = MultiTrackMixConfig.from_config_dict(audio_tracks_config)
    if not config.has_effect:
        return main_audio_path

    return mix_multi_track(ctx, main_audio_path, config, target_duration)