Blame for app.py

#--------------------------------------------------------------------------------
# 
# See Hugging Face from VSCode setup.txt for publishing instructions
#
#--------------------------------------------------------------------------------

import gradio as gr
import os
import random
import time
import numpy as np
import json

qs = {}

theme = gr.themes.Base(
    primary_hue="gray",
    secondary_hue="gray",
).set(
    background_fill_primary_dark='*neutral_950',
    background_fill_secondary_dark='*neutral_950'
)


# Define the function to handle API calls for the chatbot response
def chatbot_api(user_message):
    # Process the input and generate a response
    response = f"Chatbot response to: {user_message}"  # Example response generation
    response = print_log_file()
    return response


import uuid

# Add a function to generate a unique session ID
def generate_session_id():
    return str(uuid.uuid4())

from urllib.parse import parse_qs
def parse_query_params(query_string):
    parsed_params = parse_qs(query_string)
    simplified_params = {k: v[0] if len(v) == 1 else v for k, v in parsed_params.items()}
    return simplified_params

def app_inits():
    global qs
    

    print("DEBUG: App Init")    
    # The path to the QA text file
    filename = 'images/qa_data.txt'

    full_path = os.path.abspath(filename)
    print("Loading responces from ", full_path)

    # Open the file and load the object
    with open(filename, 'r') as file:
        qs = json.load(file)
        
        print("Loaded Q & A")
        #print(qs[0], '\n')

        # Get all keys
        keys = list(qs.keys())
        # Print the keys
        print("Keys:", keys)
        

        # https://bumk766fall2023-chatbot.hf.space/?dw=0.02&dl=0.0001&ds=1
        # print("CALL URL: http://127.0.0.1:7860/?dw=0.02&dl=0.0001")
        # https://bumk766fall2023-chatbot.hf.space/?dw=0.02&dl=0.0001&ds=0.4

        #print("CALL URL: http://127.0.0.1:7860/?dw=0.02&dl=0.0001&ds=0.4")
        #print("CALL URL: http://127.0.0.1:7860/?dw=0.2&dl=0.000001&ds=0.000004")
        #print("CALL URL: http://127.0.0.1:7860/?dw=0.002&dl=0.02&ds=0.000004")
        print("CALL URL: http://127.0.0.1:7860/?dw=0.002&dl=0.02&ds=0.000004&tv=1&dr=1&P=P1")

    return    

