# ============================================================================== # 1. IMPORTS & SETUP # ============================================================================== import os import re import difflib import pandas as pd from datetime import datetime import gradio as gr from transformers import pipeline, set_seed ,BitsAndBytesConfig # --- Model Loading --- # Read the Hugging Face token from the environment secrets hf_token = os.environ.get("HUGGING_FACE_HUB_TOKEN") # Check if the token is available if not hf_token: print("Warning: HUGGING_FACE_HUB_TOKEN not found. This may fail if the model is private.") print("Loading conversational model...") # Use the token to authenticate when loading the model bnb_config = BitsAndBytesConfig( load_in_4bit=True, # Set to True for 4-bit quantization bnb_4bit_quant_type="nf4", # "nf4" is usually the most stable bnb_4bit_use_double_quant=True, # Double quantization helps reduce memory bnb_4bit_compute_dtype="float16" # Safer than float32, faster on GPUs ) # Use the token to authenticate when loading the model generator = pipeline( 'text-generation', model='Warrior786/acc-llama-1b-merged-v10-lite', token=hf_token, device_map="auto", # Let HF infer best device usage model_kwargs={"quantization_config": bnb_config} ) set_seed(42) print("Model loaded successfully.") # --- File Paths --- # Log conversations to a local CSV file within the Hugging Face Space LOG_FILE = "ACC_chat_history.csv" # ============================================================================== # 2. CONSTANTS (Allowed Links & Emails) # ============================================================================== # List of all approved URLs for the chatbot to share ALLOWED_LINKS = [ "https://acceducate.org/", "https://acceducate.org/apply/", "https://acceducate.org/faqs/", "https://acceducate.org/programs/a-continuous-charity-acc-student-loan-refinance-program/", "https://acceducate.org/programs/", "https://acceducate.org/programs/help-a-student-with-sadaqah-jariyah/", "https://acceducate.org/programs/zakat/", "https://acceducate.org/programs/change-maker/", "https://acceducate.org/programs/break-the-chains-of-riba/", "https://acceducate.org/programs/women-empowerment/", "https://acceducate.org/programs/educational-justice/", "https://acceducate.org/ways-to-give/", "https://acceducate.org/ways-to-give/donate-stocks/", "https://acceducate.org/ways-to-give/open-an-educational-partnership-fund/", "https://acceducate.org/ways-to-give/legacy-fund/", "https://acceducate.org/ways-to-give/monthly-giving/", "https://acceducate.org/ways-to-give/support-accs-waqf-endowment-a-legacy-of-sustainable-charity/", "https://acceducate.org/ways-to-give/donate-by-zelle/", "https://acceducate.org/ways-to-give/matchinggift/", "https://acceducate.org/ways-to-give/donate-your-ira-to-acc/", "https://acceducate.org/ways-to-give/corporate-sponsored-volunteering/", "https://acceducate.org/ways-to-give/buy-coffee-give-sadaqah/", "https://acceducate.org/ways-to-give/host-an-iftar/", "https://acceducate.org/campaigns/grants/", "https://acceducate.org/student-initiated-fundraising-initiative/", "https://acceducate.org/about-us/", "https://acceducate.org/our-team/", "https://acceducate.org/financials/", "https://acceducate.org/careers/", "https://acceducate.org/why-acc/", "https://acceducate.org/invite-us/", "https://acceducate.org/events/", "https://acceducate.org/testimonials/", "https://www.facebook.com/acontinuouscharity/", "https://www.instagram.com/acontinuouscharity/", "https://www.youtube.com/channel/UCkq8wAvAqt54yzjzL_QIq3w", "https://twitter.com/accnational", ] # Mapping of keywords to specific ACC pages for better matching ACC_LINKS = { "About Us": "https://acceducate.org/about-us/", "Apply": "https://acceducate.org/apply/", "Careers": "https://acceducate.org/careers/", "Donate": "https://acceducate.org/ways-to-give/", "Events": "https://acceducate.org/events/", "FAQs": "https://acceducate.org/faqs/", "Zakat": "https://acceducate.org/programs/zakat/", "Loan Refinance": "https://acceducate.org/programs/a-continuous-charity-acc-student-loan-refinance-program/", } # Allowed email addresses ALLOWED_EMAILS = ["donate@acceducate.org"] # ============================================================================== # 3. HELPER FUNCTIONS # ============================================================================== def normalize_acc_link(url: str, user_msg: str = "") -> str: """Cleans and restricts a given URL to the official list of allowed ACC links.""" url = re.sub(r"https?://[^/\s]*acc[^/\s]*\.org", "https://acceducate.org", url) url = re.sub(r"(/(\w+))(\s*/\2)+", r"\1", url) if not url.endswith("/"): url += "/" if url in ALLOWED_LINKS: return url for keyword, link in ACC_LINKS.items(): if keyword.lower() in user_msg.lower(): return link closest = difflib.get_close_matches(url, ALLOWED_LINKS, n=1, cutoff=0.6) return closest[0] if closest else "https://acceducate.org/" def sanitize_response_links(text: str, user_msg: str = "") -> str: """Finds and normalizes all URLs in the bot's response.""" text = re.sub(r"https?://\S+", lambda m: normalize_acc_link(m.group(0), user_msg), text) def replace_md(match): display, url = match.groups() final_url = normalize_acc_link(url, user_msg) norm_display = re.sub(r"^https?://", "", final_url).rstrip("/") return f"[{norm_display}]({final_url})" text = re.sub(r"\[([^\]]+)\]\((https?://[^\)]+)\)", replace_md, text) return text def is_valid_email(email: str) -> bool: """Checks if a string is a validly formatted email address.""" pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" return re.match(pattern, email) is not None def log_to_csv(email, user_msg, bot_msg): """Saves the conversation turn to a CSV file.""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") new_row = pd.DataFrame([[timestamp, email, user_msg, bot_msg]], columns=["Time", "Email", "User", "Bot"]) try: header = not os.path.exists(LOG_FILE) new_row.to_csv(LOG_FILE, mode='a', header=header, index=False) except Exception as e: print(f"Error logging to CSV: {e}") # ============================================================================== # 4. CORE CHATBOT LOGIC # ============================================================================== def chat_fn(message: str, history: list, user_email_state: str): """Main function to handle the chatbot's logic and conversation flow.""" # --- Step 1: Handle initial state and email collection --- if user_email_state is None: if is_valid_email(message.strip()): user_email_state = message.strip() response = f"✅ Thank you! Your email `{user_email_state}` has been saved. How can I help you today?" else: response = "❌ That doesn't look like a valid email. Please enter a valid email address to continue." history.append((message, response)) # Return empty textbox, updated history, and the new email state return "", history, user_email_state # --- Step 2: Generate a response using the AI model --- # Construct a conversation history string for the model # Use only the last 2 turns to keep the context concise history_context = history[-2:] conversation_text = "" for user_turn, bot_turn in history_context: if user_turn: conversation_text += f"User: {user_turn}\n" if bot_turn: conversation_text += f"Bot: {bot_turn}\n" prompt = f"{conversation_text}User: {message}\nBot:" # Generate response generated = generator(prompt, max_length=150, num_return_sequences=1, pad_token_id=generator.tokenizer.eos_token_id) raw_response = generated[0]['generated_text'] # Extract only the last part of the response from the bot bot_response = raw_response.split("Bot:")[-1].strip() # Add a fallback if the model generates an empty response if not bot_response: bot_response = "I'm sorry, I'm not sure how to respond to that. Could you please rephrase?" # --- Step 3: Sanitize links and log the conversation --- sanitized_response = sanitize_response_links(bot_response, message) log_to_csv(user_email_state, message, sanitized_response) history.append((message, sanitized_response)) return "", history, user_email_state def welcome_message(history=None): """Initializes the chat with a welcome message.""" if history is None: history = [] history.append((None, " Welcome to the ACC Virtual Assistant! Please enter your email to start our chat.")) return history # ============================================================================== # 5. GRADIO USER INTERFACE # ============================================================================== with gr.Blocks( css=""" body, .gradio-container { background-color: #ffffff !important; color: #000000 !important; } #header { background-color: rgb(20, 90, 143); color: white; padding: 20px; border-radius: 12px; text-align: center; } #chatbot { height: 500px; } #footer { text-align: center; font-size: 14px; color: gray; margin-top: 15px; } """, title="IlmBot - ACC Assistant" ) as demo: # State object to store the user's email for the duration of the session user_email = gr.State(None) gr.Markdown( """ """ ) chatbot = gr.Chatbot(label="💬 Conversation", elem_id="chatbot") msg_box = gr.Textbox(placeholder="Enter your email first, then your message...", label="✍️ Your Message") with gr.Row(): send_btn = gr.Button("Send") clear_btn = gr.Button("Clear Chat") gr.Markdown( """ """ ) # --- Event Handlers --- def user_submit(message, history, email_state): if not message.strip(): return "", history, email_state return chat_fn(message, history, email_state) def clear_chat(): """Resets the chat and clears the user's email state.""" new_history = welcome_message([]) return "", new_history, None # Clear textbox, history, and email state # Load the initial welcome message when the app starts demo.load(welcome_message, inputs=None, outputs=[chatbot]) # Connect UI components to functions msg_box.submit(user_submit, [msg_box, chatbot, user_email], [msg_box, chatbot, user_email]) send_btn.click(user_submit, [msg_box, chatbot, user_email], [msg_box, chatbot, user_email]) clear_btn.click(clear_chat, None, [msg_box, chatbot, user_email]) # Launch the Gradio app demo.launch(debug=True,share=True)