blind/audio_1/scripts/01_gen_tts_voxcpm2.py

#!/usr/bin/env python3
"""
步骤 1：TTS v1 — VoxCPM2 批量生成
输出目录：audio_1/tts/v1_voxcpm2/

使用方法：
  # 确认 VoxCPM2 服务已启动
  python audio_1/scripts/01_gen_tts_voxcpm2.py
  python audio_1/scripts/01_gen_tts_voxcpm2.py --dry-run
  python audio_1/scripts/01_gen_tts_voxcpm2.py --role linxia
"""
import os, sys, glob, base64, json, tempfile, time, subprocess, argparse
import requests

PROJECT_DIR = "/home/xsl/blind"
VOICE_DIR   = f"{PROJECT_DIR}/voice"
OUT_DIR     = f"{PROJECT_DIR}/audio_1/tts/v1_voxcpm2"
HOST = "127.0.0.1"
PORT = "8000"
BASE_URL = f"http://{HOST}:{PORT}"

os.makedirs(OUT_DIR, exist_ok=True)

sys.path.insert(0, f"{PROJECT_DIR}/audio")
# 复用原始脚本的 ROLES 和 LINES 定义
from batch_tts_voxcpm import ROLES, LINES

# ──────────────────────────────────────────────────────────────
def find_ref_audio(role_key):
    role_dir = os.path.join(VOICE_DIR, ROLES[role_key]["dir"])
    for ext in ("*.wav", "*.mp3", "*.flac"):
        files = sorted(glob.glob(os.path.join(role_dir, ext)))
        if files:
            return files[0]
    return None

def wav_to_base64(path):
    with open(path, "rb") as f:
        return base64.b64encode(f.read()).decode()

def convert_to_wav(src, dst):
    subprocess.run(["ffmpeg", "-y", "-i", src, "-ar", "16000", "-ac", "1",
                    "-acodec", "pcm_s16le", dst],
                   check=True, capture_output=True)

def apply_speed(src, speed, dst):
    if abs(speed - 1.0) < 0.01:
        import shutil; shutil.copy2(src, dst); return
    subprocess.run(["ffmpeg", "-y", "-i", src, "-filter:a", f"atempo={speed}",
                    "-acodec", "pcm_s16le", dst],
                   check=True, capture_output=True)

def wav_to_mp3(src, dst):
    subprocess.run(["ffmpeg", "-y", "-i", src, "-codec:a", "libmp3lame",
                    "-b:a", "128k", "-ar", "24000", "-ac", "1", dst],
                   check=True, capture_output=True)

def check_server():
    try:
        return requests.get(f"{BASE_URL}/health", timeout=5).status_code == 200
    except: return False

def register_voice(wav_path):
    r = requests.post(f"{BASE_URL}/v1/voices",
                      json={"wav_base64": wav_to_base64(wav_path)}, timeout=60)
    r.raise_for_status()
    return r.json()["voice_id"]

def synthesize(voice_id, text):
    r = requests.post(f"{BASE_URL}/v1/speech",
                      json={"text": text, "voice_id": voice_id,
                            "cfg_value": 2.0, "inference_timesteps": 10},
                      timeout=120)
    r.raise_for_status()
    return r.content

# ──────────────────────────────────────────────────────────────
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--role", help="只合成指定角色")
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--resume", action="store_true", help="跳过已存在的文件", default=True)
    args = parser.parse_args()

    lines = LINES
    if args.role:
        lines = [(f,r,t) for f,r,t in lines if r == args.role]

    print(f"\n{'='*60}")
    print(f" TTS v1 — VoxCPM2  ({len(lines)} 条)")
    print(f" 输出: {OUT_DIR}")
    print(f"{'='*60}\n")

    if args.dry_run:
        for f,r,t in lines:
            out = os.path.join(OUT_DIR, f"{f}_wav_v1.mp3")
            done = "✓" if os.path.exists(out) else "·"
            print(f"  [{done}] {f}  [{r}]  {t[:40]}")
        return

    if not check_server():
        print(f"[错误] VoxCPM2 服务未响应: {BASE_URL}")
        sys.exit(1)

    # 预注册音色
    voice_ids = {}
    tmp = tempfile.mkdtemp(prefix="voxcpm_")
    for role_key in set(r for _,r,_ in lines):
        ref = find_ref_audio(role_key)
        if not ref:
            print(f"  [跳过] {role_key}: 未找到参考音频"); continue
        wav = ref if ref.endswith(".wav") else os.path.join(tmp, f"{role_key}.wav")
        if not ref.endswith(".wav"):
            convert_to_wav(ref, wav)
        print(f"  注册音色: {role_key} ... ", end="", flush=True)
        vid = register_voice(wav)
        voice_ids[role_key] = vid
        print(f"ok ({vid[:8]}...)")

    # 批量合成
    ok = fail = skip = 0
    for i, (fname, role, text) in enumerate(lines, 1):
        out_mp3 = os.path.join(OUT_DIR, f"{fname}_wav_v1.mp3")
        if args.resume and os.path.exists(out_mp3):
            print(f"  [{i:3d}/{len(lines)}] SKIP {fname}")
            skip += 1; continue

        if role not in voice_ids:
            print(f"  [{i:3d}/{len(lines)}] MISS {fname} (无音色)")
            fail += 1; continue

        speed = ROLES[role]["speed"]
        print(f"  [{i:3d}/{len(lines)}] {fname}  text={text[:30]}... ", end="", flush=True)
        t0 = time.time()
        try:
            wav_bytes = synthesize(voice_ids[role], text)
            raw_wav = os.path.join(tmp, f"{fname}_raw.wav")
            spd_wav = os.path.join(tmp, f"{fname}_spd.wav")
            with open(raw_wav, "wb") as f: f.write(wav_bytes)
            apply_speed(raw_wav, speed, spd_wav)
            wav_to_mp3(spd_wav, out_mp3)
            print(f"ok ({time.time()-t0:.1f}s  {os.path.getsize(out_mp3)//1024}KB)")
            ok += 1
        except Exception as e:
            print(f"FAIL: {e}"); fail += 1

    print(f"\n完成: {ok} ok, {skip} skip, {fail} fail → {OUT_DIR}")

if __name__ == "__main__":
    main()