Add presentation slides for OU process and Mehler formula

research/math/podcast/synth_volc_v3.py

@@ -0,0 +1,140 @@
+#!/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()