import streamlit as st from streamlit_extras.switch_page_button import switch_page from streamlit_extras.colored_header import colored_header from streamlit_extras.card import card from dataclasses import dataclass from typing import Dict, List, Optional, Tuple from duckduckgo_search import DDGS from groq import Groq import logging from enum import Enum # Set page configuration st.set_page_config( page_title="Course compass ", page_icon="📚", layout="wide", initial_sidebar_state="auto" ) st.markdown(""" """, unsafe_allow_html=True) # Define Enums class LearningFormat(str, Enum): VIDEO = "Video Lectures" INTERACTIVE = "Interactive Exercises" TEXT = "Text-based Content" PROJECT = "Project-based Learning" BLENDED = "Blended Learning" class ExperienceLevel(str, Enum): BEGINNER = "Beginner" INTERMEDIATE = "Intermediate" ADVANCED = "Advanced" EXPERT = "Expert" class CareerGoal(str, Enum): KNOWLEDGE = "Knowledge Acquisition" PROFESSIONAL = "Professional Development" CAREER = "Career Transition" ACADEMIC = "Academic Requirement" @dataclass class UserPreferences: name: str subject: str availability: str budget: int format: LearningFormat experience: ExperienceLevel goal: CareerGoal class CourseRecommender: def __init__(self, api_key: str): self.groq_client = Groq(api_key=api_key) # Use the passed api_key parameter def generate_recommendations(self, preferences: UserPreferences) -> Tuple[str, Optional[List[Dict]]]: try: prompt = self._create_prompt(preferences) response = self._get_ai_response(prompt) search_results = self._get_additional_resources(preferences.subject) return response, search_results except Exception as e: logging.error(f"Error generating recommendations: {str(e)}") return str(e), None # Return the error message as the first element def _create_prompt(self, preferences: UserPreferences) -> str: return f"""As a course recommendation expert, suggest 3-5 specific courses for someone with these preferences: - Name: {preferences.name} - Subject: {preferences.subject} - Time available: {preferences.availability} - Budget: ${preferences.budget} - Preferred format: {preferences.format} - Experience level: {preferences.experience} - Learning goal: {preferences.goal} For each course, include: 1. Course title and platform 2. Price and duration 3. Key features 4. Why it matches their preferences 5.link to redirect to the course Format the response in clear, readable markdown.""" def _get_ai_response(self, prompt: str) -> str: response = self.groq_client.chat.completions.create( messages=[ { "role": "system", "content": "You are a helpful course recommendation assistant." }, { "role": "user", "content": prompt } ], model="mixtral-8x7b-32768", temperature=0.7, max_tokens=2048 ) return response.choices[0].message.content def _get_additional_resources(self, subject: str) -> List[Dict]: try: with DDGS() as ddgs: results = list(ddgs.text( f"learn {subject} course tutorial guide", max_results=5 )) return results except Exception as e: logging.error(f"Error fetching additional resources: {str(e)}") return [] class StreamlitUI: def __init__(self): self.initialize_session_state() def initialize_session_state(self): if 'initialized' not in st.session_state: st.session_state.update({ 'initialized': True, 'step': 0, 'user_responses': {}, 'ready_for_recommendations': False, 'recommendations_generated': False, 'current_recommendations': None, 'current_results': None, 'error_message': None, 'theme_color': '#3b71ca', 'api_key_valid': False, 'api_key': None }) def validate_api_key(self, api_key: str) -> bool: """Validate if the API key is in the correct format.""" return api_key.startswith('gsk_') and len(api_key) > 20 def render_questionnaire(self) -> bool: questions = [ ("Please enter your name:", "user_name", None), ("What subject or skill would you like to learn?", "subject", None), ("How many hours per week can you dedicate to learning?", "availability", ["1-2 hours", "3-5 hours", "6-10 hours", "More than 10 hours"]), ("What is your maximum budget for this course?", "budget", None), ("What is your preferred learning format?", "format", [f.value for f in LearningFormat]), ("What is your current experience level?", "experience", [e.value for e in ExperienceLevel]), ("What is your primary goal?", "goal", [g.value for g in CareerGoal]) ] if st.session_state.step >= len(questions): return self.handle_questionnaire_completion() # Create a container for the questionnaire with st.container(): colored_header( label=f"Step {st.session_state.step + 1} of {len(questions)}", description="Tell us about your learning preferences", color_name="blue-70" ) progress = min(st.session_state.step / (len(questions) - 1), 1.0) st.progress(progress) current_question = questions[st.session_state.step] return self.handle_current_question(current_question) def handle_current_question(self, question_data: Tuple) -> bool: question, key, options = question_data with st.form(key=f"question_form_{key}"): st.write(f"### {question}") if options: user_input = st.selectbox( "Choose an option:", options, key=f"input_{key}", help=f"Select your preferred {key}" ) elif key == "budget": budget_ranges = [ {"label": "Free Courses Only", "value": 0}, {"label": "Under $50", "value": 50}, {"label": "$50 - $100", "value": 100}, {"label": "$100 - $200", "value": 200}, {"label": "$200 - $500", "value": 500}, {"label": "$500 - $1000", "value": 1000}, {"label": "Over $1000", "value": 1500} ] selected_range = st.radio( "Select your budget range:", options=[r["label"] for r in budget_ranges], key="budget_range", horizontal=True, help="Choose your preferred budget range for the course" ) # Get the corresponding value for the selected range user_input = next( (r["value"] for r in budget_ranges if r["label"] == selected_range), 100 ) else: user_input = st.text_input( "Your answer:", key=f"input_{key}", placeholder=f"Enter your {key.replace('_', ' ')}", help=f"Enter your {key.replace('_', ' ')}" ) submit_button = st.form_submit_button( "Next" if st.session_state.step < 6 else "Get Recommendations" ) if submit_button and (user_input or user_input == 0): self.save_user_input(key, user_input) st.session_state.step += 1 st.rerun() return False def save_user_input(self, key: str, user_input: str): st.session_state.user_responses[key] = user_input def handle_questionnaire_completion(self) -> bool: """Handle the completion of the questionnaire and prepare for recommendations.""" try: if not all(key in st.session_state.user_responses for key in ['user_name', 'subject', 'availability', 'budget', 'format', 'experience', 'goal']): st.error("Please complete all questions before proceeding.") self.reset_session() return False preferences = UserPreferences( name=st.session_state.user_responses['user_name'], subject=st.session_state.user_responses['subject'], availability=st.session_state.user_responses['availability'], budget=int(st.session_state.user_responses['budget']), format=st.session_state.user_responses['format'], experience=st.session_state.user_responses['experience'], goal=st.session_state.user_responses['goal'] ) with st.spinner('Generating recommendations...'): recommendations, additional_results = self.recommender.generate_recommendations(preferences) if isinstance(recommendations, str) and "Error" in recommendations: st.error(recommendations) return False st.session_state.current_recommendations = recommendations st.session_state.current_results = additional_results st.session_state.ready_for_recommendations = True st.session_state.recommendations_generated = True st.session_state.user_name = preferences.name return True except Exception as e: st.error(f"An error occurred: {str(e)}") logging.error(f"Error in handle_questionnaire_completion: {str(e)}") return False def reset_session(self): """Reset the session state to initial values.""" api_key = st.session_state.api_key # Preserve API key api_key_valid = st.session_state.api_key_valid # Preserve API key validation status st.session_state.clear() st.session_state.update({ 'step': 0, 'user_responses': {}, 'ready_for_recommendations': False, 'recommendations_generated': False, 'current_recommendations': None, 'current_results': None, 'error_message': None, 'api_key': api_key, 'api_key_valid': api_key_valid }) def display_recommendations(self): if st.session_state.error_message: st.error(st.session_state.error_message) if st.button("Try Again", use_container_width=True): self.reset_session() st.rerun() return if not st.session_state.current_recommendations: st.warning("No recommendations available. Please start a new search.") if st.button("Start New Search", use_container_width=True): self.reset_session() st.rerun() return colored_header( label=f"Personalized Course Recommendations for {st.session_state.user_name}", description="Based on your preferences, we recommend the following courses:", color_name="blue-70" ) # Display recommendations in a modern card layout col1, col2 = st.columns([7, 3]) with col1: with st.container(): st.markdown( f"""