Spaces:
Running
Running
| from datetime import datetime | |
| from typing import List, Dict | |
| import json | |
| class Message: | |
| def __init__(self, role: str, content: str, language: str = 'en'): | |
| self.role = role | |
| self.content = content | |
| self.language = language | |
| self.timestamp = datetime.now().isoformat() | |
| def to_dict(self): | |
| return { | |
| 'role': self.role, | |
| 'content': self.content, | |
| 'language': self.language, | |
| 'timestamp': self.timestamp | |
| } | |
| class ConversationManager: | |
| def __init__(self): | |
| self.conversations: Dict[str, List[Message]] = {} | |
| def add_message(self, conversation_id: str, role: str, content: str, language: str = 'en'): | |
| """إضافة رسالة للمحادثة""" | |
| if conversation_id not in self.conversations: | |
| self.conversations[conversation_id] = [] | |
| message = Message(role, content, language) | |
| self.conversations[conversation_id].append(message) | |
| def get_messages(self, conversation_id: str) -> List[Dict]: | |
| """الحصول على رسائل المحادثة""" | |
| if conversation_id not in self.conversations: | |
| return [] | |
| return [msg.to_dict() for msg in self.conversations[conversation_id]] | |
| def get_context(self, conversation_id: str, limit: int = 10) -> List[Dict]: | |
| """الحصول على السياق (آخر n رسالة)""" | |
| messages = self.get_messages(conversation_id) | |
| return messages[-limit:] if len(messages) > limit else messages | |
| def get_all_conversations(self) -> List[Dict]: | |
| """الحصول على قائمة المحادثات""" | |
| conversations = [] | |
| for conv_id, messages in self.conversations.items(): | |
| if messages: | |
| conversations.append({ | |
| 'id': conv_id, | |
| 'title': messages[0].content[:50] + '...' if len(messages[0].content) > 50 else messages[0].content, | |
| 'last_message': messages[-1].timestamp, | |
| 'message_count': len(messages) | |
| }) | |
| return sorted(conversations, key=lambda x: x['last_message'], reverse=True) | |
| def delete_conversation(self, conversation_id: str): | |
| """حذف محادثة""" | |
| if conversation_id in self.conversations: | |
| del self.conversations[conversation_id] | |
| def get_last_timestamp(self, conversation_id: str) -> str: | |
| """الحصول على آخر وقت رسالة""" | |
| if conversation_id in self.conversations and self.conversations[conversation_id]: | |
| return self.conversations[conversation_id][-1].timestamp | |
| return datetime.now().isoformat() |