接口更新：支持直接返回srt字幕内容，详见文档！2025.12.29
欢迎使用我们的微软TTS API服务。本文档旨在帮助开发者快速理解和集成我们的文本转语音接口。请在开始前仔细阅读，特别是关于认证授权的部分。
准备工作 (Prerequisites)
在调用任何接口之前，您需要准备好您在本站的用户名和密码。请注意，请勿使用注册邮箱作为user参数，应使用您的用户名。
核心概念：认证与授权
为了保障接口安全，我们所有的请求都需要进行签名和加密验证，请求前需要检查本机系统时间是否为标准北京时间，后端有时间校验，请求提交时间与后端时间相差过大会拒绝请求。
-
时间戳 (Timestamp)：所有接口都需要一个13位的毫秒级时间戳，参数名为 ts。
-
签名 (Sign)：部分接口需要对请求进行签名，签名算法为 md5(username + ts)。
-
授权头 (Authorization Header)：queryorder 和 mstts 接口需要在HTTP请求头中加入 Authorization 字段。该字段的值是通过 AES 加密生成
-
模式：ECB
-
填充：PKCS7
-
密钥 (Key)：一个16位的自定义字符串，在首次获取token时生成或指定。
-
待加密数据：一个包含user, token, ts的JSON字符串。
-
API 接口详解
基础URL (Base URL):
1. 获取Token (/gettoken)
此接口用于获取后续接口调用所需的 token 和 key。
接口地址：/gettoken
请求方法：POST
请求参数 (data)：
-
user (string): 您的网站用户名（不是邮箱）。
-
psw (string): 您的网站密码。
-
type (string): 操作类型。0表示获取token，1表示更新key和token。
-
key (string, optional): 16位自定义密钥。如果留空，系统将为您随机生成一个(建议自动生成，更安全）。
-
ts (string): 13位时间戳。
-
sign (string): 签名，计算方式为 md5(user + ts)。
返回数据示例：
{ "code": 0, "maxChar": 20000000, "date": "2024-12-31", "charNum": 3015, "key": "your_16_bit_key", "token": "your_generated_token", "msg": "success" }
2. 查询套餐信息 (/queryorder)
查询您当前账户的字符套餐余量和到期时间。
-
接口地址：/queryorder
-
请求方法：POST
-
请求头 (Headers)：Authorization: 经过AES加密后的授权字符串。
-
请求参数 (data)：user (string): 您的网站用户名。ts (string): 13位时间戳。sign (string): 签名，计算方式为 md5(user + ts)。
-
返回数据示例：
{ "code": 0, "maxChar": 20000000, "date": "2024-12-31", "charNum": 3015, "msg": "success" }
3. 微软TTS语音合成 (/mstts-ws)
核心接口，用于将SSML文本合成为语音。
-
接口地址：/mstts-ws
-
请求方法：websocket
-
接入方法参考python代码示例
重要注意事项
-
启用重定向：所有接口请求都必须启用HTTP重定向（allow_redirects=True）。
-
并发限制：接口限制最高并发为 3次/秒。超过此限制服务器将触发5-60分钟限制访问，返回 429 状态码，若一直开启无限制并发，将作封号处理。
-
SSML文档：关于如何编写SSML以控制语音情感、停顿、多角色发音等高级功能，请参考 。
API 快速测试工具(老接口）
为了方便您在编码前快速验证接口的有效性，我们提供了API测试工具。您可以在页面顶部免费下载。
-
获取token：/gettoken
-
套餐查询: /queryOrder
-
微软TTS语音合成：/mstts
Python API 调用完整示例代码
以下是一个完整的Python示例，演示了从获取Token到最终合成语音的全过程。您需要安装 requests 和 pycryptodome 库。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
z-tts /mstts-ws 流式接口 Python 客户端示例
==========================================
功能:
1. 文本按句切分, 每段 <= MAX_SEG_LEN(500) 字符(汉字/英文均按 1 字符计)
2. 连接池并行: K 条
3. 各段结果按索引归位, SRT 按该段音频实际时长做时间轴偏移, 合并输出整篇 SRT
4. 实时显示额度消耗(end 帧的 charNum/maxChar)
协议(/mstts-ws, 每条请求独立走完整校验/限流/计费):
C→S text : {"user","ssml"(base64),"quality","srt","ts","sign","authorization"}
sign = md5(user + ts), authorization = POST /gettoken 返回的 token
S→C text : {"t":"start"}
S→C bin : mp3 音频块 ×N(顺序即流顺序, 直接拼接即完整 mp3)
S→C text : {"t":"end","code":0,"srt":"<明文SRT>","charNum":..,"maxChar":..}
S→C text : {"t":"err","code":-1,"msg":"..."}
end 帧后连接保持, 可直接发下一段请求(复用免握手)
依赖: pip install websocket-client requests
用法: python mstts_ws_demo.py
"""
import base64
import hashlib
import json
import queue
import re
import time
import xml.sax.saxutils as saxutils
from concurrent.futures import ThreadPoolExecutor
import requests
import websocket
# ========================== 配置 ==========================
API_HOST = "47.96.88.233:2048" # 服务器地址(含端口)
WS_SCHEME = "ws"
USER = "" # 网站账号
PSW = "" # /gettoken 所需密码(同 HTTP 接口)
VOICE = "zh-CN-XiaoxiaoNeural" # 音色
RATE = "+0%" # 语速
PITCH = "+0Hz" # 音调
STYLE = None # 风格, 如 "cheerful"; 不用填 None
MAX_SEG_LEN = 500 # 每段最大字符数(客户端切分标准)
POOL_SIZE = 4 # 并行连接数: K 条连接经负载均衡落到不同后端; 1=单连接串行
OUT_MP3 = "output.mp3"
OUT_SRT = "output.srt"
# ==========================================================
def md5hex(s: str) -> str:
return hashlib.md5(s.encode("utf-8")).hexdigest()
def normalize_host(h: str, default_ws_scheme: str = WS_SCHEME):
"""host 配置归一化, 兼容以下几种写法:
'host:port' / 'http(s)://host:port' / 'ws(s)://host:port'
返回 (ws_scheme, http_base, bare_host)
http→ws, https→wss 自动映射, 防止把 http 前缀拼进 WS URL"""
h = (h or "").strip().rstrip("/")
m = re.match(r"^(?P<s>wss?|https?)://", h, re.IGNORECASE)
if m:
s = m.group("s").lower()
ws_scheme = {"http": "ws", "https": "wss"}.get(s, s)
http_scheme = {"ws": "http", "wss": "https"}.get(s, s)
return ws_scheme, f"{http_scheme}://{h[m.end():]}", h[m.end():]
return default_ws_scheme, f"http://{h}", h
def get_token(user: str, psw: str, base: str = None) -> dict:
"""调用 POST /gettoken 换取 token(即后续请求的 authorization 值)"""
base = base or normalize_host(API_HOST)[1]
ts = str(int(time.time()))
r = requests.post(
f"{base}/gettoken",
json={"user": user, "psw": psw, "type": "0", "ts": ts, "sign": md5hex(user + ts)},
timeout=30,
)
d = r.json()
if d.get("code") != 0:
raise RuntimeError(f"gettoken 失败: {d.get('msg')}")
return d # 含 token/key/maxChar/charNum/date
def build_ssml(text: str, voice: str, rate: str = "+0%", pitch: str = "+0Hz",
style: str = None) -> str:
body = saxutils.escape(text)
inner = f'<mstts:express-as style="{style}">{body}</mstts:express-as>' if style else body
return (
'<speak version="1.0" xmlns="http://www.w3.org/2001/10/synthesis" '
'xmlns:mstts="https://www.w3.org/2001/mstts" xml:lang="zh-CN">'
f'<voice name="{voice}">'
f'<prosody rate="{rate}" pitch="{pitch}">{inner}</prosody>'
"</voice></speak>"
)
# ---------------------- 文本分段 ----------------------
_SENT_END = "。！？!?；;…"
_CLAUSE_END = "，,、：:"
def split_text(text: str, max_len: int = MAX_SEG_LEN) -> list:
"""按句切分并贪心组包, 每段不超过 max_len 个字符。
单句超长时优先在逗号处断开, 仍超长(无标点长串)则硬切。
注意: 服务端计费口径是汉字2/英文1(calculateStringLength), 此处切分按字符数。"""
text = re.sub(r"[ \t\r\f\v]+", " ", text.strip())
if not text:
return []
# 1) 分句: 句末标点/换行处切断(保留标点)
sents, buf = [], []
for ch in text:
buf.append(ch)
if ch in _SENT_END or ch == "\n":
sents.append("".join(buf))
buf = []
if buf:
sents.append("".join(buf))
# 2) 超长单句: 优先在逗号处断
pieces = []
for s in sents:
if len(s) <= max_len:
pieces.append(s)
continue
cur = ""
for ch in s:
cur += ch
if ch in _CLAUSE_END and len(cur) >= max_len // 2:
pieces.append(cur)
cur = ""
if cur:
pieces.append(cur)
# 3) 仍超长则硬切
final = []
for p in pieces:
final.extend(p[i:i + max_len] for i in range(0, len(p), max_len))
# 4) 贪心组包
segs, cur = [], ""
for p in final:
if not cur:
cur = p
elif len(cur) + len(p) <= max_len:
cur += p
else:
segs.append(cur)
cur = p
if cur:
segs.append(cur)
return segs
# ---------------------- SRT 工具 ----------------------
def _fmt_t(sec: float) -> str:
"""秒 → SRT 时间戳 HH:MM:SS,mmm"""
ms = max(0, round(sec * 1000))
h, ms = divmod(ms, 3600000)
m, ms = divmod(ms, 60000)
s, ms = divmod(ms, 1000)
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
def _parse_t(stamp: str) -> float:
"""SRT 时间戳 → 秒"""
stamp = stamp.strip().replace(",", ".")
hh, mm, ss = stamp.split(":")
return int(hh) * 3600 + int(mm) * 60 + float(ss)
def parse_srt(srt: str) -> list:
"""SRT 文本 → [(start秒, end秒, 文本), ...]"""
entries = []
for block in re.split(r"\r?\n\s*\r?\n", srt.strip()):
lines = block.strip().splitlines()
if len(lines) >= 3 and "-->" in lines[1]:
a, b = lines[1].split("-->", 1)
entries.append((_parse_t(a), _parse_t(b), "\n".join(lines[2:]).strip()))
return entries
def merge_srt(per_segment: list) -> str:
"""per_segment: [(该段时间轴偏移秒, parse_srt结果), ...] → 整篇 SRT 文本"""
out, idx = [], 1
for offset, entries in per_segment:
for s, e, txt in entries:
out.append(f"{idx}\n{_fmt_t(s + offset)} --> {_fmt_t(e + offset)}\n{txt}\n")
idx += 1
return "\n".join(out)
# ---------------------- WS 客户端 ----------------------
class TtsError(RuntimeError):
pass
class MsttsWsClient:
OP_TEXT, OP_BINARY = 0x1, 0x2
MP3_BYTES_PER_SEC = 6000 # 固定输出 audio-24khz-48kbitrate-mono-mp3(CBR 48kbps) → 时长≈字节/6000
def __init__(self, user: str, token: str, voice: str,
scheme: str = None, host: str = API_HOST, timeout: int = 90):
ws_scheme, _http_base, bare = normalize_host(host)
self.url = f"{scheme or ws_scheme}://{bare}/mstts-ws"
self.user, self.token, self.voice = user, token, voice
self.timeout = timeout
self.ws = None
self._started = False
self._resolved_url = None
def _resolve_backend_url(self) -> str:
"""负载均衡是 307 重定向架构,
因此先发一次普通 HTTP 预检(不跟随重定向)拿 307 的 Location(真实后端地址),
把 http(s):// 换成 ws(s):// 后直连后端; 无重定向(直连后端场景)则返回原 URL。
结果缓存, 每条连接只预检一次; 池内各连接独立预检 → 算法散到不同后端"""
if self._resolved_url:
return self._resolved_url
preflight = self.url.replace("wss://", "https://", 1).replace("ws://", "http://", 1)
try:
r = requests.get(preflight, allow_redirects=False, timeout=10)
loc = r.headers.get("Location") if r.status_code in (301, 302, 303, 307, 308) else None
if loc:
m = re.match(r"^(?P<s>https?)://", loc.strip(), re.IGNORECASE)
if m:
ws_scheme = "wss" if m.group("s").lower() == "https" else "ws"
self._resolved_url = f"{ws_scheme}://{loc.strip()[m.end():]}"
return self._resolved_url
except Exception as e:
print(f"[warn] LB 预检失败, 直连原地址: {e}")
self._resolved_url = self.url
return self._resolved_url
def connect(self):
url = self._resolve_backend_url()
try:
self.ws = websocket.create_connection(url, timeout=self.timeout)
except Exception as e:
print(f"[error] WS 连接失败, 目标 URL: {url} | {type(e).__name__}: {e}")
raise
def close(self):
if self.ws:
try:
self.ws.close()
except Exception:
pass
self.ws = None
def synthesize(self, text: str, want_srt: bool = True):
"""合成一段文本 → (mp3字节, srt文本或'', end帧dict)"""
for attempt in (1, 2):
self._started = False
try:
if self.ws is None:
self.connect()
return self._round(text, want_srt)
except (websocket.WebSocketConnectionClosedException, ConnectionError, OSError) as e:
self.close()
# 仅在尚未收到 start(未开始合成、必然未扣费)时安全重试, 避免重复计费;
# 已开始的合成即使断线, 服务端仍会完成并扣费, 重发会造成双倍扣费
if attempt == 1 and not self._started:
print("[warn] 连接中断, 已重连重试(未开始合成, 无重复计费风险)")
continue
raise TtsError(f"连接中断: {e}") from e
def _round(self, text: str, want_srt: bool):
ssml = build_ssml(text, self.voice, RATE, PITCH, STYLE)
ts = str(int(time.time()))
req = {
"user": self.user,
"ssml": base64.b64encode(ssml.encode("utf-8")).decode("ascii"),
"quality": 0,
"srt": 1 if want_srt else 0,
"ts": ts,
"sign": md5hex(self.user + ts),
"authorization": self.token,
}
self.ws.send(json.dumps(req, ensure_ascii=False))
audio = bytearray()
while True:
opcode, data = self.ws.recv_data()
if opcode == self.OP_BINARY:
audio.extend(data) # mp3 音频块, 顺序拼接即可
continue
try:
m = json.loads(data.decode("utf-8"))
except (ValueError, UnicodeDecodeError) as e:
raise TtsError(f"无法解析的服务器文本帧: {e}") from e
t = m.get("t")
if t == "start":
self._started = True
elif t == "end":
if m.get("code") != 0:
raise TtsError(m.get("msg", "unknown"))
return bytes(audio), m.get("srt") or "", m
elif t == "err":
raise TtsError(m.get("msg", "unknown error"))
# 其他未知帧忽略
# ---------------------- 连接池(负载均衡并行) ----------------------
class MsttsWsPool:
"""K 条 WS 连接池: 每条新连接经负载均衡分配到不同后端, 段级并行合成。
每条连接同一时刻只跑一个请求(与服务端 busy 锁对应), 跑完归还池中复用。"""
def __init__(self, size: int, user: str, token: str, voice: str, hosts: list = None):
# hosts 缺省为 [API_HOST]*size: 生产环境全部连负载均衡地址,
# 每条 TCP 连接由 LB 独立调度到不同后端; 传多个 host 仅用于本地多后端测试
hosts = hosts or [API_HOST]
self.size = max(1, min(size, len(hosts) if len(hosts) > 1 else size))
self._q = queue.Queue()
self._clients = []
for i in range(self.size):
c = MsttsWsClient(user, token, voice, host=hosts[i % len(hosts)])
self._clients.append(c)
self._q.put(c)
def acquire(self) -> MsttsWsClient:
return self._q.get()
def release(self, client: MsttsWsClient):
self._q.put(client)
def close(self):
for c in self._clients:
c.close()
def _synth_one(client: MsttsWsClient, idx: int, total: int, seg: str):
"""合成一段并打印进度, 返回 (audio字节, srt文本, end帧)"""
audio, srt, end = client.synthesize(seg)
entries = parse_srt(srt) if srt else []
seg_dur = max(len(audio) / client.MP3_BYTES_PER_SEC,
entries[-1][1] if entries else 0.0)
print(f"[{idx + 1}/{total}] {len(seg)}字 → {len(audio)}字节(~{seg_dur:.1f}s) "
f"字幕{len(entries)}条 额度 {end.get('charNum')}/{end.get('maxChar')}")
return audio, srt, end
# ---------------------- 主流程 ----------------------
def tts_long_text(text: str, out_mp3: str = OUT_MP3, out_srt: str = OUT_SRT,
hosts: list = None, pool_size: int = POOL_SIZE):
info = get_token(USER, PSW)
print(f"[token] OK 已用/总额度: {info.get('charNum')}/{info.get('maxChar')} 有效期至: {info.get('date')}")
segs = split_text(text)
if not segs:
print("[split] 空文本, 退出")
return
print(f"[split] 共 {len(segs)} 段 ")
t0 = time.time()
n = len(segs)
results = [None] * n
if pool_size <= 1 or n == 1:
cli = MsttsWsClient(USER, info["token"], VOICE, host=(hosts or [API_HOST])[0])
try:
for i, seg in enumerate(segs):
results[i] = _synth_one(cli, i, n, seg)
finally:
cli.close()
else:
pool = MsttsWsPool(min(pool_size, n), USER, info["token"], VOICE, hosts)
print(f"[pool] {pool.size} 条连接并行合成")
def work(i, seg):
c = pool.acquire()
try:
return _synth_one(c, i, n, seg)
finally:
pool.release(c)
try:
with ThreadPoolExecutor(max_workers=pool.size) as ex:
futs = [ex.submit(work, i, s) for i, s in enumerate(segs)]
for f in futs:
f.result() # 任一段失败则整体抛出
results = [f.result() for f in futs]
finally:
pool.close()
# 时间轴合并: 按段顺序累加偏移, 与段落在哪台后端合成无关
parts, per_seg_srt, offset = [], [], 0.0
for audio, srt, _end in results:
entries = parse_srt(srt) if srt else []
per_seg_srt.append((offset, entries))
parts.append(audio)
offset += max(len(audio) / MsttsWsClient.MP3_BYTES_PER_SEC,
entries[-1][1] if entries else 0.0)
with open(out_mp3, "wb") as f:
for p in parts:
f.write(p)
with open(out_srt, "w", encoding="utf-8") as f:
f.write(merge_srt(per_seg_srt))
elapsed = time.time() - t0
eh, rem = divmod(int(elapsed), 3600)
em, es = divmod(rem, 60)
# 最新额度: 并行乱序完成, charNum 最大者即最后一次扣费后的值
latest = max((e for _a, _s, e in results if e), key=lambda e: e.get("charNum") or 0, default=None)
quota = f" 最新额度: {latest.get('charNum')}/{latest.get('maxChar')}" if latest else ""
print(f"[done] {out_mp3}({sum(len(p) for p in parts)}字节, ~{offset:.1f}s) + {out_srt} 已生成"
f" 耗时 {eh:02d}:{em:02d}:{es:02d}{quota}")
if __name__ == "__main__":
# 示例长文本(>500字, 自动演示分段与SRT合并); 替换为你的内容即可
# with open("demo.txt", "r", encoding="utf-8") as f:
# demo = f.read()
demo = ("语音合成技术让文字拥有了声音。通过流式接口，长文本可以被切分成多个片段，"
"逐段合成后再拼接成完整的音频，同时字幕的时间轴也会自动衔接，"
"不需要任何手工调整。这正是新接口相对于传统base64方案的优势所在。") * 6
tts_long_text(demo)
声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。
评论(0)