Spaces:
Sleeping
Sleeping
| """ | |
| Session Manager for Digital Islamic Therapist Chatbot | |
| Handles chat history persistence using in-memory storage | |
| """ | |
| import uuid | |
| import time | |
| from datetime import datetime, timedelta | |
| from typing import List, Dict, Optional, Any | |
| class SessionManager: | |
| """Manages chat sessions and history using in-memory storage""" | |
| def __init__(self): | |
| """Initialize the session manager with in-memory storage""" | |
| self.sessions: Dict[str, Dict[str, Any]] = {} | |
| def create_session(self, user_id: Optional[str] = None) -> str: | |
| """ | |
| Create a new chat session | |
| Args: | |
| user_id: Optional user identifier | |
| Returns: | |
| Session ID | |
| """ | |
| session_id = str(uuid.uuid4()) | |
| session_data = { | |
| "session_id": session_id, | |
| "user_id": user_id or "anonymous", | |
| "created_at": datetime.utcnow(), | |
| "last_active": datetime.utcnow(), | |
| "history": [], | |
| "expired": False | |
| } | |
| self.sessions[session_id] = session_data | |
| return session_id | |
| def get_session(self, session_id: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| Retrieve a session by ID | |
| Args: | |
| session_id: Session identifier | |
| Returns: | |
| Session data or None if not found | |
| """ | |
| return self.sessions.get(session_id) | |
| def add_message_to_history(self, session_id: str, role: str, content: str) -> bool: | |
| """ | |
| Add a message to the chat history | |
| Args: | |
| session_id: Session identifier | |
| role: Role of the message sender (user/assistant) | |
| content: Message content | |
| Returns: | |
| True if successful, False otherwise | |
| """ | |
| if session_id not in self.sessions: | |
| # Create session if it doesn't exist - use provided session_id | |
| self.sessions[session_id] = { | |
| "session_id": session_id, | |
| "user_id": "anonymous", | |
| "created_at": datetime.utcnow(), | |
| "last_active": datetime.utcnow(), | |
| "history": [], | |
| "expired": False | |
| } | |
| try: | |
| session_data = self.sessions[session_id] | |
| # Add new message to history | |
| message = { | |
| "role": role, | |
| "content": content, | |
| "timestamp": datetime.utcnow() | |
| } | |
| # Update session data | |
| session_data["history"].append(message) | |
| session_data["last_active"] = datetime.utcnow() | |
| # Keep only the last 20 messages to prevent memory bloat | |
| if len(session_data["history"]) > 20: | |
| session_data["history"] = session_data["history"][-20:] | |
| return True | |
| except Exception as e: | |
| print(f"Warning: Failed to add message to session history: {e}") | |
| return False | |
| def get_session_history(self, session_id: str) -> List[Dict[str, str]]: | |
| """ | |
| Get the chat history for a session | |
| Args: | |
| session_id: Session identifier | |
| Returns: | |
| List of message dictionaries | |
| """ | |
| session_data = self.get_session(session_id) | |
| if session_data and "history" in session_data: | |
| # Return only role and content for each message | |
| return [{"role": msg["role"], "content": msg["content"]} | |
| for msg in session_data["history"]] | |
| return [] | |
| def cleanup_expired_sessions(self, expiry_hours: int = 24) -> int: | |
| """ | |
| Clean up expired sessions | |
| Args: | |
| expiry_hours: Number of hours after which sessions expire | |
| Returns: | |
| Number of sessions cleaned up | |
| """ | |
| try: | |
| cutoff_time = datetime.utcnow() - timedelta(hours=expiry_hours) | |
| expired_session_ids = [] | |
| for session_id, session_data in self.sessions.items(): | |
| if session_data.get("last_active") < cutoff_time and not session_data.get("expired", False): | |
| expired_session_ids.append(session_id) | |
| for session_id in expired_session_ids: | |
| self.sessions[session_id]["expired"] = True | |
| return len(expired_session_ids) | |
| except Exception as e: | |
| print(f"Warning: Failed to clean up expired sessions: {e}") | |
| return 0 | |
| # Global session manager instance | |
| session_manager = SessionManager() |