Update app.py

app.py

@@ -1,77 +1,295 @@
+# import os
+# import time
+# from fastapi import FastAPI,Request
+# from fastapi.responses import HTMLResponse
+# from fastapi.staticfiles import StaticFiles
+# from llama_index.core import StorageContext, load_index_from_storage, VectorStoreIndex, SimpleDirectoryReader, ChatPromptTemplate, Settings
+# from llama_index.embeddings.huggingface import HuggingFaceEmbedding
+# from pydantic import BaseModel
+# from fastapi.responses import JSONResponse
+# import uuid  # for generating unique IDs
+# import datetime
+# from fastapi.middleware.cors import CORSMiddleware
+# from fastapi.templating import Jinja2Templates
+# from huggingface_hub import InferenceClient
+# import json
+# import re
+# from gradio_client import Client
+# from simple_salesforce import Salesforce, SalesforceLogin
+# from llama_index.llms.huggingface import HuggingFaceLLM
+# # from llama_index.llms.huggingface import HuggingFaceInferenceAPI
+
+
+# # Define Pydantic model for incoming request body
+# class MessageRequest(BaseModel):
+#     message: str
+# repo_id = "meta-llama/Meta-Llama-3-8B-Instruct"
+# llm_client = InferenceClient(
+#     model=repo_id,
+#     token=os.getenv("HF_TOKEN"),
+# )
+
+
+# os.environ["HF_TOKEN"] = os.getenv("HF_TOKEN")
+# username = os.getenv("username")
+# password = os.getenv("password")
+# security_token = os.getenv("security_token")
+# domain =  os.getenv("domain")# Using sandbox environment
+# session_id, sf_instance = SalesforceLogin(username=username, password=password, security_token=security_token, domain=domain)
+
+#     # Create Salesforce object
+# sf = Salesforce(instance=sf_instance, session_id=session_id)
+
+# app = FastAPI()
+
+
+# @app.middleware("http")
+# async def add_security_headers(request: Request, call_next):
+#     response = await call_next(request)
+#     response.headers["Content-Security-Policy"] = "frame-ancestors *; frame-src *; object-src *;"
+#     response.headers["X-Frame-Options"] = "ALLOWALL"
+#     return response
+
+
+# # Allow CORS requests from any domain
+# app.add_middleware(
+#     CORSMiddleware,
+#     allow_origins=["*"],
+#     allow_credentials=True,
+#     allow_methods=["*"],
+#     allow_headers=["*"],
+# )
+
+
+
+
+# @app.get("/favicon.ico")
+# async def favicon():
+#     return HTMLResponse("")  # or serve a real favicon if you have one
+
+
+# app.mount("/static", StaticFiles(directory="static"), name="static")
+
+# templates = Jinja2Templates(directory="static")
+# # Configure Llama index settings
+# Settings.llm = HuggingFaceLLM(
+#     model_name="meta-llama/Meta-Llama-3-8B-Instruct",
+#     tokenizer_name="meta-llama/Meta-Llama-3-8B-Instruct",
+#     context_window=3000,
+#     token=os.getenv("HF_TOKEN"),
+#     max_new_tokens=512,
+#     generate_kwargs={"temperature": 0.1},
+# )
+
+# Settings.embed_model = HuggingFaceEmbedding(
+#     model_name="BAAI/bge-small-en-v1.5"
+# )
+
+# PERSIST_DIR = "db"
+# PDF_DIRECTORY = 'data'
+
+# # Ensure directories exist
+# os.makedirs(PDF_DIRECTORY, exist_ok=True)
+# os.makedirs(PERSIST_DIR, exist_ok=True)
+# chat_history = []
+# current_chat_history = []
+# def data_ingestion_from_directory():
+#     documents = SimpleDirectoryReader(PDF_DIRECTORY).load_data()
+#     storage_context = StorageContext.from_defaults()
+#     index = VectorStoreIndex.from_documents(documents)
+#     index.storage_context.persist(persist_dir=PERSIST_DIR)
+
+# def initialize():
+#     start_time = time.time()
+#     data_ingestion_from_directory()  # Process PDF ingestion at startup
+#     print(f"Data ingestion time: {time.time() - start_time} seconds")
+# def split_name(full_name):
+#     # Split the name by spaces
+#     words = full_name.strip().split()
+    
+#     # Logic for determining first name and last name
+#     if len(words) == 1:
+#         first_name = ''
+#         last_name = words[0]
+#     elif len(words) == 2:
+#         first_name = words[0]
+#         last_name = words[1]
+#     else:
+#         first_name = words[0]
+#         last_name = ' '.join(words[1:])
+    
+#     return first_name, last_name
+
+# initialize()  # Run initialization tasks
+
+
+# def handle_query(query):
+#     chat_text_qa_msgs = [
+#         (
+#             "user",
+#             """
+#             You are the Clara Redfernstech chatbot. Your goal is to provide accurate, professional, and helpful answers to user queries based on the company's data. Always ensure your responses are clear and concise. Give response within 10-15 words only       
+#             {context_str}
+#             Question:
+#             {query_str}
+#             """
+#         )
+#     ]
+#     text_qa_template = ChatPromptTemplate.from_messages(chat_text_qa_msgs)
+    
+#     storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
+#     index = load_index_from_storage(storage_context)
+#     context_str = ""
+#     for past_query, response in reversed(current_chat_history):
+#         if past_query.strip():
+#             context_str += f"User asked: '{past_query}'\nBot answered: '{response}'\n"
+
+    
+#     query_engine = index.as_query_engine(text_qa_template=text_qa_template, context_str=context_str)
+#     answer = query_engine.query(query)
+
+#     if hasattr(answer, 'response'):
+#         response=answer.response
+#     elif isinstance(answer, dict) and 'response' in answer:
+#         response =answer['response']
+#     else:
+#         response ="Sorry, I couldn't find an answer."
+#     current_chat_history.append((query, response))
+#     return response
+# @app.get("/ch/{id}", response_class=HTMLResponse)
+# async def load_chat(request: Request, id: str):
+#     return templates.TemplateResponse("index.html", {"request": request, "user_id": id})
+# # Route to save chat history
+# @app.post("/hist/")
+# async def save_chat_history(history: dict):
+#     # Check if 'userId' is present in the incoming dictionary
+#     user_id = history.get('userId')
+#     print(user_id)
+
+#     # Ensure user_id is defined before proceeding
+#     if user_id is None:
+#         return {"error": "userId is required"}, 400
+
+#     # Construct the chat history string
+#     hist = ''.join([f"'{entry['sender']}: {entry['message']}'\n" for entry in history['history']])
+#     hist = "You are a Redfernstech summarize model. Your aim is to use this conversation to identify user interests solely based on that conversation: " + hist
+#     print(hist)
+
+#     # Get the summarized result from the client model
+#     result = hist
+
+#     try:
+#         sf.Lead.update(user_id, {'Description': result})
+#     except Exception as e:
+#         return {"error": f"Failed to update lead: {str(e)}"}, 500
+    
+#     return {"summary": result, "message": "Chat history saved"}
+# @app.post("/webhook")
+# async def receive_form_data(request: Request):
+#     form_data = await request.json()
+#     # Log in to Salesforce
+#     first_name, last_name = split_name(form_data['name'])
+#     data = {
+#     'FirstName': first_name,
+#     'LastName': last_name,
+#     'Description': 'hii',  # Static description
+#     'Company': form_data['company'],  # Assuming company is available in form_data
+#     'Phone': form_data['phone'].strip(),  # Phone from form data
+#     'Email': form_data['email'],  # Email from form data
+#     }
+#     a=sf.Lead.create(data)
+#     # Generate a unique ID (for tracking user)
+#     unique_id = a['id']
+    
+#     # Here you can do something with form_data like saving it to a database
+#     print("Received form data:", form_data)
+    
+#     # Send back the unique id to the frontend
+#     return JSONResponse({"id": unique_id})
+
+# @app.post("/chat/")
+# async def chat(request: MessageRequest):
+#     message = request.message  # Access the message from the request body
+#     response = handle_query(message)  # Process the message
+#     message_data = {
+#         "sender": "User",
+#         "message": message,
+#         "response": response,
+#         "timestamp": datetime.datetime.now().isoformat()
+#     }
+#     chat_history.append(message_data)
+#     return {"response": response}
+# @app.get("/")
+# def read_root():
+#     return {"message": "Welcome to the API"}
+
 import os
 import time
