core/session_manager.py

"""
Session Manager for ChatCal Voice - Handles user sessions in Gradio environment.

Since we're on Hugging Face without persistent storage, we'll use in-memory
session management with automatic cleanup.
"""

import time
import uuid
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta

from .session import SessionData
from .config import config


class SessionManager:
    """Manages user sessions for the voice-enabled ChatCal."""
    
    def __init__(self):
        self.sessions: Dict[str, SessionData] = {}
        self.last_cleanup = time.time()
        self.cleanup_interval = 300  # 5 minutes
    
    async def get_session(self, session_id: Optional[str] = None) -> SessionData:
        """Get or create a session."""
        # Auto-cleanup old sessions periodically
        await self._cleanup_expired_sessions()
        
        # Create new session if none provided
        if not session_id:
            session_id = self._generate_session_id()
        
        # Return existing session or create new one
        if session_id in self.sessions:
            session = self.sessions[session_id]
            session.last_activity = datetime.now()
            return session
        
        # Create new session
        session = SessionData(session_id=session_id)
        self.sessions[session_id] = session
        return session
    
    def _generate_session_id(self) -> str:
        """Generate a unique session ID."""
        timestamp = int(time.time())
        unique_id = str(uuid.uuid4())[:8]
        return f"chatcal_{timestamp}_{unique_id}"
    
    async def _cleanup_expired_sessions(self):
        """Clean up expired sessions."""
        current_time = time.time()
        
        # Only run cleanup periodically
        if current_time - self.last_cleanup < self.cleanup_interval:
            return
        
        cutoff_time = datetime.now() - timedelta(minutes=config.session_timeout_minutes)
        expired_sessions = [
            session_id for session_id, session in self.sessions.items()
            if session.last_activity < cutoff_time
        ]
        
        for session_id in expired_sessions:
            del self.sessions[session_id]
        
        if expired_sessions:
            print(f"🧹 Cleaned up {len(expired_sessions)} expired sessions")
        
        self.last_cleanup = current_time
    
    async def delete_session(self, session_id: str):
        """Delete a specific session."""
        if session_id in self.sessions:
            del self.sessions[session_id]
    
    def get_session_count(self) -> int:
        """Get the number of active sessions."""
        return len(self.sessions)
    
    def get_session_stats(self) -> Dict[str, Any]:
        """Get session statistics."""
        return {
            "active_sessions": len(self.sessions),
            "total_messages": sum(len(s.conversation_history) for s in self.sessions.values()),
            "sessions_with_user_info": sum(
                1 for s in self.sessions.values() 
                if s.user_info.get("name") or s.user_info.get("email")
            )
        }


# Global session manager instance
session_manager = SessionManager()