Spaces:
Sleeping
Sleeping
| import json | |
| import os | |
| import uuid | |
| from datetime import datetime | |
| import logging | |
| from hf_storage import HFStorageManager | |
| logger = logging.getLogger(__name__) | |
| class ChatManager: | |
| def __init__(self, history_file='chat_history.json'): | |
| self.history_file = history_file | |
| self.storage = HFStorageManager() | |
| def _load_history(self): | |
| data = self.storage.load_file(self.history_file) | |
| if data is None: | |
| return {} | |
| if isinstance(data, dict): | |
| return data | |
| return {} | |
| def _save_history(self, history): | |
| self.storage.save_file(self.history_file, history) | |
| def get_student_history(self, enrollment_no): | |
| history = self._load_history() | |
| # Return list of sessions, sorted by timestamp (newest first) | |
| sessions = history.get(str(enrollment_no), []) | |
| sessions.sort(key=lambda x: x.get('timestamp', ''), reverse=True) | |
| return sessions | |
| def create_session(self, enrollment_no, first_message): | |
| history = self._load_history() | |
| enrollment_no = str(enrollment_no) | |
| if enrollment_no not in history: | |
| history[enrollment_no] = [] | |
| # Generate title from first few words of the message | |
| title = ' '.join(first_message.split()[:5]) + '...' | |
| if len(title) > 50: | |
| title = title[:47] + '...' | |
| session_id = str(uuid.uuid4()) | |
| timestamp = datetime.now().isoformat() | |
| new_session = { | |
| 'id': session_id, | |
| 'title': title, | |
| 'timestamp': timestamp, | |
| 'messages': [] | |
| } | |
| history[enrollment_no].append(new_session) | |
| self._save_history(history) | |
| return new_session | |
| def add_message(self, enrollment_no, session_id, sender, text): | |
| history = self._load_history() | |
| enrollment_no = str(enrollment_no) | |
| if enrollment_no not in history: | |
| return None | |
| for session in history[enrollment_no]: | |
| if session['id'] == session_id: | |
| message = { | |
| 'sender': sender, | |
| 'text': text, | |
| 'timestamp': datetime.now().isoformat() | |
| } | |
| session['messages'].append(message) | |
| # Update session timestamp to bring it to top | |
| session['timestamp'] = datetime.now().isoformat() | |
| self._save_history(history) | |
| return message | |
| return None | |
| def get_session(self, enrollment_no, session_id): | |
| history = self._load_history() | |
| enrollment_no = str(enrollment_no) | |
| if enrollment_no in history: | |
| for session in history[enrollment_no]: | |
| if session['id'] == session_id: | |
| return session | |
| return None | |