-from fastapi import FastAPI,Request
-from fastapi.responses import HTMLResponse
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import HTMLResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 from llama_index.core import StorageContext, load_index_from_storage, VectorStoreIndex, SimpleDirectoryReader, ChatPromptTemplate, Settings
 from llama_index.embeddings.huggingface import HuggingFaceEmbedding
 from pydantic import BaseModel
-from fastapi.responses import JSONResponse
-import uuid  # for generating unique IDs
-import datetime
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.templating import Jinja2Templates
 from huggingface_hub import InferenceClient
 import json
-import re
-from gradio_client import Client
+import datetime
 from simple_salesforce import Salesforce, SalesforceLogin
 from llama_index.llms.huggingface import HuggingFaceLLM
-# from llama_index.llms.huggingface import HuggingFaceInferenceAPI
-
 
-# Define Pydantic model for incoming request body
+# Pydantic model for request body
 class MessageRequest(BaseModel):
     message: str
-repo_id = "meta-llama/Meta-Llama-3-8B-Instruct"
-llm_client = InferenceClient(
-    model=repo_id,
-    token=os.getenv("HF_TOKEN"),
-)
-
 
-os.environ["HF_TOKEN"] = os.getenv("HF_TOKEN")
-username = os.getenv("username")
-password = os.getenv("password")
-security_token = os.getenv("security_token")
-domain =  os.getenv("domain")# Using sandbox environment
-session_id, sf_instance = SalesforceLogin(username=username, password=password, security_token=security_token, domain=domain)
+# Validate environment variables
+required_env_vars = ["HF_TOKEN", "username", "password", "security_token", "domain"]
+for var in required_env_vars:
+    if not os.getenv(var):
+        raise EnvironmentError(f"Missing required environment variable: {var}")
 
