Tolan/backend/stt_service.py

"""
STT服务模块 - 基于 DashScope OmniRealtimeConversation API
参考官方demo实现，提供流式语音识别功能
"""
import asyncio
import time
import logging
import traceback
from typing import Optional, Callable, Awaitable
import dashscope
from dashscope.audio.qwen_omni import OmniRealtimeConversation, OmniRealtimeCallback
from dashscope.audio.qwen_omni.omni_realtime import MultiModality, TranscriptionParams
from config import settings, get_dashscope_ws_url

logger = logging.getLogger(__name__)

# 🔥 配置 DashScope SDK 的日志级别，避免空错误日志
dashscope_logger = logging.getLogger('dashscope')
dashscope_logger.setLevel(logging.WARNING)  # 只记录 WARNING 及以上级别


class STTCallback(OmniRealtimeCallback):
    """STT识别回调处理器（参考官方demo实现）"""
    
    def __init__(
        self,
        request_id: str,
        on_text_update: Optional[Callable[[str, bool], Awaitable[None]]] = None,
        on_error: Optional[Callable[[str], Awaitable[None]]] = None,
        event_loop: Optional[asyncio.AbstractEventLoop] = None
    ):
        """
        初始化STT回调处理器
        
        Args:
            request_id: 请求ID（用于日志标识）
            on_text_update: 文本更新回调 (text, is_final) -> None
            on_error: 错误回调 (error_msg) -> None
            event_loop: 事件循环（用于线程安全的异步调用）
        """
        self.request_id = request_id
        self.on_text_update = on_text_update
        self.on_error = on_error
        self.event_loop = event_loop
        
        # 识别结果存储
        self.current_text = ""
        self.is_final = False
        self.final_event = asyncio.Event() if event_loop else None
        
        # 时间统计
        self.last_update_time = 0.0
        self.first_text_time = None  # 首个文本生成时间
        self.session_id: Optional[str] = None
        
        # 语音检测统计
        self.speech_detected = False  # 是否检测到语音
        self.speech_start_count = 0   # 语音开始次数
        self.speech_stop_count = 0    # 语音停止次数
        
        # 事件接收统计（用于诊断）
        self.event_count = 0  # 收到的事件总数
        self.received_events = []  # 收到的事件类型列表（最多保存20个）
        
        # 事件处理器映射（参考官方demo）
        self.handlers = {
            'session.created': self._handle_session_created,
            'conversation.item.input_audio_transcription.completed': self._handle_final_text,
            'conversation.item.input_audio_transcription.text': self._handle_stash_text,
            'input_audio_buffer.speech_started': self._handle_speech_start,
            'input_audio_buffer.speech_stopped': self._handle_speech_stop,
            'error': self._handle_error,
        }
    
    def on_open(self):
        """连接打开回调"""
        logger.debug(
            f"[STT调试] request_id={self.request_id}, "
            f"WebSocket连接已打开"
        )
    
    def on_close(self, code: int, msg: str):
        """连接关闭回调"""
        logger.debug(
            f"[STT调试] request_id={self.request_id}, "
            f"WebSocket连接已关闭, "
            f"关闭码={code}, "
            f"关闭消息={msg}"
        )
        
        # 非正常关闭（1000是正常关闭码，None是主动关闭）
        if code is not None and code != 1000 and self.on_error:
            error_msg = f"STT连接异常关闭: {msg} (code: {code})"
            logger.error(
                f"[STT异常] request_id={self.request_id}, "
                f"错误类型=连接异常关闭, "
                f"关闭码={code}, "
                f"错误消息={msg}"
            )
            self._async_call(self.on_error(error_msg))
    
    def on_event(self, response: dict):
        """事件分发（参考官方demo）"""
        try:
            event_type = response.get('type')
            handler = self.handlers.get(event_type)
            
            # 🔥 更新事件接收统计
            self.event_count += 1
            if len(self.received_events) < 20:
                self.received_events.append(event_type)
            
            # 🔥 打印STT接口的原始返回日志（完整JSON）
            import json
            try:
                response_json = json.dumps(response, ensure_ascii=False, indent=2)
                logger.info(
                    f"[STT原始返回] request_id={self.request_id}, "
                    f"事件#{self.event_count}, "
                    f"事件类型={event_type}\n"
                    f"完整响应数据:\n{response_json}"
                )
            except Exception as json_error:
                logger.warning(
                    f"[STT原始返回] request_id={self.request_id}, "
                    f"事件#{self.event_count}, "
                    f"事件类型={event_type}, "
                    f"JSON序列化失败: {json_error}, "
                    f"原始响应={response}"
                )
            
            # 调试日志：记录所有事件（高频事件简化输出）
            if event_type in ['input_audio_buffer.speech_started', 'input_audio_buffer.speech_stopped']:
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"事件类型={event_type}"
                )
            else:
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"事件类型={event_type}, "
                    f"事件数据={response}"
                )
            
            if handler:
                handler(response)
            else:
                # 记录未处理的事件（排除常见的已知事件）
                known_events = [
                    'session.created', 
                    'session.updated', 
                    'input_audio_buffer.committed',
                    'conversation.item.created'  # 会话项创建事件
                ]
                if event_type not in known_events:
                    pass
                    
        except Exception as e:
            logger.error(
                f"[STT异常] request_id={self.request_id}, "
                f"错误类型=事件处理异常, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"事件类型={response.get('type', 'unknown')}, "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
    
    # ========== 事件处理器（参考官方demo） ==========
    
    def _handle_session_created(self, response: dict):
        """处理会话创建事件"""
        session_id = response.get('session', {}).get('id', '')
        self.session_id = session_id
        logger.debug(
            f"[STT调试] request_id={self.request_id}, "
            f"会话已创建, session_id={session_id}"
        )
    
    def _handle_final_text(self, response: dict):
        """处理最终识别结果（参考官方demo）"""
        text = response.get('transcript', '')
        
        logger.debug(
            f"[STT调试] request_id={self.request_id}, "
            f"收到最终识别结果, 文本长度={len(text)}, "
            f"文本内容={text[:50]}{'...' if len(text) > 50 else ''}"
        )
        
        # 更新状态
        self.current_text = text
        self.is_final = True
        self.last_update_time = time.time()
        
        # 触发最终结果事件
        if self.final_event and self.event_loop:
            try:
                self.event_loop.call_soon_threadsafe(self.final_event.set)
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"已触发最终结果事件"
                )
            except Exception as e:
                logger.error(
                    f"[STT异常] request_id={self.request_id}, "
                    f"错误类型=事件触发异常, "
                    f"异常类型={type(e).__name__}, "
                    f"错误消息={str(e)}, "
                    f"堆栈跟踪:\n{traceback.format_exc()}"
                )
        
        # 回调通知
        if self.on_text_update:
            self._async_call(self.on_text_update(text, True))
    
    def _handle_stash_text(self, response: dict):
        """处理中间识别结果（参考官方demo）"""
        # 合并 text（已确认部分）和 stash（正在识别部分）
        confirmed_text = response.get('text', '')
        stash_text = response.get('stash', '')
        full_text = confirmed_text + stash_text
        
        # 更新状态
        if full_text:
            self.current_text = full_text
            self.is_final = False
            current_time = time.time()
            self.last_update_time = current_time
            
            # 记录首个文本生成时间
            if self.first_text_time is None and full_text:
                self.first_text_time = current_time
                logger.info(
                    f"[STT] request_id={self.request_id}, "
                    f"首个文本生成, 文本={full_text[:30]}{'...' if len(full_text) > 30 else ''}"
                )
            
            # 调试日志：每5次中间结果记录一次（避免日志过多）
            if hasattr(self, '_stash_count'):
                self._stash_count += 1
            else:
                self._stash_count = 1
            
            if self._stash_count % 5 == 0:
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"中间识别结果 #{self._stash_count}, "
                    f"文本长度={len(full_text)}, "
                    f"文本内容={full_text[:30]}{'...' if len(full_text) > 30 else ''}"
                )
            
            # 回调通知
            if self.on_text_update:
                self._async_call(self.on_text_update(full_text, False))
    
    def _handle_speech_start(self, response: dict):
        """处理语音开始事件"""
        self.speech_detected = True
        self.speech_start_count += 1
        logger.debug(
            f"[STT调试] request_id={self.request_id}, "
            f"检测到语音开始, speech_start_count={self.speech_start_count}"
        )
    
    def _handle_speech_stop(self, response: dict):
        """处理语音停止事件"""
        self.speech_stop_count += 1
        logger.debug(
            f"[STT调试] request_id={self.request_id}, "
            f"检测到语音停止, speech_stop_count={self.speech_stop_count}"
        )
    
    def _handle_error(self, response: dict):
        """处理错误事件"""
        error = response.get('error', {})
        error_msg = error.get('message', 'Unknown error')
        error_code = error.get('code', '')
        
        # 详细记录STT API错误
        logger.error(
            f"[STT异常] request_id={self.request_id}, "
            f"错误类型=STT API错误, "
            f"错误码={error_code}, "
            f"错误消息={error_msg}, "
            f"完整响应={response}"
        )
        
        # 回调通知
        if self.on_error:
            self._async_call(self.on_error(f"{error_msg} (code: {error_code})"))
    
    # ========== 辅助方法 ==========
    
    def _async_call(self, coro):
        """线程安全的异步调用"""
        if self.event_loop:
            try:
                asyncio.run_coroutine_threadsafe(coro, self.event_loop)
            except Exception as e:
                logger.error(
                    f"[STT异常] request_id={self.request_id}, "
                    f"错误类型=异步调用异常, "
                    f"异常类型={type(e).__name__}, "
                    f"错误消息={str(e)}, "
                    f"堆栈跟踪:\n{traceback.format_exc()}"
                )
    
    def get_text(self) -> str:
        """获取当前识别文本"""
        return self.current_text
    
    def is_final_result(self) -> bool:
        """是否为最终结果"""
        return self.is_final


