2026-01-07 11:56:46 +08:00
|
|
|
|
"""
|
|
|
|
|
|
ASR 服务：语音转文字
|
|
|
|
|
|
"""
|
|
|
|
|
|
import base64
|
2026-01-18 15:57:54 +08:00
|
|
|
|
import os
|
|
|
|
|
|
|
2026-01-07 11:56:46 +08:00
|
|
|
|
from openai import OpenAI
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class ASRService:
|
|
|
|
|
|
"""ASR 服务（语音转文字）"""
|
|
|
|
|
|
|
|
|
|
|
|
def __init__(self):
|
|
|
|
|
|
api_key = os.getenv("OPENAI_API_KEY", "")
|
|
|
|
|
|
if api_key:
|
|
|
|
|
|
self.client = OpenAI(api_key=api_key)
|
|
|
|
|
|
else:
|
|
|
|
|
|
self.client = None
|
|
|
|
|
|
|
|
|
|
|
|
async def transcribe(self, audio_base64: str) -> str | None:
|
|
|
|
|
|
"""
|
|
|
|
|
|
转写音频为文字
|
|
|
|
|
|
|
|
|
|
|
|
Args:
|
|
|
|
|
|
audio_base64: Base64 编码的音频数据
|
|
|
|
|
|
|
|
|
|
|
|
Returns:
|
|
|
|
|
|
转写文本
|
|
|
|
|
|
"""
|
|
|
|
|
|
if not self.client:
|
|
|
|
|
|
# 如果没有配置 API Key，返回模拟数据
|
|
|
|
|
|
return "这是模拟的转写文本（请配置 OPENAI_API_KEY 以使用实际 ASR 功能）"
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
|
# 解码 Base64 音频
|
|
|
|
|
|
audio_bytes = base64.b64decode(audio_base64)
|
|
|
|
|
|
|
|
|
|
|
|
# 保存临时文件
|
|
|
|
|
|
import tempfile
|
|
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".m4a", delete=False) as tmp_file:
|
|
|
|
|
|
tmp_file.write(audio_bytes)
|
|
|
|
|
|
tmp_file_path = tmp_file.name
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
|
# 调用 OpenAI Whisper API
|
|
|
|
|
|
with open(tmp_file_path, "rb") as audio_file:
|
|
|
|
|
|
transcript = self.client.audio.transcriptions.create(
|
|
|
|
|
|
model="whisper-1",
|
|
|
|
|
|
file=audio_file,
|
|
|
|
|
|
language="zh" # 中文
|
|
|
|
|
|
)
|
|
|
|
|
|
return transcript.text
|
|
|
|
|
|
finally:
|
|
|
|
|
|
# 清理临时文件
|
|
|
|
|
|
import os
|
|
|
|
|
|
if os.path.exists(tmp_file_path):
|
|
|
|
|
|
os.remove(tmp_file_path)
|
|
|
|
|
|
except Exception as e:
|
|
|
|
|
|
# 出错时返回错误信息
|
|
|
|
|
|
return f"转写失败: {str(e)}"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 全局实例
|
|
|
|
|
|
asr_service = ASRService()
|
