248 lines
9.5 KiB
Python
248 lines
9.5 KiB
Python
"""讯飞录音文件转写(LFASR,raasr.xfyun.cn v2)客户端 —— 整个录音文件转文字。
|
||
|
||
接口文档: https://www.xfyun.cn/doc/asr/ifasr_new/API.html
|
||
与 `asr_xfyun`(IAT 语音听写,实时流式,单会话 60s)是**两个独立服务**:LFASR 专为
|
||
已成文件的长录音设计 —— 单文件最长 5 小时 / 500MB,服务端自行解码常见音频格式
|
||
(mp3/m4a/wav/amr/ogg/flac/...,无需本地 ffmpeg),自带说话人分离(roleType=1)。
|
||
转写是异步订单:`upload` 拿 orderId → 轮询 `get_result` 等 status=4。
|
||
|
||
凭据:env `XFYUN_APPID`(与 IAT 共用)+ `XFYUN_LFASR_SECRET_KEY`(LFASR 服务独立的
|
||
SecretKey,讯飞控制台「语音转写」页查看 —— **不是** IAT 的 APIKey/APISecret)。
|
||
签名:signa = Base64(HmacSHA1(key=secretKey, msg=MD5hex(appid+ts)))。
|
||
|
||
全 sync(httpx):调用方是 tool.execute,本就跑在 worker 线程里。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import time
|
||
from typing import Callable, Optional
|
||
|
||
import httpx
|
||
|
||
_API_BASE = "https://raasr.xfyun.cn/v2/api"
|
||
_POLL_INTERVAL_S = 10.0 # getResult 有频控,10s 一查绰绰有余
|
||
_POLL_ERR_TOLERANCE = 5 # 轮询期连续网络错误容忍次数(订单在讯飞侧,不能因抖动丢单)
|
||
|
||
# orderInfo.failType → 可读提示(文档错误码页;没列到的走通用兜底)
|
||
_FAIL_HINTS = {
|
||
1: "音频上传失败",
|
||
2: "音频转码失败(格式可能不支持,建议转成 mp3/wav 再传)",
|
||
3: "音频识别失败",
|
||
4: "音频时长超限(最长 5 小时)",
|
||
5: "音频文件校验失败(文件损坏或为空)",
|
||
6: "未检出有效语音(静音 / 纯音乐?)",
|
||
}
|
||
|
||
|
||
class LfasrError(RuntimeError):
|
||
"""转写失败(讯飞返回错误 / 连接异常),code 为讯飞错误码字符串(连接类为 None)。"""
|
||
|
||
def __init__(self, message: str, code: Optional[str] = None):
|
||
super().__init__(message)
|
||
self.code = code
|
||
|
||
|
||
class LfasrNotConfigured(LfasrError):
|
||
"""XFYUN_APPID / XFYUN_LFASR_SECRET_KEY 未配置。"""
|
||
|
||
|
||
class LfasrCancelled(LfasrError):
|
||
"""用户取消等待(讯飞侧订单不受影响,会继续转完但结果没人取)。"""
|
||
|
||
|
||
def is_configured() -> bool:
|
||
"""LFASR 凭据是否齐 —— agent_builder 据此决定挂不挂 transcribe_audio tool。"""
|
||
return all(
|
||
(os.getenv(k) or "").strip()
|
||
for k in ("XFYUN_APPID", "XFYUN_LFASR_SECRET_KEY")
|
||
)
|
||
|
||
|
||
def _load_credentials() -> tuple[str, str]:
|
||
appid = (os.getenv("XFYUN_APPID") or "").strip()
|
||
secret = (os.getenv("XFYUN_LFASR_SECRET_KEY") or "").strip()
|
||
if not (appid and secret):
|
||
raise LfasrNotConfigured(
|
||
"录音转写未配置:需在 .env 设 XFYUN_APPID / XFYUN_LFASR_SECRET_KEY"
|
||
"(后者是讯飞控制台「语音转写」服务的 SecretKey,与语音听写的 key 不同)"
|
||
)
|
||
return appid, secret
|
||
|
||
|
||
def _auth_params() -> dict:
|
||
appid, secret = _load_credentials()
|
||
ts = str(int(time.time()))
|
||
md5hex = hashlib.md5((appid + ts).encode()).hexdigest()
|
||
signa = base64.b64encode(
|
||
hmac.new(secret.encode(), md5hex.encode(), hashlib.sha1).digest()
|
||
).decode()
|
||
return {"appId": appid, "ts": ts, "signa": signa}
|
||
|
||
|
||
def _post(path: str, params: dict, content: Optional[bytes] = None,
|
||
timeout_s: float = 60) -> dict:
|
||
"""POST 一次并校验业务码;code != 000000 抛 LfasrError,返回 content 字段。"""
|
||
try:
|
||
resp = httpx.post(
|
||
f"{_API_BASE}{path}",
|
||
params={**_auth_params(), **params},
|
||
content=content,
|
||
headers={"Content-Type": "application/octet-stream"} if content else None,
|
||
timeout=timeout_s,
|
||
)
|
||
resp.raise_for_status()
|
||
body = resp.json()
|
||
except LfasrError:
|
||
raise
|
||
except Exception as e: # 连接 / 超时 / 非 JSON / HTTP 状态码
|
||
raise LfasrError(f"讯飞录音转写连接失败:{type(e).__name__}: {e}")
|
||
code = str(body.get("code") or "")
|
||
if code != "000000":
|
||
desc = body.get("descInfo") or "未知错误"
|
||
raise LfasrError(f"讯飞录音转写失败({code}):{desc}", code=code)
|
||
return body.get("content") or {}
|
||
|
||
|
||
def upload(data: bytes, file_name: str, *, language: str = "cn",
|
||
role_type: int = 1) -> tuple[str, float]:
|
||
"""上传音频建转写订单。返回 (orderId, 预估转写耗时秒)。
|
||
|
||
duration 参数按官方 demo 传定值(文档标必填但服务端不校验,只原样回显进
|
||
orderInfo.originalDuration —— 真实时长看 realDuration);roleType=1 开说话人
|
||
分离(不额外计费,会议场景直接可用)。
|
||
"""
|
||
params = {
|
||
"fileName": file_name,
|
||
"fileSize": str(len(data)),
|
||
"duration": "200",
|
||
"language": language,
|
||
"roleType": str(role_type),
|
||
}
|
||
content = _post("/upload", params, content=data, timeout_s=300)
|
||
order_id = (content.get("orderId") or "").strip()
|
||
if not order_id:
|
||
raise LfasrError("讯飞录音转写:upload 成功但未返回 orderId")
|
||
estimate_s = float(content.get("taskEstimateTime") or 0) / 1000
|
||
return order_id, estimate_s
|
||
|
||
|
||
def get_result(order_id: str) -> dict:
|
||
"""查一次订单。返回 content(含 orderInfo.status / orderResult)。"""
|
||
return _post("/getResult", {"orderId": order_id, "resultType": "transfer"},
|
||
timeout_s=60)
|
||
|
||
|
||
def parse_transcript(order_result: str) -> list[tuple[str, str]]:
|
||
"""orderResult(JSON 串)→ [(role, text)] 按时间顺序。
|
||
|
||
lattice 每项的 json_1best 是**再编码一层的 JSON 串**(lattice2 里则已是对象),
|
||
两种形态都兜。role 是说话人编号字符串("1"/"2"...,未分离时 "0" 或缺失 → "")。
|
||
"""
|
||
try:
|
||
parsed = json.loads(order_result or "{}")
|
||
except json.JSONDecodeError as e:
|
||
raise LfasrError(f"讯飞录音转写:结果 JSON 解析失败:{e}")
|
||
lattice = parsed.get("lattice") or parsed.get("lattice2") or []
|
||
segments: list[tuple[str, str]] = []
|
||
for item in lattice:
|
||
best = item.get("json_1best")
|
||
if isinstance(best, str):
|
||
try:
|
||
best = json.loads(best)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
st = (best or {}).get("st") or {}
|
||
text = "".join(
|
||
cw.get("w", "")
|
||
for rt in (st.get("rt") or [])
|
||
for ws in (rt.get("ws") or [])
|
||
for cw in (ws.get("cw") or [])
|
||
)
|
||
if not text.strip():
|
||
continue
|
||
role = str(st.get("rl") or "").strip()
|
||
segments.append(("" if role == "0" else role, text))
|
||
return segments
|
||
|
||
|
||
def format_transcript(segments: list[tuple[str, str]]) -> str:
|
||
"""[(role, text)] → 成稿文本。
|
||
|
||
识别出 ≥2 个说话人时按连续同角色合并成「【说话人N】…」段落(会议/访谈场景);
|
||
单说话人 / 未分离则纯文本拼接。
|
||
"""
|
||
roles = {r for r, _ in segments if r}
|
||
if len(roles) < 2:
|
||
return "".join(t for _, t in segments).strip()
|
||
lines: list[str] = []
|
||
cur_role, buf = None, []
|
||
for role, text in segments:
|
||
if role != cur_role:
|
||
if buf:
|
||
lines.append(f"【说话人{cur_role or '?'}】{''.join(buf)}")
|
||
cur_role, buf = role, [text]
|
||
else:
|
||
buf.append(text)
|
||
if buf:
|
||
lines.append(f"【说话人{cur_role or '?'}】{''.join(buf)}")
|
||
return "\n\n".join(lines).strip()
|
||
|
||
|
||
def transcribe_file(
|
||
data: bytes,
|
||
file_name: str,
|
||
*,
|
||
language: str = "cn",
|
||
cancel_check: Optional[Callable[[], bool]] = None,
|
||
) -> dict:
|
||
"""上传 + 轮询到订单完成。返回 {"text", "duration_s", "order_id", "elapsed_s"}。
|
||
|
||
等待上限取 max(600s, 3×讯飞预估);轮询期连续网络错误 ≤ _POLL_ERR_TOLERANCE 次
|
||
只记不抛(订单在讯飞侧,不能因一次抖动丢单)。取消抛 LfasrCancelled。
|
||
"""
|
||
started = time.monotonic()
|
||
order_id, estimate_s = upload(data, file_name, language=language)
|
||
deadline = started + max(600.0, estimate_s * 3)
|
||
consecutive_errs = 0
|
||
while True:
|
||
if cancel_check is not None and cancel_check():
|
||
raise LfasrCancelled(f"用户取消(讯飞订单 {order_id} 会继续转完,但结果未取)")
|
||
if time.monotonic() > deadline:
|
||
raise LfasrError(
|
||
f"讯飞录音转写超时(订单 {order_id} 等了 "
|
||
f"{int(time.monotonic() - started)}s 仍未完成)"
|
||
)
|
||
time.sleep(_POLL_INTERVAL_S)
|
||
try:
|
||
content = get_result(order_id)
|
||
consecutive_errs = 0
|
||
except LfasrError as e:
|
||
if e.code is not None: # 业务错误码:重试无意义,直接抛
|
||
raise
|
||
consecutive_errs += 1
|
||
if consecutive_errs > _POLL_ERR_TOLERANCE:
|
||
raise
|
||
continue
|
||
info = content.get("orderInfo") or {}
|
||
status = info.get("status")
|
||
if status == -1:
|
||
ft = info.get("failType")
|
||
hint = _FAIL_HINTS.get(ft, f"转写失败(failType={ft})")
|
||
raise LfasrError(f"讯飞录音转写失败:{hint}")
|
||
if status != 4:
|
||
continue # 0 已创建 / 3 处理中
|
||
segments = parse_transcript(content.get("orderResult") or "")
|
||
# originalDuration 只是 upload 时 duration 参数的回显(我们传的定值);
|
||
# realDuration 才是讯飞实际解出的音频时长(ms,即计费时长)—— 实测验证。
|
||
return {
|
||
"text": format_transcript(segments),
|
||
"duration_s": float(info.get("realDuration") or 0) / 1000,
|
||
"order_id": order_id,
|
||
"elapsed_s": time.monotonic() - started,
|
||
}
|