"""多轨道混音引擎 — 支持多路音频独立音量调节与混合. 基于 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)