with gr.Blocks(theme = theme) as demo:

    #chatbot = gr.Chatbot(value = [[None,'**How can I help you today?**']], elem_id="chatbot", layout = "panel", avatar_images=("images/user.jpg", "images/bot.jpg"),)
    #chatbot = gr.Chatbot(elem_id="chatbot", layout = "panel", avatar_images=("images/user.jpg", "images/bot.jpg"),)
    #chatbot = gr.Chatbot(elem_id="chatbot", layout = "panel", avatar_images=("images/user.jpg", "images/bot.jpg"),render_markdown=False)
    chatbot = gr.Chatbot(elem_id="chatbot", layout = "panel", avatar_images=("images/user.jpg", "images/bot.jpg"),latex_delimiters=[],)


    # --------------------------------------------------------------------------------------------

    def print_log_file():
        """Prints the contents of the log file 'images/log_data.txt'."""
        try:
            with open('images/log_data.txt', 'r') as log_file:
                log_content = log_file.read()
            return log_content
        except FileNotFoundError:
            return "Log file not found."

    # --------------------------------------------------------------------------------------------

    # --------------------------------------------------------------------------------------------

    def delete_log_file():
        """Deletes the log file 'images/log_data.txt' if it exists."""
        try:
            if os.path.exists('images/log_data.txt'):
                os.remove('images/log_data.txt')
                return "Log file deleted successfully."
            else:
                return "Log file does not exist."
        except Exception as e:
            return f"Error deleting log file: {e}"

    # --------------------------------------------------------------------------------------------



    # height=800,  

    #def get_params(request: gr.Request, param_hiddentextbox):
    def get_params(request: gr.Request):        

        #print("DEBUG: get_params")
        
        headers = request.headers
        host = request.client.host
        user_agent = request.headers["user-agent"]
        params = request.query_params

        session_id = generate_session_id()  # Generate a unique session ID
        return str(params), session_id

        #return {
        #    "ip": host,
        #    "user_agent": user_agent,
        #    "headers": headers,
        #    "URL_params": params,
        #}

    with gr.Row(equal_height=True):
        #msg = gr.Textbox(show_label=False, placeholder="Message ChatGPT...", max_lines = 5,container = False,)
        #msg = gr.Textbox(value="Pre-filled value", show_label=False, placeholder="Message ChatGPT...", max_lines=5, container=False, interactive=False)
        # USE FOR NOT EDITABLE VERSION
        
        #print("DEBUG: Loading preset_prompt")
        #print(gr.__version__)

        preset_prompt = ""
        msg = gr.Textbox(value=preset_prompt, show_label=False, placeholder=preset_prompt, max_lines = 5,container = False, interactive=False)



        btn = gr.Button(value="", min_width=80, size = "lg", icon="images/button.jpg", scale= 0)
        #param_hiddentextbox = gr.Textbox(show_label=False, visible=False, container = False,)
        url_params = gr.State()
        session_id = gr.State()



    #demo.load(get_params, None, url_params, queue=False)
    demo.load(get_params, None, [url_params, session_id], queue=False)
    
    # Add invisible API endpoint using these components
    invisible_input = gr.Textbox(visible=False)
    invisible_output = gr.Textbox(visible=False)
    demo.load(chatbot_api, inputs=invisible_input, outputs=invisible_output, api_name="chatbot_endpoint")


    def set_text_in_textbox(request: gr.Request):
    # Return an updated value for the Textbox
        #return gr.Textbox.update(value="Hello from page load!")
        print("DEBUG: set_text_in_textbox")    

        local_params = str(request.query_params)
       

        if local_params != "":

            local_params_parsed = parse_query_params(local_params)
            print(local_params_parsed)

            Prompt_Index = local_params_parsed.get("P", "PX")
            
            print(Prompt_Index)

        else:    
            return ("No prompt parameter specified")
        

        preset_prompt = qs.get(Prompt_Index, "Prompt Index Not Found")
        return (preset_prompt)

    demo.load(fn=set_text_in_textbox, inputs=None, outputs=msg)

    #demo.load(get_params, param_hiddentextbox, param_hiddentextbox, queue=False)

    # --------------------------------------------------------------------------------------------
    def user(user_message, history):
        print("User prompted ", user_message)
        return "", history + [[user_message, None]]
    # --------------------------------------------------------------------------------------------


    # --------------------------------------------------------------------------------------------
    def bot( history, url_params, session_id):
    
        global qs
        
        DelayBetweenWords = 0.7
        DelayBetweenLetters = 0.0001
        DelayBetweenSentences = .6
        
        TextVariant = 1
        DelayRandomizer = 1
        
        QID15 = 0
        QID16 = 0
        
        # Go fast for debugging
        DelayBetweenWords = 0.0001
        DelayBetweenLetters = 0.0001
        DelayBetweenSentences = .0001

       

        #-- Maintenance
        #
        # Check if the input prompt is "show log" or "delete log"
        text = history[-1][0].lower().strip()
        if text == "show log":
            # Display log content if prompt matches
            log_content = print_log_file()
            history[-1][1] = log_content
            yield history
            return 
        elif text == "delete log":
            # Delete log file if prompt matches
            delete_message = delete_log_file()
            history[-1][1] = delete_message
            yield history
            return 
        #
        #-----------------------------

        #time.sleep(.02)
        print("----------  url_params ------------ ")
        print(url_params)

        
        if url_params != "":
            params = parse_query_params(url_params)
            DelayBetweenWords = float(params.get("dw", DelayBetweenWords))
            DelayBetweenLetters = float(params.get("dl", DelayBetweenLetters))
            DelayBetweenSentences = float(params.get("ds", DelayBetweenSentences))

            TextVariant =  int(params.get("tv", TextVariant))
            DelayRandomizer = int(params.get("dr", DelayRandomizer))
            
            QID15 = int(params.get("QID15", QID15))
            QID16 = int(params.get("QID16", QID16))
        else:    
            print("No parameters provided. Using defaults")
        
        print("dv=", DelayBetweenWords, "dl=", DelayBetweenLetters, "ds=", DelayBetweenSentences, "tv=", TextVariant, "dr=", DelayRandomizer, "QID15=", QID15, "QID16=", QID16 )

        # Get all keys
        keywords =  list(qs.keys())
        # Print the keys
        print("Keys:", keywords )

        # String typed by user
        text = history[-1][0]
        print("User Input:", text)

        bot_message_sorry = qs.get("Sorry", "Sorry, I cannot answer that...")


        if text != "":
           search_keyword = find_best_keyword_match(keywords, text)     
        else:   
           search_keyword = "None"

        log_message("-ENTRY- [query]: " + text + " [params]: " + url_params + " [keyword]: " + search_keyword, session_id)
        
        if search_keyword == "None":
           #bot_message = "Sorry, don't know any restaurants at this location..." + " [DEBUG (params: " + url_params + ")]" 
           #bot_message = "I'm sorry, but I can only provide information about Mexican restaurants in Tucson at the moment."
           bot_message = bot_message_sorry
        else:
           output_text = qs.get(search_keyword, "Keyword identified. No information found.")
           bot_message = output_text

        if text == "debug":
           bot_message = "[DEBUG (ver. 2/4/2025) params: " + url_params + "\nKeys:\n" + "\n".join(keywords) + "]"


        # initialize  new entry
        history[-1][1] = ""
        
        # output text letter-by-letter
        for character in bot_message:
            
            if character == "^":
                history[-1][1] += " "
            elif character == "@":
                history[-1][1] += "<BR>"

            elif character != "=":
                history[-1][1] += character

            # Delay between words
            if character == " ":
                time.sleep(DelayBetweenWords)
            # Overall print speed
            #elif character == "." or character == ",":
            elif character == "=":
                time.sleep(DelayBetweenSentences)
            else: 
                time.sleep(DelayBetweenLetters)

            yield history 
    
        log_message("-EXIT - [query]: " + text + " [params]: " + url_params + " [keyword]: " + search_keyword, session_id)
       
    # --------------------------------------------------------------------------------------------
 


