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 |