用 Speechify API 合成长篇有声书：分段、429 重试与 MP3 拼接

Speechify 的 /v1/audio/speech 一次最多 2000 个计费字符，一章文字往往超过这个数。 这篇做 tts_long.py：按句子把长文切成不超过 2000 字符的段，逐段合成（遇到 429 按 Retry-After 等待），把 base64 音频解码后用 ffmpeg 无损拼成一个 MP3。

先看清成品

• 替身上跑通：2556 字符的示例章节切成 2 段（1970 + 584），中途一次 429 按 Retry-After 重试，拼成 170 秒的 MP3。
• 真实接口：无效 key 返回 401，不重试，直接提示并带上 request_id。
• 发现并修复：切句时把 “simba-3.5” 里的小数点当成句号，数字被切到两段里。pytest 4 项通过。

合规提醒：只朗读你有权使用的文字；把别人的书或文章转成音频再传播可能侵犯版权。不要用 TTS 冒充真人（声音克隆需要本人同意，本文不涉及）。本文使用 Speechify 开发者 API（api.speechify.ai），和朗读 App（speechify.com）是两个不同产品。key 在 platform.speechify.ai 创建，放在环境变量 SPEECHIFY_API_KEY 中。按字符计费，价格以官网为准（本文不列价格）。

运行说明：box 上没有 Speechify key。我用无效 key 真实调用了 POST https://api.speechify.ai/v1/audio/speech（2026-10-21 20:35 和 20:50，UTC+8），得到 HTTP 401 {"error":{"code":"unauthorized","message":"Unauthorized"},"request_id":"…"}。其余响应来自我写的本地替身：字段名按官方 API 参考，音频是 ffmpeg 生成的正弦波。没有真正合成过语音。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

78-speechify-A-src/
├── tts_long.py
├── fake_speechify.py
├── demo_local.py
├── chapter.txt
└── test_tts.py

第 1 步：主脚本

"""长文本朗读：按句子切成不超过 2000 个计费字符的段 → 逐段调用 Speechify POST /v1/audio/speech → 用 ffmpeg 拼成一个 MP3。
官方文档：https://docs.speechify.ai  key 放在环境变量 SPEECHIFY_API_KEY（在 platform.speechify.ai 创建）。"""
import argparse, base64, json, os, re, subprocess, sys, tempfile, time
from pathlib import Path
import requests

BASE = os.environ.get("SPEECHIFY_BASE", "https://api.speechify.ai")
LIMIT = 2000  # /v1/audio/speech 每次最多 2000 个计费（朗读）字符；更长用 /v1/audio/stream
RETRY = {429, 502, 503, 504}  # 官方：这些可以退避重试；4xx 要改请求

class TTSError(Exception): pass

def split_text(text, limit=LIMIT):
    # 英文句号后面必须是空白或结尾才算句末，否则 "3.5"、"e.g." 会被切开；中文句末标点直接算
    sents = re.findall(r".+?(?:[.!?]+(?=\s|$)|[。！？]+|$)\s*", text.strip(), flags=re.S) or [text]
    sents = [x for x in sents if x]
    out, cur = [], ""
    for s in sents:
        if len(s) > limit:  # 单句超长：按空格硬切
            for i in range(0, len(s), limit):
                if cur: out.append(cur.strip()); cur = ""
                out.append(s[i:i + limit].strip())
            continue
        if len(cur) + len(s) > limit: out.append(cur.strip()); cur = s
        else: cur += s
    if cur.strip(): out.append(cur.strip())
    return out

def speak(key, text, voice, *, model="simba-3.5", language=None, session=requests, sleep=None, tries=4):
    sleep = sleep or time.sleep
    body = {"input": text, "voice_id": voice, "model": model, "audio_format": "mp3",
            "options": {"loudness_normalization": True}}  # 各段响度一致，拼接后不忽大忽小
    if language: body["language"] = language
    for n in range(tries):
        r = session.post(f"{BASE}/v1/audio/speech", json=body, timeout=120,
                         headers={"Authorization": f"Bearer {key}", "User-Agent": "binbho-tutorial/1.0"})
        if r.status_code == 200:
            d = r.json(); return base64.b64decode(d["audio_data"]), d.get("billable_characters_count", 0)
        rid = ""
        try: err = r.json(); rid = err.get("request_id", ""); msg = f"{err['error'].get('code')}: {err['error'].get('message')}"
        except Exception: msg = r.text[:200]
        if r.status_code in RETRY and n < tries - 1:
            wait = float(r.headers.get("Retry-After") or 2 ** n); print(f"  {r.status_code}，{wait:g}s 后重试", file=sys.stderr); sleep(wait); continue
        hint = {401: "检查 SPEECHIFY_API_KEY", 402: "余额不足或达到 key 的月度花费上限"}.get(r.status_code, "")
        raise TTSError(f"HTTP {r.status_code} {msg} {hint} request_id={rid}".strip())

