Spaces:
Paused
Paused
| """ | |
| 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 | |
| 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) | |
| } | |
| 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 |