cang-ying/engines/audio-mixer.py

487 lines
16 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AUDIO-MIXER
混音器 配音BGM音效原视频音轨混音支持对白时自动压低BGM
功能:
1. 输入多轨音频 (对白/BGM/音效/原视频音轨)
2. 自动检测对白时段压低BGM音量 (ducking)
3. 混音输出
4. 支持批量处理
依赖:
ffmpeg (需要系统安装)
pip install numpy # 用于音频分析
用法:
python audio-mixer.py --dialogue dialogue.mp3 --bgm bgm.mp3 --output mix.mp3
python audio-mixer.py --config mix-config.json
"""
import os
import sys
import json
import argparse
import subprocess
from pathlib import Path
from datetime import datetime
PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT / "engines"))
class AudioMixer:
"""混音器"""
def __init__(self, ffmpeg_path="ffmpeg"):
self.ffmpeg_path = ffmpeg_path
self._check_ffmpeg()
def _check_ffmpeg(self):
"""检查 FFmpeg 是否可用"""
try:
result = subprocess.run(
[self.ffmpeg_path, "-version"],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
version_line = result.stdout.split("\n")[0]
print(f"✅ FFmpeg 可用: {version_line}")
return True
except Exception as e:
pass
print(f"❌ FFmpeg 不可用: {self.ffmpeg_path}")
print(f" 安装方法: brew install ffmpeg (macOS) 或 apt install ffmpeg (Ubuntu)")
return False
def mix_audio(self, dialogue=None, bgm=None, sfx=None, original=None,
output_path=None, ducking_threshold=-20, ducking_level=-10):
"""
混音
参数:
dialogue: 对白音轨路径
bgm: BGM 音轨路径
sfx: 音效音轨路径 (可选支持多个传入列表)
original: 原视频音轨路径 (可选)
output_path: 输出路径
ducking_threshold: 对白检测阈值 (dB默认 -20dB)
ducking_level: BGM 压低量 (dB默认 -10dB = 压低到原来的 1/10)
返回:
{
"success": bool,
"output_path": str,
"tracks_used": list,
"warnings": list
}
"""
print(f"\n🎵 混音")
tracks = []
warnings = []
# 检查输入文件
if dialogue and Path(dialogue).exists():
tracks.append(("dialogue", dialogue))
print(f" 对白: {Path(dialogue).name}")
elif dialogue:
warnings.append(f"对白文件不存在: {dialogue}")
if bgm and Path(bgm).exists():
tracks.append(("bgm", bgm))
print(f" BGM: {Path(bgm).name}")
elif bgm:
warnings.append(f"BGM 文件不存在: {bgm}")
if sfx:
if isinstance(sfx, str):
sfx = [sfx]
for s in sfx:
if Path(s).exists():
tracks.append(("sfx", s))
print(f" 音效: {Path(s).name}")
else:
warnings.append(f"音效文件不存在: {s}")
if original and Path(original).exists():
tracks.append(("original", original))
print(f" 原视频音轨: {Path(original).name}")
elif original:
warnings.append(f"原视频音轨不存在: {original}")
if len(tracks) == 0:
return {"success": False, "error": "没有可用的音轨"}
# 确定输出路径
if output_path is None:
# 默认输出到对白文件同目录
if dialogue:
output_path = Path(dialogue).parent / f"{Path(dialogue).stem}_mixed.mp3"
else:
output_path = PROJECT_ROOT / "outputs" / "mixed_audio.mp3"
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
# 生成 FFmpeg 命令
if len(tracks) == 1:
# 只有一条音轨,直接复制
print(f"\n ⚠️ 只有一条音轨,直接复制")
import shutil
shutil.copy2(tracks[0][1], output_path)
return {
"success": True,
"output_path": str(output_path),
"tracks_used": [t[0] for t in tracks],
"warnings": warnings
}
# 多条音轨,需要混音
print(f"\n 🔧 生成 FFmpeg 命令...")
if bgm and dialogue:
# 有 BGM 和对白 → 使用 ducking
print(f" 启用自动压低 BGM (ducking)")
print(f" 对白检测阈值: {ducking_threshold}dB")
print(f" BGM 压低量: {ducking_level}dB")
result = self._mix_with_ducking(
dialogue, bgm, sfx, original, output_path,
ducking_threshold, ducking_level
)
else:
# 无 BGM 或 无对白 → 直接混音
print(f" 直接混音 (无 ducking)")
result = self._mix_simple(
[t[1] for t in tracks],
output_path
)
return result
def _mix_with_ducking(self, dialogue, bgm, sfx, original, output_path,
ducking_threshold, ducking_level):
"""带侧链压缩(ducking) + loudnorm 响度标准化的混音"""
inputs = []
filter_complex = []
input_idx = 0
track_map = {} # track_name → filter_label
# 输入文件
if dialogue:
inputs.extend(["-i", dialogue])
track_map["dialogue"] = input_idx
input_idx += 1
if bgm:
inputs.extend(["-i", bgm])
track_map["bgm"] = input_idx
input_idx += 1
if sfx:
if isinstance(sfx, str):
sfx = [sfx]
for i, s in enumerate(sfx):
inputs.extend(["-i", s])
track_map[f"sfx{i}"] = input_idx
input_idx += 1
if original:
inputs.extend(["-i", original])
track_map["original"] = input_idx
input_idx += 1
# 侧链压缩: 对白作为 sidechainBGM 被压缩
# sidechaincompress 参数: threshold 阈值, ratio 压缩比, attack 起音, release 释音
if "dialogue" in track_map and "bgm" in track_map:
d_idx = track_map["dialogue"]
b_idx = track_map["bgm"]
# 对白音轨通过
filter_complex.append(f"[{d_idx}:a]aformat=sample_fmts=fltp:channel_layouts=stereo[d_pre]")
filter_complex.append(f"[{b_idx}:a]aformat=sample_fmts=fltp:channel_layouts=stereo[b_pre]")
# 侧链压缩: BGM 被对白压缩
filter_complex.append(
f"[b_pre][d_pre]sidechaincompress="
f"threshold={ducking_threshold/20:.3f}:ratio=4:"
f"attack=25:release=200:level_sc={ducking_level/20:.3f}[bgm_comp]"
)
filter_complex.append(f"[d_pre]volume=1[dialogue_out]")
bgm_label = "bgm_comp"
else:
bgm_label = None
# 音效/原视频音轨
mix_inputs = []
if "dialogue" in track_map:
mix_inputs.append("[dialogue_out]" if bgm_label else f"[{track_map['dialogue']}:a]")
if bgm_label:
mix_inputs.append(f"[{bgm_label}]")
elif "bgm" in track_map:
mix_inputs.append(f"[{track_map['bgm']}:a]")
for key in track_map:
if key.startswith("sfx"):
filter_complex.append(f"[{track_map[key]}:a]volume=0.8[s_{key}]")
mix_inputs.append(f"[s_{key}]")
if "original" in track_map:
filter_complex.append(f"[{track_map['original']}:a]volume=0.4[orig_out]")
mix_inputs.append("[orig_out]")
# 混音
n_inputs = len(mix_inputs)
amix_inputs = "".join(mix_inputs)
filter_complex.append(
f"{amix_inputs}amix=inputs={n_inputs}:duration=first:dropout_transition=3,"
# 响度标准化: EBU R128, 目标 -14 LUFS (短视频平台标准)
f"loudnorm=I=-14:LRA=11:TP=-1.5,"
# 峰值限幅保护
f"alimiter=limit=-1.0dB:attack=5:release=50[aout]"
)
cmd = [
self.ffmpeg_path, "-y",
*inputs,
"-filter_complex", ";".join(filter_complex),
"-map", "[aout]",
"-codec:a", "libmp3lame",
"-b:a", "192k",
str(output_path)
]
print(f" 📤 执行 FFmpeg...")
print(f" 命令: ffmpeg {' '.join(cmd[1:5])}...")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300
)
if result.returncode == 0:
print(f" ✅ 混音完成: {output_path.name}")
return {
"success": True,
"output_path": str(output_path),
"tracks_used": list(track_map.keys()),
"warnings": [],
"method": "sidechain_compress+loudnorm"
}
else:
print(f" ❌ FFmpeg 失败: {result.stderr[-500:]}")
return {
"success": False,
"error": result.stderr[-500:],
"returncode": result.returncode
}
except subprocess.TimeoutExpired:
print(f" ❌ FFmpeg 超时 (5分钟)")
return {"success": False, "error": "Timeout"}
except Exception as e:
print(f" ❌ 执行失败: {e}")
return {"success": False, "error": str(e)}
def _mix_simple(self, input_files, output_path):
"""简单混音 (无 ducking)"""
inputs = []
for f in input_files:
inputs.extend(["-i", f])
# amix 滤镜混音
filter_complex = f"amix=inputs={len(input_files)}:duration=first:dropout_transition=2"
cmd = [
self.ffmpeg_path,
"-y",
*inputs,
"-filter_complex", filter_complex,
"-codec:a", "libmp3lame",
"-b:a", "192k",
str(output_path)
]
print(f" 📤 执行 FFmpeg (简单混音)...")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300
)
if result.returncode == 0:
print(f" ✅ 混音完成: {output_path.name}")
return {
"success": True,
"output_path": str(output_path),
"method": "simple_mix"
}
else:
print(f" ❌ FFmpeg 失败: {result.stderr[-500:]}")
return {"success": False, "error": result.stderr[-500:]}
except Exception as e:
print(f" ❌ 执行失败: {e}")
return {"success": False, "error": str(e)}
def extract_audio_from_video(self, video_path, output_path=None):
"""
从视频中提取音轨
"""
if output_path is None:
output_path = Path(video_path).parent / f"{Path(video_path).stem}.mp3"
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
cmd = [
self.ffmpeg_path,
"-y",
"-i", video_path,
"-vn", # 不要视频
"-codec:a", "libmp3lame",
"-b:a", "192k",
str(output_path)
]
print(f"\n📤 从视频提取音轨: {Path(video_path).name}")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300
)
if result.returncode == 0:
print(f" ✅ 提取完成: {output_path.name}")
return {"success": True, "output_path": str(output_path)}
else:
print(f" ❌ 提取失败: {result.stderr[-500:]}")
return {"success": False, "error": result.stderr[-500:]}
except Exception as e:
print(f" ❌ 执行失败: {e}")
return {"success": False, "error": str(e)}
def batch_mix(self, config_file):
"""
批量混音 (从配置文件)
config_file JSON 格式:
{
"output_dir": "./outputs/mixed/",
"tracks": [
{
"dialogue": "dialogue/ep01-shot01.mp3",
"bgm": "bgm/ep01-theme.mp3",
"sfx": ["sfx/door-open.mp3"],
"output": "mixed/ep01-shot01.mp3"
},
...
]
}
"""
config_file = Path(config_file)
if not config_file.exists():
return {"success": False, "error": f"配置文件不存在: {config_file}"}
with open(config_file, "r", encoding="utf-8") as f:
config = json.load(f)
output_dir = Path(config.get("output_dir", "./outputs/mixed/"))
output_dir.mkdir(parents=True, exist_ok=True)
tracks = config.get("tracks", [])
print(f"\n📦 批量混音: {len(tracks)} 个任务")
results = []
for i, track_config in enumerate(tracks):
print(f"\n 进度: [{i+1}/{len(tracks)}]")
result = self.mix_audio(
dialogue=track_config.get("dialogue"),
bgm=track_config.get("bgm"),
sfx=track_config.get("sfx"),
original=track_config.get("original"),
output_path=track_config.get("output", str(output_dir / f"mixed-{i+1:03d}.mp3"))
)
results.append(result)
# 统计
success_count = sum(1 for r in results if r.get("success"))
print(f"\n✅ 批量完成: {success_count}/{len(results)} 成功")
# 保存报告
report_path = output_dir / "mix_report.json"
with open(report_path, "w", encoding="utf-8") as f:
json.dump({
"total": len(results),
"success": success_count,
"results": results,
"generated_at": datetime.now().isoformat()
}, f, ensure_ascii=False, indent=2)
print(f" 报告已保存: {report_path}")
return results
def main():
parser = argparse.ArgumentParser(description="AUDIO-MIXER")
parser.add_argument("--dialogue", type=str, help="对白音轨路径")
parser.add_argument("--bgm", type=str, help="BGM 音轨路径")
parser.add_argument("--sfx", type=str, nargs="+", help="音效音轨路径 (多个)")
parser.add_argument("--original", type=str, help="原视频音轨路径")
parser.add_argument("--output", type=str, help="输出路径")
parser.add_argument("--config", type=str, help="批量混音配置文件")
parser.add_argument("--ducking-threshold", type=float, default=-20, help="对白检测阈值 (dB)")
parser.add_argument("--ducking-level", type=float, default=-10, help="BGM 压低量 (dB)")
parser.add_argument("--extract-from-video", type=str, help="从视频提取音轨")
parser.add_argument("--ffmpeg-path", type=str, default="ffmpeg", help="FFmpeg 路径")
args = parser.parse_args()
mixer = AudioMixer(ffmpeg_path=args.ffmpeg_path)
if args.extract_from_video:
# 提取音轨模式
result = mixer.extract_audio_from_video(args.extract_from_video, args.output)
sys.exit(0 if result["success"] else 1)
if args.config:
# 批量模式
results = mixer.batch_mix(args.config)
sys.exit(0 if all(r.get("success") for r in results) else 1)
if not args.dialogue and not args.bgm and not args.original:
parser.print_help()
sys.exit(1)
# 单文件模式
result = mixer.mix_audio(
dialogue=args.dialogue,
bgm=args.bgm,
sfx=args.sfx,
original=args.original,
output_path=args.output,
ducking_threshold=args.ducking_threshold,
ducking_level=args.ducking_level
)
if result["success"]:
print(f"\n✅ 成功: {result['output_path']}")
sys.exit(0)
else:
print(f"\n❌ 失败: {result.get('error', 'Unknown error')}")
sys.exit(1)
if __name__ == "__main__":
main()