def concat(parts, out):
    with tempfile.TemporaryDirectory() as d:
        lst = Path(d) / "list.txt"; names = []
        for i, b in enumerate(parts):
            p = Path(d) / f"{i:03d}.mp3"; p.write_bytes(b); names.append(f"file '{p}'")
        lst.write_text("\n".join(names))
        subprocess.run(["ffmpeg", "-v", "error", "-y", "-f", "concat", "-safe", "0", "-i", str(lst), "-c", "copy", out], check=True)

def main(argv=None):
    ap = argparse.ArgumentParser(); ap.add_argument("file"); ap.add_argument("--voice", default="geffen_32")
    ap.add_argument("--model", default="simba-3.5"); ap.add_argument("--language"); ap.add_argument("-o", default="book.mp3")
    ap.add_argument("--limit", type=int, default=LIMIT)
    a = ap.parse_args(argv)
    key = os.environ.get("SPEECHIFY_API_KEY") or sys.exit("请先 export SPEECHIFY_API_KEY=…")
    chunks = split_text(Path(a.file).read_text(encoding="utf-8"), a.limit)
    print(f"{len(chunks)} 段，最长 {max(map(len, chunks))} 字符", file=sys.stderr)
    parts, billed = [], 0
    try:
        for i, c in enumerate(chunks, 1):
            b, n = speak(key, c, a.voice, model=a.model, language=a.language)
            parts.append(b); billed += n; print(f"  [{i}/{len(chunks)}] {len(c)} 字符 → {len(b)} 字节", file=sys.stderr)
    except TTSError as e:
        sys.exit(f"错误：{e}")
    concat(parts, a.o); print(f"已写入 {a.o}，计费字符合计 {billed}")

if __name__ == "__main__":
    main()

第 2 步：本地替身

"""本地替身（我写的）：字段名取自官方 API 参考（audio_data / audio_format / billable_characters_count / speech_marks）；
音频是 ffmpeg 生成的正弦波 MP3，时长按字符数估算。401 响应体照抄真实接口（2026-10-21 实测）。"""
import base64, json, subprocess, tempfile, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

FAIL_ONCE = {"n": 0}
def tone(seconds):
    with tempfile.NamedTemporaryFile(suffix=".mp3") as f:
        subprocess.run(["ffmpeg", "-v", "error", "-y", "-f", "lavfi", "-i", f"sine=frequency=440:duration={seconds}", "-ar", "24000", f.name], check=True)
        return open(f.name, "rb").read()

class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def _send(self, code, obj, headers=()):
        b = json.dumps(obj).encode(); self.send_response(code); self.send_header("Content-Type", "application/json")
        for k, v in headers: self.send_header(k, v)
        self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        if self.headers.get("Authorization") != "Bearer demo":
            return self._send(401, {"error": {"code": "unauthorized", "message": "Unauthorized"}, "request_id": "local"})
        if FAIL_ONCE["n"] > 0:
            FAIL_ONCE["n"] -= 1
            return self._send(429, {"error": {"code": "rate_limited", "message": "Too many requests"}}, [("Retry-After", "1")])
        n = len(body["input"])
        self._send(200, {"audio_data": base64.b64encode(tone(max(1, round(n / 15)))).decode(), "audio_format": "mp3",
                         "billable_characters_count": n, "speech_marks": {}})

def start():
    s = ThreadingHTTPServer(("127.0.0.1", 0), H); threading.Thread(target=s.serve_forever, daemon=True).start(); return s

第 3 步：演示脚本

