mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-28 10:03:54 +08:00
huozaimengli
15650e1a6c
重构所有ASR提供商的speech_to_text方法，将重复的音频解码、合并和文件保存逻辑提取到基类的speech_to_text_wrapper中。引入AudioArtifacts数据类封装PCM帧、字节数据、文件路径和临时路径，简化各提供商实现。移除各提供商中的冗余文件清理代码，由基类统一处理。 新增requires_file()和prefers_temp_file()方法允许提供商声明文件需求，优化内存和磁盘使用。保持接口兼容性的同时提高代码复用性和可维护性。
72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
import time
|
|
import os
|
|
from config.logger import setup_logging
|
|
from typing import Optional, Tuple, List
|
|
from core.providers.asr.dto.dto import InterfaceType
|
|
from core.providers.asr.base import ASRProviderBase
|
|
|
|
import requests
|
|
|
|
TAG = __name__
|
|
logger = setup_logging()
|
|
|
|
class ASRProvider(ASRProviderBase):
|
|
def __init__(self, config: dict, delete_audio_file: bool):
|
|
self.interface_type = InterfaceType.NON_STREAM
|
|
self.api_key = config.get("api_key")
|
|
self.api_url = config.get("base_url")
|
|
self.model = config.get("model_name")
|
|
self.output_dir = config.get("output_dir")
|
|
self.delete_audio_file = delete_audio_file
|
|
|
|
os.makedirs(self.output_dir, exist_ok=True)
|
|
|
|
def requires_file(self) -> bool:
|
|
return True
|
|
|
|
async def speech_to_text(self, opus_data: List[bytes], session_id: str, audio_format="opus") -> Tuple[Optional[str], Optional[str]]:
|
|
file_path = None
|
|
try:
|
|
artifacts = self.get_current_artifacts()
|
|
if artifacts is None:
|
|
return "", None
|
|
file_path = artifacts.file_path
|
|
|
|
logger.bind(tag=TAG).info(f"file path: {file_path}")
|
|
headers = {
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
}
|
|
|
|
# 使用data参数传递模型名称
|
|
data = {
|
|
"model": self.model
|
|
}
|
|
|
|
|
|
with open(file_path, "rb") as audio_file: # 使用with语句确保文件关闭
|
|
files = {
|
|
"file": audio_file
|
|
}
|
|
|
|
start_time = time.time()
|
|
response = requests.post(
|
|
self.api_url,
|
|
files=files,
|
|
data=data,
|
|
headers=headers
|
|
)
|
|
logger.bind(tag=TAG).debug(
|
|
f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {response.text}"
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
text = response.json().get("text", "")
|
|
return text, file_path
|
|
else:
|
|
raise Exception(f"API请求失败: {response.status_code} - {response.text}")
|
|
|
|
except Exception as e:
|
|
logger.bind(tag=TAG).error(f"语音识别失败: {e}")
|
|
return "", None
|
|
|