import os import wave import uuid import queue import asyncio import traceback import threading import opuslib_next import json import io import aiohttp import time import concurrent.futures from abc import ABC, abstractmethod from config.logger import setup_logging from urllib.parse import urlparse, parse_qs from typing import Optional, Tuple, List, Dict, Any from core.handle.receiveAudioHandle import startToChat from core.handle.reportHandle import enqueue_asr_report from core.utils.util import remove_punctuation_and_length from core.handle.receiveAudioHandle import handleAudioMessage TAG = __name__ logger = setup_logging() # 创建全局线程池执行器用于CPU密集型操作 executor = concurrent.futures.ThreadPoolExecutor(max_workers=4) class VoiceprintProvider: """声纹识别服务提供者""" def __init__(self, config: dict): self.original_url = config.get("url", "") self.speakers = config.get("speakers", []) self.speaker_map = self._parse_speakers() # 解析API地址和密钥 self.api_url = None self.api_key = None self.speaker_ids = [] if not self.original_url: logger.bind(tag=TAG).warning("声纹识别URL未配置，声纹识别将被禁用") self.enabled = False else: # 解析URL和key parsed_url = urlparse(self.original_url) base_url = f"{parsed_url.scheme}://{parsed_url.netloc}" # 从查询参数中提取key query_params = parse_qs(parsed_url.query) self.api_key = query_params.get('key', [''])[0] if not self.api_key: logger.bind(tag=TAG).error("URL中未找到key参数，声纹识别将被禁用") self.enabled = False else: # 构造identify接口地址 self.api_url = f"{base_url}/voiceprint/identify" # 提取speaker_ids for speaker_str in self.speakers: try: parts = speaker_str.split(",", 2) if len(parts) >= 1: speaker_id = parts[0].strip() self.speaker_ids.append(speaker_id) except Exception: continue # 检查是否有有效的说话人配置 if not self.speaker_ids: logger.bind(tag=TAG).warning("未配置有效的说话人，声纹识别将被禁用") self.enabled = False else: self.enabled = True logger.bind(tag=TAG).info(f"声纹识别已配置: API={self.api_url}, 说话人={len(self.speaker_ids)}个") def _parse_speakers(self) -> Dict[str, Dict[str, str]]: """解析说话人配置""" speaker_map = {} for speaker_str in self.speakers: try: parts = speaker_str.split(",", 2) if len(parts) >= 3: speaker_id, name, description = parts[0].strip(), parts[1].strip(), parts[2].strip() speaker_map[speaker_id] = { "name": name, "description": description } except Exception as e: logger.bind(tag=TAG).warning(f"解析说话人配置失败: {speaker_str}, 错误: {e}") return speaker_map async def identify_speaker(self, audio_data: bytes, session_id: str) -> Optional[str]: """识别说话人""" if not self.enabled or not self.api_url or not self.api_key: logger.bind(tag=TAG).debug("声纹识别功能已禁用或未配置，跳过识别") return None try: api_start_time = time.monotonic() # 准备请求头 headers = { 'Authorization': f'Bearer {self.api_key}', 'Accept': 'application/json' } # 准备multipart/form-data数据 data = aiohttp.FormData() data.add_field('speaker_ids', ','.join(self.speaker_ids)) data.add_field('file', audio_data, filename='audio.wav', content_type='audio/wav') timeout = aiohttp.ClientTimeout(total=10) # 网络请求 async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(self.api_url, headers=headers, data=data) as response: if response.status == 200: result = await response.json() speaker_id = result.get("speaker_id") score = result.get("score", 0) total_elapsed_time = time.monotonic() - api_start_time logger.bind(tag=TAG).info(f"声纹识别耗时: {total_elapsed_time:.3f}s") # 置信度检查 if score < 0.5: logger.bind(tag=TAG).warning(f"声纹识别置信度较低: {score:.3f}") if speaker_id and speaker_id in self.speaker_map: result_name = self.speaker_map[speaker_id]["name"] return result_name else: logger.bind(tag=TAG).warning(f"未识别的说话人ID: {speaker_id}") return "未知说话人" else: logger.bind(tag=TAG).error(f"声纹识别API错误: HTTP {response.status}") return None except asyncio.TimeoutError: elapsed = time.monotonic() - api_start_time logger.bind(tag=TAG).error(f"声纹识别超时: {elapsed:.3f}s") return None except Exception as e: elapsed = time.monotonic() - api_start_time logger.bind(tag=TAG).error(f"声纹识别失败: {e}") return None class ASRProviderBase(ABC): def __init__(self): self.voiceprint_provider = None def init_voiceprint(self, voiceprint_config: dict): """初始化声纹识别""" if voiceprint_config: self.voiceprint_provider = VoiceprintProvider(voiceprint_config) logger.bind(tag=TAG).info("声纹识别模块已初始化") # 打开音频通道 async def open_audio_channels(self, conn): conn.asr_priority_thread = threading.Thread( target=self.asr_text_priority_thread, args=(conn,), daemon=True ) conn.asr_priority_thread.start() # 有序处理ASR音频 def asr_text_priority_thread(self, conn): while not conn.stop_event.is_set(): try: message = conn.asr_audio_queue.get(timeout=1) future = asyncio.run_coroutine_threadsafe( handleAudioMessage(conn, message), conn.loop, ) future.result() except queue.Empty: continue except Exception as e: logger.bind(tag=TAG).error( f"处理ASR文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) continue # 接收音频 async def receive_audio(self, conn, audio, audio_have_voice): if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": have_voice = audio_have_voice else: have_voice = conn.client_have_voice conn.asr_audio.append(audio) if not have_voice and not conn.client_have_voice: conn.asr_audio = conn.asr_audio[-10:] return if conn.client_voice_stop: asr_audio_task = conn.asr_audio.copy() conn.asr_audio.clear() conn.reset_vad_states() if len(asr_audio_task) > 15: await self.handle_voice_stop(conn, asr_audio_task) # 处理语音停止 async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]): """并行处理ASR和声纹识别""" try: total_start_time = time.monotonic() # 准备音频数据 if conn.audio_format == "pcm": pcm_data = asr_audio_task else: pcm_data = self.decode_opus(asr_audio_task) combined_pcm_data = b"".join(pcm_data) # 预先准备WAV数据 wav_data = None if self.voiceprint_provider and combined_pcm_data: wav_data = self._pcm_to_wav(combined_pcm_data) # 定义ASR任务 def run_asr(): start_time = time.monotonic() try: import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete( self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format) ) end_time = time.monotonic() logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s") return result finally: loop.close() except Exception as e: end_time = time.monotonic() logger.bind(tag=TAG).error(f"ASR失败: {e}") return ("", None) # 定义声纹识别任务 def run_voiceprint(): if not wav_data: return None start_time = time.monotonic() try: import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete( self.voiceprint_provider.identify_speaker(wav_data, conn.session_id) ) return result finally: loop.close() except Exception as e: logger.bind(tag=TAG).error(f"声纹识别失败: {e}") return None # 使用线程池执行器并行运行 parallel_start_time = time.monotonic() with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: asr_future = thread_executor.submit(run_asr) if self.voiceprint_provider and wav_data: voiceprint_future = thread_executor.submit(run_voiceprint) # 等待两个线程都完成 asr_result = asr_future.result(timeout=15) voiceprint_result = voiceprint_future.result(timeout=15) results = {"asr": asr_result, "voiceprint": voiceprint_result} else: asr_result = asr_future.result(timeout=15) results = {"asr": asr_result, "voiceprint": None} parallel_execution_time = time.monotonic() - parallel_start_time # 处理结果 raw_text, file_path = results.get("asr", ("", None)) speaker_name = results.get("voiceprint", None) # 记录识别结果 if raw_text: logger.bind(tag=TAG).info(f"识别文本: {raw_text}") if speaker_name: logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}") # 性能监控 total_time = time.monotonic() - total_start_time logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s") # 检查文本长度 text_len, _ = remove_punctuation_and_length(raw_text) self.stop_ws_connection() if text_len > 0: # 构建包含说话人信息的JSON字符串 enhanced_text = self._build_enhanced_text(raw_text, speaker_name) # 使用自定义模块进行上报 await startToChat(conn, enhanced_text) enqueue_asr_report(conn, enhanced_text, asr_audio_task) except Exception as e: logger.bind(tag=TAG).error(f"处理语音停止失败: {e}") import traceback logger.bind(tag=TAG).debug(f"异常详情: {traceback.format_exc()}") def _build_enhanced_text(self, text: str, speaker_name: Optional[str]) -> str: """构建包含说话人信息的文本""" if speaker_name: return json.dumps({ "speaker": speaker_name, "content": text }, ensure_ascii=False) else: return text def _pcm_to_wav(self, pcm_data: bytes) -> bytes: """将PCM数据转换为WAV格式""" if len(pcm_data) == 0: logger.bind(tag=TAG).warning("PCM数据为空，无法转换WAV") return b"" # 确保数据长度是偶数（16位音频） if len(pcm_data) % 2 != 0: pcm_data = pcm_data[:-1] # 创建WAV文件头 wav_buffer = io.BytesIO() try: with wave.open(wav_buffer, 'wb') as wav_file: wav_file.setnchannels(1) # 单声道 wav_file.setsampwidth(2) # 16位 wav_file.setframerate(16000) # 16kHz采样率 wav_file.writeframes(pcm_data) wav_buffer.seek(0) wav_data = wav_buffer.read() return wav_data except Exception as e: logger.bind(tag=TAG).error(f"WAV转换失败: {e}") return b"" def stop_ws_connection(self): pass def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str: """PCM数据保存为WAV文件""" module_name = __name__.split(".")[-1] file_name = f"asr_{module_name}_{session_id}_{uuid.uuid4()}.wav" file_path = os.path.join(self.output_dir, file_name) with wave.open(file_path, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(b"".join(pcm_data)) return file_path @abstractmethod async def speech_to_text( self, opus_data: List[bytes], session_id: str, audio_format="opus" ) -> Tuple[Optional[str], Optional[str]]: """将语音数据转换为文本""" pass @staticmethod def decode_opus(opus_data: List[bytes]) -> List[bytes]: """将Opus音频数据解码为PCM数据""" try: decoder = opuslib_next.Decoder(16000, 1) pcm_data = [] buffer_size = 960 # 每次处理960个采样点 (60ms at 16kHz) for i, opus_packet in enumerate(opus_data): try: if not opus_packet or len(opus_packet) == 0: continue pcm_frame = decoder.decode(opus_packet, buffer_size) if pcm_frame and len(pcm_frame) > 0: pcm_data.append(pcm_frame) except opuslib_next.OpusError as e: logger.bind(tag=TAG).warning(f"Opus解码错误，跳过数据包 {i}: {e}") except Exception as e: logger.bind(tag=TAG).error(f"音频处理错误，数据包 {i}: {e}") return pcm_data except Exception as e: logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}") return []