Spaces:
Runtime error
Runtime error
| from mongoengine import * | |
| from datetime import datetime | |
| MONGODB_URI="mongodb+srv://brijrvl:[email protected]/?retryWrites=true&w=majority&appName=orderbot" | |
| DATABASE_NAME="analyticsbot" | |
| connect( | |
| db=DATABASE_NAME, | |
| host=MONGODB_URI, | |
| ) | |
| class ChatManager: | |
| def __init__(self): | |
| self.connect_to_db() | |
| def connect_to_db(self): | |
| # Connection to MongoDB | |
| connect(db=DATABASE_NAME, host=MONGODB_URI) | |
| class Session(Document): | |
| phone_number_id = StringField(required=True, unique=True) | |
| chat_history = StringField(default="") | |
| meta = { | |
| 'collection': 'sessions', | |
| 'db_alias': 'default' | |
| } | |
| class Message(Document): | |
| session_id = ReferenceField('ChatManager.Session', required=True) | |
| phone_number_id = StringField(required=True) | |
| role = StringField(choices=['USER', 'ASSISTANT'], required=True) | |
| content = StringField(required=True) | |
| timestamp = DateTimeField(default=datetime.utcnow, required=True) | |
| meta = { | |
| 'collection': 'messages', | |
| 'db_alias': 'default' | |
| } | |
| def get_or_create_session(self, phone_number): | |
| session = self.Session.objects(phone_number_id=phone_number).first() | |
| if not session: | |
| session = self.Session(phone_number_id=phone_number).save() | |
| return session | |
| def save_message(self, session_id, phone_number, role, content): | |
| print("Here 3") | |
| message = self.Message( | |
| session_id=session_id, | |
| phone_number_id=phone_number, | |
| role=role, | |
| content=content, | |
| timestamp=datetime.utcnow() | |
| ) | |
| message.save() | |
| print("Here 4") | |
| return message | |
| def get_last_n_messages(self, session, n=10): | |
| # Retrieve the last `n` messages for the given session | |
| messages = self.Message.objects(session_id=session.id).order_by('-timestamp')[:n] | |
| return messages | |
| def build_chat_history(self, messages): | |
| # Build chat history string from the retrieved messages | |
| chat_history = [] | |
| for message in messages: | |
| chat_history.append(f"{message.role}: {message.content}") | |
| return "\n".join(reversed(chat_history)) | |
| def handle_message(self, phone_number, content): | |
| session = self.get_or_create_session(phone_number) | |
| self.save_message(session.id, phone_number, 'USER', content) | |
| last_messages = self.get_last_n_messages(session) | |
| if len(last_messages) >= 10: | |
| last_three_messages = last_messages[:10] | |
| else: | |
| last_three_messages = last_messages | |
| str_chat_history = self.build_chat_history(last_three_messages) | |
| return str_chat_history | |