app.py

import gradio as gr
from transformers import pipeline
import requests
import os
import pandas as pd
import time
from dotenv import load_dotenv
from textblob import TextBlob

load_dotenv()
hf_token = os.getenv("HF_TOKEN")

# =========================
# ENV VARIABLES (use HF Secrets)
# =========================
client_id = "sb-cap1-3c4588e0trial-dev!t617058"
client_secret = "acbe78be-ead5-4b12-b3b4-32fdb27d0f5f$hFj-hDXxwHkNHC-CAvv-OKSr3KH96nLL4KqwIg7M8D8="
token_url = "https://3c4588e0trial.authentication.us10.hana.ondemand.com/oauth/token"
cap_service_url_customers = "https://3c4588e0trial-dev-cap1-srv.cfapps.us10-001.hana.ondemand.com/odata/v4/sales/Customers?$top=2"
cap_service_url_products = "https://3c4588e0trial-dev-cap1-srv.cfapps.us10-001.hana.ondemand.com/odata/v4/sales/Products?$top=2"
cap_service_url_saleorders = "https://3c4588e0trial-dev-cap1-srv.cfapps.us10-001.hana.ondemand.com/odata/v4/sales/SalesOrders?$top=2"
cap_service_url_saleorderitems = "https://3c4588e0trial-dev-cap1-srv.cfapps.us10-001.hana.ondemand.com/odata/v4/sales/SalesOrderItems?$top=2"


# =========================
# GLOBAL VARIABLES
# =========================
access_token = None
cached_customers = None
cached_products = None
cached_salesorders = None
cached_salesorderitems = None
last_refresh = 0

# =========================
# LOAD MODEL (once)
# =========================
print("Loading model...")

pipe = pipeline(
    "text-generation",
    # model="Qwen/Qwen2.5-0.5B-Instruct",
    model="Qwen/Qwen2.5-1.5B-Instruct",
    device="cpu",
    torch_dtype="auto"
)

# =========================
# TOKEN FUNCTION
# =========================
def generate_sap_xsuaa_token():
    global access_token

    print("Generating SAP token...")

    auth_response = requests.post(
        token_url,
        data={"grant_type": "client_credentials"},
        auth=(client_id, client_secret)
    )

    if auth_response.status_code != 200:
        print("Token Error:", auth_response.text)
        return None

    access_token = auth_response.json().get("access_token")
    print("Token generated!")
    return access_token

# =========================
# FETCH SAP DATA
# =========================
def fetch_sap_data():
    global access_token

    if not access_token:
        generate_sap_xsuaa_token()

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json"
    }

    res1 = requests.get(cap_service_url_customers, headers=headers)
    res2 = requests.get(cap_service_url_products, headers=headers)
    res3 = requests.get(cap_service_url_saleorders, headers=headers)
    res4 = requests.get(cap_service_url_saleorderitems, headers=headers)

    # Retry if token expired
    if res1.status_code in [400,401,403]:
        print("Token expired. Regenerating...")
        access_token = None
        generate_sap_xsuaa_token()

        headers["Authorization"] = f"Bearer {access_token}"
        res1 = requests.get(cap_service_url_customers, headers=headers)
        res2 = requests.get(cap_service_url_products, headers=headers)
        res3 = requests.get(cap_service_url_saleorders, headers=headers)
        res4 = requests.get(cap_service_url_saleorderitems, headers=headers)

    df_customers = pd.DataFrame(res1.json()["value"])
    df_products = pd.DataFrame(res2.json()["value"])
    df_saleorders = pd.DataFrame(res3.json()["value"])
    df_saleorderitems = pd.DataFrame(res4.json()["value"])

    # Keep only important columns
    df_customers = df_customers[["ID","name","country","industry"]]
    df_products = df_products[["ID","name","category","price","currency"]]
    df_saleorders = df_saleorders[["ID","customer_ID","orderDate","status"]]
    df_saleorderitems = df_saleorderitems[["ID","parent_ID","product_ID","quantity","netAmount"]]

    return df_customers, df_products, df_saleorders, df_saleorderitems

# =========================
# CACHE LOGIC
# =========================

def get_cached_data():
    global cached_customers, cached_products,cached_salesorders,cached_salesorderitems, last_refresh

    # Refresh every 5 minutes
    if time.time() - last_refresh > 3000 or cached_customers is None:
        print("Refreshing SAP data...")
        cached_customers, cached_products,cached_salesorders,cached_salesorderitems = fetch_sap_data()
        last_refresh = time.time()

    return cached_customers, cached_products,cached_salesorders,cached_salesorderitems

