Viewing File: /home/ubuntu/codegamaai-test/broker_bot/src/utils.py

import json
import os

from llama_index.llms import ChatMessage, MessageRole

def convert_chat_message(chat_history):
    """
    Convert chat history from MongoDB to the format required by the chat engine
    """

    chat = []

    for message in chat_history:
        chat.append(ChatMessage(
        role=get_history_role(message), 
        content=message['content'],
    ))

    return chat

def get_history_role(message):
    """
    Get the role of the message in the chat history
    """

    if message['role'] == 'user':
        return MessageRole.USER
    else:
        return MessageRole.ASSISTANT
    

def create_json(uuid):
     #Create a Json file and store value []
    file_path = os.path.join(os.environ['USER_DATA_DIR'], f"{uuid}.json")

    # Check if file exists
    if os.path.exists(file_path):
        pass
    else:
        data = []
        # Write the empty list to the JSON file
        with open(file_path, 'w') as json_file:
            json.dump(data, json_file)

def update_json(uuid, data):
    # Read existing JSON data from the file
    file_path = os.path.join(os.environ['USER_DATA_DIR'], f"{uuid}.json")
    try:
        with open(file_path, 'r') as json_file:
            existing_data = json.load(json_file)
    except FileNotFoundError:
        # If the file doesn't exist, initialize with an empty list
        existing_data = []

    # Append new values to the existing data
    existing_data.extend(data)

    # If length of existing data is greater than 6, only keep the last 6
    if len(existing_data) > 6:
        existing_data = existing_data[-6:]
    # Write the updated data back to the JSON file
    with open(file_path, 'w') as json_file:
        json.dump(existing_data, json_file)

def read_json(uuid):
    file_path = os.path.join(os.environ['USER_DATA_DIR'], f"{uuid}.json")
    try:
        with open(file_path, 'r') as json_file:
            data = json.load(json_file)
            return data
    except FileNotFoundError:
        print(f"File not found at location: {file_path}")
        return []
    except json.JSONDecodeError:
        print(f"Error decoding JSON from file at location: {file_path}")
        return None

# Read a Config JSON file
def read_json_config(uuid):
    with open (os.path.join(os.environ['USER_CONFIG_DIR'], f"{uuid}.json"), 'r', encoding="UTF-8") as openfile:
        json_object = json.load(openfile)
    return json_object

# Create a Config JSON file
def create_config_json(uuid,data):
    json_object = json.dumps(data, indent = 4) 
    with open(os.path.join(os.environ['USER_CONFIG_DIR'], f"{uuid}.json"), "w") as outfile: 
        outfile.write(json_object)

def save_response(response, uuid, query_text):
    
    histroy = [{'role': 'user', 'content': query_text}, {'role': 'assistant', 'content': str(response)}]

    update_json(uuid,histroy)


def check_string_2_company_string(input_string):
    keywords = ["openai", "open ai","gpt","gpt 3", "gpt 4","gpt 3.5", "gpt-3", "gpt-2", "gpt-4", "gpt-3.5-turbo", "gpt-3.5"]
    for keyword in keywords:
        if keyword in input_string.lower():
            return True 
    return False