core/session.py

"""
Session Data Model for ChatCal Voice.

Handles conversation state, user information, and session persistence
in the Hugging Face Gradio environment.
"""

from typing import Dict, List, Any, Optional
from datetime import datetime
from dataclasses import dataclass, field


@dataclass
class SessionData:
    """Data structure for user sessions."""
    
    session_id: str
    created_at: datetime = field(default_factory=datetime.now)
    last_activity: datetime = field(default_factory=datetime.now)
    
    # User information extracted from conversation
    user_info: Dict[str, Any] = field(default_factory=lambda: {
        "name": None,
        "email": None,
        "phone": None,
        "preferences": {},
        "timezone": None
    })
    
    # Conversation history
    conversation_history: List[Dict[str, str]] = field(default_factory=list)
    
    # Session state for multi-turn operations
    session_state: Dict[str, Any] = field(default_factory=lambda: {
        "pending_operation": None,  # "booking", "cancellation", "availability"
        "operation_context": {},    # Context data for operations
        "awaiting_clarification": False,
        "last_voice_input": None,
        "voice_enabled": True
    })
    
    # Booking history for this session
    booking_history: List[Dict[str, Any]] = field(default_factory=list)
    
    def add_message(self, role: str, content: str):
        """Add a message to conversation history."""
        self.conversation_history.append({
            "role": role,  # "user" or "assistant"
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
        
        # Keep only recent messages to prevent memory issues
        max_history = 50
        if len(self.conversation_history) > max_history:
            self.conversation_history = self.conversation_history[-max_history:]
        
        self.last_activity = datetime.now()
    
    def get_recent_messages(self, count: int = 10) -> List[Dict[str, str]]:
        """Get recent conversation messages."""
        return self.conversation_history[-count:] if self.conversation_history else []
    
    def update_user_info(self, **kwargs):
        """Update user information."""
        for key, value in kwargs.items():
            if key in self.user_info and value:
                self.user_info[key] = value
        self.last_activity = datetime.now()
    
    def has_required_user_info(self) -> bool:
        """Check if session has minimum required user information."""
        return (
            bool(self.user_info.get("name")) and
            (bool(self.user_info.get("email")) or bool(self.user_info.get("phone")))
        )
    
    def get_user_summary(self) -> str:
        """Get a summary of user information."""
        name = self.user_info.get("name", "Unknown")
        contact = self.user_info.get("email") or self.user_info.get("phone") or "No contact"
        return f"{name} ({contact})"
    
    def set_pending_operation(self, operation: str, context: Dict[str, Any] = None):
        """Set a pending operation with context."""
        self.session_state["pending_operation"] = operation
        self.session_state["operation_context"] = context or {}
        self.session_state["awaiting_clarification"] = False
        self.last_activity = datetime.now()
    
    def clear_pending_operation(self):
        """Clear any pending operation."""
        self.session_state["pending_operation"] = None
        self.session_state["operation_context"] = {}
        self.session_state["awaiting_clarification"] = False
        self.last_activity = datetime.now()
    
    def add_booking(self, booking_info: Dict[str, Any]):
        """Add a booking to the session history."""
        booking_info["session_id"] = self.session_id
        booking_info["timestamp"] = datetime.now().isoformat()
        self.booking_history.append(booking_info)
        self.last_activity = datetime.now()
    
    def get_session_duration_minutes(self) -> int:
        """Get session duration in minutes."""
        delta = datetime.now() - self.created_at
        return int(delta.total_seconds() / 60)
    
    def is_expired(self, timeout_minutes: int = 30) -> bool:
        """Check if session is expired."""
        delta = datetime.now() - self.last_activity
        return delta.total_seconds() > (timeout_minutes * 60)
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert session to dictionary for serialization."""
        return {
            "session_id": self.session_id,
            "created_at": self.created_at.isoformat(),
            "last_activity": self.last_activity.isoformat(),
            "user_info": self.user_info,
            "conversation_count": len(self.conversation_history),
            "session_state": self.session_state,
            "booking_count": len(self.booking_history)
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'SessionData':
        """Create session from dictionary."""
        session = cls(session_id=data["session_id"])
        session.created_at = datetime.fromisoformat(data["created_at"])
        session.last_activity = datetime.fromisoformat(data["last_activity"])
        session.user_info = data.get("user_info", {})
        session.session_state = data.get("session_state", {})
        return session