"""BGM 混音纯逻辑模块. 所有函数均为纯函数,不调用 FFmpeg、不操作文件。 便于单元测试,也方便被其他模块复用。 """ from __future__ import annotations from dataclasses import dataclass from typing import Optional @dataclass class BGMPureConfig: """BGM 混音配置(纯数据类).""" volume: float = 0.3 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 # ── 循环判断与计算 ───────────────────────────────────────────────────────────── def should_loop_bgm( bgm_duration: float, target_duration: float, loop_enabled: bool = True, ) -> bool: """判断是否需要循环 BGM. 当 BGM 时长小于目标时长的 90% 时才循环, 避免 BGM 只差一点点就铺满还要循环一次的情况。 Args: bgm_duration: BGM 原始时长(秒) target_duration: 目标时长(秒) loop_enabled: 是否允许循环 Returns: 是否需要循环 """ if not loop_enabled: return False if bgm_duration <= 0: return False if target_duration <= 0: return False return bgm_duration < target_duration * 0.9 def calculate_loop_count(bgm_duration: float, target_duration: float) -> int: """计算需要循环的次数. 多算 2 次作为余量,避免末尾因为精度问题不够长。 Args: bgm_duration: BGM 原始时长(秒) target_duration: 目标时长(秒) Returns: 循环次数,至少 1 """ if bgm_duration <= 0: return 1 if target_duration <= 0: return 1 if bgm_duration >= target_duration: return 1 return max(1, int(target_duration / bgm_duration) + 2) # ── BGM 预处理滤镜链构建 ────────────────────────────────────────────────────── def build_bgm_filter_chain( bgm_duration: float, target_duration: float, volume: float = 0.3, fade_in: float = 0.0, fade_out: float = 0.0, loop_enabled: bool = True, ) -> str: """构建 BGM 预处理滤镜链. 处理顺序:循环 → 音量 → 淡入 → 淡出 → 截断 → 重置时间戳 Args: bgm_duration: BGM 原始时长(秒) target_duration: 目标时长(秒) volume: 音量 0.0~1.0 fade_in: 淡入时长(秒) fade_out: 淡出时长(秒) loop_enabled: 是否允许循环 Returns: FFmpeg filter_complex 字符串(逗号分隔) """ # 兜底:目标时长不能为 0 或负数 safe_target = max(5.0, target_duration) if target_duration <= 0 else target_duration filter_parts: list[str] = [] # 1. 循环 needs_loop = should_loop_bgm(bgm_duration, safe_target, loop_enabled) if needs_loop: loop_count = calculate_loop_count(bgm_duration, safe_target) filter_parts.append(f"aloop=loop={loop_count}:size=0") # 2. 音量调节(钳制到 0~1) safe_volume = max(0.0, min(1.0, volume)) if abs(safe_volume - 1.0) > 0.001: filter_parts.append(f"volume={safe_volume:.3f}") # 3. 淡入 if fade_in > 0: filter_parts.append(f"afade=t=in:st=0:d={fade_in:.3f}") # 4. 淡出(从 target_duration - fade_out 开始) if fade_out > 0 and safe_target > fade_out: fade_start = safe_target - fade_out filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={fade_out:.3f}") # 5. 截断到目标时长 filter_parts.append(f"atrim=0:{safe_target:.3f}") # 6. 重置时间戳 filter_parts.append("asetpts=N/SR/TB") return ",".join(filter_parts) # ── 混音滤镜构建 ────────────────────────────────────────────────────────────── def calculate_sidechain_ratio(sidechain_ratio: float) -> float: """计算 sidechain 压缩比. sidechain_ratio 表示闪避时 BGM 音量降低比例(0~1), 映射到 FFmpeg sidechaincompress 的 ratio 参数(2:1 ~ 10:1)。 ratio = 1 / (1 - sidechain_ratio) Args: sidechain_ratio: 闪避比例 0.0~1.0 Returns: FFmpeg ratio 值(2.0 ~ 10.0) """ if sidechain_ratio <= 0: return 2.0 if sidechain_ratio >= 1.0: return 10.0 raw_ratio = 1.0 / (1.0 - sidechain_ratio) return max(2.0, min(10.0, raw_ratio)) def build_simple_mix_filter() -> str: """构建普通 amix 混音滤镜. 两路输入:[0:a] 主音频,[1:a] BGM 主音频权重 1.0,BGM 已在预处理阶段调好音量。 amix 会自动归一化,用 volume=2 补偿衰减。 Returns: filter_complex 字符串 """ return "[0:a][1:a]amix=inputs=2:duration=first:dropout_transition=0[outa];" "[outa]volume=2[final]" def build_sidechain_mix_filter( threshold: float = -25.0, ratio: float = 0.3, attack: float = 0.02, release: float = 0.5, ) -> str: """构建 sidechain 人声闪避混音滤镜. 流程: 1. BGM[1:a] 经过 sidechaincompress,用主音频[0:a]做触发 2. 主音频 + 压缩后的 BGM amix 混音 3. volume=1.5 轻微补偿 Args: threshold: 触发阈值(dB) ratio: 闪避比例 0.0~1.0(会被转换为 FFmpeg ratio) attack: 攻击时间(秒) release: 释放时间(秒) Returns: filter_complex 字符串 """ ffmpeg_ratio = calculate_sidechain_ratio(ratio) return ( f"[1:a][0:a]sidechaincompress=" f"threshold={threshold}dB:" f"ratio={ffmpeg_ratio:.1f}:" f"attack={attack:.3f}:" f"release={release:.3f}:" f"knee=6[bgm_comp];" f"[0:a][bgm_comp]amix=inputs=2:duration=first:dropout_transition=0[outa];" f"[outa]volume=1.5[final]" ) # ── 配置验证与规范化 ────────────────────────────────────────────────────────── def normalize_bgm_config(config: dict) -> dict: """规范化 BGM 配置字典. 将各种类型的输入值转换为正确的类型, 并进行边界钳制。 Args: config: 原始配置字典 Returns: 规范化后的配置字典 """ result: dict = {} # volume: 0.0 ~ 1.0 result["volume"] = max(0.0, min(1.0, float(config.get("volume", 0.3)))) # fade_in: >= 0 result["fade_in"] = max(0.0, float(config.get("fade_in", 0.0))) # fade_out: >= 0 result["fade_out"] = max(0.0, float(config.get("fade_out", 0.0))) # loop_enabled: bool result["loop_enabled"] = bool(config.get("loop_enabled", True)) # sidechain_enabled: bool result["sidechain_enabled"] = bool(config.get("sidechain_enabled", False)) # sidechain_ratio: 0.0 ~ 1.0 result["sidechain_ratio"] = max(0.0, min(1.0, float(config.get("sidechain_ratio", 0.3)))) # sidechain_attack: > 0 result["sidechain_attack"] = max(0.001, float(config.get("sidechain_attack", 0.02))) # sidechain_release: > 0 result["sidechain_release"] = max(0.01, float(config.get("sidechain_release", 0.5))) # sidechain_threshold: dB result["sidechain_threshold"] = float(config.get("sidechain_threshold", -25.0)) return result def validate_bgm_config(config: dict) -> tuple[bool, list[str]]: """验证 BGM 配置是否合法. Args: config: 配置字典 Returns: (是否合法, 错误信息列表) """ errors: list[str] = [] volume = config.get("volume", 0.3) if not isinstance(volume, (int, float)): errors.append("volume 必须是数字") elif volume < 0 or volume > 1: errors.append("volume 必须在 0~1 之间") fade_in = config.get("fade_in", 0) if not isinstance(fade_in, (int, float)): errors.append("fade_in 必须是数字") elif fade_in < 0: errors.append("fade_in 不能为负数") fade_out = config.get("fade_out", 0) if not isinstance(fade_out, (int, float)): errors.append("fade_out 必须是数字") elif fade_out < 0: errors.append("fade_out 不能为负数") sidechain_ratio = config.get("sidechain_ratio", 0.3) if not isinstance(sidechain_ratio, (int, float)): errors.append("sidechain_ratio 必须是数字") elif sidechain_ratio < 0 or sidechain_ratio > 1: errors.append("sidechain_ratio 必须在 0~1 之间") return (len(errors) == 0, errors) # ── 时长相关工具 ────────────────────────────────────────────────────────────── def calculate_fade_out_start( target_duration: float, fade_out: float, ) -> Optional[float]: """计算淡出开始时间. 如果淡出时长大于等于目标时长,返回 None(不做淡出)。 Args: target_duration: 目标时长(秒) fade_out: 淡出时长(秒) Returns: 淡出开始时间(秒),如果不需要淡出返回 None """ if fade_out <= 0: return None if target_duration <= 0: return None if fade_out >= target_duration: return None return target_duration - fade_out def estimate_bgm_processing_duration( bgm_duration: float, target_duration: float, loop_enabled: bool = True, ) -> float: """估算 BGM 预处理后的实际输出时长. 正常情况下应该等于 target_duration, 但在某些边界情况下可能不同。 Args: bgm_duration: BGM 原始时长 target_duration: 目标时长 loop_enabled: 是否允许循环 Returns: 预估输出时长(秒) """ if target_duration <= 0: return 5.0 # 兜底时长 # 不需要循环的情况:如果 BGM 够长,截断到 target_duration if not loop_enabled and bgm_duration >= target_duration: return target_duration # 需要循环或 BGM 太短:截断到 target_duration return target_duration