| 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
|
|
|
|
|
| CONTACT_NUMBER = "+91-8977513427"
|
| MODEL_FILE_PATH = 'intent_model.joblib'
|
|
|
| COURSES_PAGE_PATH = "file:///D:/internship/Project%201/AATM-aug%20first/new_all%20courses.html"
|
|
|
|
|
| DB_CONFIG = {
|
| "host":"aws-0-ap-south-1.pooler.supabase.com",
|
| "dbname":"postgres",
|
| "password":"SamuelUday@9984",
|
| "user":"postgres.piplkvydeqetbqlolzfb",
|
| "port": "6543"
|
| }
|
|
|
|
|
|
|
|
|
| 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")
|
| ]
|
|
|
| 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)
|
|
|
|
|
|
|
|
|
| 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()
|
|
|
|
|
|
|
|
|
| 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
|
|
|
|
|
|
|
|
|
| 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()
|
| |
