import psycopg2 import psycopg2.extras import re import os import joblib from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.naive_bayes import MultinomialNB from sklearn.pipeline import make_pipeline import gradio as gr # --- Configuration --- CONTACT_NUMBER = "+91-8977513427" MODEL_FILE_PATH = 'intent_model.joblib' # This is the local path to your courses page, converted to a URL format COURSES_PAGE_PATH = "file:///D:/internship/Project%201/AATM-aug%20first/new_all%20courses.html" # --- IMPORTANT: Fill in with your Supabase credentials --- DB_CONFIG = { "host":"aws-0-ap-south-1.pooler.supabase.com", "dbname":"postgres", "password":"SamuelUday@9984", "user":"postgres.piplkvydeqetbqlolzfb", "port": "6543" } # ============================================================================== # 1. INTENT CLASSIFICATION SETUP # ============================================================================== TRAINING_DATA = [ ("how much is the fee for btech", "find_fee"), ("what's the price on the mba", "find_fee"), ("how long is the course", "find_duration"), ("what is the duration", "find_duration"), ("hi there", "greeting"), ("hello good morning", "greeting"), ("hey", "greeting"), ("tell me about computer science", "general_inquiry"), ("courses in btech", "general_inquiry"), ("compare btech and mca", "compare_courses") # New intent for comparison ] def train_intent_model(): print("Training the intent recognition model...") model = make_pipeline(TfidfVectorizer(), MultinomialNB()) model.fit([item[0] for item in TRAINING_DATA], [item[1] for item in TRAINING_DATA]) joblib.dump(model, MODEL_FILE_PATH) print(f"Model trained and saved to '{MODEL_FILE_PATH}'") return model if not os.path.exists(MODEL_FILE_PATH): intent_model = train_intent_model() else: print(f"Loading existing model from '{MODEL_FILE_PATH}'") intent_model = joblib.load(MODEL_FILE_PATH) # ============================================================================== # 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" } DEGREE_KEYWORDS = ['b.tech', 'btech', 'm.tech', 'mtech', 'ph.d', 'phd', 'diploma', 'b.sc', 'bsc', 'm.sc', 'msc', 'b.a.', 'b.com', 'b.b.a', 'll.b', 'm.c.a', 'mca', 'd.voc', 'dvoc'] GREETING_WORDS = {"hello", "hi", "hey", "hlo", "yo"} QUERY_STOP_WORDS = {'courses', 'course', 'in', 'of', 'what', 'are', 'the', 'offered', 'show', 'me', 'tell', 'and', 'vs', 'versus', 'compare'} COURSE_CATALOG = [] def generate_keywords(text): 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 def initialize_course_catalog(): global COURSE_CATALOG conn = None cursor = None try: print("Connecting to Supabase database...") conn = psycopg2.connect(**DB_CONFIG) cursor = conn.cursor(cursor_factory=psycopg2.extras.DictCursor) cursor.execute("SELECT * FROM courses") courses_from_db = cursor.fetchall() print(f"Successfully loaded {len(courses_from_db)} courses from Supabase cloud database!") catalog = [] for course_data in courses_from_db: details = dict(course_data) keywords = generate_keywords(details['name']) catalog.append({"name": details['name'], "keywords": keywords, "details": details}) COURSE_CATALOG = catalog except Exception as e: print(f"FATAL: Error connecting to Supabase PostgreSQL: {e}") finally: if cursor is not None: cursor.close() if conn is not None: conn.close() print("Database connection closed.") def find_best_course_match(query: str, force_single_result=False): 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) for keyword in DEGREE_KEYWORDS: if keyword in query_keywords and keyword in course['keywords']: score += 5 if score > 0: scored_matches.append((score, course)) if not scored_matches: return None scored_matches.sort(key=lambda x: x[0], reverse=True) is_vague_query = len(query.split()) <= 2 if not force_single_result and is_vague_query and len(scored_matches) > 1 and scored_matches[0][0] == scored_matches[1][0]: return [match[1]['name'] for match in scored_matches[:5]] best_match = scored_matches[0][1] details = best_match["details"] return { "name": best_match["name"], "admission_fee": details.get("admission_fee"), "duration": details.get("duration") } def log_course_query(course_name): conn = None cursor = None try: conn = psycopg2.connect(**DB_CONFIG) cursor = conn.cursor() sql = "INSERT INTO query_logs (course_name) VALUES (%s)" cursor.execute(sql, (course_name,)) conn.commit() print(f"Logged query for: {course_name}") except Exception as e: print(f"Error logging query: {e}") finally: if cursor is not None: cursor.close() if conn is not None: conn.close() # ============================================================================== # 3. CORE CHATBOT LOGIC # ============================================================================== def get_user_intent(user_message: 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 extract_course_from_history(last_bot_message: str): if last_bot_message: match = re.search(r"details for the (.*?) course:", last_bot_message, re.IGNORECASE) if match: return match.group(1).strip() return None def process_user_query(user_message: str, history): if not COURSE_CATALOG: bot_message = "Error: The course catalog could not be loaded. Please check the terminal logs for a database connection error." history.append((user_message, bot_message)) return history intent = get_user_intent(user_message) bot_message = "" 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) if len(parts) < 2: bot_message = "Please tell me the two specific courses you want to compare. For example: 'compare btech cse and llb'." else: details_list = [] for part in parts: if part.strip(): details = find_best_course_match(part.strip(), force_single_result=True) if isinstance(details, dict): details_list.append(details) if len(details_list) < 2: bot_message = "I could only find one of the courses you mentioned. Please try again with two clear course names." else: response = f"| Feature |" for details in details_list: response += f" {details.get('name', 'N/A')} |" response += "\n|---|"+ ("---|"*len(details_list)) response += f"\n| Admission Fee |" for details in details_list: response += f" {details.get('admission_fee', 'N/A')} |" response += f"\n| Duration |" for details in details_list: response += f" {details.get('duration', 'N/A')} |" bot_message = response else: result = find_best_course_match(user_message) if isinstance(result, list): suggestions_text = "\n".join([f"{i+1}. {name}" for i, name in enumerate(result)]) bot_message = (f"I found several courses related to your query. Did you mean one of these?\n\n" f"{suggestions_text}\n\n" f"Please ask again with the full course name. For a complete list, you can visit our [courses page]({COURSES_PAGE_PATH}).") else: details = result if not details and history: course_from_history = extract_course_from_history(history[-1][1]) if course_from_history: details = find_best_course_match(course_from_history) if isinstance(details, dict): log_course_query(details["name"]) course_name = details["name"] response_parts = [f"Showing details for the {course_name} course:"] info_to_add = [] data_points = {"find_fee": ("Admission Fee", details.get("admission_fee")), "find_duration": ("Duration", details.get("duration"))} if intent in data_points: label, value = data_points[intent] if value: info_to_add.append(f"- {label}: {value}") else: for label, value in data_points.values(): if value: info_to_add.append(f"- {label}: {value}") if not info_to_add: info_to_add.append("- I don't have that specific information for this course.") response_parts.extend(info_to_add) response_parts.append(f"\nFor more assistance, please contact our admissions office at {CONTACT_NUMBER}.") bot_message = "\n".join(response_parts) else: bot_message = f"I couldn't find specific details for that course. Please try a different name or check your spelling. For direct assistance, please contact {CONTACT_NUMBER}." history.append((user_message, bot_message)) return history # ============================================================================== # 4. GRADIO INTERFACE # ============================================================================== initialize_course_catalog() iface = gr.ChatInterface( fn=process_user_query, chatbot=gr.Chatbot(height=600, type='messages', layout="bubble"), textbox=gr.Textbox(placeholder="Ask me about 'btech', 'fee for mca', or 'compare btech and llb'...", container=False, scale=7), theme="soft", examples=["btech", "fee for mca", "compare btech cse and llb"], title="AI Course Assistant ", description="Your guide to course fees and duration.", ) if __name__ == "__main__": iface.launch()