import json import gzip import uuid import asyncio import websockets from core.providers.asr.base import ASRProviderBase from config.logger import setup_logging from core.providers.asr.dto.dto import InterfaceType from typing import TYPE_CHECKING if TYPE_CHECKING: from core.connection import ConnectionHandler TAG = __name__ logger = setup_logging() class ASRProvider(ASRProviderBase): def __init__(self, config, delete_audio_file): super().__init__() self.interface_type = InterfaceType.STREAM self.config = config self.text = "" self.asr_ws = None self.forward_task = None self.is_processing = False # 添加处理状态标志 self._is_stopping = False # 添加停止标志，防止竞态条件 # 配置参数 self.appid = str(config.get("appid")) self.access_token = config.get("access_token") # 资源ID，用于区分不同的ASR模型（默认1.0模型小时版，v2版本使用seed-asr） self.resource_id = config.get("resource_id", "volc.bigasr.sauc.duration") self.boosting_table_name = config.get("boosting_table_name", "") self.correct_table_name = config.get("correct_table_name", "") self.output_dir = config.get("output_dir", "tmp/") self.delete_audio_file = delete_audio_file # 火山引擎ASR配置 enable_multilingual = config.get("enable_multilingual", False) self.enable_multilingual = ( False if str(enable_multilingual).lower() == "false" else True ) if self.enable_multilingual: self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream" else: self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_async" self.uid = config.get("uid", "streaming_asr_service") self.workflow = config.get( "workflow", "audio_in,resample,partition,vad,fe,decode,itn,nlu_punctuate" ) self.result_type = config.get("result_type", "single") self.format = config.get("format", "pcm") self.codec = config.get("codec", "pcm") self.rate = config.get("sample_rate", 16000) # language参数仅在多语种模式(bigmodel_nostream)下有效 self.language = config.get("language") if self.enable_multilingual else None self.bits = config.get("bits", 16) self.channel = config.get("channel", 1) self.auth_method = config.get("auth_method", "token") self.secret = config.get("secret", "access_secret") end_window_size = config.get("end_window_size") self.end_window_size = int(end_window_size) if end_window_size else 200 async def open_audio_channels(self, conn): await super().open_audio_channels(conn) async def receive_audio(self, conn: "ConnectionHandler", pcm_frame, audio_have_voice): # 先调用父类方法处理基础逻辑 await super().receive_audio(conn, pcm_frame, audio_have_voice) # 如果本次有声音，且之前没有建立连接 if audio_have_voice and self.asr_ws is None and not self.is_processing: try: self.is_processing = True # 建立新的WebSocket连接 headers = self.token_auth() if self.auth_method == "token" else None logger.bind(tag=TAG).info(f"正在连接ASR服务，headers: {headers}") self.asr_ws = await websockets.connect( self.ws_url, additional_headers=headers, max_size=1000000000, ping_interval=None, ping_timeout=None, close_timeout=10, ) # 发送初始化请求 request_params = self.construct_request(str(uuid.uuid4())) try: payload_bytes = str.encode(json.dumps(request_params)) payload_bytes = gzip.compress(payload_bytes) full_client_request = self.generate_header() full_client_request.extend((len(payload_bytes)).to_bytes(4, "big")) full_client_request.extend(payload_bytes) logger.bind(tag=TAG).info(f"发送初始化请求: {request_params}") await self.asr_ws.send(full_client_request) # 等待初始化响应 init_res = await self.asr_ws.recv() result = self.parse_response(init_res) logger.bind(tag=TAG).info(f"收到初始化响应: {result}") # 检查初始化响应 if "code" in result and result["code"] != 1000: error_msg = f"ASR服务初始化失败: {result.get('payload_msg', {}).get('error', '未知错误')}" logger.bind(tag=TAG).error(error_msg) raise Exception(error_msg) except Exception as e: logger.bind(tag=TAG).error(f"发送初始化请求失败: {str(e)}") if hasattr(e, "__cause__") and e.__cause__: logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}") raise e # 启动接收ASR结果的异步任务 self.forward_task = asyncio.create_task(self._forward_asr_results(conn)) # 发送缓存的音频数据 if conn.asr_audio and len(conn.asr_audio) > 0: for cached_pcm in conn.asr_audio[-10:]: try: payload = gzip.compress(cached_pcm) audio_request = bytearray( self.generate_audio_default_header() ) audio_request.extend(len(payload).to_bytes(4, "big")) audio_request.extend(payload) await self.asr_ws.send(audio_request) except Exception as e: logger.bind(tag=TAG).info( f"发送缓存音频数据时发生错误: {e}" ) except Exception as e: logger.bind(tag=TAG).error(f"建立ASR连接失败: {str(e)}") if hasattr(e, "__cause__") and e.__cause__: logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}") if self.asr_ws: await self.asr_ws.close() self.asr_ws = None self.is_processing = False return # 发送当前音频数据 if self.asr_ws and self.is_processing and not self._is_stopping: try: payload = gzip.compress(pcm_frame) audio_request = bytearray(self.generate_audio_default_header()) audio_request.extend(len(payload).to_bytes(4, "big")) audio_request.extend(payload) await self.asr_ws.send(audio_request) except Exception as e: logger.bind(tag=TAG).info(f"发送音频数据时发生错误: {e}") async def _forward_asr_results(self, conn: "ConnectionHandler"): try: while self.asr_ws and not conn.stop_event.is_set(): # 获取当前连接的音频数据 audio_data = conn.asr_audio try: response = await self.asr_ws.recv() result = self.parse_response(response) logger.bind(tag=TAG).debug(f"收到ASR结果: {result}") if "payload_msg" in result: payload = result["payload_msg"] # 检查是否是错误码1013（无有效语音） if "code" in payload and payload["code"] == 1013: # 静默处理，不记录错误日志 continue if "result" in payload: utterances = payload["result"].get("utterances", []) # 检查duration和空文本的情况 if ( not self.enable_multilingual # 注意：多语种模式不返回中间结果，需要等待最终结果 and payload.get("audio_info", {}).get("duration", 0) > 2000 and not utterances and not payload["result"].get("text") and conn.client_listen_mode != "manual" ): logger.bind(tag=TAG).error(f"识别文本：空") self.text = "" if len(audio_data) > 15: # 确保有足够音频数据 await self.handle_voice_stop(conn, audio_data) break # 专门处理没有文本的识别结果（手动模式下可能已经识别完成但是没松按键） elif not payload["result"].get("text") and not utterances: # 多语种模式会持续返回空文本，直到最后返回完整结果，所以需要排除 if self.enable_multilingual: continue if conn.client_listen_mode == "manual" and conn.client_voice_stop and len(audio_data) > 15: logger.bind(tag=TAG).debug("消息结束收到停止信号，触发处理") await self.handle_voice_stop(conn, audio_data) break for utterance in utterances: if utterance.get("definite", False): current_text = utterance["text"] logger.bind(tag=TAG).info( f"识别到文本: {current_text}" ) # 手动模式下累积识别结果 if conn.client_listen_mode == "manual": if self.text: self.text += current_text else: self.text = current_text # 在接收消息中途时收到停止信号 if conn.client_voice_stop and len(audio_data) > 0: logger.bind(tag=TAG).debug("消息中途收到停止信号，触发处理") await self.handle_voice_stop(conn, audio_data) break else: # 自动模式下直接覆盖 self.text = current_text if len(audio_data) > 15: # 确保有足够音频数据 await self.handle_voice_stop( conn, audio_data ) break elif "error" in payload: error_msg = payload.get("error", "未知错误") logger.bind(tag=TAG).error(f"ASR服务返回错误: {error_msg}") break except websockets.ConnectionClosed: logger.bind(tag=TAG).info("ASR服务连接已关闭") self.is_processing = False break except Exception as e: logger.bind(tag=TAG).error(f"处理ASR结果时发生错误: {str(e)}") if hasattr(e, "__cause__") and e.__cause__: logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}") self.is_processing = False break except Exception as e: logger.bind(tag=TAG).error(f"ASR结果转发任务发生错误: {str(e)}") if hasattr(e, "__cause__") and e.__cause__: logger.bind(tag=TAG).error(f"错误原因: {str(e.__cause__)}") finally: if self.asr_ws: await self.asr_ws.close() self.asr_ws = None self.is_processing = False self._is_stopping = False # 重置所有音频相关状态 conn.reset_audio_states() def stop_ws_connection(self): if self.asr_ws: asyncio.create_task(self.asr_ws.close()) self.asr_ws = None self.is_processing = False self._is_stopping = False async def _send_stop_request(self): """发送最后一个音频帧以通知服务器结束""" self._is_stopping = True # 先标记为停止状态，阻止后续音频发送 if self.asr_ws: try: # 发送结束标记的音频帧（gzip压缩的空数据） empty_payload = gzip.compress(b"") last_audio_request = bytearray( self.generate_last_audio_default_header() ) last_audio_request.extend(len(empty_payload).to_bytes(4, "big")) last_audio_request.extend(empty_payload) await self.asr_ws.send(last_audio_request) logger.bind(tag=TAG).debug("已发送结束音频帧") except Exception as e: logger.bind(tag=TAG).debug(f"发送结束音频帧时出错: {e}") def construct_request(self, reqid): req = { "app": { "appid": self.appid, "token": self.access_token, }, "user": {"uid": self.uid}, "request": { "reqid": reqid, "workflow": self.workflow, "show_utterances": True, "result_type": self.result_type, "sequence": 1, "end_window_size": self.end_window_size, "corpus": { "boosting_table_name": self.boosting_table_name, "correct_table_name": self.correct_table_name, } }, "audio": { "format": self.format, "codec": self.codec, "rate": self.rate, "bits": self.bits, "channel": self.channel, "sample_rate": self.rate, }, } # language参数仅在多语种模式下添加 if self.enable_multilingual and self.language: req["audio"]["language"] = self.language logger.bind(tag=TAG).debug( f"构造请求参数: {json.dumps(req, ensure_ascii=False)}" ) return req def token_auth(self): return { "X-Api-App-Key": self.appid, "X-Api-Access-Key": self.access_token, "X-Api-Resource-Id": self.resource_id, "X-Api-Connect-Id": str(uuid.uuid4()), } def generate_header( self, version=0x01, message_type=0x01, message_type_specific_flags=0x00, serial_method=0x01, compression_type=0x01, reserved_data=0x00, extension_header: bytes = b"", ): header = bytearray() header_size = int(len(extension_header) / 4) + 1 header.append((version << 4) | header_size) header.append((message_type << 4) | message_type_specific_flags) header.append((serial_method << 4) | compression_type) header.append(reserved_data) header.extend(extension_header) return header def generate_audio_default_header(self): return self.generate_header( version=0x01, message_type=0x02, message_type_specific_flags=0x00, serial_method=0x01, compression_type=0x01, ) def generate_last_audio_default_header(self): return self.generate_header( version=0x01, message_type=0x02, message_type_specific_flags=0x02, serial_method=0x01, compression_type=0x01, ) def parse_response(self, res: bytes) -> dict: try: # 检查响应长度 if len(res) < 4: logger.bind(tag=TAG).error(f"响应数据长度不足: {len(res)}") return {"error": "响应数据长度不足"} # 获取消息头 header = res[:4] message_type = header[1] >> 4 # 如果是错误响应 if message_type == 0x0F: # SERVER_ERROR_RESPONSE code = int.from_bytes(res[4:8], "big", signed=False) msg_length = int.from_bytes(res[8:12], "big", signed=False) error_msg = json.loads(res[12:].decode("utf-8")) return { "code": code, "msg_length": msg_length, "payload_msg": error_msg, } # 获取JSON数据 try: # 检查字节8-11是否为有效的JSON长度字段 # 格式：4字节头 + 4字节序列号 + 4字节长度 + JSON数据 length = int.from_bytes(res[8:12], "big") if length > 0 and length <= len(res) - 12: # 有长度字段，从字节12开始读取指定长度的JSON json_data = res[12:12 + length].decode("utf-8") else: # 无长度字段或长度无效，尝试直接解析 json_data = res[8:].decode("utf-8") result = json.loads(json_data) logger.bind(tag=TAG).debug(f"成功解析JSON响应: {result}") return {"payload_msg": result} except (UnicodeDecodeError, json.JSONDecodeError) as e: logger.bind(tag=TAG).error(f"JSON解析失败: {str(e)}") logger.bind(tag=TAG).error(f"原始数据: {res}") raise except Exception as e: logger.bind(tag=TAG).error(f"解析响应失败: {str(e)}") logger.bind(tag=TAG).error(f"原始响应数据: {res.hex()}") raise async def speech_to_text(self, opus_data, session_id, artifacts=None): result = self.text self.text = "" # 清空text return result, None async def close(self): """资源清理方法""" if self.asr_ws: await self.asr_ws.close() self.asr_ws = None if self.forward_task: self.forward_task.cancel() try: await self.forward_task except asyncio.CancelledError: pass self.forward_task = None self.is_processing = False