MohZhuang
d158afcd36
修复 - tts_build.py 走火山引擎时，project.json 里的中文音色名没被解析成音色 ID， 合成恒失败并返回 code=55000000（报错完全看不出是「名字没解析」）。 双层兜底：tts_volcano.synthesize() 入口解析 + tts_build.py 读配置后解析。 新增 - render_video.mjs：`--png-fast`（CDP optimizeForSpeed，逐像素无损，4.4×）、 `--jpeg-quality N`、`--crf N`、`--preset <名>`。最后一对把「帧容器」与 「成片码率」解耦（旧 --jpeg 会连带降 crf23/veryfast，那是草稿语义）。 默认行为完全不变。 - qc_check.py：新增实测视频流帧数校验。原先只比容器时长，而容器时长会被音轨撑起 来，帧序列缺口会导致视频流被截断却仍显示完整长度（假通过，见 lessons #57）。 文档 - lessons.md 新增 #62–#69：信息层色值契约与绘制层排查、语速标定绑定引擎与音色、 以及「PNG 编码才是瓶颈、且对并发完全免疫」的完整实测与判定路径。 - SKILL.md / references/workflow-guide.md：补截图模式选用判据（帧里有满幅照片就换模式）。 - CHANGELOG 1.3.1 段补齐 + 版本徽章同步。
406 lines
18 KiB
Python
406 lines
18 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
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 二次实战），核心五件事：
|
||
① RATE 默认 +8%（长片解说密度；短视频高密度场景可加到 +20%~+37%，
|
||
标定法：同一句话跨 RATE 对比，dur×(1+rate) 应为常数 → 线性可外推）
|
||
② 每次合成 asyncio.wait_for 硬超时 45s —— edge-tts 的 websocket 会
|
||
偶发**永久挂起**（不抛错不返回零日志），必须把挂死变成可重试失败
|
||
③ 指数退避重试 4 次（微软端点连续请求会限流抛 NoAudioReceived）
|
||
④ 句间 0.8s 间隔 + 磁盘缓存（sig 含 engine/boundary 与 trim —— 参数变了旧缓存自动失效）
|
||
⑤ 默认裁首尾静音（首回退 0.03s 保爆破音、尾保留 0.12s 保收音），
|
||
裁完段间停顿才由 GAP 显式控制
|
||
|
||
两个 edge-tts 7.x 的坑（已处理，别删）：
|
||
1) boundary 默认 'SentenceBoundary' → 整句一个事件，字幕节拍退化成
|
||
按字数插值。必须显式传 boundary='WordBoundary'。
|
||
2) 相邻数字熔成大数（「沪深300」「4479.55点」→「三百万四千…」），
|
||
数字与数字之间补中文逗号。
|
||
※ 数字补逗号只在 edge 引擎做：它会改动送进去的文本，从而影响词边界与解说词的
|
||
逐字对齐。火山引擎不熔读，原样送原文（对齐最稳）。
|
||
|
||
时长两级口径（manifest 同时写两个，下游各取所需）：
|
||
container_duration_sec 裁完后 MP3 容器时长 → **帧时长用这个**（音画同步）
|
||
speech_end_sec 词边界末尾（裁后坐标系）→ **末块字幕收尾用这个**
|
||
duration_sec 词边界末尾（原始坐标系）→ 仅排查用
|
||
|
||
用法（项目目录内）：
|
||
python tts_build.py --project . [--provider edge|volcano] [--rate +8%]
|
||
[--voice zh-CN-YunxiNeural] [--no-trim]
|
||
产出：
|
||
audio/<id>.mp3 + audio-manifest.json
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import hashlib
|
||
import inspect
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
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%"
|
||
DEFAULT_VOLUME = "+0%"
|
||
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
|
||
TRIM_THRESHOLD = 0.004
|
||
|
||
_NUMERIC_RUN = re.compile(r"[0-9]+(?:\.[0-9]+)?")
|
||
|
||
|
||
def sanitize_for_tts(text: str) -> str:
|
||
"""相邻数字之间补中文逗号，防熔读（4479.55 这种完整数字不会被拆）。"""
|
||
tokens, last = [], 0
|
||
for m in _NUMERIC_RUN.finditer(text):
|
||
tokens.append(("text", text[last:m.start()]))
|
||
tokens.append(("num", m.group(0)))
|
||
last = m.end()
|
||
tokens.append(("text", text[last:]))
|
||
|
||
out: list[str] = []
|
||
prev_kind = None
|
||
for kind, val in tokens:
|
||
if kind == "num" and prev_kind == "num":
|
||
if out and not out[-1].endswith(("，", ",", "。")):
|
||
out.append("，")
|
||
out.append(val)
|
||
prev_kind = kind
|
||
return "".join(out)
|
||
|
||
|
||
def _boundary_kwarg() -> str:
|
||
try:
|
||
p = inspect.signature(edge_tts.Communicate.__init__).parameters
|
||
return "WordBoundary" if "boundary" in p else "n/a"
|
||
except Exception:
|
||
return "n/a"
|
||
|
||
|
||
def ffmpeg_exe() -> str:
|
||
from shutil import which
|
||
p = which("ffmpeg")
|
||
if p:
|
||
return p
|
||
import imageio_ffmpeg
|
||
return imageio_ffmpeg.get_ffmpeg_exe()
|
||
|
||
|
||
def container_duration(path: str) -> float:
|
||
r = subprocess.run([ffmpeg_exe(), "-hide_banner", "-i", path],
|
||
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)
|
||
return 0.0
|
||
|
||
|
||
def decode_pcm(path: str, sr: int = 48000):
|
||
import numpy as np
|
||
r = subprocess.run([ffmpeg_exe(), "-v", "error", "-i", path, "-f", "f32le",
|
||
"-ac", "1", "-ar", str(sr), "-"], capture_output=True)
|
||
return np.frombuffer(r.stdout, dtype=np.float32)
|
||
|
||
|
||
def speech_window(path: str, sr: int = 48000):
|
||
import numpy as np
|
||
x = decode_pcm(path, sr)
|
||
if x.size == 0:
|
||
return 0.0, 0.0
|
||
idx = np.where(np.abs(x) > TRIM_THRESHOLD)[0]
|
||
if idx.size == 0:
|
||
return 0.0, 0.0
|
||
return round(float(idx[0]) / sr, 3), round(float(idx[-1]) / sr, 3)
|
||
|
||
|
||
def trim_edges(path: str, sr: int = 48000) -> float:
|
||
"""原地裁首尾静音，返回裁掉的首部长度（秒）。"""
|
||
import numpy as np
|
||
x = decode_pcm(path, sr)
|
||
if x.size == 0:
|
||
return 0.0
|
||
idx = np.where(np.abs(x) > TRIM_THRESHOLD)[0]
|
||
if idx.size == 0:
|
||
return 0.0
|
||
a = max(0, int(idx[0]) - int(TRIM_HEAD_BACKOFF * sr))
|
||
b = min(len(x), int(idx[-1]) + int(TRIM_TAIL_KEEP * sr))
|
||
if b <= a:
|
||
return 0.0
|
||
y = x[a:b]
|
||
tmp = path + ".trim.mp3"
|
||
cmd = [ffmpeg_exe(), "-hide_banner", "-loglevel", "error", "-y",
|
||
"-f", "f32le", "-ac", "1", "-ar", str(sr), "-i", "-",
|
||
"-c:a", "libmp3lame", "-b:a", "128k", tmp]
|
||
r = subprocess.run(cmd, input=y.astype("<f4").tobytes(), capture_output=True)
|
||
if r.returncode != 0 or not os.path.exists(tmp):
|
||
print(f" ⚠ trim 失败，保留原文件：{(r.stderr or b'')[-200:]}", file=sys.stderr)
|
||
return 0.0
|
||
os.replace(tmp, path)
|
||
return round(a / sr, 3)
|
||
|
||
|
||
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)
|
||
return os.path.join(d, h + ".mp3"), os.path.join(d, h + ".json")
|
||
|
||
|
||
async def synth(text, out_path, voice, rate, volume=DEFAULT_VOLUME, pitch=DEFAULT_PITCH,
|
||
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)
|
||
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, provider=provider)
|
||
if os.path.exists(cache_mp3) and os.path.exists(cache_js):
|
||
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, head_cut, trim)
|
||
|
||
async def _once_edge():
|
||
kw = {}
|
||
b = _boundary_kwarg()
|
||
if b != "n/a":
|
||
kw["boundary"] = b
|
||
comm = edge_tts.Communicate(spoken, voice, rate=rate, volume=volume, pitch=pitch, **kw)
|
||
audio = bytearray()
|
||
bounds: list[dict] = []
|
||
async for chunk in comm.stream():
|
||
if chunk["type"] == "audio":
|
||
audio.extend(chunk["data"])
|
||
elif chunk["type"] == "WordBoundary":
|
||
bounds.append({
|
||
"text": chunk["text"],
|
||
"offset_sec": round(chunk["offset"] / 10_000_000, 4),
|
||
"duration_sec": round(chunk["duration"] / 10_000_000, 4),
|
||
})
|
||
if not audio:
|
||
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 + 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)
|
||
head_cut = trim_edges(out_path) if trim else 0.0
|
||
if cache_dir:
|
||
with open(cache_mp3, "wb") as f:
|
||
f.write(open(out_path, "rb").read())
|
||
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))
|
||
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:
|
||
cd = container_duration(out_path)
|
||
ds, de = speech_window(out_path) if trimmed else (0.0, cd)
|
||
return {
|
||
"path": os.path.abspath(out_path),
|
||
"bytes": os.path.getsize(out_path) if os.path.exists(out_path) else 0,
|
||
"voice": voice, "rate": rate,
|
||
"spoken_text": spoken, "original_text": text,
|
||
# ★ 词边界是「原始坐标系」；裁剪后请用 lead_cut_sec 平移
|
||
"duration_sec": round(dur, 3),
|
||
"lead_cut_sec": round(head_cut, 3),
|
||
# ★ 帧时长口径：裁后容器时长
|
||
"container_duration_sec": cd,
|
||
# ★ 末块字幕收尾口径：裁后坐标系里的语音真实结束
|
||
# 注意 speech_window(out_path) 量的是**已经裁过**的文件，de 本身就在裁后坐标系里，
|
||
# 绝不能再减 head_cut（那会重复扣一次，症状是「重跑（走缓存）与首跑（现合成）
|
||
# 报出的 speech_end 差一个首裁量」，字幕末块提前消失）。见 lessons.md。
|
||
"speech_start_sec": ds if trimmed else 0.0,
|
||
"speech_end_sec": de if trimmed else round(dur, 3),
|
||
"tail_pad_sec": round(max(0.0, cd - (de if trimmed else dur)), 3),
|
||
"boundary_count": len(bounds),
|
||
"boundaries": bounds,
|
||
}
|
||
|
||
|
||
async def run(args) -> int:
|
||
proj = os.path.abspath(args.project)
|
||
pj = json.load(open(os.path.join(proj, "project.json"), encoding="utf-8"))
|
||
narration = json.load(open(os.path.join(proj, "narration.json"), encoding="utf-8"))
|
||
if isinstance(narration, dict):
|
||
narration = narration.get("items") or []
|
||
|
||
# 引擎优先级：命令行 --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":
|
||
# project.json 里存的是显示名（「解说小明 2.0」），但 speaker 字段只认 ID。
|
||
# 在入口就解析成 ID，后面日志/清单里也就是可直接复制的真 ID。
|
||
voice = tts_volcano.resolve_voice_token(
|
||
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"]
|
||
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 "
|
||
f"语音止 {r['speech_end_sec']:>6.3f}s 首裁 {r['lead_cut_sec']:.3f}s "
|
||
f"bounds={r['boundary_count']}", file=sys.stderr)
|
||
|
||
manifest = {
|
||
"provider": provider, "voice": voice, "rate": rate,
|
||
"trimmed": not args.no_trim,
|
||
"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")
|
||
json.dump(manifest, open(mp, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
|
||
print(f"[tts] {len(results)} 段 → audio/ + {mp}")
|
||
total = sum(r["container_duration_sec"] for r in results)
|
||
print(f"[tts] 容器合计 {total:.2f}s（未含段间 gap；timeline_build.py 会加）")
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
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="不裁首尾静音（不建议）")
|
||
p.add_argument("--no-cache", action="store_true")
|
||
args = p.parse_args()
|
||
return asyncio.run(run(args))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|