# =========================
# MAIN FUNCTION (LLM)
# =========================
# def generate_response(user_prompt):

#     try:
#         # Get cached SAP data
#         df_customers, df_products, df_saleorders, df_saleorderitems = get_cached_data()

#         # Reduce size (IMPORTANT)
#         # customers_text = str(df_customers)[:500]
#         # products_text = str(df_products)[:500]
#         # saleorders_text = str(df_saleorders)[:500]
#         # saleorderitems_text = str(df_saleorderitems)[:500]

#         customers_text = df_customers.to_string(index=False)
#         products_text = df_products.to_string(index=False)
#         saleorders_text = df_saleorders.to_string(index=False)
#         saleorderitems_text = df_saleorderitems.to_string(index=False)

#         # Build prompt

#         system_prompt = f"""
#         You are an intelligent Corporate SAP Assistant bot.
#         Your sole purpose is to answer the user's questions based strictly on the database records provided to you.
#         Customers Data: {customers_text}
#         Products Data: {products_text}
#         Sale orders Data: {saleorders_text}
#         Sale order items Data: {saleorderitems_text}
    
#         CRITICAL RULES:
#         1. NO HALLUCINATIONS: You must base your answer ONLY on the data provided. 
#         2. MISSING DATA: If the provided data does not contain the answer, do not guess. Say: "I could not find that information in the current SAP database."
#         3. FORMATTING: You must output your response in Markdown. Use bold text for important nouns and bullet points for lists to make it easy to read.
#         4. TONE: Be concise, highly professional, and helpful.
#         """
        
#         prompt = f"""
#         {system_prompt}        
#         User: {user_prompt}
#         Assistant:
#         """
       
#         # Generate response
#         result = pipe(
#             prompt,
#             max_new_tokens=100,
#             # temperature=0.3, # Temperature controls randomness
#             do_sample=False, # controls HOW the next word is selected. If False then Always picks most probable next word and No randomness in answer. And if you use True then Picks from multiple possible words
#             # repetition_penalty=1.1,
#             return_full_text=False # If return_full_text=False ensure that the model output contains only the newly generated text.
        
#         )

#         generated_text = result[0]["generated_text"]

#         # Clean output
#         response = generated_text.replace(prompt, "").strip()

#         return response

#     except Exception as e:
#         return f"Error: {str(e)}"



def generate_response(user_prompt):
    try:
        # Get cached SAP data
        df_customers, df_products, df_saleorders, df_saleorderitems = get_cached_data()

        # Reduce size
        customers_text = df_customers.to_string(index=False)
        products_text = df_products.to_string(index=False)
        saleorders_text = df_saleorders.to_string(index=False)
        saleorderitems_text = df_saleorderitems.to_string(index=False)

        # Build system prompt
        system_prompt = f"""
        Your purpose is to answer the user's questions based strictly on the database records provided to you.
        
        Customers Data: {customers_text}
        Products Data: {products_text}
        Sale orders Data: {saleorders_text}
        Sale order items Data: {saleorderitems_text}
        
        CRITICAL RULES:
        1. NO HALLUCINATIONS: You must base your answer ONLY on the data provided. 
        2. MISSING DATA: If the provided data does not contain the answer, do not guess. Say: "I could not find that information in the current SAP database."
        3. FORMATTING: You must output your response in Markdown. Use bold text for important nouns and bullet points for lists to make it easy to read.
        4. TONE: Be concise, highly professional, and helpful.
        """

        user_corrected_prompt = str(TextBlob(user_prompt).correct())
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_corrected_prompt}
        ]

        
        prompt = pipe.tokenizer.apply_chat_template(
            messages, 
            tokenize=False, 
            add_generation_prompt=True
        )

        # Generate response
        result = pipe(
            prompt,
            max_new_tokens=100,
            do_sample=True,
            temperature = 0.3,
            top_k = 3,
            top_p = 0.7,
            # use_cache = True,
            return_full_text=False 
        )
        
        generated_text = result[0]["generated_text"]
        
        # Clean output and strip any leftover end-of-turn tokens
        response = generated_text.replace(prompt, "").replace("<|im_end|>", "").strip()
        
        return response

    except Exception as e:
        return f"Error: {str(e)}"

        
# =========================
# GRADIO UI + API
# =========================
with gr.Blocks() as demo:

    user_input = gr.Textbox(label="User Question")

    output = gr.Textbox(label="Response")

    btn = gr.Button("Generate")

    btn.click(
        fn=generate_response,
        inputs=[user_input],
        outputs=output,
        api_name="predict"
    )

# REQUIRED for API exposure
demo.queue()

demo.launch()