import os, sys
import fake_speechify
srv = fake_speechify.start(); os.environ["SPEECHIFY_BASE"] = f"http://127.0.0.1:{srv.server_port}"
if "--rate-limit-once" in sys.argv: sys.argv.remove("--rate-limit-once"); fake_speechify.FAIL_ONCE["n"] = 1
import tts_long; sys.argv = ["tts_long.py"] + sys.argv[1:]; tts_long.main()

第 4 步：测试

import base64, json, subprocess, pytest
import tts_long as m

def test_no_split_inside_decimal():
    cs = m.split_text("The model simba-3.5 supports 48 languages. " * 60, 2000)
    assert all(len(c) <= 2000 for c in cs) and all(c.endswith("languages.") for c in cs)

def test_chinese_and_long_sentence():
    assert m.split_text("你好。今天下雨！要带伞吗？", 6) == ["你好。", "今天下雨！", "要带伞吗？"]
    assert all(len(c) <= 50 for c in m.split_text("x" * 120, 50))

class R:
    def __init__(self, code, body, headers=None): self.status_code, self._b, self.headers, self.text = code, body, headers or {}, json.dumps(body)
    def json(self): return self._b

def test_retry_after_then_success():
    seq = iter([R(429, {"error": {"code": "rate_limited", "message": "x"}}, {"Retry-After": "3"}),
                R(200, {"audio_data": base64.b64encode(b"MP3").decode(), "audio_format": "mp3", "billable_characters_count": 5})])
    waits, sent = [], []
    class S:
        def post(self, url, json=None, **k): sent.append(json); return next(seq)
    b, n = m.speak("k", "Hello", "geffen_32", session=S(), sleep=waits.append)
    assert b == b"MP3" and n == 5 and waits == [3.0]
    assert sent[0]["input"] == "Hello" and sent[0]["model"] == "simba-3.5" and "text" not in sent[0]

def test_401_not_retried():
    calls = []
    class S:
        def post(self, *a, **k): calls.append(1); return R(401, {"error": {"code": "unauthorized", "message": "Unauthorized"}, "request_id": "abc"})
    with pytest.raises(m.TTSError, match="401.*SPEECHIFY_API_KEY.*abc"):
        m.speak("bad", "x", "v", session=S(), sleep=lambda s: None)
    assert len(calls) == 1

第 5 步：运行

$ SPEECHIFY_API_KEY=demo python demo_local.py chapter.txt --rate-limit-once
2 段，最长 1970 字符
  429，1s 后重试
  [1/2] 1970 字符 → 524492 字节
  [2/2] 584 字符 → 156428 字节
已写入 book.mp3，计费字符合计 2554
$ ffprobe -v error -show_entries format=duration -of csv=p=0 book.mp3
170.112000
$ SPEECHIFY_API_KEY=invalid python tts_long.py chapter.txt     # 真实接口
2 段，最长 1970 字符
错误：HTTP 401 unauthorized: Unauthorized 检查 SPEECHIFY_API_KEY request_id=342b91332e4fa3e53d653853
$ python -m pytest -q
4 passed

怎么检查结果？

• 每段长度不超过 2000；计费字符合计和原文长度基本一致（本例 2554 / 2556，差的是段首尾的空格）。
• 打开 book.mp3 听段与段的衔接处；开启响度归一化后，段间音量应一致。
• 中文文本要设 --language zh-CN 并确认所选声音支持（以官方语言列表为准）。

调试示例

数字被切成两半。 第一版用 [^.!?。！？]+[.!?。！？]* 切句，测试文本是重复 60 次的 “The model simba-3.5 supports 48 languages.”：

$ python -c "import tts_long as m; cs = m.split_text('The model simba-3.5 supports 48 languages. ' * 60, 2000); print(len(cs), [c[-12:] for c in cs], [c[:12] for c in cs])"
2 ['del simba-3.', '8 languages.'] ['The model si', '5 supports 4']

第 1 段以 “simba-3.” 结尾，第 2 段以 “5 supports” 开头：小数点被当成了句号，朗读会变成“三……五”。修复：英文句号、问号、叹号后面必须是空白或文本结尾才算句末；中文句末标点直接算。修复后每段都以 “languages.” 结尾。

常见问题

模板规则

• 按句子切段，不在数字中间切。
• 429 按 Retry-After 等待，4xx 不重试。
• 显式指定 audio_format 和 model。
• 只朗读有权使用的文字。

项目资源

长文切得好，听起来才连贯。