refactor(asr): 统一音频预处理逻辑并引入AudioArtifacts

重构所有ASR提供商的speech_to_text方法，将重复的音频解码、合并和文件保存逻辑提取到基类的speech_to_text_wrapper中。引入AudioArtifacts数据类封装PCM帧、字节数据、文件路径和临时路径，简化各提供商实现。移除各提供商中的冗余文件清理代码，由基类统一处理。

新增requires_file()和prefers_temp_file()方法允许提供商声明文件需求，优化内存和磁盘使用。保持接口兼容性的同时提高代码复用性和可维护性。

main/xiaozhi-server/core/providers/asr/base.py

@@ -8,14 +8,16 @@ import queue
 import asyncio
 import traceback
 import threading
+import shutil
 import opuslib_next
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
-from typing import Optional, Tuple, List
+from typing import Optional, Tuple, List, NamedTuple
 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
+import tempfile
 
 TAG = __name__
 logger = setup_logging()
@@ -23,7 +25,8 @@ logger = setup_logging()
 
 class ASRProviderBase(ABC):
     def __init__(self):
-        pass
+        self._current_artifacts: Optional[ASRProviderBase.AudioArtifacts] = None
+        """当前正在处理的音频 artifact"""
 
     # 打开音频通道
     async def open_audio_channels(self, conn):
@@ -93,10 +96,14 @@ class ASRProviderBase(ABC):
                 wav_data = self._pcm_to_wav(combined_pcm_data)
 
             # 定义ASR任务
-            asr_task = self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
+            asr_task = self.speech_to_text_wrapper(
+                asr_audio_task, conn.session_id, conn.audio_format
+            )
 
             if conn.voiceprint_provider and wav_data:
-                voiceprint_task = conn.voiceprint_provider.identify_speaker(wav_data, conn.session_id)
+                voiceprint_task = conn.voiceprint_provider.identify_speaker(
+                    wav_data, conn.session_id
+                )
                 # 并发等待两个结果
                 asr_result, voiceprint_result = await asyncio.gather(
                     asr_task, voiceprint_task, return_exceptions=True
@@ -160,19 +167,19 @@ class ASRProviderBase(ABC):
                 # 使用自定义模块进行上报
                 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:
         """构建包含说话人信息的文本（仅用于纯文本ASR）"""
         if speaker_name and speaker_name.strip():
-            return json.dumps({
-                "speaker": speaker_name,
-                "content": text
-            }, ensure_ascii=False)
+            return json.dumps(
+                {"speaker": speaker_name, "content": text}, ensure_ascii=False
+            )
         else:
             return text
 
@@ -181,23 +188,23 @@ class ASRProviderBase(ABC):
         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位
+            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}")
@@ -206,6 +213,37 @@ class ASRProviderBase(ABC):
     def stop_ws_connection(self):
         pass
 
+    class AudioArtifacts(NamedTuple):
+        pcm_frames: List[bytes]
+        pcm_bytes: bytes
+        file_path: Optional[str]
+        temp_path: Optional[str]
+
+    def get_current_artifacts(self) -> Optional["ASRProviderBase.AudioArtifacts"]:
+        return self._current_artifacts
+
+    def requires_file(self) -> bool:
+        """是否需要文件输入"""
+        return False
+
+    def prefers_temp_file(self) -> bool:
+        """是否优先使用临时文件"""
+        return False
+
+    def build_temp_file(self, pcm_bytes: bytes) -> Optional[str]:
+        try:
+            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
+                temp_path = temp_file.name
+            with wave.open(temp_path, "wb") as wav_file:
+                wav_file.setnchannels(1)
+                wav_file.setsampwidth(2)
+                wav_file.setframerate(16000)
+                wav_file.writeframes(pcm_bytes)
+            return temp_path
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"临时音频文件生成失败: {e}")
+            return None
+
     def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
         """PCM数据保存为WAV文件"""
         module_name = __name__.split(".")[-1]
@@ -220,6 +258,59 @@ class ASRProviderBase(ABC):
 
         return file_path
 
+    async def speech_to_text_wrapper(
+        self, opus_data: List[bytes], session_id: str, audio_format="opus"
+    ) -> Tuple[Optional[str], Optional[str]]:
+        file_path = None
+        temp_path = None
+        try:
+            if audio_format == "pcm":
+                pcm_data = opus_data
+            else:
+                pcm_data = self.decode_opus(opus_data)
+            combined_pcm_data = b"".join(pcm_data)
+
+            free_space = shutil.disk_usage(self.output_dir).free
+            if free_space < len(combined_pcm_data) * 2:
+                raise OSError("磁盘空间不足")
+
+            if self.requires_file() and self.prefers_temp_file():
+                temp_path = self.build_temp_file(combined_pcm_data)
+
+            if (hasattr(self, "delete_audio_file") and not self.delete_audio_file) or (
+                self.requires_file() and not self.prefers_temp_file()
+            ):
+                file_path = self.save_audio_to_file(pcm_data, session_id)
+
+            self._current_artifacts = ASRProviderBase.AudioArtifacts(
+                pcm_frames=pcm_data,
+                pcm_bytes=combined_pcm_data,
+                file_path=file_path,
+                temp_path=temp_path,
+            )
+
+            text, _ = await self.speech_to_text(opus_data, session_id, audio_format)
+            return text, file_path
+        except OSError as e:
+            logger.bind(tag=TAG).error(f"文件操作错误: {e}")
+            return None, None
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"语音识别失败: {e}")
+            return None, None
+        finally:
+            try:
+                if temp_path and os.path.exists(temp_path):
+                    os.unlink(temp_path)
+                if (
+                    hasattr(self, "delete_audio_file")
+                    and self.delete_audio_file
+                    and file_path
+                    and os.path.exists(file_path)
+                ):
+                    os.remove(file_path)
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"文件清理失败: {e}")
+
     @abstractmethod
     async def speech_to_text(
         self, opus_data: List[bytes], session_id: str, audio_format="opus"
@@ -235,23 +326,23 @@ class ASRProviderBase(ABC):
             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 []