m0_batch_tts.py

# -*- coding: utf-8 -*-
"""
M0 抢救性批量合成脚本：沪语正字句库 → qwen3-tts-flash (Jada) → 本地 wav 缓存

用法：
  1. pip install dashscope requests
  2. set DASHSCOPE_API_KEY=sk-xxx        (Windows)   /  export DASHSCOPE_API_KEY=sk-xxx (Linux/Mac)
     可选：set DASHSCOPE_BASE_URL=https://llm-xxxx.cn-beijing.maas.aliyuncs.com/api/v1
     （注意是 /api/v1 结尾，不是 /compatible-mode/v1）
  3. 准备 sentences.csv（UTF-8，两列：id,text，text 为沪语正字）
  4. python m0_batch_tts.py

特性：断点续跑（已存在的 wav 跳过）、失败重试 3 次、生成 manifest.csv 映射表。
"""
import csv
import os
import sys
import time
from pathlib import Path

import dashscope
import requests

# ---------- 配置 ----------
MODEL = "qwen3-tts-flash"          # 稳定版主线；下线前 3 个月有通知期
VOICE = "Jada"                     # 沪语音色，仅存于 qwen3-tts / qwen-tts 老系列
SENTENCES_CSV = "sentences.csv"    # 输入：id,text（沪语正字）
OUT_DIR = Path("audio_cache")      # 输出目录
MANIFEST = OUT_DIR / "manifest.csv"
MAX_RETRY = 3
SLEEP_BETWEEN = 0.5                # 请求间隔（秒），避免触发限流
# --------------------------

dashscope.base_http_api_url = os.getenv(
    "DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/api/v1"
)
API_KEY = os.getenv("DASHSCOPE_API_KEY")
if not API_KEY:
    sys.exit("请先设置环境变量 DASHSCOPE_API_KEY（不要把 Key 写进代码或聊天记录）")


def synthesize(text: str, out_path: Path) -> None:
    """合成一句并立即下载落盘（返回的 URL 仅 24 小时有效）。"""
    resp = dashscope.MultiModalConversation.call(
        model=MODEL,
        api_key=API_KEY,
        text=text,
        voice=VOICE,
        language_type="Chinese",
        stream=False,
    )
    if resp is None or resp.output is None or resp.output.audio is None:
        raise RuntimeError(f"API 返回异常: {resp}")
    audio_url = resp.output.audio["url"]
    r = requests.get(audio_url, timeout=60)
    r.raise_for_status()
    tmp = out_path.with_suffix(".part")
    tmp.write_bytes(r.content)
    tmp.rename(out_path)  # 原子写入，避免半截文件被当成已完成


def main() -> None:
    OUT_DIR.mkdir(exist_ok=True)
    rows = []
    with open(SENTENCES_CSV, encoding="utf-8-sig") as f:
        for row in csv.DictReader(f):
            rows.append((row["id"].strip(), row["text"].strip()))
    print(f"共 {len(rows)} 句，输出目录 {OUT_DIR.resolve()}")

    ok, skip, fail = 0, 0, []
    for sid, text in rows:
        out_path = OUT_DIR / f"{sid}.wav"
        if out_path.exists():
            skip += 1
            continue
        for attempt in range(1, MAX_RETRY + 1):
            try:
                synthesize(text, out_path)
                ok += 1
                print(f"[OK] {sid}: {text}")
                break
            except Exception as e:
                print(f"[重试 {attempt}/{MAX_RETRY}] {sid}: {e}")
                time.sleep(2 * attempt)
        else:
            fail.append((sid, text))
        time.sleep(SLEEP_BETWEEN)

    # 写映射表（全量重写，含之前已缓存的）
    with open(MANIFEST, "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["id", "text", "audio_file", "model", "voice", "status"])
        for sid, text in rows:
            exists = (OUT_DIR / f"{sid}.wav").exists()
            w.writerow([sid, text, f"{sid}.wav", MODEL, VOICE,
                        "ok" if exists else "failed"])

    print(f"\n完成：新合成 {ok}，已跳过 {skip}，失败 {len(fail)}")
    if fail:
        print("失败清单（可直接重跑脚本续传）：")
        for sid, text in fail:
            print(f"  {sid}: {text}")


if __name__ == "__main__":
    main()