import base64 import hashlib import hmac import json import time from datetime import datetime, timezone import os from typing import Optional, Tuple, List from core.providers.asr.dto.dto import InterfaceType import requests from core.providers.asr.base import ASRProviderBase from config.logger import setup_logging TAG = __name__ logger = setup_logging() class ASRProvider(ASRProviderBase): API_URL = "https://asr.tencentcloudapi.com" API_VERSION = "2019-06-14" FORMAT = "pcm" # 支持的音频格式：pcm, wav, mp3 def __init__(self, config: dict, delete_audio_file: bool = True): super().__init__() self.interface_type = InterfaceType.NON_STREAM self.secret_id = config.get("secret_id") self.secret_key = config.get("secret_key") self.output_dir = config.get("output_dir") self.delete_audio_file = delete_audio_file # 确保输出目录存在 os.makedirs(self.output_dir, exist_ok=True) async def speech_to_text( self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" if not opus_data: logger.bind(tag=TAG).warning("音频数据为空！") return None, None file_path = None try: # 检查配置是否已设置 if not self.secret_id or not self.secret_key: logger.bind(tag=TAG).error("腾讯云语音识别配置未设置，无法进行识别") return None, file_path # 将Opus音频数据解码为PCM if audio_format == "pcm": pcm_data = opus_data else: pcm_data = self.decode_opus(opus_data) combined_pcm_data = b"".join(pcm_data) # 判断是否保存为WAV文件 if self.delete_audio_file: pass else: self.save_audio_to_file(pcm_data, session_id) # 将音频数据转换为Base64编码 base64_audio = base64.b64encode(combined_pcm_data).decode("utf-8") # 构建请求体 request_body = self._build_request_body(base64_audio) # 获取认证头 timestamp, authorization = self._get_auth_headers(request_body) # 发送请求 start_time = time.time() result = self._send_request(request_body, timestamp, authorization) if result: logger.bind(tag=TAG).debug( f"腾讯云语音识别耗时: {time.time() - start_time:.3f}s | 结果: {result}" ) return result, file_path except Exception as e: logger.bind(tag=TAG).error(f"处理音频时发生错误！{e}", exc_info=True) return None, file_path def _build_request_body(self, base64_audio: str) -> str: """构建请求体""" request_map = { "ProjectId": 0, "SubServiceType": 2, # 一句话识别 "EngSerViceType": "16k_zh", # 中文普通话通用 "SourceType": 1, # 音频数据来源为语音文件 "VoiceFormat": self.FORMAT, # 音频格式 "Data": base64_audio, # Base64编码的音频数据 "DataLen": len(base64_audio), # 数据长度 } return json.dumps(request_map) def _get_auth_headers(self, request_body: str) -> Tuple[str, str]: """获取认证头""" try: # 获取当前UTC时间戳 now = datetime.now(timezone.utc) timestamp = str(int(now.timestamp())) date = now.strftime("%Y-%m-%d") # 服务名称必须是 "asr" service = "asr" # 拼接凭证范围 credential_scope = f"{date}/{service}/tc3_request" # 使用TC3-HMAC-SHA256签名方法 algorithm = "TC3-HMAC-SHA256" # 构建规范请求字符串 http_request_method = "POST" canonical_uri = "/" canonical_query_string = "" # 注意：头部信息需要按照ASCII升序排列，且key和value都转为小写 # 必须包含content-type和host头部 content_type = "application/json; charset=utf-8" host = "asr.tencentcloudapi.com" action = "SentenceRecognition" # 接口名称 # 构建规范头部信息，注意顺序和格式 canonical_headers = ( f"content-type:{content_type.lower()}\n" + f"host:{host.lower()}\n" + f"x-tc-action:{action.lower()}\n" ) signed_headers = "content-type;host;x-tc-action" # 请求体哈希值 payload_hash = self._sha256_hex(request_body) # 构建规范请求字符串 canonical_request = ( f"{http_request_method}\n" + f"{canonical_uri}\n" + f"{canonical_query_string}\n" + f"{canonical_headers}\n" + f"{signed_headers}\n" + f"{payload_hash}" ) # 计算规范请求的哈希值 hashed_canonical_request = self._sha256_hex(canonical_request) # 构建待签名字符串 string_to_sign = ( f"{algorithm}\n" + f"{timestamp}\n" + f"{credential_scope}\n" + f"{hashed_canonical_request}" ) # 计算签名密钥 secret_date = self._hmac_sha256(f"TC3{self.secret_key}", date) secret_service = self._hmac_sha256(secret_date, service) secret_signing = self._hmac_sha256(secret_service, "tc3_request") # 计算签名 signature = self._bytes_to_hex( self._hmac_sha256(secret_signing, string_to_sign) ) # 构建授权头 authorization = ( f"{algorithm} " + f"Credential={self.secret_id}/{credential_scope}, " + f"SignedHeaders={signed_headers}, " + f"Signature={signature}" ) return timestamp, authorization except Exception as e: logger.bind(tag=TAG).error(f"生成认证头失败: {e}", exc_info=True) raise RuntimeError(f"生成认证头失败: {e}") def _send_request( self, request_body: str, timestamp: str, authorization: str ) -> Optional[str]: """发送请求到腾讯云API""" headers = { "Content-Type": "application/json; charset=utf-8", "Host": "asr.tencentcloudapi.com", "Authorization": authorization, "X-TC-Action": "SentenceRecognition", "X-TC-Version": self.API_VERSION, "X-TC-Timestamp": timestamp, "X-TC-Region": "ap-shanghai", } try: response = requests.post(self.API_URL, headers=headers, data=request_body) if not response.ok: raise IOError(f"请求失败: {response.status_code} {response.reason}") response_json = response.json() # 检查是否有错误 if "Response" in response_json and "Error" in response_json["Response"]: error = response_json["Response"]["Error"] error_code = error["Code"] error_message = error["Message"] raise IOError(f"API返回错误: {error_code}: {error_message}") # 提取识别结果 if "Response" in response_json and "Result" in response_json["Response"]: return response_json["Response"]["Result"] else: logger.bind(tag=TAG).warning(f"响应中没有识别结果: {response_json}") return "" except Exception as e: logger.bind(tag=TAG).error(f"发送请求失败: {e}", exc_info=True) return None def _sha256_hex(self, data: str) -> str: """计算字符串的SHA256哈希值""" digest = hashlib.sha256(data.encode("utf-8")).digest() return self._bytes_to_hex(digest) def _hmac_sha256(self, key, data: str) -> bytes: """计算HMAC-SHA256""" if isinstance(key, str): key = key.encode("utf-8") return hmac.new(key, data.encode("utf-8"), hashlib.sha256).digest() def _bytes_to_hex(self, bytes_data: bytes) -> str: """字节数组转十六进制字符串""" return "".join(f"{b:02x}" for b in bytes_data)