先跑通一句短文本再谈批量——SparkAudio 1.0 接入前要确认的顺序

把 SparkAudio 1.0 接进自己的流程时，最容易出错的地方往往在顺序：还没确认一句短文本能否返回可播放的音频，就先压几十段甚至上百段批量任务。一旦失败，到底是密钥不对、参数填错，还是自己这一侧的队列限制了并发，就说不清了。建议的顺序是：最小请求、格式确认、超时重试、批量队列，四步各自留下可以复现的证据，再往下走。

接入前的判断标准是“一句十来个字的文本能稳定返回可解码的音频”。做法上先用最小请求一次验完鉴权、地址和必填参数，再用 ffprobe 和播放器各验一次容器与采样率，然后加超时重试跑十段观察真实耗时，最后才上队列并限制并发。边界在于：接口字段名、错误码和限额需要以你实际拿到的接口说明为准，下面的骨架只作占位，不能直接当成官方接口。

用一句十来个字的文本跑通第一次请求

第一次请求的目标是同时验证鉴权、服务地址和必填参数，因此文本要短、变量要少。密钥不要写进代码，走环境变量占位；地址、模型或音色名先用占位符，跑通后再替换成真实值。下面这段骨架放在本地脚本里跑即可，字段名以你实际接口为准。

import os, requests

BASE = os.environ["SPARKAUDIO_BASE_URL"]   # 例如 https://host/v1
KEY  = os.environ["SPARKAUDIO_API_KEY"]    # 不写进仓库
TEXT = "今天下午三点开会，请准时参加。"

resp = requests.post(
    f"{BASE}/tts",
    headers={"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"},
    json={"text": TEXT, "voice": "default", "format": "wav"},
    timeout=(5, 30),
)
print("status:", resp.status_code)
print("content-type:", resp.headers.get("Content-Type"))
print("request-id:", resp.headers.get("X-Request-Id"))
print("body-head:", resp.text[:300])

失败时先看四个地方：状态码是 401/403 还是 400/422，能区分是密钥问题还是参数问题；响应体里的错误码和错误信息；响应头中的请求标识，便于后续排查；返回的 Content-Type 是 JSON 还是音频流，返回 JSON 却在代码里直接写文件，会得到一个能打开但放不出声音的文件。文本长度、音色名、音频格式这三个必填项建议单独改一次再跑，确认哪个字段真正生效。

确认返回的音频容器与采样率能被播放器识别

这一步不是多余：如果批量跑完才发现文件放不出来，返工成本最高。先把第一次请求的响应按 Content-Type 保存成对应扩展名的文件，再用 ffprobe 读一遍元信息。

ffprobe -v error -show_entries format=format_name,duration \
  -show_entries stream=codec_name,sample_rate,channels \
  -of default=noprint_wrappers=1 out.wav

核对项包括：容器头是否符合预期（WAV 应有 RIFF/WAVE 标识，MP3 应有帧同步或 ID3 头）；采样率与声道数是否和你下游链路一致，比如播放器、ASR、拼接工具是否接受 24kHz 单声道；时长是否与文本长度大致匹配，过短往往意味着被截断；响度是否接近静音。如果接口返回的是 JSON 内嵌 base64 音频，要先解码再保存，否则 ffprobe 会直接报格式错误。

验证方式建议做两次：一次用系统播放器或浏览器打开听一遍，确认不是噪音、不是空白；一次用解码库读一遍，确认下游代码能正常加载。

python -c "import soundfile as sf; d, sr = sf.read('out.wav'); print(d.shape, sr)"

播放器能放但解码库报错，通常是容器扩展名与实际编码不一致；两者都失败，则回到上一步检查响应体是不是错误 JSON。

加上超时与重试再跑十段文本

单次成功不代表网络条件稳定。用同一段短文本准备十句变体（长度相近、标点不同即可），在加了超时和重试的客户端下跑一遍，观察失败出现在第几段、耗时集中在哪个区间。超时要分开设置：连接超时防止地址或网络不通时长时间挂起，读取超时防止服务端生成慢导致阻塞。设置位置通常在 HTTP 客户端的 timeout 参数或会话适配器里。

from requests.adapters import HTTPAdapter, Retry

session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=Retry(
    total=2, connect=2, read=2,
    backoff_factor=1.5,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["POST"],
)))

重试次数建议先给到 2 次、退避从 1 到 2 秒起步，再结合你自己的限额和使用条款确认。合成类 POST 请求重试可能产生重复任务或重复计费，如果接口支持幂等键，把同一段的标识带上去更稳妥；不支持时，重试只针对连接失败和 5xx，不要对 4xx 无脑重试。每次失败都落一条记录：段号、开始与结束时间、耗时、状态码、错误码、请求标识。这份记录是后面判断“是接口慢还是自己队列堵”的唯一依据。

把短文本流程扩到批量任务队列

十段能跑通，再上队列。关键不是队列本身，而是并发度和限速要由你这一侧主动控制，避免把自己的出口带宽、下游拼接任务或配额一起压垮。先用配置文件把并发写死，跑一段再调。

job:
  workers: 2          # 同时在跑的请求数
  queue_size: 50      # 排队上限，满了就先暂停投递
  rate_limit_per_min: 20
  max_attempts: 3
  backoff_seconds: 2
  on_failure: keep_and_continue
  out_dir: ./out
  failed_log: ./failed.jsonl

执行侧按文件逐行读取分段文本，每段带一个稳定编号，结果按编号落盘，避免乱序覆盖。限速可以按每分钟投递数做，也可以用固定间隔投递，两者选一种并写进配置，不要在代码里散落 sleep。失败段单独写进 failed.jsonl，重跑入口只读这个文件，不要重跑全量：

python run_tts.py `--input` segments.jsonl `--out` ./out `--config` job.yaml
python run_tts.py `--only-failed` ./failed.jsonl `--out` ./out `--config` job.yaml

并发从 1 或 2 开始，观察失败记录里状态码的分布再往上加；如果失败集中在 429，先降速而不是加重试。

把可用参数固化成一份配置文件

跑通之后，把所有试出来的可用值收进一份配置文件，后续调用只改文本。字段清单建议包含：接口地址、模型或音色名、输出格式、采样率、连接与读取超时、重试次数与退避、并发度、限速、输出目录、失败日志路径。密钥不要写进这份文件，用环境变量或本地未入库的 .env 注入，配置里只留变量名，这样配置可以随代码一起走，密钥不会。

endpoint: ${SPARKAUDIO_BASE_URL}
auth_key_env: SPARKAUDIO_API_KEY
voice: default
format: wav
sample_rate: 24000
timeout: {connect: 5, read: 30}
retry: {max_attempts: 3, backoff_seconds: 2, status_forcelist: [429, 500, 502, 503, 504]}
concurrency: {workers: 2, rate_limit_per_min: 20}
output: {dir: ./out, failed_log: ./failed.jsonl}

固化后用同一段短文本做一次回归：跑一次请求，再用同样的 ffprobe 命令核对容器、采样率、声道是否与固化前一致，时长是否在同一量级；然后跑一次十段，确认失败记录为空或只有可解释的条目。回归通过后，配置文件就作为接入基线，后续换音色、换格式都在这份文件上改一处并重新回归，而不是散在脚本里逐个试。