240a184da8
- 新增tencent_asr_service.py腾讯云一句话识别 - 优化asr_service.py - 更新services/__init__.py按ASR_PROVIDER切换whisper/tencent - 更新requirements.txt Co-authored-by: Cursor <cursoragent@cursor.com>
114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
"""
|
||
ASR 服务：语音转文字（腾讯云一句话识别）
|
||
使用腾讯云 ASR API 进行语音识别
|
||
"""
|
||
import base64
|
||
import logging
|
||
import os
|
||
import uuid
|
||
from typing import Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class TencentASRService:
|
||
"""
|
||
ASR 服务（语音转文字）
|
||
使用腾讯云一句话识别 API（SentenceRecognition）
|
||
文档：https://cloud.tencent.com/document/product/1093/35646
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._secret_id = os.getenv("TENCENT_SECRET_ID", "")
|
||
self._secret_key = os.getenv("TENCENT_SECRET_KEY", "")
|
||
self._app_id = os.getenv("TENCENT_ASR_APP_ID", "")
|
||
self._ready = bool(self._secret_id and self._secret_key)
|
||
self._client = None
|
||
|
||
if not self._ready:
|
||
logger.warning("腾讯云 ASR 未配置：缺少 TENCENT_SECRET_ID 或 TENCENT_SECRET_KEY")
|
||
|
||
def _get_client(self):
|
||
"""懒加载腾讯云 ASR 客户端"""
|
||
if self._client is not None:
|
||
return self._client
|
||
|
||
try:
|
||
from tencentcloud.common import credential
|
||
from tencentcloud.common.profile.client_profile import ClientProfile
|
||
from tencentcloud.common.profile.http_profile import HttpProfile
|
||
from tencentcloud.asr.v20190614 import asr_client
|
||
|
||
cred = credential.Credential(self._secret_id, self._secret_key)
|
||
|
||
http_profile = HttpProfile()
|
||
http_profile.endpoint = "asr.tencentcloudapi.com"
|
||
|
||
client_profile = ClientProfile()
|
||
client_profile.httpProfile = http_profile
|
||
|
||
self._client = asr_client.AsrClient(cred, "", client_profile)
|
||
logger.info("腾讯云 ASR 客户端初始化成功")
|
||
return self._client
|
||
except Exception as e:
|
||
logger.error(f"腾讯云 ASR 客户端初始化失败: {e}", exc_info=True)
|
||
return None
|
||
|
||
def ensure_ready(self) -> bool:
|
||
"""
|
||
确保 ASR 服务已就绪。
|
||
腾讯云 ASR 是远程 API，无需预加载模型，仅检查凭证配置。
|
||
"""
|
||
if not self._ready:
|
||
return False
|
||
# 尝试初始化客户端，验证 SDK 可用
|
||
return self._get_client() is not None
|
||
|
||
def is_ready(self) -> bool:
|
||
"""检查 ASR 服务是否可用。"""
|
||
return self._ready
|
||
|
||
async def transcribe(self, audio_base64: str) -> Optional[str]:
|
||
"""
|
||
转写音频为文字（腾讯云一句话识别）
|
||
|
||
Args:
|
||
audio_base64: Base64 编码的音频数据
|
||
|
||
Returns:
|
||
转写文本，失败时返回错误信息
|
||
"""
|
||
if not self._ready:
|
||
return "转写失败: 腾讯云 ASR 未配置"
|
||
|
||
client = self._get_client()
|
||
if not client:
|
||
return "转写失败: 腾讯云 ASR 客户端初始化失败"
|
||
|
||
try:
|
||
from tencentcloud.asr.v20190614 import models
|
||
|
||
req = models.SentenceRecognitionRequest()
|
||
req.EngSerViceType = "16k_zh" # 16k 中文普通话
|
||
req.SourceType = 1 # 1 = 语音数据的 Base64 编码
|
||
req.VoiceFormat = "m4a" # 音频格式
|
||
req.Data = audio_base64
|
||
req.DataLen = len(base64.b64decode(audio_base64))
|
||
|
||
resp = client.SentenceRecognition(req)
|
||
result = resp.Result
|
||
|
||
logger.info(
|
||
f"腾讯云 ASR 转写完成: 文本长度={len(result) if result else 0}"
|
||
)
|
||
|
||
return result.strip() if result else ""
|
||
|
||
except Exception as e:
|
||
logger.error(f"腾讯云 ASR 转写失败: {e}", exc_info=True)
|
||
return f"转写失败: {str(e)}"
|
||
|
||
|
||
# 全局实例
|
||
tencent_asr_service = TencentASRService()
|