Tolan/backend/request_logger.py

"""
请求日志记录器 - 记录每个语音对话请求的详细性能数据
使用JSON Lines格式，按天轮转日志文件
"""
import json
import time
import queue
import threading
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any
from dataclasses import dataclass
from config import settings


@dataclass
class PerformanceMetrics:
    """性能指标数据类"""
    # 用户说话
    user_speech_start: Optional[float] = None
    user_speech_end: Optional[float] = None
    user_text: str = ""
    
    # STT相关
    stt_first_audio_received: Optional[float] = None
    stt_first_text_generated: Optional[float] = None
    stt_last_audio_received: Optional[float] = None
    stt_completed: Optional[float] = None
    stt_audio_chunks: int = 0  # 音频块数量
    stt_total_bytes: int = 0  # 总字节数
    
    # LLM提示词准备
    llm_prompt_prepare_start: Optional[float] = None
    llm_prompt_prepare_end: Optional[float] = None
    llm_prompt_length: int = 0  # 提示词长度
    llm_context_round: int = 0  # 上下文轮次
    llm_full_prompt: str = ""  # 完整prompt（system_prompt + user_input）
    
    # LLM生成
    llm_call_start: Optional[float] = None
    llm_first_text_output: Optional[float] = None
    llm_all_text_completed: Optional[float] = None
    llm_call_end: Optional[float] = None
    llm_emotion: str = "neutral"  # 情感
    llm_sentences: int = 0  # 句子数
    
    # TTS生成
    tts_call_start: Optional[float] = None
    tts_first_audio_output: Optional[float] = None
    tts_all_audio_completed: Optional[float] = None
    tts_call_end: Optional[float] = None
    tts_voice: str = "Cherry"  # 音色
    tts_audio_chunks: int = 0  # 音频块数量
    tts_total_bytes: int = 0  # 总字节数
    
    # AI响应
    ai_response_text: str = ""
    
    # 消息ID（数据库关联）
    user_message_id: Optional[str] = None  # 用户消息的 message_id
    assistant_message_id: Optional[str] = None  # AI消息的 message_id
    
    def calculate_user_speech_duration_ms(self) -> Optional[float]:
        """计算用户说话时长(ms)"""
        if self.user_speech_start and self.user_speech_end:
            return (self.user_speech_end - self.user_speech_start) * 1000
        return None
    
    def calculate_stt_latency_ms(self) -> Optional[float]:
        """计算STT延迟(ms): STT完成时间 - 收到最后一个音频时间"""
        if self.stt_completed and self.stt_last_audio_received:
            return (self.stt_completed - self.stt_last_audio_received) * 1000
        return None
    
    def calculate_llm_prompt_latency_ms(self) -> Optional[float]:
        """计算LLM提示词准备延迟(ms)"""
        if self.llm_prompt_prepare_end and self.llm_prompt_prepare_start:
            return (self.llm_prompt_prepare_end - self.llm_prompt_prepare_start) * 1000
        return None
    
    def calculate_llm_generation_latency_ms(self) -> Optional[float]:
        """计算LLM生成延迟(ms)"""
        if self.llm_first_text_output and self.llm_call_start:
            return (self.llm_first_text_output - self.llm_call_start) * 1000
        return None
    
    def calculate_tts_latency_ms(self) -> Optional[float]:
        """计算TTS延迟(ms)"""
        if self.tts_first_audio_output and self.tts_call_start:
            return (self.tts_first_audio_output - self.tts_call_start) * 1000
        return None
    
    def calculate_total_latency_ms(self) -> Optional[float]:
        """计算总延迟(ms): 开始输出第一个音频时间 - 用户结束说话时间"""
        if self.tts_first_audio_output and self.user_speech_end:
            return (self.tts_first_audio_output - self.user_speech_end) * 1000
        return None


