zcbot/core/asr_xfyun.py

158 lines
6.4 KiB
Python

"""讯飞语音听写(IAT,流式版)WebSocket 客户端 — 整段音频转写。
接口文档: https://www.xfyun.cn/doc/asr/voicedictation/API.html
- 鉴权: HMAC-SHA256 签名拼进 wss URL query(authorization/date/host)
- 音频: 仅收 16kHz/16bit/单声道 PCM(raw),单会话上限 60 秒
- 凭据: env XFYUN_APPID / XFYUN_API_KEY / XFYUN_API_SECRET(换正式 key 只改 .env)
本模块只做"整段 PCM → 最终文本"(web 语音输入按钮用);未开 dwa 动态修正 ——
那是实时逐字上屏才需要的,整段转写直接按序拼接 result.ws 即可。将来做流式
上屏时复用 build_auth_url() 签临时 URL 给前端直连。
"""
from __future__ import annotations
import asyncio
import base64
import hashlib
import hmac
import json
import os
import time
from contextlib import suppress
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time
XFYUN_HOST = "iat-api.xfyun.cn"
XFYUN_PATH = "/v2/iat"
SAMPLE_RATE = 16000
BYTES_PER_SECOND = SAMPLE_RATE * 2 # 16bit mono
MAX_SECONDS = 60 # 讯飞单会话硬上限
MAX_PCM_BYTES = MAX_SECONDS * BYTES_PER_SECOND
# 每帧 9600B(=300ms 音频),base64 后 12800B < 讯飞单帧上限 13000B;
# 帧间 40ms 为文档推荐节奏,整段上传比实时快 7.5 倍(60s 音频 ≈ 8s 传完)
_FRAME_BYTES = 9600
_FRAME_INTERVAL = 0.04
_RECV_TIMEOUT = 30 # 单次等响应上限;讯飞 10s 无数据也会主动断
# 常见错误码 → 给用户/运维看的中文提示(全量见文档错误码页)
_ERR_HINTS = {
10105: "appid 不合法(检查 XFYUN_APPID)",
10107: "参数值非法",
10110: "无授权许可(服务未开通或已过期)",
10114: "会话超时(单次音频超 60 秒)",
10163: "业务参数校验失败",
10165: "会话句柄无效",
10313: "appid 不能为空或与 apikey 不匹配",
11200: "无权限或用量已用完(检查讯飞控制台服务量)",
11201: "当日流控超限",
}
class XfyunASRError(RuntimeError):
"""转写失败(讯飞返回错误码 / 连接异常),code 为讯飞错误码(连接类异常为 None)。"""
def __init__(self, message: str, code: int | None = None):
super().__init__(message)
self.code = code
class XfyunASRNotConfigured(XfyunASRError):
"""XFYUN_* env 未配置 — 上层应返回 501 而非 502。"""
def build_auth_url(api_key: str, api_secret: str) -> str:
"""按讯飞规范生成带签名的 wss URL(date 用 RFC1123 GMT)。"""
date = format_date_time(time.time())
signature_origin = f"host: {XFYUN_HOST}\ndate: {date}\nGET {XFYUN_PATH} HTTP/1.1"
signature = base64.b64encode(
hmac.new(api_secret.encode(), signature_origin.encode(), hashlib.sha256).digest()
).decode()
authorization_origin = (
f'api_key="{api_key}", algorithm="hmac-sha256", '
f'headers="host date request-line", signature="{signature}"'
)
authorization = base64.b64encode(authorization_origin.encode()).decode()
query = urlencode({"authorization": authorization, "date": date, "host": XFYUN_HOST})
return f"wss://{XFYUN_HOST}{XFYUN_PATH}?{query}"
def _load_credentials() -> tuple[str, str, str]:
appid = (os.getenv("XFYUN_APPID") or "").strip()
api_key = (os.getenv("XFYUN_API_KEY") or "").strip()
api_secret = (os.getenv("XFYUN_API_SECRET") or "").strip()
if not (appid and api_key and api_secret):
raise XfyunASRNotConfigured(
"语音识别未配置:需在 .env 设 XFYUN_APPID / XFYUN_API_KEY / XFYUN_API_SECRET"
)
return appid, api_key, api_secret
async def transcribe(pcm: bytes, *, language: str = "zh_cn") -> str:
"""整段 PCM(16k/16bit/mono, 小端)→ 识别文本。空音频/纯静音返回 """""
appid, api_key, api_secret = _load_credentials()
if not pcm:
return ""
if len(pcm) > MAX_PCM_BYTES:
raise XfyunASRError(f"音频超长:最多 {MAX_SECONDS}")
import websockets # uvicorn[standard] 附带;requirements 亦显式声明
audio_meta = {"format": f"audio/L16;rate={SAMPLE_RATE}", "encoding": "raw"}
async def _send_frames(ws) -> None:
first = True
for off in range(0, len(pcm), _FRAME_BYTES):
chunk = base64.b64encode(pcm[off : off + _FRAME_BYTES]).decode()
frame: dict = {"data": {"status": 1 if not first else 0, **audio_meta, "audio": chunk}}
if first:
frame["common"] = {"app_id": appid}
# vad_eos 拉到上限 10s:整段转写不希望句间停顿被判"说完了"截尾
frame["business"] = {
"language": language,
"domain": "iat",
"accent": "mandarin",
"vad_eos": 10000,
"ptt": 1,
}
first = False
await ws.send(json.dumps(frame))
await asyncio.sleep(_FRAME_INTERVAL)
await ws.send(json.dumps({"data": {"status": 2, **audio_meta, "audio": ""}}))
pieces: list[str] = []
try:
async with websockets.connect(build_auth_url(api_key, api_secret), open_timeout=10) as ws:
send_task = asyncio.create_task(_send_frames(ws))
try:
while True:
raw = await asyncio.wait_for(ws.recv(), timeout=_RECV_TIMEOUT)
msg = json.loads(raw)
code = msg.get("code")
if code:
hint = _ERR_HINTS.get(code, msg.get("message") or "未知错误")
raise XfyunASRError(f"讯飞识别失败({code}):{hint}", code=code)
data = msg.get("data") or {}
result = data.get("result") or {}
pieces.append(
"".join(
cw.get("w", "")
for w in (result.get("ws") or [])
for cw in (w.get("cw") or [])
)
)
if data.get("status") == 2:
break
finally:
send_task.cancel()
with suppress(asyncio.CancelledError):
await send_task
except XfyunASRError:
raise
except asyncio.TimeoutError:
raise XfyunASRError("讯飞识别超时(30s 无响应)")
except Exception as e: # 握手 401/连接断开等
raise XfyunASRError(f"讯飞连接失败:{type(e).__name__}: {e}")
return "".join(pieces).strip()