# study_planner.py import json from datetime import datetime, timedelta, date from typing import Dict, List, Any import logging from groq import Groq import google.generativeai as genai import os logger = logging.getLogger(__name__) # Initialize AI clients (reuse from main app) groq_client = None genai_client = None try: GROQ_API_KEY = os.environ.get("GROQ_API_KEY") if GROQ_API_KEY: groq_client = Groq(api_key=GROQ_API_KEY) except Exception as e: logger.warning(f"Groq client not available: {e}") try: GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY") if GEMINI_API_KEY: genai.configure(api_key=GEMINI_API_KEY) genai_client = genai except Exception as e: logger.warning(f"Gemini client not available: {e}") class StudyPlanner: def __init__(self): self.level_progression = { 'A1': {'next': 'A2', 'weeks': 12, 'focus': ['basic vocabulary', 'present tense', 'introductions']}, 'A2': {'next': 'B1', 'weeks': 16, 'focus': ['past tense', 'future tense', 'everyday situations']}, 'B1': {'next': 'B2', 'weeks': 20, 'focus': ['conditional', 'complex sentences', 'opinions']}, 'B2': {'next': 'C1', 'weeks': 24, 'focus': ['subjunctive', 'formal writing', 'presentations']}, 'C1': {'next': 'C2', 'weeks': 28, 'focus': ['nuanced expressions', 'academic writing', 'debates']}, 'C2': {'next': 'C2', 'weeks': 32, 'focus': ['native-like fluency', 'specialized topics', 'literature']} } self.activity_types = { 'reading': { 'icon': '📚', 'min_duration': 20, 'max_duration': 45, 'difficulty_scaling': True, 'description': 'Read articles and texts' }, 'flashcards': { 'icon': '🃏', 'min_duration': 10, 'max_duration': 25, 'difficulty_scaling': False, 'description': 'Review vocabulary flashcards' }, 'conversation': { 'icon': '💬', 'min_duration': 15, 'max_duration': 30, 'difficulty_scaling': True, 'description': 'Practice speaking and conversation' }, 'writing': { 'icon': '✍️', 'min_duration': 15, 'max_duration': 40, 'difficulty_scaling': True, 'description': 'Complete writing exercises' }, 'listening': { 'icon': '🎧', 'min_duration': 15, 'max_duration': 30, 'difficulty_scaling': True, 'description': 'Listen to audio content' }, 'grammar': { 'icon': '📝', 'min_duration': 10, 'max_duration': 25, 'difficulty_scaling': True, 'description': 'Study grammar rules and patterns' } } def generate_personalized_plan(self, user_data: Dict[str, Any]) -> Dict[str, Any]: """Generate a comprehensive study plan based on user data""" try: current_level = user_data.get('english_level', 'B1') target_level = user_data.get('target_level', 'B2') weekly_hours = user_data.get('weekly_hours', 5) interests = user_data.get('interests', {}) context_focus = user_data.get('context_focus', 'General/Social') study_goals = user_data.get('study_goals', []) # Calculate timeline timeline = self._calculate_study_timeline(current_level, target_level, weekly_hours) # Generate weekly structure weekly_structure = self._create_weekly_structure(weekly_hours, current_level, context_focus) # Create specific activities activities = self._generate_weekly_activities( weekly_structure, interests, current_level, context_focus, study_goals ) # Generate AI-powered study tips study_tips = self._generate_ai_study_tips(user_data) plan = { 'id': f"plan_{datetime.now().strftime('%Y%m%d_%H%M%S')}", 'created_at': datetime.now().isoformat(), 'current_level': current_level, 'target_level': target_level, 'weekly_hours': weekly_hours, 'estimated_weeks': timeline['weeks'], 'completion_date': timeline['completion_date'], 'weekly_structure': weekly_structure, 'activities': activities, 'study_tips': study_tips, 'milestones': self._create_milestones(current_level, target_level, timeline['weeks']), 'adaptations': self._suggest_adaptations(user_data) } return {'success': True, 'plan': plan} except Exception as e: logger.error(f"Error generating study plan: {e}") return {'success': False, 'error': str(e)} def _calculate_study_timeline(self, current_level: str, target_level: str, weekly_hours: int) -> Dict[str, Any]: """Calculate realistic timeline for reaching target level""" try: current_info = self.level_progression.get(current_level, self.level_progression['B1']) base_weeks = current_info['weeks'] # Adjust based on weekly hours (baseline is 5 hours/week) hour_multiplier = 5 / max(weekly_hours, 1) adjusted_weeks = int(base_weeks * hour_multiplier) # If targeting multiple levels ahead, add additional time level_order = ['A1', 'A2', 'B1', 'B2', 'C1', 'C2'] current_idx = level_order.index(current_level) if current_level in level_order else 2 target_idx = level_order.index(target_level) if target_level in level_order else 3 if target_idx > current_idx + 1: # Multiple levels - add 20% more time adjusted_weeks = int(adjusted_weeks * 1.2 * (target_idx - current_idx)) completion_date = (datetime.now() + timedelta(weeks=adjusted_weeks)).date() return { 'weeks': adjusted_weeks, 'completion_date': completion_date.isoformat(), 'intensity': 'High' if weekly_hours > 7 else 'Medium' if weekly_hours > 4 else 'Light' } except Exception as e: logger.error(f"Error calculating timeline: {e}") return {'weeks': 16, 'completion_date': (datetime.now() + timedelta(weeks=16)).date().isoformat()} def _create_weekly_structure(self, weekly_hours: int, level: str, context: str) -> Dict[str, Any]: """Create optimal weekly study structure""" try: # Base distribution percentages distributions = { 'A1': {'reading': 0.25, 'flashcards': 0.30, 'conversation': 0.20, 'writing': 0.15, 'grammar': 0.10}, 'A2': {'reading': 0.30, 'flashcards': 0.25, 'conversation': 0.20, 'writing': 0.15, 'grammar': 0.10}, 'B1': {'reading': 0.30, 'flashcards': 0.20, 'conversation': 0.25, 'writing': 0.20, 'listening': 0.05}, 'B2': {'reading': 0.25, 'flashcards': 0.15, 'conversation': 0.25, 'writing': 0.25, 'listening': 0.10}, 'C1': {'reading': 0.30, 'flashcards': 0.10, 'conversation': 0.25, 'writing': 0.25, 'listening': 0.10}, 'C2': {'reading': 0.35, 'flashcards': 0.05, 'conversation': 0.25, 'writing': 0.25, 'listening': 0.10} } base_dist = distributions.get(level, distributions['B1']) # Adjust based on context if context == 'Professional/Business': base_dist['writing'] = min(base_dist['writing'] + 0.10, 0.40) base_dist['reading'] = max(base_dist['reading'] - 0.05, 0.15) base_dist['conversation'] = max(base_dist['conversation'] - 0.05, 0.15) elif context == 'Technical/IT': base_dist['reading'] = min(base_dist['reading'] + 0.10, 0.45) base_dist['flashcards'] = min(base_dist['flashcards'] + 0.05, 0.35) base_dist['conversation'] = max(base_dist['conversation'] - 0.10, 0.15) # Convert to actual hours weekly_structure = {} total_minutes = weekly_hours * 60 for activity, percentage in base_dist.items(): minutes = int(total_minutes * percentage) if minutes >= self.activity_types[activity]['min_duration']: weekly_structure[activity] = { 'minutes_per_week': minutes, 'sessions_per_week': max(1, minutes // 30), # Aim for 30-min sessions 'minutes_per_session': minutes // max(1, minutes // 30) } return weekly_structure except Exception as e: logger.error(f"Error creating weekly structure: {e}") return {} def _generate_weekly_activities(self, structure: Dict, interests: Dict, level: str, context: str, goals: List) -> List[Dict]: """Generate specific weekly activities""" activities = [] try: for activity_type, schedule in structure.items(): activity_info = self.activity_types[activity_type] for session in range(schedule['sessions_per_week']): activity = { 'id': f"{activity_type}_{session + 1}", 'type': activity_type, 'icon': activity_info['icon'], 'title': f"{activity_info['description']}", 'duration_minutes': schedule['minutes_per_session'], 'difficulty': level, 'context': context, 'day_of_week': (session * 2) % 7, # Spread throughout week 'specific_tasks': self._generate_specific_tasks(activity_type, level, context, interests, goals) } activities.append(activity) # Sort by day of week activities.sort(key=lambda x: x['day_of_week']) return activities except Exception as e: logger.error(f"Error generating activities: {e}") return [] def _generate_specific_tasks(self, activity_type: str, level: str, context: str, interests: Dict, goals: List) -> List[str]: """Generate specific tasks for each activity type""" tasks = [] try: interest_topics = list(interests.keys())[:3] if interests else ['general topics'] if activity_type == 'reading': tasks = [ f"Read a {context.lower()} article about {topic}" for topic in interest_topics ] + [ f"Practice reading comprehension with {level}-level texts", "Identify new vocabulary and create flashcards" ] elif activity_type == 'flashcards': tasks = [ "Review previous day's vocabulary", "Practice new words from recent reading", f"Focus on {context.lower()} terminology" ] elif activity_type == 'conversation': tasks = [ f"Discuss {topic} using {level}-level vocabulary" for topic in interest_topics[:2] ] + [ "Practice pronunciation with AI feedback", f"Role-play {context.lower()} scenarios" ] elif activity_type == 'writing': tasks = [ f"Write a short text about {topic}" for topic in interest_topics[:1] ] + [ f"Practice {context.lower()} writing format s", "Get AI feedback on grammar and style" ] elif activity_type == 'listening': tasks = [ f"Listen to content about {topic}" for topic in interest_topics[:2] ] + [ "Practice with different accents", "Take notes while listening" ] elif activity_type == 'grammar': level_grammar = { 'A1': ['present tense', 'basic sentence structure', 'personal pronouns'], 'A2': ['past tense', 'future tense', 'comparatives'], 'B1': ['present perfect', 'conditional sentences', 'passive voice'], 'B2': ['subjunctive mood', 'complex sentences', 'reported speech'], 'C1': ['advanced tenses', 'nuanced expressions', 'formal structures'], 'C2': ['idiomatic expressions', 'stylistic variations', 'literary devices'] } tasks = [f"Study {topic}" for topic in level_grammar.get(level, level_grammar['B1'])] return tasks[:3] # Limit to 3 tasks per activity except Exception as e: logger.error(f"Error generating specific tasks: {e}") return ["Complete activity as planned"] def _generate_ai_study_tips(self, user_data: Dict) -> List[str]: """Generate personalized study tips using AI""" try: if not groq_client and not genai_client: return self._get_default_tips(user_data.get('english_level', 'B1')) prompt = f""" Generate 5 personalized English study tips for a user with these characteristics: - Current Level: {user_data.get('english_level', 'B1')} - Target Level: {user_data.get('target_level', 'B2')} - Weekly Study Time: {user_data.get('weekly_hours', 5)} hours - Context Focus: {user_data.get('context_focus', 'General/Social')} - Interests: {', '.join(user_data.get('interests', {}).keys())} Provide practical, actionable tips that are specific to their level and interests. Format as a simple list of tips, each starting with an emoji. """ response_text = None if groq_client: response = groq_client.chat.completions.create( model="llama-3.1-8b-instant", messages=[{"role": "user", "content": prompt}], temperature=0.7 ) response_text = response.choices[0].message.content elif genai_client: model = genai_client.GenerativeModel('gemini-2.5-flash-latest') response = model.generate_content(prompt) response_text = response.text if response_text: # Extract tips from response tips = [line.strip() for line in response_text.split('\n') if line.strip() and ('📚' in line or '💡' in line or '🎯' in line or '⭐' in line or '🚀' in line)] return tips[:5] if tips else self._get_default_tips(user_data.get('english_level', 'B1')) return self._get_default_tips(user_data.get('english_level', 'B1')) except Exception as e: logger.error(f"Error generating AI study tips: {e}") return self._get_default_tips(user_data.get('english_level', 'B1')) def _get_default_tips(self, level: str) -> List[str]: """Get default study tips based on level""" tips_by_level = { 'A1': [ "📚 Start with basic vocabulary - 10 new words daily", "🎯 Focus on present tense in daily conversations", "💡 Use picture dictionaries for visual learning", "⭐ Practice pronunciation with simple audio materials", "🚀 Don't worry about mistakes - communication is key!" ], 'A2': [ "📚 Read simple news articles and stories", "🎯 Practice past and future tenses regularly", "💡 Join basic English conversation groups", "⭐ Use language learning apps for daily practice", "🚀 Watch movies with subtitles in your language" ], 'B1': [ "📚 Read intermediate articles on topics you enjoy", "🎯 Practice expressing opinions and preferences", "💡 Start writing short paragraphs daily", "⭐ Listen to podcasts at normal speed", "🚀 Try to think in English for simple tasks" ], 'B2': [ "📚 Read longer articles and opinion pieces", "🎯 Practice formal and informal writing styles", "💡 Engage in debates and discussions", "⭐ Watch news programs without subtitles", "🚀 Set specific goals for each study session" ], 'C1': [ "📚 Read academic and professional texts", "🎯 Practice nuanced expressions and idioms", "💡 Write formal reports and presentations", "⭐ Listen to academic lectures and conferences", "🚀 Focus on specialized vocabulary for your field" ], 'C2': [ "📚 Read literature and complex analytical texts", "🎯 Master subtle language differences", "💡 Write with stylistic sophistication", "⭐ Engage with native speakers in professional contexts", "🚀 Aim for native-like fluency in all skills" ] } return tips_by_level.get(level, tips_by_level['B1']) def _create_milestones(self, current_level: str, target_level: str, weeks: int) -> List[Dict]: """Create progress milestones""" milestones = [] try: milestone_intervals = max(2, weeks // 4) # Create 4 milestones for i in range(1, 5): week = milestone_intervals * i if week <= weeks: milestone = { 'week': week, 'title': f"Milestone {i}", 'description': self._get_milestone_description(i, current_level, target_level), 'target_date': (datetime.now() + timedelta(weeks=week)).date().isoformat(), 'completed': False } milestones.append(milestone) return milestones except Exception as e: logger.error(f"Error creating milestones: {e}") return [] def _get_milestone_description(self, milestone_num: int, current_level: str, target_level: str) -> str: """Get description for milestone""" descriptions = { 1: f"Complete foundation review and establish study routine", 2: f"Reach intermediate proficiency between {current_level} and {target_level}", 3: f"Demonstrate advanced skills approaching {target_level} level", 4: f"Achieve {target_level} level proficiency in all skills" } return descriptions.get(milestone_num, f"Progress checkpoint {milestone_num}") def _suggest_adaptations(self, user_data: Dict) -> List[str]: """Suggest plan adaptations based on user data""" adaptations = [] try: weekly_hours = user_data.get('weekly_hours', 5) context = user_data.get('context_focus', 'General/Social') level = user_data.get('english_level', 'B1') if weekly_hours < 4: adaptations.append("💡 Consider increasing study time to 4+ hours/week for faster progress") if weekly_hours > 8: adaptations.append("⚠️ Ensure you don't burn out - quality over quantity") if context == 'Professional/Business': adaptations.append("📊 Focus extra time on business writing and presentation skills") if context == 'Technical/IT': adaptations.append("💻 Include technical documentation reading in your routine") if level in ['C1', 'C2']: adaptations.append("🎯 Consider specialized courses or certification preparation") return adaptations[:3] # Limit to 3 adaptations except Exception as e: logger.error(f"Error suggesting adaptations: {e}") return [] # Global instance study_planner = StudyPlanner()