File Manager

Editing File: /home/ubuntu/codegamaai-test/efimarket_bot/app.py

from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI,Request
from supertokens_fastapi import get_cors_allowed_headers
import uvicorn
import os
import json
from src.load_test import *
from src.utils import *
from src.training import *
# from src.web_scraping import *
from src.help import *
# from conversation_data import *
# from TTS import *
# from STT import *
from concurrent.futures import ThreadPoolExecutor
import asyncio
# from src.token_count import read_and_count_tokens_in_folder
from src.constants import redis_context_memory_URL
from src.constants import *

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"] + get_cors_allowed_headers(),
)

def process_request(data):
    user_id = data['user_id']
    bot_id = data['bot_id']
    query = data['query']
    context_id = data['context_id']

# bot_name = data['bot_name']
    # prompt = data['prompt']
    # Iscontact_details_availabel = data['contact_status']
    # error_message = data['error_message']
    # general_ai = data['general_ai']
    mail = "abhinavv@abhi.com"
    bot_name = "Pixie The Helper"
    prompt = "You are a Tax Assistent that provides Tax Related solution to user"
    Iscontact_details_availabel = 3
    error_message = "I'm sorry, but I don't have the information needed to answer your question right now. for more specialized support please write a mail to "+mail
    general_ai = 1

if query == "haive-delete-context":
        data = {"id": context_id}
        response = requests.delete(redis_context_memory_URL, json=data)
        if response.status_code == 200:
            return "context Delete request successful."
        else:
            return response.status_code
    try:
        mail = mail
    except:
        mail = "dummy@gmail.com"
    response = load_test(user_id=user_id, bot_id=bot_id, query=query, context_id=context_id, bot_name=bot_name,
                         custom_instruction=prompt, mail=mail, Iscontact_details_availabel=Iscontact_details_availabel,error_message=error_message,general_ai=general_ai)
    return response

async def process_request_async(data):
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor() as executor:
        response = await loop.run_in_executor(None, process_request, data)
        return response

@app.post("/api/v1/query")
async def create_item(request: Request):
    try:
        b_json = await request.body()
        data = json.loads(b_json)
        print(data)
        
        response = await process_request_async(data)
        return {"message": 'success', "response": response, "status_code": 200}
    except Exception as e:
        print(e)
        return {"message": 'query api failed', "status_code": 500}

# @app.post("/api/v1/query")
# async def create_item(request: Request):
#     b_json = await request.body()
#     data = json.loads(b_json)
#     print(data)
    
#     response = await process_request_async(data)
#     return {"message": 'success', "response": response, "status_code": 200}

@app.post("/api/v1/train")
async def create_item2(request: Request):
    # try:
    b_json = await request.body()
    data = json.loads(b_json)
    user_id = data['user_id']
    bot_id = data['bot_id']
    
    user_folder = os.path.join(os.environ['DB_DIR'], user_id)
    user_bot_folder = os.path.join(os.environ['DB_DIR'], user_id, bot_id)
    user_bot_data = os.path.join(os.environ['DB_DIR'], user_id, bot_id, 'data')
    user_bot_qanda = os.path.join(os.environ['DB_DIR'], user_id, bot_id, 'q_a_data')
    user_bot_KB = os.path.join(os.environ['DB_DIR'], user_id, bot_id, 'knowledge_base')

print(data)
    if not os.path.exists(user_folder):
        os.makedirs(user_folder)
    if not os.path.exists(user_bot_folder):
        os.makedirs(user_bot_folder)
    if not os.path.exists(user_bot_data):
        os.makedirs(user_bot_data)
    if not os.path.exists(user_bot_KB):
        os.makedirs(user_bot_KB)
    else:
        shutil.rmtree(user_bot_KB)
        os.makedirs(user_bot_KB)
    if not os.path.exists(user_bot_qanda):
        os.makedirs(user_bot_qanda)
    train_and_save(user_id, bot_id)
    return {"message":"Training started successfully","status_code":200}
    # except:
        # return {"message":'Training API failed',"status_code":500}

@app.post("/api/v1/answer_suggestion")
async def create_item4(request: Request):
    try:
        b_json = await request.body()
        data = json.loads(b_json)
        print(data)
        query = data['query']
        response = answer_suggestion(query)
        return {"message":"success","response":response,"status_code":200}
    except:
        return {"message":'answer_suggestion API failed',"status_code":500}

# if __name__ == "__main__":
#     uvicorn.run(app, host="0.0.0.0",port=5003)
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0",port=5012,ssl_keyfile="privkey.pem",ssl_certfile="fullchain.pem")

# while True:
#     query = input("Enter your query: ")
#     if query == "exit":1
#         break
#     else:
#         user_id = "69"
#         bot_id = "69"
#         query = "What is the New Regime Tax Rates in India"
#         context_id = "U-8-64ccd6e96867eaakaaaak77kgggFgg"
#         model = "gpt-4"
#         bot_name = "Haive"
#         prompt = "You are a Tax Assistent that provides Tax Related solution to user"
#         mail = "abhinavv@abhi.com"
#         Iscontact_details_availabel = 3
#         error_message = "I'm sorry, but I don't have the information needed to answer your question right now. for more specialized support please write a mail to "+mail
#         general_ai = 1

#         response = load_test(user_id=user_id, bot_id=bot_id, query=query, context_id=context_id, bot_name=bot_name,
#                             custom_instruction=prompt, mail=mail, Iscontact_details_availabel=Iscontact_details_availabel,error_message=error_message,general_ai=general_ai)
        
#         print(response)

Back to Directory File Manager