class STTSession:
    """STT识别会话管理器（参考官方demo实现）"""
    
    def __init__(
        self,
        request_id: str,
        on_text_update: Optional[Callable[[str, bool], Awaitable[None]]] = None,
        on_error: Optional[Callable[[str], Awaitable[None]]] = None,
        event_loop: Optional[asyncio.AbstractEventLoop] = None
    ):
        """
        初始化STT会话
        
        Args:
            request_id: 请求ID
            on_text_update: 文本更新回调
            on_error: 错误回调
            event_loop: 事件循环
        """
        self.request_id = request_id
        self.callback = STTCallback(
            request_id=request_id,
            on_text_update=on_text_update,
            on_error=on_error,
            event_loop=event_loop
        )
        self.conversation: Optional[OmniRealtimeConversation] = None
        self.is_connected = False
        self.is_closed = False
    
    async def connect(self, max_retries: int = 2) -> bool:
        """
        连接STT服务并初始化会话（带重试机制）
        
        🔥 每次请求都创建新的 WebSocket 会话，确保可以随时建立请求
        不会因为连接复用而阻塞，每个 request_id 都有独立的会话
        
        Args:
            max_retries: 最大重试次数（默认2次，总共尝试3次）
        
        Returns:
            是否连接成功
        """
        last_error = None
        
        for attempt in range(max_retries + 1):
            try:
                start_time = time.time()
                
                # 设置API Key
                dashscope.api_key = settings.active_dashscope_api_key
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"连接尝试 {attempt + 1}/{max_retries + 1}, "
                    f"API Key已设置, 使用国际版={settings.use_dashscope_intl}"
                )
                
                # 🔥 创建新的会话对象（每次请求独立创建，参考官方demo）
                def create_conversation():
                    ws_url = get_dashscope_ws_url()
                    logger.debug(
                        f"[STT调试] request_id={self.request_id}, "
                        f"创建会话对象, URL={ws_url}"
                    )
                    
                    conversation = OmniRealtimeConversation(
                        model=settings.STT_MODEL,
                        url=ws_url,
                        callback=self.callback
                    )
                    # 注入自身引用（回调中可能需要）
                    self.callback.conversation = conversation
                    logger.debug(
                        f"[STT调试] request_id={self.request_id}, "
                        f"会话对象已创建"
                    )
                    return conversation
                
                # 异步创建会话（每次都是新连接）
                create_start = time.time()
                self.conversation = await asyncio.to_thread(create_conversation)
                create_duration = (time.time() - create_start) * 1000
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"会话对象创建完成, 耗时={create_duration:.1f}ms"
                )
                
                # 连接WebSocket（添加超时保护）
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"开始连接WebSocket..."
                )
                connect_start = time.time()
                try:
                    await asyncio.wait_for(
                        asyncio.to_thread(self.conversation.connect),
                        timeout=10.0  # 10秒连接超时
                    )
                    connect_duration = (time.time() - connect_start) * 1000
                    logger.debug(
                        f"[STT调试] request_id={self.request_id}, "
                        f"WebSocket连接成功, 耗时={connect_duration:.1f}ms"
                    )
                except asyncio.TimeoutError:
                    raise Exception("STT WebSocket连接超时（10秒）")
                
                # 配置转录参数（参考官方demo）
                # ⚠️ 重要：language参数是必需的！
                # 官方支持: 'zh'(中文), 'en'(英文), 'ja'(日文), 'ko'(韩文)
                # 尝试支持: 'auto'(自动检测) - 需要测试是否有效
                stt_language = getattr(settings, 'STT_LANGUAGE', 'zh')
                
                transcription_params = TranscriptionParams(
                    language=stt_language,
                    sample_rate=16000,
                    input_audio_format="pcm"
                    # corpus_text=""  # 可选：辅助识别的语料
                )
                
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"配置转录参数: language={stt_language}, "
                    f"sample_rate=16000, format=pcm"
                )
                
                # 更新会话配置（禁用VAD，用户通过按钮控制录音）
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"开始更新会话配置..."
                )
                config_start = time.time()
                try:
                    await asyncio.wait_for(
                        asyncio.to_thread(
                            self.conversation.update_session,
                            output_modalities=[MultiModality.TEXT],
                            enable_input_audio_transcription=True,
                            transcription_params=transcription_params,
                            enable_turn_detection=False
                        ),
                        timeout=5.0  # 5秒配置超时
                    )
                    config_duration = (time.time() - config_start) * 1000
                    logger.debug(
                        f"[STT调试] request_id={self.request_id}, "
                        f"会话配置完成, 耗时={config_duration:.1f}ms"
                    )
                except asyncio.TimeoutError:
                    raise Exception("STT会话配置超时（5秒）")
                
                self.is_connected = True
                connect_duration = (time.time() - start_time) * 1000
                logger.info(
                    f"[STT连接成功] request_id={self.request_id}, "
                    f"尝试次数={attempt + 1}, "
                    f"耗时={connect_duration:.1f}ms"
                )
                
                return True
                
            except Exception as e:
                last_error = e
                connect_duration = (time.time() - start_time) * 1000 if 'start_time' in locals() else 0
                
                # 详细记录错误
                error_details = {
                    "request_id": self.request_id,
                    "错误类型": "STT连接失败",
                    "异常类型": type(e).__name__,
                    "错误消息": str(e),
                    "尝试次数": f"{attempt + 1}/{max_retries + 1}",
                    "耗时": f"{connect_duration:.1f}ms",
                    "API Key": f"{settings.active_dashscope_api_key[:10]}... (已脱敏)",
                    "使用国际版": settings.use_dashscope_intl,
                }
                
                # 如果是最后一次尝试，记录完整堆栈
                if attempt == max_retries:
                    error_details["堆栈跟踪"] = traceback.format_exc()
                    logger.error(f"[STT异常] {error_details}")
                else:
                    logger.warning(f"[STT连接重试] {error_details}")
                
                # 如果不是最后一次尝试，等待后重试
                if attempt < max_retries:
                    wait_time = (attempt + 1) * 0.5  # 递增等待：0.5s, 1s
                    logger.info(f"[STT连接重试] request_id={self.request_id}, {wait_time:.1f}秒后重试...")
                    await asyncio.sleep(wait_time)
                else:
                    # 最后一次尝试失败，通知错误回调
                    if self.callback.on_error:
                        await self.callback.on_error(f"连接失败（已重试{max_retries}次）: {str(e)}")
        
        return False
    
    async def send_audio(self, audio_data: bytes) -> bool:
        """
        发送音频数据（参考官方demo，带超时保护）
        
        Args:
            audio_data: PCM音频数据（原始bytes）
            
        Returns:
            是否发送成功
        """
        if not self.is_connected or not self.conversation:
            logger.warning(
                f"[STT警告] request_id={self.request_id}, "
                f"连接状态异常: is_connected={self.is_connected}, "
                f"conversation={'存在' if self.conversation else '不存在'}"
            )
            return False
        
        try:
            # 🔥 音频数据验证和统计（用于诊断问题）
            import struct
            try:
                # 将PCM字节数据转换为Int16数组
                sample_count = len(audio_data) // 2
                if sample_count > 0:
                    samples = struct.unpack(f'<{sample_count}h', audio_data)
                    max_amplitude = max(abs(s) for s in samples) if samples else 0
                    avg_amplitude = sum(abs(s) for s in samples) // len(samples) if samples else 0
                    zero_count = sum(1 for s in samples if s == 0)
                    zero_ratio = zero_count / len(samples) if samples else 0.0
                    
                    # 记录音频统计信息（仅第一个chunk和异常情况）
                    if not hasattr(self, '_audio_chunk_count'):
                        self._audio_chunk_count = 0
                    self._audio_chunk_count += 1
                    
                    # 第一个chunk或异常情况（全零或振幅过低）时记录详细信息
                    if self._audio_chunk_count == 1 or max_amplitude < 100 or zero_ratio > 0.9:
                        logger.info(
                            f"[STT音频统计] request_id={self.request_id}, "
                            f"chunk={self._audio_chunk_count}, "
                            f"长度={len(audio_data)} bytes, "
                            f"采样数={sample_count}, "
                            f"最大振幅={max_amplitude}, "
                            f"平均振幅={avg_amplitude}, "
                            f"零值比例={zero_ratio:.2%}, "
                            f"是否静音={'是' if max_amplitude < 100 or zero_ratio > 0.9 else '否'}"
                        )
            except Exception as stats_error:
                logger.warning(
                    f"[STT音频统计] request_id={self.request_id}, "
                    f"统计计算失败: {stats_error}"
                )
            
            # Base64编码（API要求）
            import base64
            audio_b64 = base64.b64encode(audio_data).decode('ascii')
            
            # 使用INFO级别记录音频准备发送（便于排查问题）
            logger.info(
                f"[STT] request_id={self.request_id}, "
                f"准备发送音频: 原始长度={len(audio_data)} bytes, "
                f"Base64长度={len(audio_b64)} chars"
            )
            
            # 发送音频帧（参考官方demo，添加超时保护）
            send_start = time.time()
            try:
                await asyncio.wait_for(
                    asyncio.to_thread(self.conversation.append_audio, audio_b64),
                    timeout=5.0  # 5秒发送超时
                )
                send_duration = (time.time() - send_start) * 1000
                # 使用INFO级别记录音频发送成功（便于排查问题）
                logger.info(
                    f"[STT] request_id={self.request_id}, "
                    f"音频发送成功, 原始长度={len(audio_data)} bytes, "
                    f"耗时={send_duration:.1f}ms"
                )
            except asyncio.TimeoutError:
                raise Exception("发送音频数据超时（5秒）")
            
            return True
            
        except Exception as e:
            # 标记连接可能已断开
            error_str = str(e).lower()
            if 'timeout' in error_str or 'closed' in error_str or 'disconnect' in error_str:
                self.is_connected = False
                logger.error(
                    f"[STT异常] request_id={self.request_id}, "
                    f"错误类型=连接可能已断开, "
                    f"异常类型={type(e).__name__}, "
                    f"错误消息={str(e)}, "
                    f"音频数据长度={len(audio_data)} bytes"
                )
            else:
                logger.error(
                    f"[STT异常] request_id={self.request_id}, "
                    f"错误类型=发送音频失败, "
                    f"异常类型={type(e).__name__}, "
                    f"错误消息={str(e)}, "
                    f"音频数据长度={len(audio_data)} bytes, "
                    f"连接状态=is_connected={self.is_connected}, "
                    f"堆栈跟踪:\n{traceback.format_exc()}"
                )
            
            if self.callback.on_error:
                await self.callback.on_error(f"发送音频失败: {e}")
            return False
    
    async def commit_audio(self) -> bool:
        """
        提交音频以触发识别（禁用VAD时需要手动调用，带超时保护）
        
        Returns:
            是否提交成功
        """
        if not self.is_connected or not self.conversation:
            logger.warning(
                f"[STT警告] request_id={self.request_id}, "
                f"连接状态异常，无法提交音频: "
                f"is_connected={self.is_connected}, "
                f"conversation={'存在' if self.conversation else '不存在'}"
            )
            return False
        
        try:
            logger.debug(
                f"[STT调试] request_id={self.request_id}, "
                f"开始提交音频..."
            )
            
            # 添加超时保护
            commit_start = time.time()
            try:
                await asyncio.wait_for(
                    asyncio.to_thread(self.conversation.commit),
                    timeout=5.0  # 5秒提交超时
                )
                commit_duration = (time.time() - commit_start) * 1000
                logger.info(
                    f"[STT] request_id={self.request_id}, "
                    f"音频提交成功, 耗时={commit_duration:.1f}ms"
                )
            except asyncio.TimeoutError:
                raise Exception("提交音频超时（5秒）")
            
            return True
        except Exception as e:
            # 标记连接可能已断开
            error_str = str(e).lower()
            if 'timeout' in error_str or 'closed' in error_str or 'disconnect' in error_str:
                self.is_connected = False
            
            logger.error(
                f"[STT异常] request_id={self.request_id}, "
                f"错误类型=提交音频失败, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"连接状态=is_connected={self.is_connected}, "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
            return False
    
    async def wait_for_final_result(self, timeout: float = 3.0) -> Optional[str]:
        """
        等待最终识别结果
        
        Args:
            timeout: 超时时间（秒）
            
        Returns:
            识别文本，超时或失败返回None
        """
        logger.debug(
            f"[STT调试] request_id={self.request_id}, "
            f"开始等待最终识别结果, 超时时间={timeout}秒"
        )
        
        if not self.callback.final_event:
            logger.debug(
                f"[STT调试] request_id={self.request_id}, "
                f"未设置事件循环，直接返回当前文本"
            )
            return self.callback.get_text()
        
        wait_start = time.time()
        try:
            await asyncio.wait_for(self.callback.final_event.wait(), timeout=timeout)
            wait_duration = (time.time() - wait_start) * 1000
            result_text = self.callback.get_text()
            logger.info(
                f"[STT] request_id={self.request_id}, "
                f"收到最终识别结果, 等待耗时={wait_duration:.1f}ms, "
                f"文本长度={len(result_text)}, "
                f"文本内容={result_text[:50]}{'...' if len(result_text) > 50 else ''}"
            )
            return result_text
            
        except asyncio.TimeoutError:
            # 超时后，检查是否有中间结果和语音检测状态
            wait_duration = (time.time() - wait_start) * 1000
            current_text = self.callback.get_text()
            speech_detected = self.callback.speech_detected
            speech_start_count = self.callback.speech_start_count
            speech_stop_count = self.callback.speech_stop_count
            
            # 🔥 输出事件接收统计（用于诊断）
            event_count = self.callback.event_count
            received_events = self.callback.received_events
            
            logger.warning(
                f"[STT警告] request_id={self.request_id}, "
                f"错误类型=等待识别结果超时, "
                f"超时时间={timeout}秒, "
                f"等待耗时={wait_duration:.1f}ms, "
                f"当前识别文本={current_text}, "
                f"语音检测状态=speech_detected={speech_detected}, "
                f"speech_start_count={speech_start_count}, "
                f"speech_stop_count={speech_stop_count}, "
                f"收到事件总数={event_count}, "
                f"收到的事件类型={received_events}"
            )
            return current_text
        except Exception as e:
            logger.error(
                f"[STT异常] request_id={self.request_id}, "
                f"错误类型=等待识别结果异常, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"超时时间={timeout}秒, "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
            return None
    
    async def close(self):
        """关闭STT会话（参考官方demo）"""
        if self.is_closed:
            logger.debug(
                f"[STT调试] request_id={self.request_id}, "
                f"会话已关闭，跳过"
            )
            return
        
        logger.debug(
            f"[STT调试] request_id={self.request_id}, "
            f"开始关闭STT会话, "
            f"连接状态=is_connected={self.is_connected}, "
            f"conversation={'存在' if self.conversation else '不存在'}"
        )
        
        try:
            if self.conversation:
                # 异步关闭连接
                close_start = time.time()
                await asyncio.to_thread(self.conversation.close)
                close_duration = (time.time() - close_start) * 1000
                logger.debug(
                    f"[STT调试] request_id={self.request_id}, "
                    f"会话关闭完成, 耗时={close_duration:.1f}ms"
                )
            
            self.is_connected = False
            self.is_closed = True
            
        except Exception as e:
            logger.error(
                f"[STT异常] request_id={self.request_id}, "
                f"错误类型=关闭会话异常, "
                f"异常类型={type(e).__name__}, "
                f"错误消息={str(e)}, "
                f"连接状态=is_connected={self.is_connected}, is_closed={self.is_closed}, "
                f"堆栈跟踪:\n{traceback.format_exc()}"
            )
    
    def get_current_text(self) -> str:
        """获取当前识别文本"""
        return self.callback.get_text()
    
    def get_last_update_time(self) -> float:
        """获取最后更新时间"""
        return self.callback.last_update_time
    
    def get_first_text_time(self) -> Optional[float]:
        """获取首个文本生成时间"""
        return self.callback.first_text_time


class STTService:
    """STT服务管理器（单例）"""
    
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self):
        """初始化STT服务"""
        if not hasattr(self, '_initialized'):
            self._initialized = True
            
            # 设置API Key
            dashscope.api_key = settings.active_dashscope_api_key
            
            # 国际版endpoint配置
            if settings.use_dashscope_intl:
                from config import get_dashscope_domain
                dashscope_domain = get_dashscope_domain()
                dashscope.base_http_api_url = f"https://{dashscope_domain}/api/v1"
                dashscope.base_websocket_api_url = f"wss://{dashscope_domain}/api-ws/v1"
    
    async def create_session(
        self,
        request_id: str,
        on_text_update: Optional[Callable[[str, bool], Awaitable[None]]] = None,
        on_error: Optional[Callable[[str], Awaitable[None]]] = None,
        event_loop: Optional[asyncio.AbstractEventLoop] = None
    ) -> STTSession:
        """
        创建STT识别会话
        
        Args:
            request_id: 请求ID
            on_text_update: 文本更新回调
            on_error: 错误回调
            event_loop: 事件循环
            
        Returns:
            STT会话对象
        """
        session = STTSession(
            request_id=request_id,
            on_text_update=on_text_update,
            on_error=on_error,
            event_loop=event_loop
        )
        
        # 自动连接
        success = await session.connect()
        
        return session


# 全局STT服务实例
stt_service = STTService()