import json import re import threading import gradio as gr # ============================================================================== # 1. CONFIGURATION # ============================================================================== # --- General Configuration --- CONTACT_NUMBER = "+91-8977513427" COURSE_CATALOG = [] # UPDATED: The script now reads your nested file directly. JSON_FILE_PATH = "final_merged_course_data.json" # ============================================================================== # 2. COURSE CATALOG AND SEARCH LOGIC # ============================================================================== ABBREVIATION_MAP = { "cs": "computer science", "cse": "computer science engineering", "llb": "bachelor laws", "mca": "master computer applications", "dvoc": "diploma vocation" } QUERY_STOP_WORDS = {'courses', 'course', 'in', 'of', 'what', 'are', 'the', 'offered', 'show', 'me', 'tell', 'and', 'vs', 'versus', 'compare'} def generate_keywords(text: str) -> set: if not isinstance(text, str): return set() text = re.sub(r'[^\w\s.-]', '', text).lower().replace('.', '') words = text.split() words = [word for word in words if word not in QUERY_STOP_WORDS] expanded_words = set(words) for word in words: if word in ABBREVIATION_MAP: expanded_words.update(ABBREVIATION_MAP[word].split()) return expanded_words # ============================================================================== # NEW DATA LOADING LOGIC TO HANDLE NESTED JSON # ============================================================================== def parse_nested_courses(parent_name, data, course_list): """ A recursive function to navigate the nested dictionary and extract course info. """ if isinstance(data, dict): # BASE CASE: We've found a course entry if it has these keys. if 'duration' in data and 'admission_fee' in data: course_entry = { "name": parent_name.strip(), "duration": data.get("duration"), "admission_fee": data.get("admission_fee") } course_list.append(course_entry) return # RECURSIVE STEP: It's a category, so go deeper. for key, value in data.items(): # Create a meaningful name by combining the parent and current keys. if parent_name: # Avoid redundant names like "B.Com (Commerce)" if key is already in parent. if key.lower() in parent_name.lower() or key.lower() == 'general': new_name = parent_name else: new_name = f"{parent_name} ({key})" else: # This is for the top-level keys new_name = key parse_nested_courses(new_name, value, course_list) def initialize_course_catalog(): """ Initializes the course catalog by loading and flattening the nested JSON file. """ global COURSE_CATALOG print(f"Attempting to initialize course catalog from nested file '{JSON_FILE_PATH}'...") try: with open(JSON_FILE_PATH, 'r', encoding='utf-8') as f: nested_json_data = json.load(f) # This list will hold the simple, flattened course data. flat_course_list = [] # Start the recursive parsing process. parse_nested_courses("", nested_json_data, flat_course_list) if not flat_course_list: print(f"🔴 FATAL ERROR: Could not find any valid course entries in '{JSON_FILE_PATH}'.") return # Now, build the final COURSE_CATALOG with keywords, just like before. catalog = [] for course_data in flat_course_list: if isinstance(course_data, dict) and 'name' in course_data: details = course_data keywords = generate_keywords(details['name']) catalog.append({"name": details['name'], "keywords": keywords, "details": details}) else: print(f"⚠️ WARNING: Skipping invalid course entry after processing: {course_data}") COURSE_CATALOG = catalog print(f"✅ Successfully loaded and processed {len(COURSE_CATALOG)} courses from the nested file.") except FileNotFoundError: print(f"🔴 FATAL ERROR: The file '{JSON_FILE_PATH}' was not found.") raise except json.JSONDecodeError: print(f"🔴 FATAL ERROR: Could not decode JSON from '{JSON_FILE_PATH}'. Please check for syntax errors.") raise except Exception as e: print(f"🔴 FATAL ERROR during course initialization: {e}") raise e # ============================================================================== # (The rest of the code is unchanged) # ============================================================================== def find_best_course_match(query: str, force_single_result: bool = False) -> dict | None: if not COURSE_CATALOG: return None query_keywords = generate_keywords(query) if not query_keywords: return None scored_matches = [] for course in COURSE_CATALOG: intersection = query_keywords.intersection(course['keywords']) score = len(intersection) * 10 if query_keywords.issubset(course['keywords']): score += 20 score -= len(course['keywords']) - len(intersection) if score > 0: scored_matches.append((score, course)) if not scored_matches: return None scored_matches.sort(key=lambda x: x[0], reverse=True) best_score, best_match_course = scored_matches[0] result = {"best_match": best_match_course["details"]} is_ambiguous = (not force_single_result and len(scored_matches) > 1 and (best_score - scored_matches[1][0] < 5)) if is_ambiguous: result["suggestions"] = [match[1]['name'] for match in scored_matches] return result # ============================================================================== # 4. CORE CHATBOT LOGIC # ============================================================================== GREETING_WORDS = {"hello", "hi", "hey", "hlo", "yo"} def get_user_intent(user_message: str) -> str: message = user_message.lower() if any(word in message for word in ["compare", "vs", "versus"]): return "compare_courses" words = set(message.split()) if len(words) <= 2 and words.intersection(GREETING_WORDS): return "greeting" if any(word in message for word in ["fee", "fees", "cost", "price", "much"]): return "find_fee" if any(word in message for word in ["duration", "long", "length", "years"]): return "find_duration" return "general_inquiry" def process_user_query(user_message: str, history: list) -> tuple: print(f"\nReceived new query: '{user_message}'") if not COURSE_CATALOG: bot_message = "🔴 Error: The course catalog isn't loaded. Please check the server logs for errors." history.append((user_message, bot_message)) return "", history intent = get_user_intent(user_message) print(f"Detected intent: '{intent}'") bot_message = "" try: if intent == "greeting": bot_message = "Hello! I am your AI Course Assistant. How can I help?" elif intent == "compare_courses": parts = re.split(r'\s+(?:and|vs|versus)\s+', user_message, flags=re.IGNORECASE) if len(parts) < 2: bot_message = "Please tell me the two courses you want to compare. E.g., 'compare btech cse and llb'." else: details_list = [] for part in parts: if part.strip(): result = find_best_course_match(part.strip(), force_single_result=True) if result and result.get("best_match"): details_list.append(result["best_match"]) if len(details_list) < 2: bot_message = "I had trouble finding distinct details for both courses. Please try again." else: header = "| Feature | " + " | ".join([d.get('name', 'N/A') for d in details_list]) + " |" separator = "|---|" + "---|" * len(details_list) fee_row = "| Admission Fee | " + " | ".join([str(d.get('admission_fee', 'N/A')) for d in details_list]) + " |" duration_row = "| Duration | " + " | ".join([d.get('duration', 'N/A') for d in details_list]) + " |" bot_message = f"{header}\n{separator}\n{fee_row}\n{duration_row}" else: result = find_best_course_match(user_message) if not result: bot_message = "I couldn't find any courses matching your query. Please try a different name." elif result.get("suggestions"): suggestions_text = "\n".join([f"• {name}" for name in result["suggestions"][:5]]) bot_message = f"I found a few related courses. Did you mean one of these?\n\n{suggestions_text}\n\nPlease ask again with the full name for details." elif result.get("best_match"): details = result["best_match"] course_name = details.get("name", "N/A") response_parts = [f"Showing details for **{course_name}**:"] if intent == "find_fee" and details.get("admission_fee"): response_parts.append(f"- **Admission Fee:** {details['admission_fee']}") elif intent == "find_duration" and details.get("duration"): response_parts.append(f"- **Duration:** {details['duration']}") else: if details.get("admission_fee"): response_parts.append(f"- **Admission Fee:** {details['admission_fee']}") if details.get("duration"): response_parts.append(f"- **Duration:** {details['duration']}") if len(response_parts) == 1: response_parts.append("- No specific fee or duration info found.") response_parts.append(f"\nFor more help, contact admissions at **{CONTACT_NUMBER}**.") bot_message = "\n".join(response_parts) history.append((user_message, bot_message)) print(f"Generated response: '{bot_message[:100]}...'") return "", history except Exception as e: print(f"🔴 CRITICAL ERROR in process_user_query: {e}") error_message = "I'm sorry, I encountered a critical error on my server. Please try again later." history.append((user_message, error_message)) return "", history # ============================================================================== # 5. GRADIO INTERFACE AND APP LAUNCH # ============================================================================== initialize_course_catalog() with gr.Blocks(theme="soft", title="AATM Course Assistant") as iface: gr.Markdown("## 🤖 AI Course Assistant\nYour guide to course fees, duration, and comparisons.") chatbot = gr.Chatbot(height=600, bubble_full_width=False, label="AATM Chat", elem_id="chatbot") with gr.Row(): msg = gr.Textbox( show_label=False, placeholder="E.g., 'fee for mca', or 'compare btech cse and llb'...", container=False, scale=7, elem_id="chatbot_input" ) clear = gr.ClearButton([msg, chatbot], scale=1) msg.submit(process_user_query, [msg, chatbot], [msg, chatbot], api_name="process_user_query") if __name__ == "__main__": iface.launch()