aiData/Util/ASRClient.py

import asyncio
from http import HTTPStatus
from dashscope.audio.asr import Recognition
import dashscope
import logging
from Config import Config

# 初始化日志记录器
logger = logging.getLogger(__name__)

# 若没有配置环境变量，请用百炼API Key将下行替换为：dashscope.api_key = "sk-xxx"
dashscope.api_key = Config.ALY_LLM_API_KEY

class ASRClient:
    """
    阿里云语音识别客户端，用于处理语音文件的转写任务
    使用 Recognition API 支持本地文件实时转写
    """
    
    def __init__(self, api_key=None):
        """
        初始化ASR客户端
        
        Args:
            api_key: 阿里云DashScope API密钥，若不提供则使用配置文件中的密钥
        """
        logger.info("开始初始化ASR客户端")
        
        try:
            self.api_key = api_key or Config.ALY_LLM_API_KEY
            dashscope.api_key = self.api_key
            logger.info("ASR客户端初始化完成")
        except Exception as e:
            logger.error(f"初始化ASR客户端失败: {str(e)}", exc_info=True)
            raise
    
    def transcribe_file_sync(self, file_path):
        """
        转写本地音频文件 (同步版本)
        
        Args:
            file_path: 本地音频文件路径
            
        Returns:
            str: 转写后的文本，如果失败返回None
        """
        logger.info(f"开始转写文件(Sync): {file_path}")
        
        try:
            recognition = Recognition(
                model='paraformer-realtime-v1',
                format='mp3',
                sample_rate=16000,
                callback=None
            )
            
            result = recognition.call(file_path)
            
            if result.status_code == HTTPStatus.OK:
                sentences = []
                if 'sentence' in result.output:
                    for s in result.output['sentence']:
                        sentences.append(s['text'])
                text = "".join(sentences)
                logger.info("转写成功")
                return text
            else:
                logger.error(f"转写失败: {result.code} - {result.message}")
                return None
                
        except Exception as e:
            logger.error(f"转写过程出错: {str(e)}", exc_info=True)
            return None

    async def transcribe_file(self, file_path):
        """
        转写本地音频文件
        
        Args:
            file_path: 本地音频文件路径
            
        Returns:
            str: 转写后的文本，如果失败返回None
        """
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, self.transcribe_file_sync, file_path)