163 lines
6.7 KiB
Python
163 lines
6.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""爆音修复（R6）：处理 scan_clicks.py 标记的 sfx/bgm 资产。
|
||
|
||
修复策略（只修 sfx/bgm，voice 不动）：
|
||
- one-shot sfx：去直流（逐声道减均值）→ 首尾 8ms cos² 淡入淡出 → 峰值归一 -3dBFS
|
||
- loop（bgm/ 与 sfx/amb_*）：只去直流（首尾交叉淡化工艺保证边界连续，加淡入淡出会破坏循环）
|
||
- rpg one-shot 优先从 staging（tools/gen_rpg_sfx_wav/ 原始 SA3 wav / gen_rpg_syn.py 重出的
|
||
/tmp/rpg_syn）重做，避免二次有损编码；其余从现有 ogg 解码修复。
|
||
处理前原件备份到 tools/asset_fix_backup/。处理前后指标记录到 /tmp/fix_report.jsonl。
|
||
"""
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
|
||
import numpy as np
|
||
import soundfile as sf
|
||
|
||
ROOT = "/home/xsl/SoundGame"
|
||
WEB = os.path.join(ROOT, "web")
|
||
BACKUP = os.path.join(ROOT, "tools", "asset_fix_backup")
|
||
FADE_MS = 8
|
||
PEAK_TARGET = 10 ** (-3.0 / 20) # -3dBFS
|
||
|
||
RPG_SA3 = os.path.join(ROOT, "tools", "gen_rpg_sfx_wav")
|
||
RPG_SYN = "/tmp/rpg_syn"
|
||
# rpg one-shot 的定长裁剪参数（同 R4 工艺）
|
||
RPG_TRIM = {
|
||
"tel_wolf": 1.0, "tel_bandit": 1.0, "tel_spider": 1.0, "tel_elite": 1.0,
|
||
"tel_boss1": 1.0, "tel_croc": 1.0, "tel_shaman": 1.0, "tel_puppet": 1.0,
|
||
"tel_boss2": 1.0, "atk_swing": 0.5, "atk_hit": 0.4, "parry": 0.5,
|
||
"skill_slash": 0.8, "enemy_die": 1.0, "potion": 1.2, "chest": 1.5,
|
||
"drink_water": 1.5, "rest": 2.0, "door_heavy": 3.0,
|
||
}
|
||
|
||
|
||
def is_loop(rel):
|
||
return "/bgm/" in "/" + rel or os.path.basename(rel).startswith("amb_")
|
||
|
||
|
||
def metrics(y, sr):
|
||
mono = y.mean(axis=1) if y.ndim > 1 else y
|
||
n5 = max(1, int(sr * 5 / 1000))
|
||
peak = float(np.max(np.abs(mono)))
|
||
return {
|
||
"peak_db": round(20 * np.log10(max(peak, 1e-9)), 2),
|
||
"dc": round(float(np.mean(mono)), 5),
|
||
"head5ms_db": round(20 * np.log10(max(float(np.max(np.abs(mono[:n5]))), 1e-9)), 2),
|
||
"tail5ms_db": round(20 * np.log10(max(float(np.max(np.abs(mono[-n5:]))), 1e-9)), 2),
|
||
}
|
||
|
||
|
||
def fix_array(y, sr, loop):
|
||
y = y.astype(np.float64)
|
||
if not loop:
|
||
n = int(sr * FADE_MS / 1000)
|
||
ramp = np.sin(np.linspace(0, np.pi / 2, n)) ** 2
|
||
y[:n] *= ramp[:, None] if y.ndim > 1 else ramp
|
||
y[-n:] *= ramp[::-1][:, None] if y.ndim > 1 else ramp[::-1]
|
||
# 峰值归一 -3dBFS
|
||
peak = np.max(np.abs(y))
|
||
if peak > 0:
|
||
y = y / peak * PEAK_TARGET
|
||
# 去直流放最后（淡入淡出/归一会重新引入微小偏置；常数平移不影响循环连续性）
|
||
y = y - y.mean(axis=0)
|
||
return y.astype(np.float32)
|
||
|
||
|
||
def encode(wav_path, base):
|
||
"""编码 ogg+m4a；对 ogg 做解码回读真峰值校验——MDCT 重建对高密度/近 Nyquist
|
||
内容会过冲数 dB（R6 爆音根因），过冲则按实测降增益重编（最多 3 次迭代，
|
||
首轮附带 18kHz 低通消振铃）。返回解码后实测峰值 dB。"""
|
||
subprocess.run(["ffmpeg", "-y", "-v", "error", "-i", wav_path, "-ar", "44100",
|
||
"-c:a", "aac", "-b:a", "128k", base + ".m4a"], check=True)
|
||
cur = wav_path
|
||
peak_db = None
|
||
for attempt in range(3):
|
||
subprocess.run(["ffmpeg", "-y", "-v", "error", "-i", cur, "-ar", "44100",
|
||
"-c:a", "libvorbis", "-q:a", "3", base + ".ogg"], check=True)
|
||
dec, _sr = sf.read(base + ".ogg", always_2d=True)
|
||
peak_db = 20 * np.log10(max(float(np.max(np.abs(dec))), 1e-9))
|
||
if peak_db <= -2.7:
|
||
break
|
||
gain_db = -3.2 - peak_db
|
||
nxt = wav_path + f".tp{attempt}.wav"
|
||
af = ("lowpass=f=18000," if attempt == 0 else "") + f"volume={gain_db:.2f}dB"
|
||
subprocess.run(["ffmpeg", "-y", "-v", "error", "-i", cur,
|
||
"-af", af, "-c:a", "pcm_f32le", nxt], check=True)
|
||
cur = nxt
|
||
return peak_db
|
||
|
||
|
||
def main():
|
||
flagged = []
|
||
for line in open("/tmp/scan_all.jsonl"):
|
||
r = json.loads(line)
|
||
if "/voice/" in r["path"]:
|
||
continue # 语音响度策略不动
|
||
if r["hot_peak"] or r["dc_bad"] or ((r["click_head"] or r["click_tail"]) and not r["loop"]):
|
||
flagged.append(r["path"])
|
||
print(f"flagged sfx/bgm: {len(flagged)}")
|
||
|
||
report = []
|
||
for rel in flagged:
|
||
game = rel.split("/")[0]
|
||
name = os.path.basename(rel)[:-4]
|
||
ogg = os.path.join(WEB, rel)
|
||
base = ogg[:-4]
|
||
loop = is_loop(rel)
|
||
|
||
# 选源：rpg one-shot 用 staging 重做；其余用现有 ogg
|
||
src = None
|
||
trim = None
|
||
if game == "rpg" and not loop:
|
||
if name in RPG_TRIM and os.path.exists(os.path.join(RPG_SA3, name + ".wav")):
|
||
src, trim = os.path.join(RPG_SA3, name + ".wav"), RPG_TRIM[name]
|
||
elif os.path.exists(os.path.join(RPG_SYN, name + ".wav")):
|
||
src = os.path.join(RPG_SYN, name + ".wav")
|
||
if src:
|
||
y, sr = sf.read(src, always_2d=True)
|
||
y = y.astype(np.float64)
|
||
if trim: # 去头静音 + 定长（ffmpeg silenceremove 的 python 等价：阈值 -45dB）
|
||
mono = np.abs(y.mean(axis=1))
|
||
win = int(sr * 0.005)
|
||
env = np.convolve(mono, np.ones(win) / win, mode="same")
|
||
start = int(np.argmax(env > 10 ** (-45 / 20)))
|
||
y = y[start:start + int(sr * trim)]
|
||
else:
|
||
# 有备份读备份（初代无损源），避免多次有损转码
|
||
bak = os.path.join(BACKUP, rel)
|
||
y, sr = sf.read(bak if os.path.exists(bak) else ogg, always_2d=True)
|
||
|
||
before = metrics(y, sr)
|
||
# 备份原件（ogg+m4a）
|
||
for ext in (".ogg", ".m4a"):
|
||
dst = os.path.join(BACKUP, rel[:-4] + ext)
|
||
os.makedirs(os.path.dirname(dst), exist_ok=True)
|
||
if os.path.exists(base + ext):
|
||
shutil.copy2(base + ext, dst)
|
||
|
||
y2 = fix_array(y, sr, loop)
|
||
after = metrics(y2, sr)
|
||
tmp = f"/tmp/fix_{game}_{name}.wav"
|
||
sf.write(tmp, y2, sr)
|
||
dec_peak = encode(tmp, base) # ogg 解码回读真峰值（含过冲迭代修正）
|
||
after["decoded_peak_db"] = round(dec_peak, 2)
|
||
report.append({"path": rel, "loop": loop, "src": "staging" if src else "ogg",
|
||
"before": before, "after": after})
|
||
print(f"[fix] {rel} ({'loop' if loop else 'shot'}, src={'staging' if src else 'ogg'}) "
|
||
f"peak {before['peak_db']}→{after['peak_db']} (decoded {round(dec_peak, 2)}) "
|
||
f"dc {before['dc']}→{after['dc']} head {before['head5ms_db']}→{after['head5ms_db']}")
|
||
|
||
with open("/tmp/fix_report.jsonl", "w") as f:
|
||
for r in report:
|
||
f.write(json.dumps(r, ensure_ascii=False) + "\n")
|
||
print(f"fixed {len(report)} files, backup in {BACKUP}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|