feat(v1.3.0): 新增火山引擎语音合成 2.0 作为可选配音引擎

流水线开跑前先问用户用哪个 TTS。两引擎产出的 audio-manifest.json
结构完全一致,故 timeline_build / subs / 渲染器零改动。

新增
- scripts/tts_volcano.py  接口包,唯一读 tts.env 的地方。零第三方依赖
                          (只用标准库 urllib),HTTP Chunked 单向流式,
                          与 edge-tts「发一次收一串」形态同构
- scripts/tts_setup.py    向导:选方案 → 配密钥 → 测连接 → 选音色,
                          输出 NEXT_ACTION=… 供智能体识别
- tts.env.example         密钥模板(真实 tts.env 已 gitignore)
- references/volcano-tts.md  密钥纪律 / 配置步骤 / 接口规格 / 音色表 / 排查表

安全
- Key 只存 tts.env,只有 tts_volcano.py 读它;异常过脱敏,只抹本次实际
  加载的密钥值与 AKLT… 形态,刻意不用「长字符串即密钥」的宽规则
- check_integrity.py 新增密钥防线检查(含「疑似把模板改名」提示)
- package_skill.py 排除 .env / .key / .pem,修掉一个真实外泄口

修复(真机联调暴露,均只在实跑时复现)
- enable_subtitle 必须显式开启,否则 words 恒为空数组且完全不报错,
  下游会静默退回按字数插值、字幕开始飘
- usage 挂在结束标记那一行,提前 continue 导致计费恒为 0
- 缓存未存首裁量,命中缓存时 lead_cut_sec=0,重跑字幕比首跑晚一帧
- CLI 把中文音色名原样当 speaker 发出,服务端报 resource ID mismatched

文档
- README 中英双语新增「配音:两个引擎」章节,按用户要求不写命令
  (那是智能体的职责),只讲「会问你什么 / 你唯一要动手的地方」
- 疑难排查补 6 条火山症状;lessons 34 → 56 条;流水线图加 tts_setup 节点
This commit is contained in:
MohZhuang committed 2026-09-23 10:04:13 +08:00
1 parent 09aa089354
commit 0d74a1097d
16 files changed
+1769 -45

No files matched your search

