# -*- coding:utf-8 -*- # 讯飞开放平台 大模型中文语音识别 WebSocket Demo # 运行环境：Python 3.7+ # 依赖库：websocket-client import _thread as thread import base64 import hashlib import hmac import json import ssl import time from datetime import datetime from time import mktime from urllib import parse from urllib.parse import urlencode from wsgiref.handlers import format_date_time import websocket # ====== 配置信息 ====== APPId = "xxxxx" APIKey = "xxxxxxxxxxxxxxxxx" APISecret = "xxxxxxxxxxxxxxx" AudioFile = r"xxxxxxxxxx" RequestUrl = "wss://iat.cn-huabei-1.xf-yun.com/v1" STATUS_FIRST_FRAME = 0 # 第一帧标识 STATUS_CONTINUE_FRAME = 1 # 中间帧标识 STATUS_LAST_FRAME = 2 # 最后一帧标识 result_text = "" # 累积识别结果 class Ws_Param(object): # 初始化 def __init__(self, APPID, APIKey, APISecret, AudioFile): self.APPID = APPID self.APIKey = APIKey self.APISecret = APISecret self.AudioFile = AudioFile self.iat_params = { "language": "zh_cn", "accent": "mandarin", "domain": "spark_asr_v2", "eos":60000, "result": { "encoding": "utf8", "compress": "raw", "format": "json" } } # 生成鉴权 url def create_url(self): url = RequestUrl url_result = parse.urlparse(url) now = datetime.now() date = format_date_time(mktime(now.timetuple())) #print("date:",date) # 拼接签名原文 signature_origin = "host: {}\ndate: {}\nGET {} HTTP/1.1".format( url_result.hostname, date, url_result.path ) #print("signature_origin:", signature_origin) # HMAC-SHA256 加密 signature_sha = hmac.new( self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'), digestmod=hashlib.sha256 ).digest() signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8') #print("signature_sha:", signature_sha) authorization_origin = 'api_key="%s", algorithm="%s", headers="%s", signature="%s"' % ( self.APIKey, "hmac-sha256", "host date request-line", signature_sha ) #print("authorization_origin:", authorization_origin) authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8') #print("authorization:", authorization) v = { "host": url_result.hostname, "date": date, "authorization": authorization } #print("url:",url + "?" + urlencode(v)) return url + "?" + urlencode(v) # 收到 websocket 消息的处理 def on_message(ws, message): global result_text try: print("原始返回JSON:", message) # 打印完整原始返回 message = json.loads(message) header = message.get("header", {}) code = header.get("code", -1) status = header.get("status", -1) sid = header.get("sid", "") if code != 0: print(f"请求错误 code={code}, sid={sid}") ws.close() return payload = message.get("payload") if payload: result = payload.get("result", {}) text_b64 = result.get("text") if text_b64: text_json = json.loads(base64.b64decode(text_b64).decode("utf-8")) #print(text_json) ws_list = text_json.get("ws", []) for item in ws_list: for cw in item.get("cw", []): result_text += cw.get("w", "") print("当前累积结果:", result_text) if status == STATUS_LAST_FRAME: print("最终识别结果:", result_text) ws.close() except Exception as e: print("on_message error:", e) # 收到 websocket 错误的处理 def on_error(ws, error): print("### error:", error) # 收到 websocket 关闭的处理 def on_close(ws, close_status_code, close_msg): print("### 执行结束，连接自动关闭 ###") # 收到 websocket 连接建立的处理 def on_open(ws): def run(*args): frame_size = 1024 # 每帧音频大小（字节） interval = 0.04 # 发送音频间隔（秒） status = STATUS_FIRST_FRAME seq = 0 try: with open(wsParam.AudioFile, "rb") as fp: # 跳过 WAV 文件头（44字节） wav_marker = fp.read(4) if wav_marker == b'RIFF': fp.seek(44, 0) else: fp.seek(0, 0) while True: buf = fp.read(frame_size) # 文件读取结束 if not buf: status = STATUS_LAST_FRAME audio_b64 = base64.b64encode(buf).decode('utf-8') if buf else "" # 第一帧 if status == STATUS_FIRST_FRAME: d = { "header": { "app_id": wsParam.APPID, "status": STATUS_FIRST_FRAME, "res_id": "" }, "parameter": {"iat": wsParam.iat_params}, "payload": { "audio": { "encoding": "lame", "sample_rate": 16000, "channels": 1, "bit_depth": 16, "status": STATUS_FIRST_FRAME, "seq": seq, "audio": audio_b64, } } } ws.send(json.dumps(d)) #print(json.dumps(d)) status = STATUS_CONTINUE_FRAME # 中间帧 elif status == STATUS_CONTINUE_FRAME: d = { "header": { "app_id": wsParam.APPID, "status": STATUS_CONTINUE_FRAME }, "parameter": {"iat": wsParam.iat_params}, "payload": { "audio": { "encoding": "lame", "sample_rate": 16000, "channels": 1, "bit_depth": 16, "status": STATUS_CONTINUE_FRAME, "seq": seq, "audio": audio_b64, } } } ws.send(json.dumps(d)) #print(json.dumps(d)) # 最后一帧 elif status == STATUS_LAST_FRAME: d = { "header": { "app_id": wsParam.APPID, "status": STATUS_LAST_FRAME }, "parameter": {"iat": wsParam.iat_params}, "payload": { "audio": { "encoding": "lame", "sample_rate": 16000, "channels": 1, "bit_depth": 16, "status": STATUS_LAST_FRAME, "seq": seq, "audio": audio_b64, } } } ws.send(json.dumps(d)) #print(json.dumps(d)) break seq += 1 time.sleep(interval) except FileNotFoundError: print(f"音频文件不存在: {wsParam.AudioFile}") ws.close() except Exception as e: print("发送数据异常:", e) ws.close() thread.start_new_thread(run, ()) if __name__ == "__main__": wsParam = Ws_Param( APPID=APPId, APIKey=APIKey, APISecret=APISecret, AudioFile=AudioFile ) websocket.enableTrace(False) wsUrl = wsParam.create_url() ws = websocket.WebSocketApp( wsUrl, on_message=on_message, on_error=on_error, on_close=on_close ) ws.on_open = on_open ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})