class RequestLogger:
    """请求日志记录器（线程安全，异步写入）"""
    
    def __init__(self, log_dir: str = "logs"):
        """
        初始化日志记录器
        
        Args:
            log_dir: 日志目录
        """
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)
        
        # 当前日志文件路径
        self.current_log_file: Optional[Path] = None
        self.current_date: Optional[str] = None
        
        # 异步写入队列
        self.log_queue: queue.Queue = queue.Queue()
        self.writer_thread: Optional[threading.Thread] = None
        self.running = True
        
        # 启动后台写入线程
        self._start_writer_thread()
    
    def _get_log_file_path(self, date_str: str) -> Path:
        """获取指定日期的日志文件路径"""
        return self.log_dir / f"request_{date_str}.log"
    
    def _get_current_log_file(self) -> Path:
        """获取当前日期的日志文件路径（自动轮转）"""
        today = datetime.now().strftime("%Y-%m-%d")
        
        # 如果日期变化，切换日志文件
        if self.current_date != today:
            self.current_date = today
            self.current_log_file = self._get_log_file_path(today)
        
        return self.current_log_file
    
    def _start_writer_thread(self):
        """启动后台写入线程"""
        def writer_loop():
            """后台写入循环"""
            while self.running:
                try:
                    # 从队列获取日志条目（超时1秒）
                    log_entry = self.log_queue.get(timeout=1.0)
                    
                    # 获取当前日志文件
                    log_file = self._get_current_log_file()
                    
                    # 写入JSON格式日志文件（追加模式）
                    with open(log_file, 'a', encoding='utf-8') as f:
                        f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
                    
                    # 如果是请求日志，同时写入conversation_log.txt
                    if log_entry.get("类型") == "请求":
                        self._write_conversation_log(log_entry)
                    
                    self.log_queue.task_done()
                    
                except queue.Empty:
                    continue
                except Exception as e:
                    # 写入失败，记录到stderr（避免循环）
                    import sys
                    print(f"[RequestLogger] 写入日志失败: {e}", file=sys.stderr)
        
        self.writer_thread = threading.Thread(target=writer_loop, daemon=True)
        self.writer_thread.start()
    
    def _format_timestamp(self, timestamp: Optional[float]) -> Optional[str]:
        """格式化时间戳为ISO 8601格式"""
        if timestamp is None:
            return None
        dt = datetime.fromtimestamp(timestamp)
        return dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    
    def _format_iso_time_to_conversation_log(self, iso_time: Optional[str]) -> str:
        """将ISO时间戳格式化为conversation_log.txt格式 (HH:MM:SS.mmm)"""
        if not iso_time:
            return "N/A"
        try:
            # 从ISO格式解析: 2025-11-11T22:42:46.123Z
            dt = datetime.fromisoformat(iso_time.replace('Z', '+00:00'))
            return dt.strftime("%H:%M:%S.%f")[:-3]
        except:
            return "N/A"
    
    def _write_conversation_log(self, log_entry: Dict[str, Any]):
        """写入conversation_log.txt格式的日志"""
        try:
            conversation_log_path = Path("conversation_log.txt")
            
            stages = log_entry.get("阶段", {})
            summary = log_entry.get("总结", {})
            request_id = log_entry.get("请求ID", "unknown")
            timestamp = log_entry.get("时间戳", "")
            
            # 解析时间戳获取日期时间
            if timestamp:
                try:
                    # 从ISO格式解析: 2025-11-11T22:42:46.123Z
                    dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
                    date_str = dt.strftime("%Y-%m-%d %H:%M:%S")
                except:
                    date_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            else:
                date_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            
            lines = []
            lines.append(f"[{date_str}] [{request_id}]")
            
            # 用户说话
            user_speech = stages.get("用户说话", {})
            user_text = ""
            if "语音识别" in stages and "输出" in stages["语音识别"]:
                user_text = stages["语音识别"]["输出"].get("文本", "")
            
            user_start = user_speech.get("开始时间")
            user_end = user_speech.get("结束时间")
            user_duration = user_speech.get("时长_毫秒")
            
            user_start_str = self._format_iso_time_to_conversation_log(user_start)
            user_end_str = self._format_iso_time_to_conversation_log(user_end)
            user_duration_str = f"{int(user_duration)}ms" if user_duration else "N/A"
            
            lines.append(f"用户说话: {user_start_str}, {user_text}, {user_end_str}, {user_duration_str}")
            
            # STT
            stt = stages.get("语音识别", {})
            stt_first_audio = stt.get("首次收到音频")
            stt_first_text = stt.get("首次生成文本")
            stt_last_audio = stt.get("最后收到音频")
            stt_completed = stt.get("完成时间")
            stt_latency = stt.get("延迟_毫秒")
            
            stt_first_audio_str = self._format_iso_time_to_conversation_log(stt_first_audio)
            stt_first_text_str = self._format_iso_time_to_conversation_log(stt_first_text)
            stt_last_audio_str = self._format_iso_time_to_conversation_log(stt_last_audio)
            stt_completed_str = self._format_iso_time_to_conversation_log(stt_completed)
            stt_latency_str = f"{int(stt_latency)}ms" if stt_latency else "N/A"
            
            lines.append(f"STT: {stt_first_audio_str}, {stt_first_text_str}, {stt_last_audio_str}, {stt_completed_str}, {stt_latency_str}")
            
            # LLM提示词
            llm_prompt = stages.get("LLM提示词准备", {})
            llm_prompt_start = llm_prompt.get("开始时间")
            llm_prompt_end = llm_prompt.get("结束时间")
            llm_prompt_duration = llm_prompt.get("时长_毫秒")
            
            llm_prompt_start_str = self._format_iso_time_to_conversation_log(llm_prompt_start)
            llm_prompt_end_str = self._format_iso_time_to_conversation_log(llm_prompt_end)
            llm_prompt_duration_str = f"{int(llm_prompt_duration)}ms" if llm_prompt_duration else "N/A"
            
            lines.append(f"LLM提示词: {llm_prompt_start_str}, {llm_prompt_end_str}, {llm_prompt_duration_str}")
            
            # LLM生成
            llm = stages.get("LLM生成", {})
            llm_call_start = llm.get("调用开始")
            llm_first_text = llm.get("首次文本输出")
            llm_all_completed = llm.get("全部文本完成")
            llm_call_end = llm.get("调用结束")
            llm_latency = llm.get("延迟_毫秒")
            
            llm_call_start_str = self._format_iso_time_to_conversation_log(llm_call_start)
            llm_first_text_str = self._format_iso_time_to_conversation_log(llm_first_text)
            llm_all_completed_str = self._format_iso_time_to_conversation_log(llm_all_completed)
            llm_call_end_str = self._format_iso_time_to_conversation_log(llm_call_end)
            llm_latency_str = f"{int(llm_latency)}ms" if llm_latency else "N/A"
            
            lines.append(f"LLM生成: {llm_call_start_str}, {llm_first_text_str}, {llm_all_completed_str}, {llm_call_end_str}, {llm_latency_str}")
            
            # TTS生成
            tts = stages.get("TTS生成", {})
            tts_call_start = tts.get("调用开始")
            tts_first_audio = tts.get("首次音频输出")
            tts_all_completed = tts.get("全部音频完成")
            tts_call_end = tts.get("调用结束")
            tts_latency = tts.get("延迟_毫秒")
            
            tts_call_start_str = self._format_iso_time_to_conversation_log(tts_call_start)
            tts_first_audio_str = self._format_iso_time_to_conversation_log(tts_first_audio)
            tts_all_completed_str = self._format_iso_time_to_conversation_log(tts_all_completed)
            tts_call_end_str = self._format_iso_time_to_conversation_log(tts_call_end)
            tts_latency_str = f"{int(tts_latency)}ms" if tts_latency else "N/A"
            
            lines.append(f"TTS生成: {tts_call_start_str}, {tts_first_audio_str}, {tts_all_completed_str}, {tts_call_end_str}, {tts_latency_str}")
            
            # 总结
            ai_response_text = ""
            if "LLM生成" in stages and "输出" in stages["LLM生成"]:
                ai_response_text = stages["LLM生成"]["输出"].get("文本", "")
            
            total_latency = summary.get("首次音频延迟_毫秒")
            total_latency_str = f"{int(total_latency)}ms" if total_latency else "N/A"
            
            lines.append(f"总结: {ai_response_text}, {total_latency_str}")
            lines.append("-" * 120)
            lines.append("")
            
            # 写入文件
            with open(conversation_log_path, 'a', encoding='utf-8') as f:
                f.write('\n'.join(lines) + '\n')
                
        except Exception as e:
            # 写入失败不影响主流程
            import sys
            print(f"[RequestLogger] 写入conversation_log.txt失败: {e}", file=sys.stderr)
    
    def _build_request_log(
        self,
        request_id: str,
        user_id: str,
        metrics: PerformanceMetrics
    ) -> Dict[str, Any]:
        """构建请求日志JSON对象"""
        
        # 构建各阶段数据
        stages = {}
        
        # 用户说话阶段
        if metrics.user_speech_start and metrics.user_speech_end:
            stages["用户说话"] = {
                "开始时间": self._format_timestamp(metrics.user_speech_start),
                "结束时间": self._format_timestamp(metrics.user_speech_end),
                "时长_毫秒": metrics.calculate_user_speech_duration_ms()
            }
        
        # STT阶段
        if metrics.stt_first_audio_received:
            stt_stage = {
                "首次收到音频": self._format_timestamp(metrics.stt_first_audio_received),
                "最后收到音频": self._format_timestamp(metrics.stt_last_audio_received),
                "完成时间": self._format_timestamp(metrics.stt_completed),
                "输入": {
                    "音频块数": metrics.stt_audio_chunks,
                    "总字节数": metrics.stt_total_bytes
                },
                "输出": {
                    "文本": metrics.user_text,
                    "是否最终": True
                }
            }
            if metrics.stt_first_text_generated:
                stt_stage["首次生成文本"] = self._format_timestamp(metrics.stt_first_text_generated)
            if metrics.stt_completed and metrics.stt_last_audio_received:
                stt_stage["延迟_毫秒"] = metrics.calculate_stt_latency_ms()
            stages["语音识别"] = stt_stage
        
        # LLM提示词准备阶段
        if metrics.llm_prompt_prepare_start and metrics.llm_prompt_prepare_end:
            stages["LLM提示词准备"] = {
                "开始时间": self._format_timestamp(metrics.llm_prompt_prepare_start),
                "结束时间": self._format_timestamp(metrics.llm_prompt_prepare_end),
                "时长_毫秒": metrics.calculate_llm_prompt_latency_ms()
            }
        
        # LLM生成阶段
        if metrics.llm_call_start:
            llm_stage = {
                "调用开始": self._format_timestamp(metrics.llm_call_start),
                "调用结束": self._format_timestamp(metrics.llm_call_end),
                "输入": {
                    "提示词长度": metrics.llm_prompt_length,
                    "上下文轮次": metrics.llm_context_round,
                    "完整提示词": metrics.llm_full_prompt if metrics.llm_full_prompt else ""
                },
                "输出": {
                    "文本": metrics.ai_response_text,
                    "情感": metrics.llm_emotion,
                    "句子数": metrics.llm_sentences
                }
            }
            if metrics.llm_first_text_output:
                llm_stage["首次文本输出"] = self._format_timestamp(metrics.llm_first_text_output)
            if metrics.llm_all_text_completed:
                llm_stage["全部文本完成"] = self._format_timestamp(metrics.llm_all_text_completed)
            if metrics.llm_first_text_output and metrics.llm_call_start:
                llm_stage["延迟_毫秒"] = metrics.calculate_llm_generation_latency_ms()
            stages["LLM生成"] = llm_stage
        
        # TTS阶段
        if metrics.tts_call_start:
            tts_stage = {
                "调用开始": self._format_timestamp(metrics.tts_call_start),
                "调用结束": self._format_timestamp(metrics.tts_call_end),
                "输入": {
                    "文本": metrics.ai_response_text,
                    "情感": metrics.llm_emotion,
                    "音色": metrics.tts_voice
                },
                "输出": {
                    "音频块数": metrics.tts_audio_chunks,
                    "总字节数": metrics.tts_total_bytes
                }
            }
            if metrics.tts_first_audio_output:
                tts_stage["首次音频输出"] = self._format_timestamp(metrics.tts_first_audio_output)
            if metrics.tts_all_audio_completed:
                tts_stage["全部音频完成"] = self._format_timestamp(metrics.tts_all_audio_completed)
            if metrics.tts_first_audio_output and metrics.tts_call_start:
                tts_stage["延迟_毫秒"] = metrics.calculate_tts_latency_ms()
            stages["TTS生成"] = tts_stage
        
        # 构建总结
        summary = {}
        if metrics.tts_first_audio_output and metrics.user_speech_end:
            summary["首次音频延迟_毫秒"] = metrics.calculate_total_latency_ms()
        if metrics.tts_all_audio_completed and metrics.user_speech_start:
            total_duration = (metrics.tts_all_audio_completed - metrics.user_speech_start) * 1000
            summary["总延迟_毫秒"] = total_duration
        
        # 构建完整日志对象
        log_entry = {
            "类型": "请求",
            "时间戳": self._format_timestamp(time.time()),
            "请求ID": request_id,
            "用户ID": user_id,
            "阶段": stages
        }
        
        if summary:
            log_entry["总结"] = summary
        
        # 添加消息ID（数据库关联）
        if metrics.user_message_id:
            log_entry["用户消息ID"] = metrics.user_message_id
        if metrics.assistant_message_id:
            log_entry["助手消息ID"] = metrics.assistant_message_id
        
        return log_entry
    
    def log_request(
        self,
        user_id: str,
        request_id: str,
        metrics: PerformanceMetrics
    ):
        """
        记录一条对话请求
        
        Args:
            user_id: 用户ID
            request_id: 请求ID
            metrics: 性能指标对象
        """
        try:
            log_entry = self._build_request_log(request_id, user_id, metrics)
            self.log_queue.put(log_entry)
        except Exception as e:
            # 记录失败不影响主流程
            import sys
            print(f"[RequestLogger] 记录请求日志失败: {e}", file=sys.stderr)
    
    def log_error(
        self,
        user_id: str,
        request_id: str,
        error_code: str,
        error_message: str,
        stage: Optional[str] = None,
        user_text: Optional[str] = None
    ):
        """
        记录错误请求
        
        Args:
            user_id: 用户ID
            request_id: 请求ID
            error_code: 错误代码
            error_message: 错误信息
            stage: 错误发生的阶段（可选）
            user_text: 用户输入文本（可选）
        """
        try:
            log_entry = {
                "类型": "错误",
                "时间戳": self._format_timestamp(time.time()),
                "请求ID": request_id,
                "用户ID": user_id,
                "错误": {
                    "代码": error_code,
                    "消息": error_message
                }
            }
            
            if stage:
                log_entry["错误"]["阶段"] = stage
            
            if user_text:
                log_entry["上下文"] = {
                    "用户文本": user_text,
                    "阶段": stage or "unknown"
                }
            
            self.log_queue.put(log_entry)
        except Exception as e:
            import sys
            print(f"[RequestLogger] 记录错误日志失败: {e}", file=sys.stderr)
    
    def shutdown(self):
        """关闭日志记录器（等待队列清空）"""
        self.running = False
        if self.writer_thread:
            self.log_queue.join()  # 等待队列清空
            self.writer_thread.join(timeout=5.0)


# 全局单例
request_logger = RequestLogger(log_dir=settings.LOG_DIR)