Spaces:
Runtime error
Runtime error
File size: 2,862 Bytes
360b354 |
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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 |
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
|