Files
worldmodel/research/math/podcast/synth_volc_v3.py
T
gaojie 920439e044
Sync to site1 / sync (push) Has been cancelled
Add presentation slides for OU process and Mehler formula
2026-06-03 04:08:01 +08:00

141 lines
4.6 KiB
Python
Raw 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
"""用火山引擎 v3 单向流式 TTS/api/v3/tts/unidirectional)把双人对话脚本合成为一个 MP3。
鉴权:请求头 x-api-key + X-Api-Resource-Id(新版方式,区别于旧版 Bearer;{token})。
用法:
VOLC_API_KEY=<你的key> python3 synth_volc_v3.py \
--script-file hermite_polynomials-script.json \
--output-file output/hermite_polynomials-podcast.mp3
"""
import argparse
import base64
import json
import os
import sys
import time
import uuid
import requests
ENDPOINT = "https://openspeech.bytedance.com/api/v3/tts/unidirectional"
RESOURCE_ID = "volc.service_type.10029"
# 男女两个大模型情感音色(v2 mars bigtts
VOICE_MALE = "zh_male_beijingxiaoye_emo_v2_mars_bigtts"
VOICE_FEMALE = "zh_female_roumeinvyou_emo_v2_mars_bigtts"
ADDITIONS = json.dumps(
{
"disable_markdown_filter": True,
"enable_language_detector": True,
"enable_latex_tn": True,
"disable_default_bit_rate": True,
"max_length_to_filter_parenthesis": 0,
}
)
def synth_line(api_key: str, text: str, speaker: str, retries: int = 3) -> bytes:
"""合成一句,返回 mp3 字节。失败抛异常。"""
headers = {
"x-api-key": api_key,
"X-Api-Resource-Id": RESOURCE_ID,
"Connection": "keep-alive",
"Content-Type": "application/json",
}
payload = {
"req_params": {
"text": text,
"speaker": speaker,
"additions": ADDITIONS,
"audio_params": {"format": "mp3", "sample_rate": 24000},
}
}
last_err = None
for attempt in range(retries):
try:
r = requests.post(ENDPOINT, headers=headers, json=payload, timeout=60)
if r.status_code != 200:
last_err = f"HTTP {r.status_code}: {r.text[:200]}"
time.sleep(1.0 + attempt)
continue
# 响应体可能是单个 JSON,也可能是多段(流式)。逐段解析 data 累积。
audio = b""
text_body = r.text.strip()
# 尝试按行分割的多 JSON(流式 unidirectional 常见)
chunks = []
for line in text_body.splitlines():
line = line.strip()
if not line:
continue
chunks.append(line)
if not chunks:
chunks = [text_body]
ok = False
for c in chunks:
try:
obj = json.loads(c)
except json.JSONDecodeError:
continue
if obj.get("code") not in (0, None):
last_err = f"code={obj.get('code')} msg={obj.get('message')}"
continue
d = obj.get("data")
if d:
audio += base64.b64decode(d)
ok = True
if ok and audio:
return audio
last_err = last_err or "no audio data in response"
time.sleep(1.0 + attempt)
except Exception as e: # noqa
last_err = str(e)
time.sleep(1.0 + attempt)
raise RuntimeError(f"合成失败: {last_err}")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--script-file", required=True)
ap.add_argument("--output-file", required=True)
args = ap.parse_args()
api_key = os.getenv("VOLC_API_KEY")
if not api_key:
print("ERROR: 需要环境变量 VOLC_API_KEY", file=sys.stderr)
sys.exit(2)
with open(args.script_file, encoding="utf-8") as f:
script = json.load(f)
lines = script["lines"]
total = len(lines)
print(f"加载脚本:{script.get('title','')},共 {total}")
os.makedirs(os.path.dirname(args.output_file) or ".", exist_ok=True)
all_audio = b""
ok_count = 0
for i, line in enumerate(lines, 1):
speaker = VOICE_MALE if line["speaker"] == "male" else VOICE_FEMALE
text = line["paragraph"]
try:
audio = synth_line(api_key, text, speaker)
all_audio += audio
ok_count += 1
print(f"[{i}/{total}] {line['speaker']:6s} ok ({len(audio)} bytes)")
except Exception as e: # noqa
print(f"[{i}/{total}] {line['speaker']:6s} FAIL: {e}")
time.sleep(0.15) # 轻微限速
if not all_audio:
print("ERROR: 没有任何一句合成成功", file=sys.stderr)
sys.exit(1)
with open(args.output_file, "wb") as f:
f.write(all_audio)
print(f"\n完成:{ok_count}/{total} 句成功,已写出 {args.output_file}{len(all_audio)} bytes")
if __name__ == "__main__":
main()