#!/usr/bin/env python3 # -*- coding: utf-8 -*- """按 docs/语音播报内容设计文档_V1.0.md 生成 h5/assets/voice/*.mp3。 音色：Qwen3-TTS CustomVoice · Serena（温暖柔和年轻女声）+ 甜美亲和 instruct。 后处理：裁首尾静音 ≤150ms → 响度 −16 LUFS → MP3 48kbps CBR 单声道 22050Hz。 依赖 SoundGame 的 .venv-tts 与模型缓存（见 /home/xsl/SoundGame/env.d/tts.sh）。 """ from __future__ import annotations import json import os import subprocess import sys import tempfile from pathlib import Path os.environ.setdefault("HF_HOME", "/home/xsl/SoundGame/models/hf") os.environ.setdefault("HF_ENDPOINT", "https://hf-mirror.com") import numpy as np import soundfile as sf import torch from qwen_tts import Qwen3TTSModel ROOT = Path(__file__).resolve().parents[1] OUT_DIR = ROOT / "h5" / "assets" / "voice" STAGING = ROOT / "tools" / "voice_wav_tmp" SPEAKER = "Serena" # 甜美、亲和、语速适中偏缓；全表同一 instruct，保证步骤/姿态/倒计时同音色同气质 INSTRUCT = ( "甜美柔和的年轻女声，亲切温暖，像贴心的美肤顾问，" "语速适中偏缓，吐字清晰，带着微笑感" ) # (文件名不含扩展名, 文案) — 与设计文档第 2 节逐字一致 LINES: list[tuple[str, str]] = [ ("step1-front", "请单人正脸出镜,面部对准取景框"), ("step2-rightface", "请保持单人出镜,慢慢向左转头,露出右脸"), ("step3-leftface", "请慢慢向右转头,露出左脸"), ("hint-noface", "未检测到人脸,请正对镜头"), ("hint-near", "请靠近一点"), ("hint-far", "请远一点"), ("hint-front", "请正对屏幕"), ("hint-turn-left", "请向左转头"), ("hint-turn-right", "请向右转头"), ("hint-hold", "保持不动"), ("countdown-321", "三、二、一"), ("shot-done", "拍摄完成"), ("analyzing", "正在生成您的肌肤报告,请稍候"), ("analyze-fail", "检测失败,请调整后重新拍摄"), ] PEAK_NORM = 0.89 MAX_EDGE_SILENCE_MS = 150 SILENCE_THRESH = 0.01 def trim_edges(wav: np.ndarray, sr: int) -> np.ndarray: """裁掉过长首尾静音，保留 ≤150ms 边沿，避免提示显迟钝。""" abs_w = np.abs(wav) above = np.where(abs_w > SILENCE_THRESH)[0] if len(above) == 0: return wav keep = int(sr * MAX_EDGE_SILENCE_MS / 1000) start = max(0, int(above[0]) - keep) end = min(len(wav), int(above[-1]) + 1 + keep) return wav[start:end] def to_mp3(wav_path: Path, mp3_path: Path) -> None: """响度归一 −16 LUFS + 转 MP3 48k CBR mono 22050Hz。""" # 两遍 loudnorm：先测再应用。探测不能用 -v error，否则 JSON 统计会被压掉。 probe = subprocess.run( [ "ffmpeg", "-hide_banner", "-nostats", "-i", str(wav_path), "-af", "loudnorm=I=-16:TP=-1.5:LRA=11:print_format=json", "-f", "null", "-", ], capture_output=True, text=True, check=True, ) err = probe.stderr start = err.rfind("{") end = err.rfind("}") if start < 0 or end < 0: raise RuntimeError(f"loudnorm probe failed for {wav_path}: {err[-500:]}") stats = json.loads(err[start : end + 1]) af = ( f"loudnorm=I=-16:TP=-1.5:LRA=11:" f"measured_I={stats['input_i']}:" f"measured_TP={stats['input_tp']}:" f"measured_LRA={stats['input_lra']}:" f"measured_thresh={stats['input_thresh']}:" f"offset={stats['target_offset']}:" f"linear=true" ) subprocess.run( [ "ffmpeg", "-y", "-v", "error", "-i", str(wav_path), "-af", af, "-ar", "22050", "-ac", "1", "-c:a", "libmp3lame", "-b:a", "48k", str(mp3_path), ], check=True, ) def normalize_clip(wav: np.ndarray, sr: int) -> np.ndarray: wav = trim_edges(np.asarray(wav, dtype=np.float32).reshape(-1), sr) peak = float(np.max(np.abs(wav))) if peak < 0.05: raise RuntimeError(f"peak too low ({peak:.4f})") wav = wav / peak * PEAK_NORM dur = len(wav) / sr if dur < 0.35: wav = np.concatenate([wav, np.zeros(int((0.35 - dur) * sr), dtype=np.float32)]) return wav def write_mp3(name: str, wav: np.ndarray, sr: int) -> None: wav_path = STAGING / f"{name}.wav" mp3_path = OUT_DIR / f"{name}.mp3" sf.write(str(wav_path), wav, sr) to_mp3(wav_path, mp3_path) dur = len(wav) / sr print(f"[ok] {name}.mp3 {dur:.2f}s {mp3_path.stat().st_size} bytes", flush=True) def gen_countdown(model) -> tuple[np.ndarray, int]: """倒计时单独按字生成再拼接，避免「三、二、一」整句被读成播报腔、与提示音气质漂移。 与提示音共用同一 speaker + instruct；句间留约 0.35s，总长约 2.5s。 """ digits = ["三", "二", "一"] # 同批带上一条短提示作音色锚点，丢弃锚点音频 texts = ["保持不动", *digits] wavs, sr = model.generate_custom_voice( text=texts, language=["Chinese"] * len(texts), speaker=[SPEAKER] * len(texts), instruct=[INSTRUCT] * len(texts), max_new_tokens=2048, ) gap = np.zeros(int(0.35 * sr), dtype=np.float32) parts: list[np.ndarray] = [] for i, digit in enumerate(digits): clip = normalize_clip(wavs[i + 1], sr) parts.append(clip) if i < len(digits) - 1: parts.append(gap) print(f" countdown digit {digit}: {len(clip) / sr:.2f}s", flush=True) return np.concatenate(parts), sr def main() -> int: import argparse ap = argparse.ArgumentParser() ap.add_argument( "--only", default="", help="逗号分隔的 id，只生成这些；空则全量", ) args = ap.parse_args() only = {x.strip() for x in args.only.split(",") if x.strip()} todo = [(n, t) for n, t in LINES if not only or n in only] if only and not todo: print(f"unknown --only ids: {sorted(only)}", flush=True) return 1 OUT_DIR.mkdir(parents=True, exist_ok=True) STAGING.mkdir(parents=True, exist_ok=True) print("loading Qwen3-TTS CustomVoice ...", flush=True) model = Qwen3TTSModel.from_pretrained( "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice", device_map="cuda:0", dtype=torch.bfloat16, ) # 倒计时走专用拼接路径，其余整句生成 normal = [(n, t) for n, t in todo if n != "countdown-321"] do_countdown = any(n == "countdown-321" for n, _ in todo) if normal: names = [n for n, _ in normal] texts = [t for _, t in normal] print(f"generating {len(names)} lines with speaker={SPEAKER} ...", flush=True) wavs, sr = model.generate_custom_voice( text=texts, language=["Chinese"] * len(names), speaker=[SPEAKER] * len(names), instruct=[INSTRUCT] * len(names), max_new_tokens=2048, ) for name, wav in zip(names, wavs): try: write_mp3(name, normalize_clip(wav, sr), sr) except RuntimeError as e: print(f"[FAIL] {name}: {e}", flush=True) return 1 if do_countdown: print(f"generating countdown-321 (digit stitch, speaker={SPEAKER}) ...", flush=True) wav, sr = gen_countdown(model) # 整段再峰值归一一次，保证与提示音响度一致 peak = float(np.max(np.abs(wav))) wav = wav / peak * PEAK_NORM write_mp3("countdown-321", wav, sr) # 清单始终覆盖文档全量已落地文件（缺文件则跳过写入名） existing = [ n for n, _ in LINES if (OUT_DIR / f"{n}.mp3").is_file() ] manifest = {"files": existing} (OUT_DIR / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) print(f"wrote {OUT_DIR / 'manifest.json'} ({len(existing)} files)", flush=True) return 0 if __name__ == "__main__": sys.exit(main())