import json import time import uuid import hmac import base64 import hashlib import asyncio import requests import websockets from urllib import parse from datetime import datetime from config.logger import setup_logging from core.providers.asr.base import ASRProviderBase 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 AccessToken: @staticmethod def _encode_text(text): encoded_text = parse.quote_plus(text) return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") @staticmethod def _encode_dict(dic): keys = dic.keys() dic_sorted = [(key, dic[key]) for key in sorted(keys)] encoded_text = parse.urlencode(dic_sorted) return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") @staticmethod def create_token(access_key_id, access_key_secret): parameters = { "AccessKeyId": access_key_id, "Action": "CreateToken", "Format": "JSON", "RegionId": "cn-shanghai", "SignatureMethod": "HMAC-SHA1", "SignatureNonce": str(uuid.uuid1()), "SignatureVersion": "1.0", "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "Version": "2019-02-28", } query_string = AccessToken._encode_dict(parameters) string_to_sign = ( "GET" + "&" + AccessToken._encode_text("/") + "&" + AccessToken._encode_text(query_string) ) secreted_string = hmac.new( bytes(access_key_secret + "&", encoding="utf-8"), bytes(string_to_sign, encoding="utf-8"), hashlib.sha1, ).digest() signature = base64.b64encode(secreted_string) signature = AccessToken._encode_text(signature) full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % (signature, query_string) response = requests.get(full_url) if response.ok: root_obj = response.json() if "Token" in root_obj: return root_obj["Token"]["Id"], root_obj["Token"]["ExpireTime"] return None, None 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.server_ready = False # 服务器准备状态 # 基础配置 self.access_key_id = config.get("access_key_id") self.access_key_secret = config.get("access_key_secret") self.appkey = config.get("appkey") self.token = config.get("token") self.host = config.get("host", "nls-gateway-cn-shanghai.aliyuncs.com") # 如果配置的是内网地址（包含-internal.aliyuncs.com），则使用ws协议，默认是wss协议 if "-internal." in self.host: self.ws_url = f"ws://{self.host}/ws/v1" else: # 默认使用wss协议 self.ws_url = f"wss://{self.host}/ws/v1" self.max_sentence_silence = config.get("max_sentence_silence") self.output_dir = config.get("output_dir", "./audio_output") self.delete_audio_file = delete_audio_file self.expire_time = None self.task_id = uuid.uuid4().hex # Token管理 if self.access_key_id and self.access_key_secret: self._refresh_token() elif not self.token: raise ValueError("必须提供access_key_id+access_key_secret或者直接提供token") def _refresh_token(self): """刷新Token""" self.token, expire_time_str = AccessToken.create_token(self.access_key_id, self.access_key_secret) if not self.token: raise ValueError("无法获取有效的访问Token") try: expire_str = str(expire_time_str).strip() if expire_str.isdigit(): expire_time = datetime.fromtimestamp(int(expire_str)) else: expire_time = datetime.strptime(expire_str, "%Y-%m-%dT%H:%M:%SZ") self.expire_time = expire_time.timestamp() - 60 except: self.expire_time = None def _is_token_expired(self): """检查Token是否过期""" return self.expire_time and time.time() > self.expire_time async def open_audio_channels(self, conn): await super().open_audio_channels(conn) async def receive_audio(self, conn, pcm_frame, audio_have_voice): # 先调用父类方法处理基础逻辑 await super().receive_audio(conn, pcm_frame, audio_have_voice) # 只在有声音且没有连接时建立连接（排除正在停止的情况） if audio_have_voice and not self.is_processing and not self.asr_ws: try: await self._start_recognition(conn) except Exception as e: logger.bind(tag=TAG).error(f"开始识别失败: {str(e)}") await self._cleanup() return if self.asr_ws and self.is_processing and self.server_ready: try: await self.asr_ws.send(pcm_frame) except Exception as e: logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}") await self._cleanup() async def _start_recognition(self, conn: "ConnectionHandler"): """开始识别会话""" if self._is_token_expired(): self._refresh_token() # 建立连接 headers = {"X-NLS-Token": self.token} self.asr_ws = await websockets.connect( self.ws_url, additional_headers=headers, max_size=1000000000, ping_interval=None, ping_timeout=None, close_timeout=5, ) self.task_id = uuid.uuid4().hex logger.bind(tag=TAG).debug(f"WebSocket连接建立成功, task_id: {self.task_id}") self.is_processing = True self.server_ready = False # 重置服务器准备状态 self.forward_task = asyncio.create_task(self._forward_results(conn)) # 发送开始请求 start_request = { "header": { "namespace": "SpeechTranscriber", "name": "StartTranscription", "message_id": uuid.uuid4().hex, "task_id": self.task_id, "appkey": self.appkey }, "payload": { "format": "pcm", "sample_rate": 16000, "enable_intermediate_result": True, "enable_punctuation_prediction": True, "enable_inverse_text_normalization": True, "max_sentence_silence": self.max_sentence_silence, "enable_voice_detection": False, } } await self.asr_ws.send(json.dumps(start_request, ensure_ascii=False)) logger.bind(tag=TAG).debug("已发送开始请求，等待服务器准备...") async def _forward_results(self, conn: "ConnectionHandler"): """转发识别结果""" try: while not conn.stop_event.is_set(): # 获取当前连接的音频数据 audio_data = conn.asr_audio try: response = await self.asr_ws.recv() result = json.loads(response) header = result.get("header", {}) payload = result.get("payload", {}) message_name = header.get("name", "") status = header.get("status", 0) if status != 20000000: if status == 40010004: logger.bind(tag=TAG).warning(f"请在服务端响应完成后再关闭链接，状态码: {status}") break if status in [40000004, 40010003]: # 连接超时或客户端断开 logger.bind(tag=TAG).warning(f"连接问题，状态码: {status}") break elif status in [40270002, 40270003]: # 音频问题 logger.bind(tag=TAG).warning(f"音频处理问题，状态码: {status}") continue else: logger.bind(tag=TAG).error(f"识别错误，状态码: {status}, 消息: {header.get('status_text', '')}") continue # 收到TranscriptionStarted表示服务器准备好接收音频数据 if message_name == "TranscriptionStarted": self.server_ready = True logger.bind(tag=TAG).debug("服务器已准备，开始发送缓存音频...") # 发送缓存音频 if conn.asr_audio: for cached_pcm in conn.asr_audio[-10:]: try: await self.asr_ws.send(cached_pcm) except Exception as e: logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}") break continue elif message_name == "SentenceEnd": # 句子结束（每个句子都会触发） text = payload.get("result", "") if text: logger.bind(tag=TAG).info(f"识别到文本: {text}") # 手动模式下累积识别结果 if conn.client_listen_mode == "manual": if self.text: self.text += text else: self.text = text # 手动模式下，只有在收到stop信号后才触发处理（仅处理一次） if conn.client_voice_stop: logger.bind(tag=TAG).debug("收到最终识别结果，触发处理") await self.handle_voice_stop(conn, audio_data) break else: # 自动模式下直接覆盖 self.text = text await self.handle_voice_stop(conn, audio_data) break except asyncio.TimeoutError: logger.bind(tag=TAG).error("接收结果超时") 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"处理结果失败: {str(e)}") break except Exception as e: logger.bind(tag=TAG).error(f"结果转发失败: {str(e)}") finally: # 清理连接的音频缓存 await self._cleanup() conn.reset_audio_states() async def _send_stop_request(self): """发送停止识别请求（不关闭连接）""" if self.asr_ws: try: # 先停止音频发送 self.is_processing = False stop_msg = { "header": { "namespace": "SpeechTranscriber", "name": "StopTranscription", "message_id": uuid.uuid4().hex, "task_id": self.task_id, "appkey": self.appkey } } logger.bind(tag=TAG).debug("停止识别请求已发送") await self.asr_ws.send(json.dumps(stop_msg, ensure_ascii=False)) except Exception as e: logger.bind(tag=TAG).error(f"发送停止识别请求失败: {e}") async def _cleanup(self): """清理资源（关闭连接）""" logger.bind(tag=TAG).debug(f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}") # 状态重置 self.is_processing = False self.server_ready = False logger.bind(tag=TAG).debug("ASR状态已重置") # 关闭连接 if self.asr_ws: try: logger.bind(tag=TAG).debug("正在关闭WebSocket连接") await asyncio.wait_for(self.asr_ws.close(), timeout=2.0) logger.bind(tag=TAG).debug("WebSocket连接已关闭") except Exception as e: logger.bind(tag=TAG).error(f"关闭WebSocket连接失败: {e}") finally: self.asr_ws = None # 清理任务引用 self.forward_task = None logger.bind(tag=TAG).debug("ASR会话清理完成") async def speech_to_text(self, opus_data, session_id, artifacts=None): """获取识别结果""" result = self.text self.text = "" return result, None async def close(self): """关闭资源""" await self._cleanup()