Spaces:
Sleeping
Sleeping
from config import settings | |
from langchain_mongodb.chat_message_histories import MongoDBChatMessageHistory | |
import json | |
import random | |
def create_new_session(user_id): | |
mongodb_chatbot_message_collection = settings.mongodb_db.get_collection(settings.MONGODB_DB_USER_SESSIONS_COLLECTION_NAME) | |
# print("********************************************************** NEW SESSION *******************************************************************") | |
try: | |
user_sessions = mongodb_chatbot_message_collection.find_one({"user_id": user_id}) | |
# print(user_sessions, "LAST") | |
if user_sessions: | |
session_ids = user_sessions['session_id'] | |
session_id = user_id + "_bmoxi_" + str(int(session_ids[-1].split("_")[-1])+1) | |
# print("session_id:", session_id) | |
mongodb_chatbot_message_collection.update_one({ "user_id": user_id },{"$push": {"session_id":session_id}}) | |
return session_id | |
else: | |
mongodb_chatbot_message_collection.insert_one({ "user_id": user_id,"session_id":[user_id + "_bmoxi_1"]}) | |
return user_id + "_bmoxi_1" | |
except: | |
mongodb_chatbot_message_collection.insert_one({ "user_id": user_id,"session_id":[user_id + "_bmoxi_1"]}) | |
return user_id + "_bmoxi_1" | |
def get_last_session(user_id="user_1"): | |
mongodb_chatbot_message_collection = settings.mongodb_db.get_collection( | |
settings.MONGODB_DB_USER_SESSIONS_COLLECTION_NAME) | |
sessions_cursor = mongodb_chatbot_message_collection.find_one( | |
{"user_id": user_id}) | |
# print(sessions_cursor) | |
sessions_list = sessions_cursor['session_id'] | |
second_last_session_id = None | |
if len(sessions_list) >= 2: | |
second_last_session_id = sessions_list[-2] | |
change_use_tool_status(sessions_list[-1]) | |
return {"last_session_id": sessions_list[-1], "second_last_session_id": second_last_session_id if second_last_session_id else None} | |
def get_chat_history(session_id="bmoxinew"): | |
# Set up MongoDB for storing chat history | |
chat_history = MongoDBChatMessageHistory( | |
connection_string=settings.MONGODB_CONNECTION_STRING, | |
database_name=settings.MONGODB_DB_NAME, # Specify the database name here | |
collection_name=settings.MONGODB_DB_CHAT_COLLECTION_NAME, | |
session_id=session_id, | |
) | |
return chat_history | |
def get_chat_bot_name(user_id="user_1"): | |
# print(settings.MONGODB_CONNECTION_STRING) | |
# print(settings.mongodb_chatbot_name_collection) | |
result = settings.mongodb_chatbot_name_collection.find_one( | |
{"user_id": user_id}) | |
# print("CHATBOT RESULT", result, type(result)) | |
if result: | |
# print(result) | |
return result['chat_bot_name'] | |
return settings.CHATBOT_NAME | |
def get_last_conversion(last_session_id, second_last_session_id): | |
mongodb_chatbot_message_collection = settings.mongodb_db.get_collection( | |
settings.MONGODB_DB_CHAT_COLLECTION_NAME) | |
collection_count = mongodb_chatbot_message_collection.count_documents({"SessionId": last_session_id}) | |
# print("******************************** data********************888") | |
# print(collection_count) | |
# print(last_session_id) | |
# print("*********************************") | |
if collection_count <=2: | |
sessions_cursor = mongodb_chatbot_message_collection.find({"SessionId": second_last_session_id}) # Sort by timestamp descending and limit to 2 results | |
# print(sessions_cursor) | |
sessions_list = list(sessions_cursor) | |
# print(sessions_list) | |
conversation = """""" | |
for document in sessions_list: | |
# print("MY document") | |
# print(document) | |
if "History" in document: | |
history = json.loads(document['History']) | |
# print(history) | |
# print(history['type']) | |
# print(history['data']) | |
# print(history['data']['content']) | |
conversation += f"""{history['type']}: {history['data']['content']}\n""" | |
return conversation | |
else: | |
return None | |
def set_chat_bot_name(name, user_id): | |
# Insert document into collection | |
insert_result = settings.mongodb_chatbot_name_collection.update_one({"user_id": user_id}, { "$set": { "chat_bot_name": name } }, upsert=True) | |
# print("done successfully...") | |
return name | |
def save_mood_summary(data,user_id): | |
mongodb_chatbot_mood_collection = settings.mongodb_db.get_collection( | |
settings.MONGODB_DB_CHAT_BOT_MOOD_COLLECTION_NAME) | |
mongodb_chatbot_mood_collection.insert_one(({"user_id": user_id, "mood_summary": data })) | |
def get_mood_data(user_id): | |
mongodb_chatbot_mood_collection = settings.mongodb_db.get_collection(settings.MONGODB_DB_CHAT_BOT_MOOD_COLLECTION_NAME) | |
data = mongodb_chatbot_mood_collection.find_one({"user_id":user_id}) | |
if data: | |
return data['mood_summary'] | |
else: | |
return "" | |
def change_use_tool_status(last_session_id): | |
mongodb_chatbot_tools_collection = settings.mongodb_db.get_collection(settings.MONGODB_DB_CHAT_BOT_TOOLS_COLLECTION_NAME) | |
if mongodb_chatbot_tools_collection.count_documents({"session_id": last_session_id})==0: | |
mongodb_chatbot_tools_collection.insert_one({"session_id": last_session_id,"set_tools": False} ) | |
def isFirstSession(user_id): | |
mongodb_chatbot_message_collection = settings.mongodb_db.get_collection(settings.MONGODB_DB_USER_SESSIONS_COLLECTION_NAME) | |
user_sessions = mongodb_chatbot_message_collection.find_one({"user_id": user_id}) | |
if user_sessions: | |
return len(user_sessions['session_id'])==1 | |
else: | |
return False |