Spaces:
Runtime error
Runtime error
| #-------------------------------------------------------------------------------- | |
| # | |
| # 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) | |