+73
View File
@@ -14,6 +14,9 @@ check_integrity.py —— 仓库自洽性检查(CI 与本地都能跑,零第
—— 授权信息缺失会让 THIRD_PARTY_NOTICES 失真。
5. 帧模板 / 封面模板不得出现外链字体或 CDN 资源
—— 离线渲染硬要求,见 frame-contract.md 契约 2。
6. 密钥防线:.gitignore 必须忽略 tts.env / *.env / *.key,且仓库内不得出现
真实的火山引擎密钥(非空赋值 / AKLT 形态 token)
—— 火山 TTS 是本技能唯一需要密钥的功能,泄露即永久失守。
用法:
python scripts/check_integrity.py # 以脚本所在技能的根目录为准
@@ -159,6 +162,75 @@ def check_no_remote_assets(root: str, fails: list[str]) -> None:
print(f"{OK} 模板无外链资源:{len(targets)} 个模板检查通过")
def check_no_secrets(root: str, fails: list[str]) -> None:
"""密钥防线:tts.env 必须被忽略,且仓库里不得出现真实密钥。
对应 2026-09-23 加入的火山引擎 TTS 支持 —— 那是本技能唯一需要密钥的功能。
密钥纪律是「agent 不读、仓库不记」,所以这里做两道自动化检查:
a) .gitignore 覆盖 tts.env / *.env / *.key,且 tts.env.example 模板仍被跟踪;
b) 草签文件里不得出现非空的密钥赋值或 AKLT 形态的 token。
"""
gi = os.path.join(root, ".gitignore")
if os.path.isfile(gi):
text = read(gi)
need = ["tts.env", "*.env", "*.key"]
missing = [n for n in need if n not in text]
if missing:
fails.append(f"secret: .gitignore 未忽略 {', '.join(missing)}"
f" —— 密钥可能被提交进公开仓库")
else:
print(f"{OK} 密钥防线:.gitignore 已忽略 tts.env / *.env / *.key")
else:
fails.append("secret: 没有 .gitignore —— 密钥无保护")
example = os.path.join(root, "tts.env.example")
if not os.path.isfile(example):
if os.path.isfile(os.path.join(root, "tts.env")):
# 实证过这种踩法:用户把仓库里的 tts.env.example **改名**成 tts.env 去填 key,
# 于是模板没了、仓库里多了一份(被忽略的)密钥文件。给个能指向原因的说法。
fails.append("secret: 缺 tts.env.example 模板,但同目录有 tts.env —— "
"疑似把模板文件改名成了 tts.env。请恢复 tts.env.example"
"(模板要留在仓库),tts.env 另行复制")
else:
fails.append("secret: 缺 tts.env.example 模板(用户没有可照抄的密钥模板)")
# 真实密钥形态:非空赋值 / AKLT 前缀
SECRET_ASSIGN = re.compile(
r"^\s*(VOLC_API_KEY|VOLC_ACCESS_KEY|X_Api_Key|X-Api-Key)\s*[=:]\s*[\"']?([^\s\"'#]+)",
re.M)
AKLT = re.compile(r"AKLT[A-Za-z0-9_\-]{8,}")
SKIP_DIRS = {".git", "node_modules", "__pycache__", "dist", ".venv", "venv"}
SKIP_EXT = {".mp4", ".mp3", ".wav", ".png", ".jpg", ".jpeg", ".ttf", ".otf",
".woff", ".woff2", ".zip", ".ico"}
hits: list[str] = []
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
for fn in filenames:
if os.path.splitext(fn)[1].lower() in SKIP_EXT:
continue
p = os.path.join(dirpath, fn)
rel = os.path.relpath(p, root)
# tts.env 本体跳过:它是用户本地文件,本检查只保证它不入库
if fn in ("tts.env",) or rel in ("tts.env",):
continue
try:
body = read(p)
except (UnicodeDecodeError, OSError):
continue
for m in AKLT.finditer(body):
hits.append(f"{rel}: AKLT 形态 token")
if fn.endswith(".example"):
continue
for m in SECRET_ASSIGN.finditer(body):
if m.group(2).strip(): # 有值才算泄露,空模板是合规的
hits.append(f"{rel}: {m.group(1)} 有非空值")
if hits:
for h in sorted(set(hits)):
fails.append(f"secret: 疑似真实密钥 —— {h}")
else:
print(f"{OK} 密钥防线:仓库内未发现真实密钥(模板里的空赋值不算)")
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--root", default=None, help="仓库根目录(默认 = 本脚本所在技能的根)")
@@ -173,6 +245,7 @@ def main() -> int:
count = check_catalog(root, fails)
check_catalog_count(root, count, fails)
check_no_remote_assets(root, fails)
check_no_secrets(root, fails)
print("-" * 40)
if fails:
+3
View File
@@ -59,6 +59,9 @@ def main() -> int:
"order": [], # 写完 narration.json 后把 id 按顺序填进来
"theme": "violet",
"progress": True,
# 配音引擎:edge = 免费免密钥;volcano = 火山引擎语音合成 2.0(需 tts.env)。
# ★ 建项目后先问用户用哪个 —— python <skill>/scripts/tts_setup.py --project .
"provider": "edge",
"voice": "zh-CN-YunxiNeural" if args.lang == "zh" else "en-US-GuyNeural",
"rate": "+8%",
}
+8
View File
@@ -54,6 +54,14 @@ def main() -> int:
order = pj.get("order") or []
gap = args.gap if args.gap >= 0 else float(pj.get("gap", 0.35))
if not order:
# 不拦这一下的话,下游 ffmpeg 会报 `Failed to set value ';concat=n=0:...'`
# —— 看不出根因在哪。直接说清楚该改哪个文件的哪个字段。
print("✗ project.json 的 order 是空的 —— 它是场景顺序表,先填上再跑。\n"
" 形如:\"order\": [\"intro\", \"point1\", \"ending\"](= narration.json 里 id 的顺序)",
file=sys.stderr)
return 1
man_path = os.path.join(proj, "audio-manifest.json")
if not os.path.exists(man_path):
print("✗ 先跑 tts_build.py(缺 audio-manifest.json)", file=sys.stderr)
+129 -26
View File
@@ -1,7 +1,19 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
tts_build.py —— html-explainer 的配音与时间轴数据源(edge-tts)。
tts_build.py —— html-explainer 的配音与时间轴数据源。
【两个 TTS 引擎】`--provider` / 环境变量 TTS_PROVIDER / project.json 的 provider 字段
(优先级从左到右)。**跑之前先问用户用哪个**(SKILL.md 确认点 3):
edge 默认。edge-tts 云端合成,免费、免密钥、有词级边界。
VOICE 默认 zh-CN-YunxiNeural(云希)。
volcano 火山引擎(豆包)语音合成 2.0。音质更好,**需要 API Key**。
密钥由 scripts/tts_volcano.py 从 tts.env 读取 —— 本文件从不接触密钥值。
音色默认 zh_male_m191_uranus_bigtts(云舟 2.0)。
设置向导:python <skill>/scripts/tts_setup.py --project .
★ 两个引擎产出的 manifest 结构完全一致(boundaries 都是「秒 + 原始坐标系」),
timeline_build.py / subs.py / 渲染器零改动。
【血统与口径】工程化外壳移植自 anything2explainer 的 tts_build.py(经
html-video-workbuddy-driver 的 tts_edges.py 二次实战),核心五件事:
@@ -10,7 +22,7 @@ html-video-workbuddy-driver 的 tts_edges.py 二次实战),核心五件事
② 每次合成 asyncio.wait_for 硬超时 45s —— edge-tts 的 websocket 会
偶发**永久挂起**(不抛错不返回零日志),必须把挂死变成可重试失败
③ 指数退避重试 4 次(微软端点连续请求会限流抛 NoAudioReceived)
④ 句间 0.8s 间隔 + 磁盘缓存(sig 含 boundary 与 trim —— 参数变了旧缓存自动失效)
④ 句间 0.8s 间隔 + 磁盘缓存(sig 含 engine/boundary 与 trim —— 参数变了旧缓存自动失效)
⑤ 默认裁首尾静音(首回退 0.03s 保爆破音、尾保留 0.12s 保收音),
裁完段间停顿才由 GAP 显式控制
@@ -19,6 +31,8 @@ html-video-workbuddy-driver 的 tts_edges.py 二次实战),核心五件事
按字数插值。必须显式传 boundary='WordBoundary'。
2) 相邻数字熔成大数(「沪深300」「4479.55点」→「三百万四千…」),
数字与数字之间补中文逗号。
※ 数字补逗号只在 edge 引擎做:它会改动送进去的文本,从而影响词边界与解说词的
逐字对齐。火山引擎不熔读,原样送原文(对齐最稳)。
时长两级口径(manifest 同时写两个,下游各取所需):
container_duration_sec 裁完后 MP3 容器时长 → **帧时长用这个**(音画同步)
@@ -26,7 +40,8 @@ html-video-workbuddy-driver 的 tts_edges.py 二次实战),核心五件事
duration_sec 词边界末尾(原始坐标系)→ 仅排查用
用法(项目目录内):
python tts_build.py --project . [--rate +8%] [--voice zh-CN-YunxiNeural] [--no-trim]
python tts_build.py --project . [--provider edge|volcano] [--rate +8%]
[--voice zh-CN-YunxiNeural] [--no-trim]
产出:
audio/<id>.mp3 + audio-manifest.json
"""
@@ -44,6 +59,9 @@ import sys
import edge_tts
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import tts_volcano # noqa: E402
DEFAULT_VOICE_ZH = "zh-CN-YunxiNeural" # 云希,男声
DEFAULT_VOICE_EN = "en-US-GuyNeural"
DEFAULT_RATE = "+8%"
@@ -52,6 +70,8 @@ DEFAULT_PITCH = "+0Hz"
DEFAULT_TIMEOUT = 45.0
DEFAULT_RETRIES = 4
RETRY_SLEEP = 0.8
VOLCANO_TIMEOUT = 60.0
VOLCANO_RETRIES = 3
TRIM_HEAD_BACKOFF = 0.03
TRIM_TAIL_KEEP = 0.12
@@ -152,8 +172,18 @@ def trim_edges(path: str, sr: int = 48000) -> float:
return round(a / sr, 3)
def cache_paths(text, voice, rate, volume, pitch, cache_dir, trim):
sig = f"{voice}|{rate}|{volume}|{pitch}|{_boundary_kwarg()}|{int(trim)}|{text}"
def _rate_to_int(rate) -> int:
"""'+8%' / '8' / 0.08 / 8 → 8。火山引擎的 speech_rate 是整数百分比([-50,100])。
实现在 tts_volcano.parse_rate_percent —— 语速换算只保留一份,
tts_setup 做格式清洗时用的是同一份逻辑。
"""
return tts_volcano.parse_rate_percent(rate)
def cache_paths(text, voice, rate, volume, pitch, cache_dir, trim, provider="edge"):
sig = (f"{provider}|{voice}|{rate}|{volume}|{pitch}|"
f"{_boundary_kwarg() if provider == 'edge' else 'chars'}|{int(trim)}|{text}")
h = hashlib.sha1(sig.encode("utf-8")).hexdigest()[:16]
d = os.path.join(cache_dir, "trim") if trim else cache_dir
os.makedirs(d, exist_ok=True)
@@ -161,21 +191,41 @@ def cache_paths(text, voice, rate, volume, pitch, cache_dir, trim):
async def synth(text, out_path, voice, rate, volume=DEFAULT_VOLUME, pitch=DEFAULT_PITCH,
cache_dir="", timeout=DEFAULT_TIMEOUT, retries=DEFAULT_RETRIES, trim=True) -> dict:
cache_dir="", timeout=None, retries=None, trim=True,
provider="edge", project_dir="") -> dict:
"""合成一段。provider='edge' 走 edge-tts,'volcano' 走火山引擎(密钥由接口包自理)。"""
os.makedirs(os.path.dirname(os.path.abspath(out_path)) or ".", exist_ok=True)
spoken = sanitize_for_tts(text)
if provider == "volcano":
# 火山不熔读相邻数字 → 原样送原文,词边界与解说词逐字对齐最稳
spoken = text
timeout = VOLCANO_TIMEOUT if timeout is None else timeout
retries = VOLCANO_RETRIES if retries is None else retries
else:
spoken = sanitize_for_tts(text)
timeout = DEFAULT_TIMEOUT if timeout is None else timeout
retries = DEFAULT_RETRIES if retries is None else retries
cache_mp3 = cache_js = ""
if cache_dir:
cache_mp3, cache_js = cache_paths(text, voice, rate, volume, pitch, cache_dir, trim)
cache_mp3, cache_js = cache_paths(text, voice, rate, volume, pitch, cache_dir,
trim, provider=provider)
if os.path.exists(cache_mp3) and os.path.exists(cache_js):
bounds = json.load(open(cache_js, encoding="utf-8"))
cached = json.load(open(cache_js, encoding="utf-8"))
# 缓存里必须连「首裁了多少」一起存。词边界是原始坐标系,下游 subs.py 靠
# lead_cut_sec 平移;命中缓存时若把它当 0,重跑出来的字幕会比首跑晚约一帧
# (实测 0.037s)。见 lessons.md #51。
if isinstance(cached, dict):
bounds = cached.get("boundaries") or []
head_cut = float(cached.get("lead_cut_sec") or 0.0)
else: # 旧版缓存是裸列表,没有首裁记录
bounds = cached
head_cut = 0.0
with open(cache_mp3, "rb") as f:
open(out_path, "wb").write(f.read())
dur = bounds[-1]["offset_sec"] + bounds[-1]["duration_sec"] if bounds else 0.0
return _result(text, spoken, out_path, voice, rate, bounds, dur, 0.0, trim)
return _result(text, spoken, out_path, voice, rate, bounds, dur, head_cut, trim)
async def _once():
async def _once_edge():
kw = {}
b = _boundary_kwarg()
if b != "n/a":
@@ -196,10 +246,28 @@ async def synth(text, out_path, voice, rate, volume=DEFAULT_VOLUME, pitch=DEFAUL
raise RuntimeError("empty audio stream")
return bytes(audio), bounds
# 接口包在「要了字幕却没拿到时间戳」时会返回 warning;必须往上抛,
# 否则下游只能看到「没有词边界」,排查时不知道是音色不支持还是接口出问题。
warns: list[str] = []
async def _once_volcano():
# 同步 HTTP 调用丢到线程里;接口包负责读 tts.env、脱敏、切片与重试
r = await asyncio.to_thread(
tts_volcano.synthesize, spoken, voice, _rate_to_int(rate),
"mp3", 24000, 160000, timeout, 1, None, project_dir)
if r.get("warning"):
warns.append(str(r["warning"]))
return r["audio"], r["boundaries"]
_once = _once_volcano if provider == "volcano" else _once_edge
last: Exception | None = None
for attempt in range(retries):
try:
audio, bounds = await asyncio.wait_for(_once(), timeout=timeout)
audio, bounds = await asyncio.wait_for(_once(), timeout=timeout + 15)
for w in warns:
print(f" ⚠ {w}", file=sys.stderr)
warns.clear()
with open(out_path, "wb") as f:
f.write(audio)
await asyncio.sleep(RETRY_SLEEP)
@@ -207,16 +275,27 @@ async def synth(text, out_path, voice, rate, volume=DEFAULT_VOLUME, pitch=DEFAUL
if cache_dir:
with open(cache_mp3, "wb") as f:
f.write(open(out_path, "rb").read())
json.dump(bounds, open(cache_js, "w", encoding="utf-8"), ensure_ascii=False)
json.dump({"boundaries": bounds, "lead_cut_sec": round(head_cut, 4)},
open(cache_js, "w", encoding="utf-8"), ensure_ascii=False)
dur = bounds[-1]["offset_sec"] + bounds[-1]["duration_sec"] if bounds else 0.0
return _result(text, spoken, out_path, voice, rate, bounds, dur, head_cut, trim)
except Exception as e: # noqa: BLE001
last = e
print(f" [retry {attempt + 1}/{retries}] {type(e).__name__}: {e}", file=sys.stderr, flush=True)
if getattr(e, "retryable", True) is False:
break # 永久性错误(密钥/音色/参数),重试是白等
await asyncio.sleep(1.5 * (attempt + 1))
raise SystemExit(
f"edge-tts 连续 {retries} 次失败:{type(last).__name__}: {last}\n"
f"已合成段落走缓存不会重复耗时,稍后重跑即可。")
label = "火山引擎" if provider == "volcano" else "edge-tts"
if getattr(last, "retryable", True) is False:
# 永久性错误(密钥/音色/参数):只试了一次就别谎报「连续 N 次失败」
detail = f"{label}:{last}"
else:
detail = f"{label} 连续 {attempt + 1} 次失败:{type(last).__name__}: {last}"
if provider == "volcano":
# 抛普通异常(不是 SystemExit):SystemExit 不是 Exception 子类,
# 会被 run() 里给火山准备的友好提示漏掉,用户就只看到一串原始报错。
raise tts_volcano.VolcanoTTSError(detail)
raise SystemExit(detail + "\n已合成段落走缓存不会重复耗时,稍后重跑即可。")
def _result(text, spoken, out_path, voice, rate, bounds, dur, head_cut, trimmed) -> dict:
@@ -251,18 +330,39 @@ async def run(args) -> int:
if isinstance(narration, dict):
narration = narration.get("items") or []
voice = args.voice or pj.get("voice") or DEFAULT_VOICE_ZH
rate = args.rate or pj.get("rate") or DEFAULT_RATE
# 引擎优先级:命令行 --provider > 环境变量 TTS_PROVIDER > project.json
provider = (args.provider or os.environ.get("TTS_PROVIDER")
or pj.get("provider") or "edge").strip().lower()
if provider not in ("edge", "volcano"):
print(f"✗ 未知 provider:{provider}(只支持 edge / volcano)", file=sys.stderr)
return 1
if provider == "volcano":
voice = args.voice or pj.get("voice") or tts_volcano.DEFAULT_VOICE
rate = args.rate or pj.get("rate") or 0
else:
voice = args.voice or pj.get("voice") or DEFAULT_VOICE_ZH
rate = args.rate or pj.get("rate") or DEFAULT_RATE
out_dir = os.path.join(proj, "audio")
cache = os.path.join(out_dir, "cache")
results = []
print(f"[tts] 引擎 {provider} · 音色 {voice} · 语速 {rate}", file=sys.stderr)
for item in narration:
key, text = item["id"], item["text"]
r = await synth(text, os.path.join(out_dir, f"{key}.mp3"),
voice=item.get("voice", voice), rate=item.get("rate", rate),
cache_dir="" if args.no_cache else cache,
trim=not args.no_trim)
try:
r = await synth(text, os.path.join(out_dir, f"{key}.mp3"),
voice=item.get("voice", voice), rate=item.get("rate", rate),
cache_dir="" if args.no_cache else cache,
trim=not args.no_trim, provider=provider, project_dir=proj)
except Exception as e: # noqa: BLE001
if provider == "volcano":
print(f"✗ [{key}] 火山引擎合成失败:\n{e}\n"
f" 排查:python <skill>/scripts/tts_setup.py --project . --check",
file=sys.stderr)
return 1
raise
r["id"] = key
results.append(r)
print(f" [{key}] {r['bytes']:>8}B 容器 {r['container_duration_sec']:>7.3f}s "
@@ -270,10 +370,11 @@ async def run(args) -> int:
f"bounds={r['boundary_count']}", file=sys.stderr)
manifest = {
"voice": voice, "rate": rate,
"provider": provider, "voice": voice, "rate": rate,
"trimmed": not args.no_trim,
"engine": f"edge-tts {edge_tts.__version__}",
"boundary_mode": _boundary_kwarg(),
"engine": ("volcano doubao-tts-2.0 (seed-tts-2.0)" if provider == "volcano"
else f"edge-tts {edge_tts.__version__}"),
"boundary_mode": _boundary_kwarg() if provider == "edge" else "char-timestamps",
"items": results,
}
mp = os.path.join(proj, "audio-manifest.json")
@@ -285,8 +386,10 @@ async def run(args) -> int:
def main() -> int:
p = argparse.ArgumentParser(description="html-explainer 配音合成(edge-tts)")
p = argparse.ArgumentParser(description="html-explainer 配音合成(edge-tts / 火山引擎)")
p.add_argument("--project", default=".")
p.add_argument("--provider", default="", choices=["", "edge", "volcano"],
help="edge=免费免密钥(默认);volcano=火山引擎语音合成 2.0(需 tts.env)")
p.add_argument("--voice", default="")
p.add_argument("--rate", default="")
p.add_argument("--no-trim", action="store_true", help="不裁首尾静音(不建议)")
+335
View File
@@ -0,0 +1,335 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
tts_setup.py —— 配音方案设置向导(选 TTS 引擎 → 配密钥 → 测连接 → 选音色)。
【它在流程里的位置】SKILL.md 的「确认点 3」要求先问用户用哪个 TTS。
本脚本把那次对话落成可复现的三步,并保证**密钥永不经过对话窗口**:
① 选方案 edge-tts(免费、免密钥、开箱可用) / 火山引擎语音合成 2.0(音质更好,需 API Key)
② 配密钥 选火山但缺 tts.env → 生成带注释的空模板 → 打印指引 → 退出(NEXT_ACTION=fill_env)
用户填好回来说一声 → 重跑本脚本(或 --check)→ 测连接
③ 选音色 列出内置常用音色(可自定义 ID)→ 写进 project.json
【密钥纪律】本脚本**从不打印、不落盘、不回显**密钥值;状态摘要走 tts_volcano.describe_env()
(内部已脱敏)。也不会因为测连接失败而把密钥打进日志。
用法:
python tts_setup.py --project . --status # 当前配置 + 脱敏密钥状态
python tts_setup.py --project . --provider volcano # 选火山(缺 env 则生成模板后退出)
python tts_setup.py --project . --list-voices # 列出候选音色
python tts_setup.py --project . --check # 测连接(火山走真合成;edge 查依赖)
python tts_setup.py --project . --provider volcano --voice "解说小明 2.0" # 一步到位
python tts_setup.py --project . # 交互式(人在终端时用)
退出码:0 完成;2 = 需要用户手工填 tts.env 后再来;1 = 出错。
"""
from __future__ import annotations
import argparse
import json
import os
import shutil
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import tts_volcano # noqa: E402
ENV_FILENAME = "tts.env"
ENV_TEMPLATE = """\
# ---------------------------------------------------------------
# 火山引擎(豆包)语音合成 2.0 —— 密钥配置
# 本文件由 tts_setup.py 生成,已被 .gitignore 忽略,不会进 git 仓库。
# ★ agent 不需要、也不应该读取本文件;接口包 scripts/tts_volcano.py 会自己读。
# ---------------------------------------------------------------
# 【必填】API Key —— 火山引擎控制台 → 语音技术 → API Key 管理
# https://console.volcengine.com/speech/new/setting/apikeys
VOLC_API_KEY=
# 资源 ID:seed-tts-2.0 = 预置音色;seed-icl-2.0 = 声音复刻音色
VOLC_RESOURCE_ID=seed-tts-2.0
# 【可选】旧版鉴权:若你拿到的是 APP ID + Access Token 而不是 API Key,填这两项
VOLC_APP_ID=
VOLC_ACCESS_KEY=
"""
# edge-tts 常用音色(名称, voice, 说明)。完整列表:edge-tts --list-voices
EDGE_VOICES: list[tuple[str, str, str]] = [
("云希(默认)", "zh-CN-YunxiNeural", "中文男声 · 年轻,解说常用"),
("云健", "zh-CN-YunjianNeural", "中文男声 · 浑厚,纪录片感"),
("晓晓", "zh-CN-XiaoxiaoNeural", "中文女声 · 通用"),
("晓伊", "zh-CN-XiaoyiNeural", "中文女声 · 甜"),
("云扬", "zh-CN-YunyangNeural", "中文男声 · 新闻播报"),
("云夏", "zh-CN-YunxiaNeural", "中文男声 · 少年"),
("辽宁小北", "zh-CN-liaoning-XiaobeiNeural", "中文女声 · 东北方言"),
("Guy(英文)", "en-US-GuyNeural", "英文男声"),
("Aria(英文)", "en-US-AriaNeural", "英文女声"),
]
EDGE_DEFAULT_ZH = "zh-CN-YunxiNeural"
EDGE_DEFAULT_EN = "en-US-GuyNeural"
def catalog(provider: str) -> list[tuple[str, str, str]]:
return tts_volcano.VOICES if provider == "volcano" else EDGE_VOICES
def resolve_voice(provider: str, token: str) -> str:
"""名称 / 序号 / 原始 ID 都能接受。返回 "" 表示「用户没指定」。"""
token = (token or "").strip()
if not token:
return ""
if provider == "volcano":
# 交给接口包解析,避免两处规则各写一份、慢慢跑偏
return tts_volcano.resolve_voice_token(token)
cat = catalog(provider)
if token.isdigit() and 1 <= int(token) <= len(cat):
return cat[int(token) - 1][1]
for name, vid, _ in cat:
if token == name or token == vid:
return vid
# 允许带括注的名字,如「云希(默认)」
for name, vid, _ in cat:
if name.startswith(token) or token.startswith(name):
return vid
return token # 自定义 ID 原样返回
def _normalize_rate(provider: str, value) -> object:
"""把语速写成该引擎的表示法:火山 = int 百分比,edge = '+N%'。
new_project.py 默认写 '+8%'(那是 edge 的格式)。用户后来切到火山时若不清洗,
火山项目的 project.json 里就一直留着 edge 字符串 —— 下游换算虽然能兜住,
但 --status 打出来会让人以为配错了引擎。
"""
if value is None or value == "":
return 0 if provider == "volcano" else "+8%"
pct = tts_volcano.parse_rate_percent(value)
if provider == "volcano":
return max(-50, min(100, pct))
return f"{pct:+d}%"
def print_voices(provider: str) -> None:
label = "火山引擎语音合成 2.0" if provider == "volcano" else "edge-tts"
print(f"{label} 候选音色:")
for i, (name, vid, note) in enumerate(catalog(provider), 1):
print(f" {i:>2}. {name:<22} {note}")
print(f" {vid}")
print(" (也可以直接给音色 ID 自定义)")
def load_project(project: str) -> tuple[str, dict]:
proj = os.path.abspath(project or ".")
pj_path = os.path.join(proj, "project.json")
if not os.path.isfile(pj_path):
print(f"✗ 找不到 {pj_path}(--project 要指向视频项目目录)", file=sys.stderr)
raise SystemExit(1)
return proj, json.load(open(pj_path, encoding="utf-8"))
def save_project(proj: str, pj: dict) -> None:
with open(os.path.join(proj, "project.json"), "w", encoding="utf-8") as f:
json.dump(pj, f, ensure_ascii=False, indent=2)
def ensure_env_file(proj: str) -> tuple[str, bool]:
"""返回 (路径, 是否本次新建)。已存在则原样返回,**不读内容**。
优先级与 tts_volcano.env_candidates() 保持一致:项目目录 → 技能目录 → 新建于项目目录。
不按同序查找会出现「向导认技能目录的、接口包认项目目录的」这种静默错位。
"""
project_env = os.path.join(proj, ENV_FILENAME)
if os.path.isfile(project_env):
return project_env, False
skill_env = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ENV_FILENAME)
if os.path.isfile(skill_env):
return skill_env, False
with open(project_env, "w", encoding="utf-8") as f:
f.write(ENV_TEMPLATE)
return project_env, True
def cmd_status(proj: str, pj: dict) -> int:
provider = pj.get("provider") or os.environ.get("TTS_PROVIDER") or "edge"
rate = pj.get("rate")
if rate is None or rate == "":
rate_shown = "(未设置,将用默认)"
elif provider == "volcano":
rate_shown = f"{rate}%(快慢百分比,0 = 原速)"
else:
rate_shown = str(rate)
print(f"当前方案:{provider}")
print(f"当前音色:{pj.get('voice') or '(未设置,将用默认)'}")
print(f"当前语速:{rate_shown}")
if provider == "volcano":
st = tts_volcano.describe_env(proj)
if st["status"] == "ok":
print(f"tts.env :已就绪 {st['env_file']}")
print(f"密钥 :{st['key_masked']}(脱敏显示)· 鉴权 {st['auth']} · 资源 {st['resource_id']}")
else:
print(f"tts.env :未就绪 —— {st['detail']}")
else:
try:
import edge_tts # noqa: F401
print("edge-tts:已安装(免费,无需密钥)")
except ImportError:
print("edge-tts:未安装 → pip install edge-tts")
return 0
def cmd_check(proj: str, provider: str, voice: str) -> int:
print(f"测试方案:{provider}", flush=True) # flush:否则与 stderr 的错误信息交错乱序
if provider == "volcano":
st = tts_volcano.check(proj, voice)
if st["ok"]:
print(f"✓ 火山接口连通 · 音色 {st['voice']} · 音频 {st['audio_bytes']}B · "
f"时间戳 {st['boundaries']} 个 · 计费 {st['chars_billed']} 字")
print(f" tts.env {st['env_file']} · 密钥 {st['key_masked']}")
return 0
print(f"✗ 火山接口未通过({st['stage']})\n{st['detail']}", file=sys.stderr)
return 1
try:
import edge_tts # noqa: F401
except ImportError:
print("✗ edge-tts 未安装:pip install edge-tts", file=sys.stderr)
return 1
if shutil.which("ffmpeg") is None:
try:
import imageio_ffmpeg # noqa: F401
except ImportError:
print("✗ 缺 ffmpeg(edge 引擎裁静音/量时长要用):pip install imageio-ffmpeg",
file=sys.stderr)
return 1
print("✓ edge-tts 与 ffmpeg 就绪(edge 不需要密钥)")
return 0
def interactive_ask(prompt: str, options: list[tuple[str, str]], default_idx: int = 0) -> str:
print(prompt)
for i, (label, note) in enumerate(options, 1):
mark = "(默认)" if i - 1 == default_idx else ""
print(f" {i}. {label}{mark} {note}")
while True:
raw = input(f"选择 [1-{len(options)}],直接回车用默认:").strip()
if not raw:
return options[default_idx][0]
if raw.isdigit() and 1 <= int(raw) <= len(options):
return options[int(raw) - 1][0]
print(" 输入不在范围内,再试一次。")
def run(args) -> int:
proj, pj = load_project(args.project)
provider = (args.provider or os.environ.get("TTS_PROVIDER")
or pj.get("provider") or "").lower()
if args.status:
return cmd_status(proj, pj)
# ---- ① 选方案 ----
if provider not in ("edge", "volcano"):
if not sys.stdin.isatty():
print("当前未设置配音方案。可选:", file=sys.stderr)
print(" --provider edge 免费、免密钥、开箱可用(默认)", file=sys.stderr)
print(" --provider volcano 火山引擎语音合成 2.0,音质更好,需 API Key", file=sys.stderr)
print("NEXT_ACTION=ask_user_provider")
return 1
choice = interactive_ask(
"配音用哪个 TTS?",
[("edge-tts", "免费、免密钥、开箱可用"),
("火山引擎语音合成 2.0", "音质更好、需 API Key(约 1 分钟配置)")])
provider = "edge" if choice.startswith("edge") else "volcano"
# ★ 方案一旦确定立刻落盘。用户填 key 会中断流程,若不落盘,
# 回来跑 --check 时 project.json 还写着 edge,会静默去测 edge(踩过这个坑)。
if pj.get("provider") != provider:
pj["provider"] = provider
save_project(proj, pj)
# ---- ② 密钥(仅火山) ----
if provider == "volcano":
env_path, created = ensure_env_file(proj)
if created:
print(f"✓ 已生成密钥模板:{env_path}")
print(" (该文件已被 .gitignore 忽略,不会进仓库;我也不会读它的内容)")
print()
print("请手动完成两步:")
print(" 1. 打开上面这个文件,把 VOLC_API_KEY= 后面填上你的 API Key")
print(" API Key 获取:https://console.volcengine.com/speech/new/setting/apikeys")
print(" 2. 存盘后回来告诉我「填好了」,我再测连接并继续")
print()
print("NEXT_ACTION=fill_env")
return 2
voice_for_check = resolve_voice("volcano", args.voice or pj.get("voice", ""))
st = tts_volcano.check(proj, voice_for_check)
if not st["ok"]:
print(f"✗ 密钥已存在但连接未通过({st['stage']}):\n{st['detail']}", file=sys.stderr)
print(f" tts.env:{env_path}", file=sys.stderr)
return 1
print(f"✓ 连接正常 · 音色 {st['voice']} · 密钥 {st['key_masked']}")
# ---- ③ 选音色 ----
name_or_id = args.voice or ""
if not name_or_id:
if not sys.stdin.isatty():
default = tts_volcano.DEFAULT_VOICE if provider == "volcano" else EDGE_DEFAULT_ZH
print(f"未指定音色,使用默认 {default}")
print("NEXT_ACTION=ask_user_voice")
name_or_id = default
else:
print_voices(provider)
name_or_id = input("音色(填序号 / 名称 / 直接给 ID,回车用默认):").strip() or ""
default_voice = tts_volcano.DEFAULT_VOICE if provider == "volcano" else EDGE_DEFAULT_ZH
voice = resolve_voice(provider, name_or_id) or default_voice
# ---- 写回 project.json ----
pj["provider"] = provider
pj["voice"] = voice
pj["rate"] = _normalize_rate(provider, args.rate or pj.get("rate"))
save_project(proj, pj)
print()
print(f"✓ 已写入 {os.path.join(proj, 'project.json')}")
rate_shown = f"{pj['rate']}%(快慢百分比)" if provider == "volcano" else pj["rate"]
print(f" 方案 {provider} · 音色 {voice} · 语速 {rate_shown}")
if provider == "volcano":
print(" 下一步:python <skill>/scripts/tts_build.py --project .")
else:
print(" 下一步:python <skill>/scripts/tts_build.py --project .")
print("NEXT_ACTION=done")
return 0
def main() -> int:
p = argparse.ArgumentParser(description="html-explainer 配音方案设置向导")
p.add_argument("--project", default=".")
p.add_argument("--provider", default="", choices=["", "edge", "volcano"])
p.add_argument("--voice", default="")
p.add_argument("--rate", default="")
p.add_argument("--status", action="store_true", help="显示当前配置与脱敏密钥状态")
p.add_argument("--check", action="store_true", help="只测连接")
p.add_argument("--list-voices", action="store_true", help="列出候选音色")
p.add_argument("--env-file", default="", help="显式指定 tts.env(仅本次)")
args = p.parse_args()
if args.env_file:
os.environ["TTS_ENV_FILE"] = args.env_file
proj, pj = load_project(args.project)
provider = (args.provider or os.environ.get("TTS_PROVIDER")
or pj.get("provider") or "edge").lower()
if args.list_voices:
print_voices(provider)
return 0
if args.check:
return cmd_check(proj, provider, resolve_voice(provider, args.voice))
return run(args)
if __name__ == "__main__":
raise SystemExit(main())
+598
View File
@@ -0,0 +1,598 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
tts_volcano.py —— 火山引擎(豆包)语音合成 2.0 接口包。
【为什么单独成一个包】密钥只在这一个文件里被读取,调用方(含 AI agent)
只传「文本 + 音色 + 语速」,永远拿不到也不需要密钥值。本包保证:
★ 不打印、不记录、不回显密钥;异常信息一律脱敏(见 _redact)。
★ 不把密钥写进任何文件;只读 tts.env,从不写 tts.env。
★ 上游返回体里也没有密钥,缓存与 manifest 里也不会出现密钥。
→ agent 只需要会调用本包,不需要(也不应该)读 tts.env。
【接口】与 tts_build.py 的 edge 引擎同构,下游 timing 逻辑零改动:
synthesize(text, voice, rate=0, ...) -> {
"audio": bytes, # mp3 字节
"boundaries": [{"text","offset_sec","duration_sec"}], # 原始坐标系
"usage": {"text_words": int} | {},
"requests": int, # 实际请求数(长文本会切分)
}
★ boundaries 的时间单位是**秒**,坐标系 = 返回音频的起点,
与 edge-tts 的 WordBoundary 语义一致(下游 subs.py 按 offset_sec - lead_cut 平移)。
【接口规格】(2026-09 核对官方文档)
POST https://openspeech.bytedance.com/api/v3/tts/unidirectional
Header: X-Api-Key(必选)/ X-Api-Resource-Id(必选,TTS2.0 = seed-tts-2.0)
/ X-Api-Request-Id(必选,uuid)/ Content-Type: application/json
Body: {user:{uid}, req_params:{text, speaker, audio_params:{format,sample_rate,bit_rate,speech_rate}}}
响应: NDJSON,逐行 JSON;音频块 code=0 + data(base64);
**sentence.words[] 直接给字级时间戳**(word / startTime / endTime / confidence,单位秒);
结束标记 code=20000000 且 data=null(★ 不是 code=0,误判会丢最后一段音频)。
【本机注意】openspeech.bytedance.com 是境内端点,默认**绕过系统代理**直连
(本机常驻代理会让境内域名绕远甚至失败)。要强制走代理时设 VOLC_PROXY。
用法:
python tts_volcano.py --check # 连通性自检(用一句话试合成)
python tts_volcano.py --voices # 列出内置常用音色
python tts_volcano.py --synth "测试文本" out.mp3 # 手动合成一段
python tts_volcano.py --env-file <路径> ... # 指定 tts.env(默认自动探测)
被 tts_build.py 作为库调用(provider=volcano 时)。
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
import uuid
ENDPOINT = "https://openspeech.bytedance.com/api/v3/tts/unidirectional"
DEFAULT_RESOURCE_ID = "seed-tts-2.0" # TTS 2.0 预置音色
CLONE_RESOURCE_ID = "seed-icl-2.0" # 声音复刻音色
END_CODE = 20000000 # ★ 结束标记;不是 0
MAX_TEXT_LEN = 200 # 单请求文本上限(超了按标点切分)
DEFAULT_TIMEOUT = 60.0
DEFAULT_RETRIES = 3
RETRY_SLEEP = 1.0
# 音色 ID 末尾都是 _uranus_bigtts = 豆包语音合成模型 2.0
DEFAULT_VOICE = "zh_male_m191_uranus_bigtts" # 云舟 2.0
ENV_FILENAME = "tts.env"
# 内置常用音色(名称, voice_type, 说明)。完整列表见官方音色文档。
VOICES: list[tuple[str, str, str]] = [
# —— 男声:解说 / 通用 ——
("云舟 2.0", "zh_male_m191_uranus_bigtts", "男声 · 沉稳通用(默认,最稳)"),
("温暖阿虎 2.0", "zh_male_wennuanahu_uranus_bigtts", "男声 · 温暖"),
("解说小明 2.0", "zh_male_jieshuoxiaoming_uranus_bigtts", "男声 · 解说"),
("磁性解说男声 2.0", "zh_male_cixingjieshuonan_uranus_bigtts", "男声 · 磁性解说"),
("悬疑解说 2.0", "zh_male_xuanyijieshuo_uranus_bigtts", "男声 · 悬疑叙事"),
("广告解说 2.0", "zh_male_guanggaojieshuo_uranus_bigtts", "男声 · 广告调"),
("儒雅青年 2.0", "zh_male_ruyaqingnian_uranus_bigtts", "男声 · 青年儒雅"),
("少年梓辛 2.0", "zh_male_shaonianzixin_uranus_bigtts", "男声 · 少年"),
# —— 女声 ——
("小何 2.0", "zh_female_xiaohe_uranus_bigtts", "女声 · 通用(默认女声)"),
("Vivi 2.0", "zh_female_vv_uranus_bigtts", "女声 · 活泼通用"),
("知性灿灿 2.0", "zh_female_cancan_uranus_bigtts", "女声 · 知性"),
("甜美桃子 2.0", "zh_female_tianmeitaozi_uranus_bigtts", "女声 · 甜美"),
("邻家女孩 2.0", "zh_female_linjianvhai_uranus_bigtts", "女声 · 邻家"),
("温柔淑女 2.0", "zh_female_wenroushunv_uranus_bigtts", "女声 · 温柔"),
("深夜播客 2.0", "zh_male_shenyeboke_uranus_bigtts", "男声 · 深夜播客/低语"),
# —— 英文 ——
("Tim", "en_male_tim_uranus_bigtts", "美式英语 · 男声"),
("Dacey", "en_female_dacey_uranus_bigtts", "美式英语 · 女声"),
]
VOICE_BY_NAME = {name: vid for name, vid, _ in VOICES}
VOICE_IDS = {vid for _, vid, _ in VOICES}
def resolve_voice_token(token: str, default: str = "") -> str:
"""把「名称 / 序号 / 原始 ID」统一成音色 ID。空 token → default。
必需:speaker 字段只认 ID。直接传中文显示名会被服务端拒掉,
报的还是 `55000000 resource ID is mismatched with speaker related resource`
这种看不出是「名字没解析」的错误。
"""
token = (token or "").strip()
if not token:
return default or DEFAULT_VOICE
if token.isdigit() and 1 <= int(token) <= len(VOICES):
return VOICES[int(token) - 1][1]
if token in VOICE_BY_NAME:
return VOICE_BY_NAME[token]
if token in VOICE_IDS:
return token
for name, vid, _ in VOICES: # 允许带括注的名字,如「云舟 2.0(默认)」
if name.startswith(token) or token.startswith(name):
return vid
return token # 自定义 ID(如复刻音色)原样返回
def parse_rate_percent(value) -> int:
"""把各种语速写法统一成整数百分比:'+8%' / '8' / 0.08 / 8 → 8。
火山 `speech_rate` 与 edge `rate` 语义相同(都表示「快百分之几」),
只是表示法不同:火山要 int,edge 要 '+N%' 字符串。
两个下游脚本共用这一个换算,避免各写一份慢慢跑偏。
"""
if isinstance(value, bool) or value is None or value == "":
return 0
if isinstance(value, (int, float)):
v = float(value)
return int(round(v * 100 if abs(v) <= 1.5 else v))
m = re.search(r"([-+]?\d+(?:\.\d+)?)", str(value))
if not m:
return 0
v = float(m.group(1))
if "%" in str(value):
return int(round(v))
return int(round(v * 100 if abs(v) <= 1.5 else v))
# 形如 AKLT... 的火山 AccessKey —— 只认这种明确形态,避免把普通 request-id 也抹掉
_SECRET_PAT = re.compile(r"AKLT[A-Za-z0-9_\-]{6,}")
class VolcanoTTSConfigError(RuntimeError):
"""缺 tts.env 或字段不全 —— 由 tts_setup.py 负责引导用户补齐。"""
class VolcanoTTSError(RuntimeError):
"""调用失败(已脱敏,可安全打印)。
retryable=False 表示**重试没有意义**(密钥错、音色 ID 错、参数错这类 4xx)。
退避重试只对网络抖动、超时、5xx、429 有意义 —— 对 401 重试三次只是白等两轮。
"""
def __init__(self, message: str, retryable: bool = True):
super().__init__(message)
self.retryable = retryable
# ---------------------------------------------------------------- 密钥读取
def _skill_root() -> str:
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def env_candidates(project_dir: str = "") -> list[str]:
"""tts.env 探测顺序:显式指定 → 项目目录 → 技能目录 → 用户目录。"""
out: list[str] = []
explicit = os.environ.get("TTS_ENV_FILE", "").strip()
if explicit:
out.append(explicit)
if project_dir:
out.append(os.path.join(os.path.abspath(project_dir), ENV_FILENAME))
out.append(os.path.join(_skill_root(), ENV_FILENAME))
out.append(os.path.join(os.path.expanduser("~"), ".workbuddy", ENV_FILENAME))
seen, uniq = set(), []
for p in out:
if p not in seen:
seen.add(p)
uniq.append(p)
return uniq
def find_env_file(project_dir: str = "") -> str:
for p in env_candidates(project_dir):
if os.path.isfile(p):
return p
return ""
def parse_env(path: str) -> dict[str, str]:
"""极简 KEY=VALUE 解析(与 python-dotenv 兼容的最小子集),不执行任何代码。"""
vals: dict[str, str] = {}
with open(path, encoding="utf-8-sig") as f:
for raw in f:
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
k, v = k.strip(), v.strip()
if len(v) >= 2 and v[0] == v[-1] and v[0] in "\"'":
v = v[1:-1]
vals[k] = v
return vals
def load_credentials(project_dir: str = "") -> dict[str, str]:
"""读取密钥。返回的 dict 里含密钥,**调用方不得打印或落盘**。"""
path = find_env_file(project_dir)
if not path:
raise VolcanoTTSConfigError(
"没有找到 tts.env。跑到项目根目录执行:python <skill>/scripts/tts_setup.py\n"
"(它会生成一份带注释的空模板,你手工填入 API Key 即可。)")
vals = parse_env(path)
api_key = vals.get("VOLC_API_KEY", "").strip()
app_id = vals.get("VOLC_APP_ID", "").strip()
access_key = vals.get("VOLC_ACCESS_KEY", "").strip()
if not api_key and not (app_id and access_key):
raise VolcanoTTSConfigError(
f"{path} 里还没有填密钥。请打开该文件填入 VOLC_API_KEY 后告诉我,我来测连接。\n"
" API Key 获取:火山引擎控制台 → 语音技术 → API Key 管理\n"
" https://console.volcengine.com/speech/new/setting/apikeys")
return {
"path": path,
"api_key": api_key,
"app_id": app_id,
"access_key": access_key,
"resource_id": vals.get("VOLC_RESOURCE_ID", "").strip() or DEFAULT_RESOURCE_ID,
}
def describe_env(project_dir: str = "") -> dict[str, str]:
"""给 agent/用户看的**脱敏**状态摘要 —— 这个函数的输出可以安全打印。"""
try:
c = load_credentials(project_dir)
except VolcanoTTSConfigError as e:
return {"status": "missing", "detail": str(e).splitlines()[0]}
mode = "X-Api-Key" if c["api_key"] else "X-Api-App-Id + X-Api-Access-Key"
return {
"status": "ok",
"env_file": c["path"],
"auth": mode,
"key_masked": _mask(c["api_key"] or c["access_key"]),
"resource_id": c["resource_id"],
}
def _mask(secret: str) -> str:
if not secret:
return "(未设置)"
if len(secret) <= 8:
return "*" * len(secret)
return f"{secret[:4]}{'*' * 6}{secret[-4:]}(长度 {len(secret)})"
def _redact(text: str) -> str:
"""把可能混进错误信息的密钥抹掉。异常与日志一律过这一层。
只做两件事:抹掉**本次实际加载的**密钥值,以及 AKLT 形态的 token。
刻意不用「长字符串就当密钥」这种宽规则 —— 那会把 reqid、logid 一起抹掉,
排查问题时就看不到定位线索了(而密钥只可能从下面这两个来源进入输出)。
"""
try:
c = load_credentials("")
for k in ("api_key", "access_key"):
v = c.get(k) or ""
if len(v) >= 8 and v in text:
text = text.replace(v, _mask(v))
except Exception:
pass
return _SECRET_PAT.sub(lambda m: _mask(m.group(0)), text)
# ---------------------------------------------------------------- 请求构造
def _headers(cred: dict[str, str]) -> dict[str, str]:
h = {
"Content-Type": "application/json",
"Accept-Encoding": "identity", # 不要 gzip,NDJSON 逐行读
"X-Api-Resource-Id": cred["resource_id"],
"X-Api-Request-Id": str(uuid.uuid4()),
# 不加这个头,响应里就没有 usage.text_words(计费字数恒为 0)
"X-Control-Require-Usage-Tokens-Return": "*",
}
if cred["api_key"]:
h["X-Api-Key"] = cred["api_key"]
else: # 旧版鉴权兜底
h["X-Api-App-Id"] = cred["app_id"]
h["X-Api-Access-Key"] = cred["access_key"]
return h
def _opener() -> urllib.request.OpenerDirector:
"""默认不走代理:境内端点直连更快更稳;需要时用 VOLC_PROXY 显式指定。"""
proxy = os.environ.get("VOLC_PROXY", "").strip()
if proxy:
return urllib.request.build_opener(
urllib.request.ProxyHandler({"http": proxy, "https": proxy}))
return urllib.request.build_opener(urllib.request.ProxyHandler({}))
def split_long_text(text: str, limit: int = MAX_TEXT_LEN) -> list[str]:
"""按标点切分到 <=limit 字,保证不把一个词从中间劈开。"""
if len(text) <= limit:
return [text]
parts = re.split(r"(?<=[。!?;,、,.!?;::\n])", text)
out: list[str] = []
for p in parts:
if not p:
continue
if len(p) <= limit and (not out or len(out[-1]) + len(p) > limit):
out.append(p)
elif len(p) <= limit:
out[-1] += p
else: # 单句仍超长 → 硬切
for i in range(0, len(p), limit):
out.append(p[i:i + limit])
return [s for s in out if s.strip()] or [text]
def _one_request(text: str, voice: str, cred: dict[str, str], rate: int,
fmt: str, sample_rate: int, bit_rate: int,
timeout: float, enable_subtitle: bool = True,
context_texts: list[str] | None = None,
uid: str = "") -> tuple[bytes, list[dict], dict]:
"""单次请求 → (mp3 字节, boundaries, usage)。"""
audio_params: dict = {"format": fmt, "sample_rate": sample_rate}
if fmt == "mp3" and bit_rate:
audio_params["bit_rate"] = bit_rate
# ★ 不传这个开关,响应里的 sentence.words 恒为空数组 —— 字级时间戳拿不到。
# 实测(2026-09-23):同文本 false→words=0,true→words=10。
# 官方文档示例里默认带词时间戳,容易让人以为「不用开」,实际必须显式打开。
# 仅豆包语音合成大模型 2.0 音色、仅中英文语种支持。
if enable_subtitle:
audio_params["enable_subtitle"] = True
if rate:
audio_params["speech_rate"] = int(rate) # [-50,100] → 0.5x~2.0x
req_params: dict = {"text": text, "speaker": voice, "audio_params": audio_params}
if context_texts:
req_params["context_texts"] = list(context_texts)
body = {"user": {"uid": uid or f"html-explainer-{uuid.uuid4().hex[:12]}"},
"req_params": req_params}
req = urllib.request.Request(
ENDPOINT, data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
headers=_headers(cred), method="POST")
chunks: list[bytes] = []
boundaries: list[dict] = []
usage: dict = {}
try:
with _opener().open(req, timeout=timeout) as resp:
for raw in resp: # NDJSON:逐行
line = raw.strip()
if not line:
continue
try:
obj = json.loads(line.decode("utf-8"))
except Exception:
continue
code = obj.get("code")
data = obj.get("data")
# ★ usage 挂在**结束行**上(code=20000000、data=null 那一行)。
# 先判结束标记再读 usage,会把计费字数永远读成 0(踩过)。
if obj.get("usage"):
usage = obj["usage"]
if code == END_CODE: # ★ 结束标记,且不含音频
continue
if code not in (0, None):
raise VolcanoTTSError(
f"接口返回 code={code} message={obj.get('message')!r}")
if data:
try:
chunks.append(base64.b64decode(data))
except Exception:
raise VolcanoTTSError("音频块 base64 解码失败(流可能被截断)")
sent = obj.get("sentence") or {}
for w in (sent.get("words") or []):
txt = (w.get("word") or "").strip()
if not txt:
continue
st = float(w.get("startTime") or 0.0)
et = float(w.get("endTime") or st)
boundaries.append({
"text": txt,
"offset_sec": round(st, 4),
"duration_sec": round(max(0.0, et - st), 4),
})
except urllib.error.HTTPError as e:
detail = ""
try:
detail = e.read().decode("utf-8", "replace")[:400]
except Exception:
pass
hint = ""
# 4xx 基本都是永久性错误(密钥/音色/参数),重试无益;429 与 5xx 才值得重试
retryable = e.code == 429 or e.code >= 500
if e.code in (401, 403):
hint = "\n → 鉴权失败:核对 tts.env 里的 VOLC_API_KEY 是否正确、是否已开通该服务。"
elif e.code == 404:
hint = "\n → 音色或资源 ID 不对:预置音色要配 seed-tts-2.0。"
raise VolcanoTTSError(
f"HTTP {e.code} {e.reason}{hint}\n {_redact(detail)}", retryable=retryable)
except urllib.error.URLError as e:
raise VolcanoTTSError(
f"网络不可达:{_redact(str(e.reason))}\n"
" → 本包默认绕过系统代理直连。若你的网络确实需要代理,设 VOLC_PROXY=http://127.0.0.1:<port>。")
except TimeoutError:
raise VolcanoTTSError(f"请求超时({timeout}s)")
if not chunks:
raise VolcanoTTSError("没有收到任何音频块(检查音色 ID 与文本是否为空)")
return b"".join(chunks), boundaries, usage
def _looks_like_mp3(b: bytes) -> bool:
if len(b) < 4:
return False
head = b[:4].hex()
return head.startswith("494433") or head.startswith("fff3") or head.startswith("fffb") \
or head.startswith("fffa") or head.startswith("fff2")
def synthesize(text: str, voice: str = DEFAULT_VOICE, rate: int = 0,
fmt: str = "mp3", sample_rate: int = 24000, bit_rate: int = 160000,
timeout: float = DEFAULT_TIMEOUT, retries: int = DEFAULT_RETRIES,
context_texts: list[str] | None = None,
project_dir: str = "",
enable_subtitle: bool = True) -> dict:
"""合成一段文本 → {audio, boundaries, usage, requests}。失败抛 VolcanoTTSError。
enable_subtitle=True(默认)会要求服务端返回字级时间戳。关掉只在
「音色不支持字幕(非中英文)」时有意义 —— 那种情况下 words 必为空。
"""
text = (text or "").strip()
if not text:
raise VolcanoTTSError("文本为空")
cred = load_credentials(project_dir)
last: Exception | None = None
for attempt in range(max(1, retries)):
try:
pieces = split_long_text(text)
audio_all: list[bytes] = []
bounds_all: list[dict] = []
usage: dict = {}
# 长文本切分时,后续片段的词边界要按前面片段的实际时长平移
for i, piece in enumerate(pieces):
audio, bounds, u = _one_request(
piece, voice, cred, rate, fmt, sample_rate, bit_rate,
timeout, enable_subtitle, context_texts)
if i:
offset = _bytes_duration(b"".join(audio_all), fmt)
bounds = [{**b, "offset_sec": round(b["offset_sec"] + offset, 4)}
for b in bounds]
audio_all.append(audio)
bounds_all.extend(bounds)
if u:
usage = u
audio = b"".join(audio_all)
if not _looks_like_mp3(audio) and fmt == "mp3":
raise VolcanoTTSError("返回的音频不是有效 MP3 头(可能被截断)")
result = {"audio": audio, "boundaries": bounds_all,
"usage": usage, "requests": len(pieces)}
# ★ 要了字幕却一个时间戳都没回来 —— 大概率踩到「音色不支持字幕」。
# 静默放行会让下游字幕整段塌掉,所以这里显式记一条可打印的警告。
if enable_subtitle and not bounds_all:
result["warning"] = (
"未返回字级时间戳(sentence.words 为空)。"
"该能力仅豆包语音合成大模型 2.0 的中英文音色支持;"
"当前音色若不支持,字幕将退化为按整段均分——建议换成 zh_*_uranus_bigtts。")
return result
except VolcanoTTSConfigError:
raise
except VolcanoTTSError as e:
if not e.retryable:
raise # 密钥/参数错,重试是白等
last = e
if attempt + 1 < max(1, retries):
time.sleep(RETRY_SLEEP * (attempt + 1))
except Exception as e: # noqa: BLE001
last = e
if attempt + 1 < max(1, retries):
time.sleep(RETRY_SLEEP * (attempt + 1))
raise VolcanoTTSError(_redact(f"连续 {retries} 次失败:{type(last).__name__}: {last}"))
def _bytes_duration(data: bytes, fmt: str) -> float:
"""用 ffmpeg 量已合成字节的时长(长文本切片拼接时算偏移用)。"""
import subprocess
import tempfile
from shutil import which
ff = which("ffmpeg")
if not ff:
try:
import imageio_ffmpeg
ff = imageio_ffmpeg.get_ffmpeg_exe()
except Exception:
return 0.0
fd, tmp = tempfile.mkstemp(suffix=f".{fmt}")
try:
with os.fdopen(fd, "wb") as f:
f.write(data)
r = subprocess.run([ff, "-hide_banner", "-i", tmp],
capture_output=True, text=True, errors="replace")
for line in r.stderr.splitlines():
if "Duration:" in line:
t = line.split("Duration:")[1].split(",")[0].strip()
h, m, s = t.split(":")
return round(int(h) * 3600 + int(m) * 60 + float(s), 3)
finally:
try:
os.remove(tmp)
except OSError:
pass
return 0.0
def check(project_dir: str = "", voice: str = "", timeout: float = 30.0) -> dict:
"""连通性自检:用一句短文本试合成。返回可安全打印的状态 dict。"""
try:
cred = load_credentials(project_dir)
except VolcanoTTSConfigError as e:
return {"ok": False, "stage": "env", "detail": str(e)}
v = resolve_voice_token(voice)
try:
r = synthesize("连接测试,这句话只用来验证接口是否可用。",
voice=v, timeout=timeout, retries=1, project_dir=project_dir)
return {
"ok": True, "stage": "api", "voice": v,
"env_file": cred["path"],
"key_masked": _mask(cred["api_key"] or cred["access_key"]),
"resource_id": cred["resource_id"],
"audio_bytes": len(r["audio"]),
"boundaries": len(r["boundaries"]),
"chars_billed": (r.get("usage") or {}).get("text_words", 0),
}
except VolcanoTTSError as e:
return {"ok": False, "stage": "api", "voice": v, "env_file": cred["path"],
"detail": str(e)}
# ---------------------------------------------------------------- CLI
def _print_voices() -> None:
print("内置常用音色(完整列表见官方音色文档):")
for i, (name, vid, note) in enumerate(VOICES, 1):
print(f" {i:>2}. {name:<18} {note}\n {vid}")
def main() -> int:
p = argparse.ArgumentParser(description="火山引擎语音合成 2.0 接口包")
p.add_argument("--project", default="", help="项目目录(用于探测 tts.env)")
p.add_argument("--env-file", default="", help="显式指定 tts.env 路径")
p.add_argument("--check", action="store_true", help="连通性自检")
p.add_argument("--voices", action="store_true", help="列出内置音色")
p.add_argument("--synth", default="", help="合成一段文本")
p.add_argument("--out", default="volcano_tts_test.mp3")
p.add_argument("--voice", default="")
p.add_argument("--rate", type=int, default=0)
args = p.parse_args()
if args.env_file:
os.environ["TTS_ENV_FILE"] = args.env_file
if args.voices:
_print_voices()
return 0
if args.check:
st = check(args.project, args.voice)
if st["ok"]:
print(f"✓ 连接正常 · 音色 {st['voice']} · 音频 {st['audio_bytes']}B · "
f"时间戳 {st['boundaries']} 个 · 计费 {st['chars_billed']} 字")
print(f" tts.env: {st['env_file']}")
print(f" 密钥: {st['key_masked']} · 资源 {st['resource_id']}")
return 0
print(f"✗ 未通过({st['stage']})\n{st['detail']}", file=sys.stderr)
return 1
if args.synth:
try:
r = synthesize(args.synth, voice=resolve_voice_token(args.voice),
rate=args.rate, project_dir=args.project)
except (VolcanoTTSError, VolcanoTTSConfigError) as e:
print(f"✗ {e}", file=sys.stderr)
return 1
with open(args.out, "wb") as f:
f.write(r["audio"])
print(f"✓ {args.out} · {len(r['audio'])}B · 时间戳 {len(r['boundaries'])} 个 "
f"· 请求 {r['requests']} 次")
if r.get("warning"):
print(f"⚠ {r['warning']}", file=sys.stderr)
return 0
p.print_help()
return 0
if __name__ == "__main__":
raise SystemExit(main())