-    # Create Salesforce object
-sf = Salesforce(instance=sf_instance, session_id=session_id)
+# Salesforce configuration
+try:
+    session_id, sf_instance = SalesforceLogin(
+        username=os.getenv("username"),
+        password=os.getenv("password"),
+        security_token=os.getenv("security_token"),
+        domain=os.getenv("domain")
+    )
+    sf = Salesforce(instance=sf_instance, session_id=session_id)
+except Exception as e:
+    raise Exception(f"Failed to initialize Salesforce: {str(e)}")
 
+# FastAPI setup
 app = FastAPI()
 
-
+# Security headers middleware
 @app.middleware("http")
 async def add_security_headers(request: Request, call_next):
     response = await call_next(request)
-    response.headers["Content-Security-Policy"] = "frame-ancestors *; frame-src *; object-src *;"
-    response.headers["X-Frame-Options"] = "ALLOWALL"
+    response.headers.update({
+        "Content-Security-Policy": "default-src 'self'; frame-ancestors 'self';",
+        "X-Frame-Options": "DENY",
+        "X-Content-Type-Options": "nosniff",
+        "Referrer-Policy": "strict-origin-when-cross-origin"
+    })
     return response
 
-
-# Allow CORS requests from any domain
+# CORS configuration
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],
+    allow_origins=["*"],  # Consider specifying allowed origins in production
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-
-
-
-@app.get("/favicon.ico")
-async def favicon():
-    return HTMLResponse("")  # or serve a real favicon if you have one
-
-
+# Static files and templates
 app.mount("/static", StaticFiles(directory="static"), name="static")
-
 templates = Jinja2Templates(directory="static")