# Events handling

#    btn.click(user, [msg, chatbot], [msg, chatbot], queue=False, show_progress=False).then(
#        bot, [chatbot, url_params, session_id], chatbot, concurrency_limit=50, api_name="chatbot_api"
#    )

    def hide_button():
    # Hide the button entirely
        return gr.Button(visible=bool(0))

    btn.click(user, [msg, chatbot], [msg, chatbot], queue=False, show_progress=False).then(
    hide_button,   # hide_button
    inputs=None,
    outputs=btn    # Apply the updates to the same button
    ).then(
        lambda: "", None, msg
    ).then(
        bot, [chatbot, url_params, session_id], chatbot, concurrency_limit=50
    )



    msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False, show_progress=False).then(
        bot, [chatbot, url_params, session_id], chatbot, concurrency_limit=50
    )


    # btn.click(user, [msg, chatbot], [msg, chatbot], queue=False, show_progress=False).then(
    #     bot, [chatbot, url_params], chatbot, concurrency_limit=50
    # )
    
    # msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False, show_progress=False).then(
    #     bot, [chatbot, url_params], chatbot, concurrency_limit=50
    # )
    

# --------------------------------------------------------------------------------------------

def normalize(s):
    """Lowercase and strip the string to normalize it."""
    return s.lower().strip()

def levenshtein_distance(s1, s2):
    """Calculate the Levenshtein distance between two strings."""
    if len(s1) < len(s2):
        return levenshtein_distance(s2, s1)

    if len(s2) == 0:
        return len(s1)

    previous_row = range(len(s2) + 1)
    for i, c1 in enumerate(s1):
        current_row = [i + 1]
        for j, c2 in enumerate(s2):
            insertions = previous_row[j + 1] + 1
            deletions = current_row[j] + 1
            substitutions = previous_row[j] + (c1 != c2)
            current_row.append(min(insertions, deletions, substitutions))
        previous_row = current_row
    
    return previous_row[-1]


def find_best_keyword_match(keywords, text, max_distance=3):
    """Find the best keyword match in the text, allowing for some misspelling."""
    text_normalized = normalize(text)
    best_match = None
    lowest_distance = float('inf')
    
    for keyword in keywords:
        keyword_normalized = normalize(keyword)
        # Calculate the Levenshtein distance for composite keywords by comparing against slices of the text
        if ' ' in keyword_normalized:
            slice_length = len(keyword_normalized)
            for i in range(len(text_normalized) - slice_length + 1):
                text_slice = text_normalized[i:i+slice_length]
                distance = levenshtein_distance(text_slice, keyword_normalized)
                if distance < lowest_distance:
                    best_match = keyword
                    lowest_distance = distance
        else:
            # For single words, compare against each word in the text
            for word in text_normalized.split():
                distance = levenshtein_distance(word, keyword_normalized)
                if distance < lowest_distance:
                    best_match = keyword
                    lowest_distance = distance

    if lowest_distance <= max_distance:
        return best_match
    return "None"


from datetime import datetime
import threading

# Initialize a global threading lock
log_lock = threading.Lock()

# The path to the log file
filename = 'images/log_data.txt'

def log_message(param, session_id):
    # Get current timestamp
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    # Define the log file name based on session ID
    #filename = f'{session_id}_log_data.txt'
    session_id_text = f'{session_id}'

    # Prepare log message
    log_msg = f'{session_id_text}: {timestamp}: : {param}\n'

    # Open the log file, append the message, and ensure the file is closed properly
    # Use the lock to ensure that only one thread writes to the file at a time
    with log_lock:
        with open(filename, 'a') as log_file:
            log_file.write(log_msg)

    # Get the absolute path of the log file
    abs_path = os.path.abspath(filename)
    
    # Print the absolute path for debugging purposes
    print(f'Writing to Log file: {abs_path}')


# --------------------------------------------------------------------------------------------
# Launching
# --------------------------------------------------------------------------------------------

app_inits()    
demo.queue()
demo.launch(allowed_paths=["."], max_threads=40, show_error=True)