import pandas as pd import re from sentence_transformers import SentenceTransformer, util import streamlit as st from nltk.corpus import stopwords import nltk import numpy as np from datasets import load_dataset # Download necessary NLTK data nltk.download('stopwords') nltk.download('punkt') # Load the dataset from Hugging Face try: dataset = load_dataset("Sasivar/Analytics_Vidhya_Free_Course_Data") df = pd.DataFrame(dataset['train']) # Assuming the data is in the 'train' split except Exception as e: st.error(f"Error loading dataset: {e}") # Clean and preprocess text stop_words = set(stopwords.words('english')) # Abbreviation to full form mapping abbreviation_map = { "ml": "machine learning", "dl": "deep learning", "llm": "large language model", "ai": "artificial intelligence", "nlp": "natural language processing", "cv": "computer vision", "knn": "k nearest neighbors", "regression": "regression analysis" } def expand_abbreviation(query): """Expand abbreviations to their full forms.""" query = query.lower() # Convert to lowercase for consistency for abbreviation, full_form in abbreviation_map.items(): query = re.sub(rf"\b{abbreviation}\b", full_form, query) # Word boundary match to avoid partial matches return query def clean_text(text): """Clean and preprocess text (remove non-word characters and stopwords).""" text = text.lower() # Convert to lowercase for consistency text = re.sub(r'\W', ' ', text) # Remove non-word characters (punctuation, etc.) tokens = text.split() tokens = [word for word in tokens if word not in stop_words] # Remove stopwords return ' '.join(tokens) # Handle missing ratings def handle_missing_ratings(df, strategy="average"): """Fill missing ratings based on the chosen strategy.""" df['Rating'] = df['Rating'].replace('Unknown', np.nan) # Replace 'Unknown' with NaN if strategy == "average": # Fill with average rating average_rating = df['Rating'].astype(float).mean() df['Rating'] = df['Rating'].fillna(average_rating) elif strategy == "mode": # Fill with mode rating mode_rating = df['Rating'].mode()[0] df['Rating'] = df['Rating'].fillna(mode_rating) elif strategy == "random": # Fill with random rating in the range of existing ratings min_rating = df['Rating'].min() max_rating = df['Rating'].max() random_ratings = np.random.randint(min_rating, max_rating + 1, size=df['Rating'].isna().sum()) df.loc[df['Rating'].isna(), 'Rating'] = random_ratings return df # Add 'Course Info' column by combining key information fields df = handle_missing_ratings(df, strategy="average") # Choose your strategy here: "average", "mode", "random" df['Course Info'] = ( df['Course Title'] + ' ' + df['Description'].fillna('') + ' ' + df['Level'].fillna('Unknown') ).apply(lambda x: expand_abbreviation(clean_text(x))) # Clean and expand abbreviations # Load and cache model @st.cache_resource def load_model(): return SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2') # Cache course embeddings @st.cache_data def get_course_embeddings(_model): return _model.encode(df['Course Info'].tolist(), convert_to_tensor=True) # Load model and encode courses model = load_model() course_embeddings = get_course_embeddings(model) # Define search function with improved processing def search_courses(query, level_filter="All", rating_filter=0.0, threshold=0.5): """Search for the most relevant courses based on the user's query and filters.""" try: # Expand abbreviations in the query query = expand_abbreviation(query) # Clean the expanded query text and compute its embedding query = clean_text(query) query_embedding = model.encode(query, convert_to_tensor=True) # Calculate similarity scores using cosine similarity similarities = util.pytorch_cos_sim(query_embedding, course_embeddings)[0] df['Score'] = similarities.cpu().numpy() # Apply threshold and filter by score sorted_df = df[df['Score'] > threshold].sort_values(by='Score', ascending=False) # Apply level filter if a specific level is selected, include 'Unknown' in the filtering process if level_filter != "All": sorted_df = sorted_df[(sorted_df['Level'] == level_filter) | (sorted_df['Level'] == "Unknown")] else: sorted_df = sorted_df[sorted_df['Level'] != "Unknown"] # Exclude 'Unknown' if 'All' is selected # Apply rating filter sorted_df = sorted_df[sorted_df['Rating'] >= rating_filter] # Remove duplicates based on the course title and score sorted_df = sorted_df.drop_duplicates(subset=['Course Title']) # Select top results top_courses = sorted_df.head(5)[['Course Title', 'Duration', 'Course Link', 'Description', 'Rating', 'Level']] # Round the rating to 2 decimal places for cleaner display top_courses['Rating'] = top_courses['Rating'].round(2) # Prepare results for display results = [{ "Title": row['Course Title'], "Duration": row['Duration'], "Link": row['Course Link'], "Description": row['Description'], "Rating": row['Rating'], "Level": row['Level'] } for _, row in top_courses.iterrows()] return results except Exception as e: st.error(f"Error in search_courses function: {e}") return [] # Streamlit app interface def run_app(): st.title("Smart Course Search") st.write("Find the best free courses tailored to your search query!") # Input form for user query and filters query = st.text_input("Enter your search query:") level_filter = st.selectbox("Select Course Level", options=["All", "Beginner", "Intermediate", "Advanced"]) rating_filter = st.slider("Minimum Rating", 0.0, 5.0, 0.0) if query: results = search_courses(query, level_filter, rating_filter) # Display results with enhanced formatting if results: st.subheader("Search Results:") for result in results: st.markdown(f"### {result['Title']}") st.write( f"**Duration:** {result['Duration']} | **Rating:** {result['Rating']} ⭐ | **Level:** {result['Level']}") st.markdown(f"**Description:** {result['Description']}") st.markdown(f"[Course Link]({result['Link']})") st.write("---") else: st.write("No results found.") else: st.write("Please enter a query to search for courses.") # Run the Streamlit app if __name__ == "__main__": run_app()