File size: 6,288 Bytes
5feba25 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | """
Session Management for Medical Predictor Chatbot
Handles: session creation, persistence, cleanup, metadata tracking
"""
import json
import logging
from pathlib import Path
from datetime import datetime
from typing import Dict, Optional, Any
logger = logging.getLogger(__name__)
# Session storage locations
SESSION_DIR = Path("/tmp/deepsense_sessions")
METADATA_FILE = SESSION_DIR / "session_metadata.json"
# Ensure directories exist
SESSION_DIR.mkdir(exist_ok=True)
class SessionManager:
"""Manages session lifecycle, persistence, and metadata tracking"""
def __init__(self, session_dir: Path = SESSION_DIR):
self.session_dir = session_dir
self.metadata_file = METADATA_FILE
self._load_all_metadata()
def _load_all_metadata(self):
"""Load all session metadata from file"""
try:
if self.metadata_file.exists():
with open(self.metadata_file, 'r') as f:
self.all_metadata = json.load(f)
else:
self.all_metadata = {}
except Exception as e:
logger.warning(f"⚠️ Could not load session metadata: {e}")
self.all_metadata = {}
def _save_all_metadata(self):
"""Persist all session metadata to file"""
try:
with open(self.metadata_file, 'w') as f:
json.dump(self.all_metadata, f, indent=2)
except Exception as e:
logger.error(f"❌ Error saving session metadata: {e}")
def create_session(self, session_id: str, user_name: str = "User") -> Dict[str, Any]:
"""
Create a new session with metadata
Args:
session_id: Unique session identifier
user_name: User name (extracted from greeting)
Returns:
Session metadata dict
"""
metadata = {
"session_id": session_id,
"user_name": user_name,
"created_at": datetime.now().isoformat(),
"last_accessed": datetime.now().isoformat(),
"features_collected": 0,
"status": "active",
"reset_count": 0
}
self.all_metadata[session_id] = metadata
self._save_all_metadata()
logger.info(f"✅ Created session: {session_id} for {user_name}")
return metadata
def get_session_metadata(self, session_id: str) -> Optional[Dict[str, Any]]:
"""Get metadata for a specific session"""
return self.all_metadata.get(session_id)
def update_session(self, session_id: str, features_collected: int):
"""Update session metadata with current stats"""
if session_id in self.all_metadata:
self.all_metadata[session_id]["last_accessed"] = datetime.now().isoformat()
self.all_metadata[session_id]["features_collected"] = features_collected
self._save_all_metadata()
def reset_session(self, session_id: str) -> bool:
"""
Reset a session: delete session file and update metadata
Args:
session_id: Session to reset
Returns:
True if successful, False otherwise
"""
try:
# Delete session state file
session_file = self.session_dir / f"{session_id}.json"
if session_file.exists():
session_file.unlink()
logger.info(f"🗑️ Deleted session file: {session_file}")
# Update metadata
if session_id in self.all_metadata:
self.all_metadata[session_id]["last_accessed"] = datetime.now().isoformat()
self.all_metadata[session_id]["features_collected"] = 0
self.all_metadata[session_id]["reset_count"] += 1
self.all_metadata[session_id]["status"] = "reset"
self._save_all_metadata()
reset_count = self.all_metadata[session_id]["reset_count"]
logger.info(f"🔄 Session {session_id} reset (reset count: {reset_count})")
return True
except Exception as e:
logger.error(f"❌ Error resetting session {session_id}: {e}")
return False
def cleanup_old_sessions(self, max_age_hours: int = 24):
"""
Clean up session files older than max_age_hours
Args:
max_age_hours: Sessions older than this are deleted
"""
try:
import time
current_time = time.time()
deleted_count = 0
for session_id, metadata in list(self.all_metadata.items()):
session_file = self.session_dir / f"{session_id}.json"
if session_file.exists():
file_age_hours = (current_time - session_file.stat().st_mtime) / 3600
if file_age_hours > max_age_hours:
session_file.unlink()
logger.info(f"🗑️ Cleaned up old session: {session_id}")
deleted_count += 1
if deleted_count > 0:
logger.info(f"🧹 Cleaned up {deleted_count} old session(s)")
except Exception as e:
logger.warning(f"⚠️ Error during session cleanup: {e}")
def get_all_active_sessions(self) -> Dict[str, Dict[str, Any]]:
"""Get all active sessions"""
return {sid: meta for sid, meta in self.all_metadata.items()
if meta.get("status") == "active"}
def list_sessions(self) -> str:
"""Get formatted list of all sessions"""
if not self.all_metadata:
return "No active sessions"
lines = ["Active Sessions:"]
for sid, meta in self.all_metadata.items():
lines.append(
f" • {meta['session_id']}: {meta['user_name']} "
f"({meta['features_collected']}/16 features, "
f"reset {meta['reset_count']} times)"
)
return "\n".join(lines)
# Global session manager instance
_session_manager: Optional[SessionManager] = None
def get_session_manager() -> SessionManager:
"""Get or create the global session manager"""
global _session_manager
if _session_manager is None:
_session_manager = SessionManager()
return _session_manager
|