-# Configure Llama index settings
+
+# LlamaIndex configuration
 Settings.llm = HuggingFaceLLM(
     model_name="meta-llama/Meta-Llama-3-8B-Instruct",
     tokenizer_name="meta-llama/Meta-Llama-3-8B-Instruct",
@@ -85,140 +303,132 @@ Settings.embed_model = HuggingFaceEmbedding(
     model_name="BAAI/bge-small-en-v1.5"
 )
 
+# Directory constants
 PERSIST_DIR = "db"
-PDF_DIRECTORY = 'data'
+PDF_DIRECTORY = "data"
 
-# Ensure directories exist
+# Initialize directories
 os.makedirs(PDF_DIRECTORY, exist_ok=True)
 os.makedirs(PERSIST_DIR, exist_ok=True)
+
+# Chat history storage
 chat_history = []
 current_chat_history = []
+
 def data_ingestion_from_directory():
-    documents = SimpleDirectoryReader(PDF_DIRECTORY).load_data()
-    storage_context = StorageContext.from_defaults()
-    index = VectorStoreIndex.from_documents(documents)
-    index.storage_context.persist(persist_dir=PERSIST_DIR)
+    try:
+        documents = SimpleDirectoryReader(PDF_DIRECTORY).load_data()
+        storage_context = StorageContext.from_defaults()
+        index = VectorStoreIndex.from_documents(documents)
+        index.storage_context.persist(persist_dir=PERSIST_DIR)
+    except Exception as e:
+        raise Exception(f"Data ingestion failed: {str(e)}")
 
 def initialize():
     start_time = time.time()
-    data_ingestion_from_directory()  # Process PDF ingestion at startup
-    print(f"Data ingestion time: {time.time() - start_time} seconds")
-def split_name(full_name):
-    # Split the name by spaces
+    data_ingestion_from_directory()
+    print(f"Data ingestion completed in {time.time() - start_time:.2f} seconds")
+
+def split_name(full_name: str) -> tuple:
     words = full_name.strip().split()
-    
-    # Logic for determining first name and last name
     if len(words) == 1:
-        first_name = ''
-        last_name = words[0]
+        return "", words[0]
     elif len(words) == 2:
-        first_name = words[0]
-        last_name = words[1]
-    else:
-        first_name = words[0]
-        last_name = ' '.join(words[1:])
-    
-    return first_name, last_name
+        return words[0], words[1]
+    return words[0], " ".join(words[1:])
 
-initialize()  # Run initialization tasks
+# Run initialization
+initialize()
 
-
-def handle_query(query):
+def handle_query(query: str) -> str:
     chat_text_qa_msgs = [
         (
             "user",
             """
-            You are the Clara Redfernstech chatbot. Your goal is to provide accurate, professional, and helpful answers to user queries based on the company's data. Always ensure your responses are clear and concise. Give response within 10-15 words only       
+            You are the Clara Redfernstech chatbot. Provide accurate, professional answers in 10-15 words.
             {context_str}
-            Question:
-            {query_str}
+            Question: {query_str}
             """
         )
     ]
     text_qa_template = ChatPromptTemplate.from_messages(chat_text_qa_msgs)
     
-    storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
-    index = load_index_from_storage(storage_context)
-    context_str = ""
-    for past_query, response in reversed(current_chat_history):
-        if past_query.strip():
-            context_str += f"User asked: '{past_query}'\nBot answered: '{response}'\n"
+    try:
+        storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
+        index = load_index_from_storage(storage_context)
+        
+        context_str = ""
+        for past_query, response in reversed(current_chat_history[-5:]):  # Limit context to last 5 interactions
+            if past_query.strip():
+                context_str += f"User: '{past_query}'\nBot: '{response}'\n"
+        
+        query_engine = index.as_query_engine(text_qa_template=text_qa_template)
+        answer = query_engine.query(query)
+        
+        response = getattr(answer, 'response', answer.get('response', "Sorry, I couldn't find an answer."))
+        current_chat_history.append((query, response))
+        return response
+    except Exception as e:
+        return f"Error processing query: {str(e)}"
+
+@app.get("/favicon.ico")
+async def favicon():
+    return HTMLResponse(status_code=204)
 
-    
-    query_engine = index.as_query_engine(text_qa_template=text_qa_template, context_str=context_str)
-    answer = query_engine.query(query)
-
-    if hasattr(answer, 'response'):
-        response=answer.response
-    elif isinstance(answer, dict) and 'response' in answer:
-        response =answer['response']
-    else:
-        response ="Sorry, I couldn't find an answer."
-    current_chat_history.append((query, response))
-    return response
 @app.get("/ch/{id}", response_class=HTMLResponse)
 async def load_chat(request: Request, id: str):
     return templates.TemplateResponse("index.html", {"request": request, "user_id": id})
-# Route to save chat history
+
 @app.post("/hist/")
 async def save_chat_history(history: dict):
-    # Check if 'userId' is present in the incoming dictionary
     user_id = history.get('userId')
-    print(user_id)
-
-    # Ensure user_id is defined before proceeding
-    if user_id is None:
-        return {"error": "userId is required"}, 400
-
-    # Construct the chat history string
-    hist = ''.join([f"'{entry['sender']}: {entry['message']}'\n" for entry in history['history']])
-    hist = "You are a Redfernstech summarize model. Your aim is to use this conversation to identify user interests solely based on that conversation: " + hist
-    print(hist)
-
-    # Get the summarized result from the client model
-    result = hist
+    if not user_id:
+        raise HTTPException(status_code=400, detail="userId is required")
 
     try:
-        sf.Lead.update(user_id, {'Description': result})
+        hist = ''.join([f"'{entry['sender']}: {entry['message']}'\n" for entry in history['history']])
+        summary = f"Conversation summary for user interest analysis:\n{hist}"
+        
+        sf.Lead.update(user_id, {'Description': summary})
+        return {"summary": summary, "message": "Chat history saved"}
     except Exception as e:
-        return {"error": f"Failed to update lead: {str(e)}"}, 500
-    
-    return {"summary": result, "message": "Chat history saved"}
+        raise HTTPException(status_code=500, detail=f"Failed to update lead: {str(e)}")
+
 @app.post("/webhook")
 async def receive_form_data(request: Request):
-    form_data = await request.json()
-    # Log in to Salesforce
-    first_name, last_name = split_name(form_data['name'])
-    data = {
-    'FirstName': first_name,
-    'LastName': last_name,
-    'Description': 'hii',  # Static description
-    'Company': form_data['company'],  # Assuming company is available in form_data
-    'Phone': form_data['phone'].strip(),  # Phone from form data
-    'Email': form_data['email'],  # Email from form data
-    }
-    a=sf.Lead.create(data)
-    # Generate a unique ID (for tracking user)
-    unique_id = a['id']
-    
-    # Here you can do something with form_data like saving it to a database
-    print("Received form data:", form_data)
-    
-    # Send back the unique id to the frontend
-    return JSONResponse({"id": unique_id})
+    try:
+        form_data = await request.json()
+        first_name, last_name = split_name(form_data.get('name', ''))
+        
+        data = {
+            'FirstName': first_name,
+            'LastName': last_name,
+            'Description': 'New lead from webhook',
+            'Company': form_data.get('company', 'Unknown'),
+            'Phone': form_data.get('phone', '').strip(),
+            'Email': form_data.get('email', ''),
+        }
+        
+        result = sf.Lead.create(data)
+        return JSONResponse({"id": result['id']})
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to process webhook: {str(e)}")
 
 @app.post("/chat/")
 async def chat(request: MessageRequest):
-    message = request.message  # Access the message from the request body
-    response = handle_query(message)  # Process the message
-    message_data = {
-        "sender": "User",
-        "message": message,
-        "response": response,
-        "timestamp": datetime.datetime.now().isoformat()
-    }
-    chat_history.append(message_data)
-    return {"response": response}
+    try:
+        response = handle_query(request.message)
+        message_data = {
+            "sender": "User",
+            "message": request.message,
+            "response": response,
+            "timestamp": datetime.datetime.now().isoformat()
+        }
+        chat_history.append(message_data)
+        return {"response": response}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Chat processing failed: {str(e)}")
+
 @app.get("/")
 def read_root():
-    return {"message": "Welcome to the API"}
+    return {"message": "Welcome to the Redfernstech API"}