Spaces:
Runtime error
Runtime error
File size: 4,669 Bytes
eeb0f9c |
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 |
"""
Session Store - Persistent storage for conversation memory
Saves and loads user sessions across app restarts
"""
import json
import os
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any
class SessionStore:
"""Manages persistent storage of conversation sessions"""
def __init__(self, storage_dir: str = "sessions"):
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(parents=True, exist_ok=True)
def save_session(self, user_id: str, session_data: Dict[str, Any]) -> None:
"""
Save user session to disk
Args:
user_id: Unique user identifier
session_data: Session data to save (memory state)
"""
session_file = self.storage_dir / f"{user_id}.json"
# Add metadata
session_data['last_updated'] = datetime.now().isoformat()
session_data['user_id'] = user_id
with open(session_file, 'w', encoding='utf-8') as f:
json.dump(session_data, f, ensure_ascii=False, indent=2)
def load_session(self, user_id: str) -> Optional[Dict[str, Any]]:
"""
Load user session from disk
Args:
user_id: Unique user identifier
Returns:
Session data or None if not found
"""
session_file = self.storage_dir / f"{user_id}.json"
if not session_file.exists():
return None
try:
with open(session_file, 'r', encoding='utf-8') as f:
return json.load(f)
except (json.JSONDecodeError, IOError) as e:
print(f"⚠️ Error loading session for {user_id}: {e}")
return None
def delete_session(self, user_id: str) -> bool:
"""
Delete user session
Args:
user_id: Unique user identifier
Returns:
True if deleted, False if not found
"""
session_file = self.storage_dir / f"{user_id}.json"
if session_file.exists():
session_file.unlink()
return True
return False
def list_sessions(self) -> list:
"""
List all stored sessions
Returns:
List of user IDs with sessions
"""
return [f.stem for f in self.storage_dir.glob("*.json")]
def get_session_info(self, user_id: str) -> Optional[Dict[str, Any]]:
"""
Get session metadata without loading full data
Args:
user_id: Unique user identifier
Returns:
Session metadata or None
"""
session_file = self.storage_dir / f"{user_id}.json"
if not session_file.exists():
return None
try:
with open(session_file, 'r', encoding='utf-8') as f:
data = json.load(f)
return {
'user_id': data.get('user_id'),
'last_updated': data.get('last_updated'),
'has_profile': bool(data.get('user_profile')),
'current_agent': data.get('current_agent'),
'conversation_count': len(data.get('conversation_history', []))
}
except (json.JSONDecodeError, IOError):
return None
def cleanup_old_sessions(self, days: int = 30) -> int:
"""
Delete sessions older than specified days
Args:
days: Number of days to keep sessions
Returns:
Number of sessions deleted
"""
from datetime import timedelta
cutoff_date = datetime.now() - timedelta(days=days)
deleted_count = 0
for session_file in self.storage_dir.glob("*.json"):
try:
with open(session_file, 'r', encoding='utf-8') as f:
data = json.load(f)
last_updated = datetime.fromisoformat(data.get('last_updated', ''))
if last_updated < cutoff_date:
session_file.unlink()
deleted_count += 1
except (json.JSONDecodeError, IOError, ValueError):
continue
return deleted_count
# Global instance
_session_store = None
def get_session_store() -> SessionStore:
"""Get global session store instance"""
global _session_store
if _session_store is None:
_session_store = SessionStore()
return _session_store
|