import json import time from datetime import datetime from pathlib import Path from typing import Optional import threading class ConversationLogger: """Logs conversation details including session config and message exchanges""" def __init__(self, logs_dir: str = "logs"): self.logs_dir = Path(logs_dir).absolute() self.logs_dir.mkdir(exist_ok=True, parents=True) self.current_session: Optional[dict] = None self.lock = threading.Lock() print(f"ConversationLogger initialized. Logs directory: {self.logs_dir}") print(f"Directory exists: {self.logs_dir.exists()}") print(f"Directory writable: {self.logs_dir.is_dir()}") def start_session( self, room_name: str, instructions: str, voice_id: str, language: str, llm_model: str, first_message: Optional[str] = None, ): """Start a new conversation session""" with self.lock: # Force end any existing session before starting a new one if self.current_session: print(f"Warning: Starting new session while one is active. Ending previous session.") self._end_session_unsafe() session_id = f"{room_name}_{int(time.time() * 1000)}" # Use milliseconds for uniqueness config = { "instructions": instructions, "voice_id": voice_id, "language": language, "llm_model": llm_model, } if first_message: config["first_message"] = first_message self.current_session = { "session_id": session_id, "room_name": room_name, "start_time": datetime.now().isoformat(), "config": config, "messages": [], } print(f"Started new conversation session: {session_id}") return session_id def _save_current_session(self): """Save the current session to file immediately (internal, no lock)""" if not self.current_session: return None session_id = self.current_session["session_id"] log_file = self.logs_dir / f"{session_id}.json" try: # Update end time on every save self.current_session["end_time"] = datetime.now().isoformat() # Calculate total duration start = datetime.fromisoformat(self.current_session["start_time"]) end = datetime.fromisoformat(self.current_session["end_time"]) self.current_session["total_duration"] = (end - start).total_seconds() with open(log_file, "w", encoding="utf-8") as f: json.dump(self.current_session, f, indent=2, ensure_ascii=False) return str(log_file) except Exception as e: print(f"ERROR saving session: {e}") import traceback traceback.print_exc() return None def log_user_message( self, transcript: str, duration: Optional[float] = None ): """Log a user message""" if not self.current_session: print("Warning: log_user_message called but no active session") return with self.lock: message = { "timestamp": datetime.now().isoformat(), "type": "user", "transcript": transcript, "duration": duration, } self.current_session["messages"].append(message) print(f"Logged user message: {transcript[:50]}... (total messages: {len(self.current_session['messages'])})") # Save immediately after each message log_file = self._save_current_session() if log_file: print(f"Auto-saved to: {log_file}") def log_bot_message( self, response: str, duration: Optional[float] = None, llm_ms: Optional[float] = None, ): """Log a bot response. Optionally include LLM latency (ms). TTS/e2e can be set via update_last_bot_latency.""" if not self.current_session: print("Warning: log_bot_message called but no active session") return with self.lock: message = { "timestamp": datetime.now().isoformat(), "type": "bot", "response": response, "duration": duration, } if llm_ms is not None: message["latency_llm_ms"] = round(llm_ms, 0) self.current_session["messages"].append(message) print(f"Logged bot message: {response[:50]}... (total messages: {len(self.current_session['messages'])})") # Save immediately after each message log_file = self._save_current_session() if log_file: print(f"Auto-saved to: {log_file}") def update_last_bot_latency( self, tts_ms: Optional[float] = None, e2e_ms: Optional[float] = None, ): """Add TTS and/or end-to-end latency (ms) to the most recent bot message that does not yet have latency_tts_ms. This avoids race with log_bot_message (thread pool): we attach to the message for this turn, not the previous one.""" if not self.current_session or not self.current_session["messages"]: return with self.lock: for m in reversed(self.current_session["messages"]): if m.get("type") == "bot" and "latency_tts_ms" not in m: if tts_ms is not None: m["latency_tts_ms"] = round(tts_ms, 0) if e2e_ms is not None: m["latency_e2e_ms"] = round(e2e_ms, 0) self._save_current_session() return def _end_session_unsafe(self): """Internal method to end session without acquiring lock""" if not self.current_session: return None # Save one final time log_file = self._save_current_session() if log_file: print(f"Session ended - Final save: {log_file}") print(f" - Messages: {len(self.current_session['messages'])}") print(f" - Duration: {self.current_session.get('total_duration', 0):.2f}s") self.current_session = None return log_file def end_session(self): """End the current session and save to file""" if not self.current_session: print("Warning: end_session called but no active session") return None with self.lock: return self._end_session_unsafe() def get_all_sessions(self): """Get all conversation sessions""" sessions = [] for log_file in self.logs_dir.glob("*.json"): try: with open(log_file, "r", encoding="utf-8") as f: sessions.append(json.load(f)) except Exception as e: print(f"Error reading {log_file}: {e}") # Sort by start time, newest first sessions.sort(key=lambda x: x.get("start_time", ""), reverse=True) return sessions def get_session(self, session_id: str): """Get a specific session by ID""" log_file = self.logs_dir / f"{session_id}.json" if log_file.exists(): with open(log_file, "r", encoding="utf-8") as f: return json.load(f) return None # Global logger instance - initialize it properly print("=" * 60) print("Initializing ConversationLogger...") logger = ConversationLogger() print("ConversationLogger ready") print("=" * 60)