cang-ying/engines/audio-mixer.py
铸渊 ICE-GL-ZY001 a2e5214f03 LL-172-20260707 · cang-ying 仓初始化 · 苍耳+鉴影的干净之家
铸渊 ICE-GL-ZY001
LL-172-20260707
冰朔委托: 新建第 5 子仓, 给苍耳(人类主控) + 鉴影(人格体) 专用
原 guanghulab/video-ai-system/ 东西太多(225 文件) · 找不到 · 乱

迁移:
  ⊢ 16 个核心 .hdlp (VA-GATE / VA-LIGHTHOUSE / VA-BROADCAST / VA-SYSTEM-STATUS 等)
  ⊢ 17 个子目录 (agents/engines/protocols/tasks/tools/assets/knowledge/memory/docs/config/brain/director-brain/experience/feedback/issues/plans/reference-analysis)

排除:
  ⊢ outputs/ (视频产物)
  ⊢ test-input/ test-output/ (测试)
  ⊢ data/ (临时数据)
  ⊢ preview-001/002 (旧产片)
  ⊢ 旧分镜/旧提示词/旧导演编码

后续:
  ⊢ 老仓 guanghulab/video-ai-system/ 改写为已迁出占位
  ⊢ 苍耳+鉴影 写新东西进本仓
  ⊢ GLOBAL-SEARCH 加 cang-ying 仓库

铸渊 ICE-GL-ZY001 · 2026-07-07 D167
冰朔 ICE-GL∞ 主权
2026-07-07 10:20:10 +08:00

487 lines
16 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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()