| 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() |