File size: 2,734 Bytes
3df5f14 | 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 | 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() |