#-------------------------------------------------------------------------------- # # 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] += "
" 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)