import re import json import random from textblob import TextBlob try: from mistralai import Mistral as _MistralClient _MISTRAL_AVAILABLE = True except ImportError: _MISTRAL_AVAILABLE = False _MistralClient = None class _MistralResponse: """Wraps a Mistral chat response to expose a `.text` attribute.""" def __init__(self, content: str): self.text = content class MistralModelWrapper: """Drop-in replacement for `genai.GenerativeModel`. Exposes `generate_content(content)` so all existing callers keep working without modification. `content` may be either: - a plain string prompt, or - a list [prompt_str, {'mime_type': ..., 'data': ...}] (legacy multimodal multimodal – the audio part is silently dropped because Mistral is text-only; the text prompt is still sent as usual). """ def __init__(self, client, model_name: str): self._client = client self._model_name = model_name def generate_content(self, content) -> _MistralResponse: if isinstance(content, list): # Extract the text part from a multimodal list; ignore binary blobs text_parts = [p for p in content if isinstance(p, str)] prompt = ' '.join(text_parts) if text_parts else '' else: prompt = str(content) response = self._client.chat.complete( model=self._model_name, messages=[{'role': 'user', 'content': prompt}], ) text = response.choices[0].message.content return _MistralResponse(text) from pydub import AudioSegment import os import shutil from app.models.topic import ComprehensionQuestion, SpeakingQuestion from app.models.activity import Activity from app.models.quiz import VocabularyQuiz, QuizQuestion from flask import current_app from flask_login import current_user import logging import nltk from nltk.tokenize import sent_tokenize, word_tokenize from nltk.corpus import wordnet from nltk.stem import WordNetLemmatizer from datetime import datetime # Download NLTK data if needed try: nltk.download('punkt', quiet=True) nltk.download('wordnet', quiet=True) nltk.download('averaged_perceptron_tagger', quiet=True) except: pass # If download fails, continue anyway logger = logging.getLogger(__name__) # Define global functions before the class to avoid circular imports def get_ai_assessor(): """ Gets a singleton instance of the AIAssessment class. """ from flask import current_app # Always check if we need to update the configuration if hasattr(current_app, 'ai_assessor'): # Check if AI transcription config has changed use_ai = bool(current_app.config.get('USE_AI_TRANSCRIPTION', False)) current_assessor = current_app.ai_assessor if getattr(current_assessor, '_use_ai_transcription', False) != use_ai: logger.info(f"Updating AI transcription setting to: {use_ai}") current_assessor.enable_ai_transcription(use_ai) if not hasattr(current_app, 'ai_assessor'): assessor = AIAssessment() # Enable AI transcription via app config if requested try: use_ai = bool(current_app.config.get('USE_AI_TRANSCRIPTION', False)) assessor.enable_ai_transcription(use_ai) logger.info(f"Initialized AI assessor with transcription: {use_ai}") except Exception as e: logger.warning(f"Failed to configure AI transcription: {e}") current_app.ai_assessor = assessor return current_app.ai_assessor def assess_submission(submission, feedback_text): """ Process teacher's assessment of a student submission. Args: submission: The Submission object feedback_text: The teacher's feedback text Returns: dict: A dictionary with assessment results """ assessor = get_ai_assessor() # Validate feedback text feedback = assessor.validate_word_count(feedback_text) # Simple structure with teacher feedback assessment = { 'feedback': feedback, 'score': submission.score, # Use the score from the submission 'timestamp': submission.reviewed_at.isoformat() if submission.reviewed_at else None } return assessment class AIAssessment: def __init__(self): # Initialize self.model = None # By default do not use AI transcription. Enable with enable_ai_transcription(True) self._use_ai_transcription = False # last transcription confidence recorded by _transcribe_audio (0.0-1.0) self._last_transcription_confidence = None # Replace language tool initialization with TextBlob (no init required) try: logger.info("Using TextBlob for grammar checking instead of LanguageTool") except Exception as e: logger.error(f"Error initializing text analysis tools: {str(e)}") # Set FFmpeg paths with explicit error checking try: ffmpeg_path = current_app.config.get('FFMPEG_PATH') ffprobe_path = current_app.config.get('FFPROBE_PATH') if ffmpeg_path and ffprobe_path and os.path.exists(ffmpeg_path) and os.path.exists(ffprobe_path): logger.info(f"FFmpeg found at: {ffmpeg_path}") AudioSegment.converter = ffmpeg_path AudioSegment.ffmpeg = ffmpeg_path AudioSegment.ffprobe = ffprobe_path else: # Try to find in PATH logger.warning("FFmpeg not found at configured path, searching PATH...") ffmpeg_in_path = shutil.which('ffmpeg') ffprobe_in_path = shutil.which('ffprobe') if ffmpeg_in_path and ffprobe_in_path: logger.info(f"Found FFmpeg in PATH: {ffmpeg_in_path}") AudioSegment.converter = ffmpeg_in_path AudioSegment.ffmpeg = ffmpeg_in_path AudioSegment.ffprobe = ffprobe_in_path else: logger.error("FFmpeg/FFprobe not found in configured paths or in PATH") except Exception as e: logger.error(f"Error setting FFmpeg paths: {str(e)}") def enable_ai_transcription(self, enabled: bool): """Toggle AI-based transcription. When enabled, speaking assessment is sent to the configured Mistral model.""" self._use_ai_transcription = bool(enabled) def _check_subject_verb_agreement(self, subject, verb): """Basic subject-verb agreement check""" singular_subjects = ['he', 'she', 'it'] if subject.lower() in singular_subjects and verb.lower() in ['are', 'were']: return False return True def _fix_subject_verb_agreement(self, sentence): """Basic attempt to fix subject-verb agreement""" for subject in ['he', 'she', 'it']: sentence = sentence.replace(f"{subject} are", f"{subject} is") sentence = sentence.replace(f"{subject} were", f"{subject} was") return sentence def personalize_content(self, content): """ Replace placeholders in content with user-specific information Args: content (str): The content with placeholders Returns: str: Personalized content with placeholders replaced """ if not content: return "" # Check if we have access to current_user try: if current_user and current_user.is_authenticated and hasattr(current_user, 'username'): # Replace [name] with the student's username content = content.replace('[name]', current_user.username) # Replace [first_name] with the student's first name if available if hasattr(current_user, 'first_name') and current_user.first_name: content = content.replace('[first_name]', current_user.first_name) else: content = content.replace('[first_name]', current_user.username) except Exception as e: logging.warning(f"Could not personalize content: {e}") return content def _configure_genai(self): # Configure Mistral API only when needed if not hasattr(self, 'genai_configured'): if not _MISTRAL_AVAILABLE: logger.error("mistralai package not installed. Run: pip install mistralai") self.genai_configured = False return False api_key = current_app.config.get('MISTRAL_API_KEY', '') # Check if the API key is valid (not empty and not the default placeholder) if not api_key or api_key == 'your-mistral-api-key-here': logger.error("Invalid or missing Mistral API key. Please set a valid MISTRAL_API_KEY in your .env file.") self.genai_configured = False return False try: self._mistral_client = _MistralClient(api_key=api_key) self.genai_configured = True logger.debug("Mistral API configured successfully") return True except Exception as e: logger.error(f"Error configuring Mistral API: {str(e)}") self.genai_configured = False return False return self.genai_configured def get_model(self): """Lazy initialization of Mistral model wrapper.""" if self.model is None: if not self._configure_genai(): logger.error("Failed to configure Mistral API") return None # Models to try in order of preference (cost/speed vs quality) models_to_try = [ 'mistral-small-latest', # Fast, cost-effective 'mistral-medium-latest', # Balanced 'mistral-large-latest', # Highest quality fallback ] for model_name in models_to_try: try: logger.info(f"Attempting to initialize Mistral model: {model_name}") wrapper = MistralModelWrapper(self._mistral_client, model_name) # Quick smoke-test test_response = wrapper.generate_content("Hello, world!") logger.info(f"Successfully initialized and tested Mistral model: {model_name}") self.model = wrapper return self.model except Exception as e: logger.warning(f"Failed to initialize Mistral model {model_name}: {str(e)}") self.model = None # Reset for next attempt logger.error("All Mistral models failed to initialize") return None return self.model def _transcribe_audio(self, audio_path): """AI-only transcription wrapper kept for compatibility. If AI transcription is enabled, delegate to the multimodal AI path which returns both transcription and assessment. Otherwise return an empty string to indicate no transcription. """ try: if getattr(self, '_use_ai_transcription', False) and self._configure_genai() and self.get_model(): # Use the AI path to transcribe (without performing the full assessment here) try: result = self._ai_transcribe_and_assess(audio_path) if result and isinstance(result, dict): # store confidence and return transcript try: self._last_transcription_confidence = float(result.get('scores', {}).get('overall', 0) / 100.0) except Exception: pass return result.get('transcription', '') except Exception as e: logger.warning(f"AI transcription wrapper failed: {e}") # If AI transcription disabled or failed, return empty (we removed local recognizer) return "" except Exception as e: logger.error(f"_transcribe_audio error: {e}") return "" def generate_speaking_questions(self, response_text_or_topic): """Return a list of 5 SpeakingQuestion objects. If `response_text_or_topic` looks like long reading content, call the AI model to generate 5 short speaking prompts related to the topic. Otherwise treat the input as already parsed response text (one prompt per line). """ # Determine input text safely. If a Topic-like object is passed, extract its reading_content. text = '' try: # If it's an object with reading_content attribute if hasattr(response_text_or_topic, 'reading_content'): text = (response_text_or_topic.reading_content or '').strip() logger.info(f"Extracted reading_content from topic object: {len(text)} characters") # If it's a dict with reading_content elif isinstance(response_text_or_topic, dict) and 'reading_content' in response_text_or_topic: text = (response_text_or_topic.get('reading_content') or '').strip() logger.info(f"Extracted reading_content from dict: {len(text)} characters") else: text = (response_text_or_topic or '').strip() logger.info(f"Using input as text directly: {len(text)} characters") # Log the first 200 characters of content for debugging if text: preview = text[:200] + "..." if len(text) > 200 else text logger.debug(f"Content preview: {preview}") else: logger.warning("No reading content available for question generation") except Exception as e: logger.error(f"Error extracting reading content: {e}") text = '' question_texts = [] # Prefer AI generation whenever topic content is provided and model is available if text and self._configure_genai() and self.get_model(): try: model = self.get_model() prompt = ( "You are an English instructor creating speaking questions for BEGINNER students based on the reading content below.\n\n" "🎯 YOUR MISSION:\n" "Generate EXACTLY 5 speaking questions that are DIRECTLY BASED on the SPECIFIC content from the reading.\n" "Questions must ask students to discuss, explain, or share information ABOUT THE READING CONTENT.\n\n" "📚 READING CONTENT:\n" "---\n" + text + "\n---\n\n" "⚠️ CRITICAL REQUIREMENTS:\n" "1. ALL questions MUST start with 'Tell me' or 'Tell me about'\n" "2. Questions MUST be about SPECIFIC people, places, facts, or topics mentioned in the reading above\n" "3. Questions should ask students to explain or discuss what they learned from the reading\n" "4. Each question should be 5-12 words long (short and simple)\n" "5. Questions should prompt 15-30 seconds of speech\n" "6. Use simple, beginner-friendly language (A1-A2 level)\n" "7. Return ONLY a JSON array of 5 question strings - no other text\n\n" "✅ GOOD EXAMPLES (if reading is about 'John, a student at MIT who studies engineering'):\n" "- \"Tell me about John\"\n" "- \"Tell me about what John studies\"\n" "- \"Tell me about John's university\"\n" "- \"Tell me what John does at MIT\"\n" "- \"Tell me about engineering students\"\n\n" "❌ BAD EXAMPLES (too generic, not about the reading):\n" "- \"Tell me about your hobbies\" (not in reading)\n" "- \"Tell me about what you do for entertainment\" (not related)\n" "- \"Tell me about your family\" (not mentioned in reading)\n" "- \"Read and speak these sentences\" (forbidden format)\n\n" "📋 HOW TO GENERATE QUESTIONS:\n" "Step 1: Read the content carefully and identify:\n" " - Main people/characters mentioned\n" " - Key places or locations\n" " - Important facts or information\n" " - Main topics or themes\n" " - Activities or events described\n" "\n" "Step 2: Create 'Tell me' questions using this format:\n" " - Tell me about [specific person from reading]\n" " - Tell me about [specific place from reading]\n" " - Tell me what [person/character] does\n" " - Tell me about [specific topic mentioned]\n" " - Tell me about [specific fact from content]\n\n" "🚫 FORBIDDEN (DO NOT USE):\n" "- Generic questions not based on the reading\n" "- Questions about the student's personal life (unless directly related to reading)\n" "- 'Read and speak', 'Read aloud', 'Repeat after me'\n" "- Questions longer than 12 words\n" "- Any question not starting with 'Tell me'\n\n" "📤 OUTPUT FORMAT:\n" "Return ONLY a JSON array of exactly 5 strings:\n" '[\"Tell me about [specific element from reading]\", \"Tell me about [another specific element]\", ...]\n\n' "IMPORTANT: Your questions must be SO SPECIFIC to the reading that someone who hasn't read it won't know how to answer!\n\n" "RESPOND WITH JSON ARRAY NOW:" ) logger.info("Requesting AI-generated speaking prompts for topic (length=%d)", len(text)) response = model.generate_content(prompt) response_text = getattr(response, 'text', str(response)) # Log raw AI response to help debug if the prompts are not topic-aware logger.debug("AI generate_speaking_questions raw response: %s", (response_text[:1000] + '...') if len(response_text) > 1000 else response_text) # Persist raw responses to instance folder for offline inspection try: # Write raw AI responses to the project's top-level instance/ folder. # __file__ is app/services/ai_assessment.py -> climb up to project root. project_root = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) log_dir = os.path.join(project_root, 'instance') os.makedirs(log_dir, exist_ok=True) log_path = os.path.join(log_dir, 'ai_generate_speaking_raw.log') with open(log_path, 'a', encoding='utf-8') as lf: lf.write("--- %s ---\n" % datetime.utcnow().isoformat()) lf.write(response_text + "\n\n") except Exception as e: logger.debug(f"Could not persist AI raw response to instance log: {e}") # Prefer JSON array responses try: parsed_arr = json.loads(response_text) if isinstance(parsed_arr, list) and len(parsed_arr) >= 1: question_texts = [str(x).strip() for x in parsed_arr if str(x).strip()] except Exception: # fallback to line-splitting question_texts = [line.strip() for line in response_text.split('\n') if line.strip()] # Detect generic 'read aloud' prompts and rewrite them from reading content if present generic_patterns = [ r"read and speak", r"read aloud", r"repeat after me", r"speak these sentences", r"listen and repeat", r"read.*loudly", r"speak.*loudly", r"read.*out loud", r"practice reading", r"read the following", r"pronounce.*sentences" ] rewritten = False for i, q in enumerate(list(question_texts)): if any(re.search(pat, q, re.I) for pat in generic_patterns): logger.warning(f"Detected generic read-aloud prompt: '{q}' - rewriting using topic content") # Create "Tell me" replacement prompts based on content try: # Extract key topics from content for "Tell me" questions text_lower = text.lower() if any(word in text_lower for word in ['university', 'faculty', 'department', 'study']): replacement = "Tell me about your faculty and department." elif any(word in text_lower for word in ['family', 'parents', 'relatives']): replacement = "Tell me about your family background." elif any(word in text_lower for word in ['hobby', 'interests', 'activities']): replacement = "Tell me about your hobbies and interests." elif any(word in text_lower for word in ['work', 'job', 'career']): replacement = "Tell me about your career goals." else: replacement = "Tell me about your thoughts on this topic." question_texts[i] = replacement rewritten = True logger.info(f"Rewrote generic prompt to: '{replacement}'") except Exception: question_texts[i] = "Tell me about your opinion on this topic." rewritten = True if rewritten: logger.info("Completed rewriting generic prompts") logger.debug("Final prompts after rewriting: %s", question_texts) # If after rewriting we still have generic or no useful prompts, derive prompts locally def _contains_generic(prompts): generic_indicators = ['read and speak', 'read aloud', 'repeat after me', 'speak these sentences', 'listen and repeat'] for q in prompts: # Check for generic phrases if any(indicator.lower() in q.lower() for indicator in generic_indicators): return True # Check if question doesn't start with "Tell me" if not q.strip().lower().startswith('tell me'): return True return False if (not question_texts or _contains_generic(question_texts)) and text: try: logger.warning("AI returned generic or non-'Tell me' prompts, using local derivation from reading content") question_texts = self._derive_prompts_from_text(text, n=5) logger.debug("Locally derived prompts: %s", question_texts) except Exception as e: logger.error(f"Local prompt derivation failed: {e}") except Exception as e: logger.warning(f"AI generate_speaking_questions failed: {e}") question_texts = [] # If no reading content or AI failed completely, use local fallback if not question_texts: if text: # We have content but AI failed - use local derivation try: logger.info("No AI prompts generated, falling back to local content-based generation") question_texts = self._derive_prompts_from_text(text, n=5) except Exception as e: logger.error(f"Local fallback failed: {e}") # Use generic fallback questions question_texts = [ "Tell me about yourself and your interests.", "Tell me about your hobbies and activities.", "Tell me about your daily routine and lifestyle.", "Tell me about your goals and future plans.", "Tell me about what you enjoy learning in English." ] else: # No reading content - determine topic type and use appropriate questions logger.warning("No reading content provided, using topic-based fallback questions") # Check if this might be a "myself" topic based on common patterns topic_name = str(response_text_or_topic).lower() if response_text_or_topic else '' if 'myself' in topic_name or 'introduce' in topic_name or 'about me' in topic_name: # Create varied "myself" questions all starting with "Tell me" myself_questions = [ "Tell me about yourself and your background.", "Tell me about your faculty and department.", "Tell me about your family and where you're from.", "Tell me about your hobbies and interests.", "Tell me about your goals for learning English.", "Tell me about your daily routine and activities.", "Tell me about your hometown and culture.", "Tell me about your dreams and future plans.", "Tell me about what makes you unique.", "Tell me about your favorite subjects and studies." ] # Randomly select 5 questions for variety question_texts = random.sample(myself_questions, 5) else: # Generic topic questions with "Tell me" format generic_questions = [ "Tell me about your opinion on this topic.", "Tell me about how this subject relates to your life.", "Tell me about what you learned from this content.", "Tell me about why this topic interests you.", "Tell me about your experience with this subject.", "Tell me about what questions you have on this topic.", "Tell me about how you would explain this to someone.", "Tell me about what you find most important here.", "Tell me about your thoughts on this subject.", "Tell me about how this topic affects people today." ] question_texts = random.sample(generic_questions, 5) # Remove numbering if present (e.g., "1. What is...") question_texts = [re.sub(r'^\d+\.\s*', '', q) for q in question_texts] # Remove duplicates while preserving order seen = set() unique_texts = [] for q in question_texts: if q.lower() not in seen: unique_texts.append(q) seen.add(q.lower()) # Ensure we have exactly 5 unique questions while len(unique_texts) < 5: filler = "What do you think about this topic?" if filler.lower() not in seen: unique_texts.append(filler) seen.add(filler.lower()) else: # Add a numbered filler if needed unique_texts.append("Share something about yourself.") limited_texts = unique_texts[:5] questions = [] for i, text in enumerate(limited_texts): expected_duration = 30 # Default 30 seconds for beginners if len(text.split()) < 5: expected_duration = 20 elif len(text.split()) > 10: expected_duration = 45 question = SpeakingQuestion( text=text, expected_duration=expected_duration ) questions.append(question) return questions def _generate_fallback_speaking_questions(self, reading_content): """Generate basic speaking questions when AI is unavailable.""" try: # Ensure content is personalized (in case this method is called directly) reading_content = self.personalize_content(reading_content) # Extract topic and keywords from reading content text = reading_content.lower() sentences = sent_tokenize(text) words = word_tokenize(text) # Try to determine the general topic common_topics = { 'introduction': ['introduce', 'myself', 'name', 'background'], 'education': ['school', 'university', 'study', 'learn', 'education', 'student', 'teacher', 'classroom'], 'work': ['job', 'career', 'profession', 'work', 'company', 'business'], 'family': ['family', 'parent', 'child', 'brother', 'sister', 'mother', 'father'], 'travel': ['travel', 'trip', 'journey', 'country', 'city', 'visit', 'tourist'], 'hobbies': ['hobby', 'interest', 'free time', 'leisure', 'activity', 'enjoy'], 'technology': ['technology', 'computer', 'internet', 'phone', 'app', 'digital'], 'health': ['health', 'exercise', 'diet', 'fitness', 'medical'], 'environment': ['environment', 'climate', 'nature', 'pollution', 'sustainable'], 'culture': ['culture', 'tradition', 'custom', 'celebration', 'festival'] } topic_scores = {topic: 0 for topic in common_topics} for topic, keywords in common_topics.items(): for keyword in keywords: if keyword in text: topic_scores[topic] += 1 # Default topic in case nothing matches main_topic = max(topic_scores.items(), key=lambda x: x[1])[0] if any(topic_scores.values()) else 'general' # Create generic questions based on topic generic_questions = { 'introduction': [ "What is your name?", "Where are you from?", "How old are you?", "What do you like to do?", "Tell me about your family." ], 'education': [ "Which faculty are you studying in? (For example: Faculty of Engineering, Faculty of Arts)", "What is the name of your department?", "What subjects do you study in your department?", "What is your class schedule like? Tell us about your daily classes.", "Do you study in the morning or evening shift?" ], 'work': [ "Do you have a job?", "What job do you want in the future?", "What time do you start work?", "Do you like your job?", "Is your job easy or hard?" ], 'family': [ "Tell me about your family.", "What family traditions do you have?", "How do you spend time with your family?", "What values did you learn from your family?", "Describe a memorable family event or celebration." ], 'travel': [ "Describe a place you've visited that you enjoyed.", "Where would you like to travel in the future and why?", "What do you usually do when you travel to a new place?", "How do you prepare for a trip?", "What was your most memorable travel experience?" ], 'hobbies': [ "What do you like to do for fun?", "Do you play any sports?", "Can you play music?", "What games do you like?", "Do you like to read books?" ], 'technology': [ "How do you use technology in your daily life?", "What technological developments are you most excited about?", "How has technology changed the way you work or study?", "What are the advantages and disadvantages of modern technology?", "Describe your relationship with social media." ], 'health': [ "How do you maintain a healthy lifestyle?", "What do you do to stay physically active?", "Describe your typical diet and eating habits.", "How do you manage stress in your life?", "What changes would you like to make to be healthier?" ], 'environment': [ "What environmental issues concern you the most?", "How do you try to reduce your environmental impact?", "How has the environment changed in your area over time?", "What can individuals do to protect the environment?", "Describe your connection to nature and outdoor activities." ], 'culture': [ "Describe some important aspects of your culture.", "What cultural traditions do you participate in?", "How has your cultural background influenced you?", "What interests you about other cultures?", "Describe a cultural celebration or festival that's important to you." ], 'general': [ "What do you like to do?", "Do you have any pets?", "What is your favorite food?", "What did you do yesterday?", "Do you like to watch TV?" ] } # Get questions for the identified topic topic_questions = generic_questions.get(main_topic, generic_questions['general']) # Reword questions to encourage longer answers (at least 20 seconds) def make_longer(q): return f"{q} Please give a detailed answer and try to speak for at least 20 seconds." seen = set() unique_questions = [] for q in topic_questions: if q.lower() not in seen: unique_questions.append(make_longer(q)) seen.add(q.lower()) # Ensure we have exactly 5 unique questions while len(unique_questions) < 5: filler = make_longer("What do you think about this topic?") if filler.lower() not in seen: unique_questions.append(filler) seen.add(filler.lower()) else: unique_questions.append(make_longer("Share something about yourself.")) limited_questions = unique_questions[:5] questions = [] for i, text in enumerate(limited_questions): expected_duration = 20 # Minimum 20 seconds for all fallback questions question = SpeakingQuestion( text=text, expected_duration=expected_duration ) questions.append(question) return questions except Exception as e: logger.error(f"Error generating fallback speaking questions: {str(e)}") # Return extremely basic questions if all else fails questions = [ SpeakingQuestion(text="What is your name?", expected_duration=20), SpeakingQuestion(text="Where are you from?", expected_duration=20), SpeakingQuestion(text="What do you like to do?", expected_duration=30), SpeakingQuestion(text="Do you have any pets?", expected_duration=30), SpeakingQuestion(text="What is your favorite food?", expected_duration=30) ] return questions def _derive_prompts_from_text(self, text, n=5): """Derive n content-specific speaking prompts from reading text using 'Tell me' format. This is a local fallback when AI returns unusable prompts. Questions are based on SPECIFIC content from the reading, not generic topics. """ prompts = [] try: import re sents = sent_tokenize(text) if not sents: raise ValueError("No sentences found in text") text_lower = text.lower() # Strategy 1: Extract proper nouns (actual names, not common words) # Find capitalized words that are likely proper nouns capitalized_words = re.findall(r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b', text) # Remove common sentence starters and pronouns common_caps = { 'The', 'This', 'That', 'These', 'Those', 'I', 'My', 'He', 'She', 'It', 'They', 'We', 'His', 'Her', 'Their', 'Our', 'You', 'Your', 'There', 'Here', 'When', 'Where', 'What', 'Who', 'Why', 'How', 'A', 'An', 'Person', 'People', 'Thing', 'Things', 'Place', 'Places' # Generic words } proper_nouns = [] for word in capitalized_words: # Keep only if it's not a common word and appears to be a name/place if word not in common_caps and len(word) > 2: proper_nouns.append(word) # Remove duplicates while preserving order proper_nouns = list(dict.fromkeys(proper_nouns)) # Strategy 2: Extract key information from first sentence first_sentence = sents[0].strip() # Look for patterns like "X is a/an [role]" or "X studies/works at [place]" # Example: "John is a student at MIT" name_role_pattern = r'\b([A-Z][a-z]+)\s+(?:is|was)\s+(?:a|an)\s+(\w+)' matches = re.findall(name_role_pattern, first_sentence) if matches: name, role = matches[0] if name not in common_caps: prompts.append(f"Tell me about {name}.") if role not in ['person', 'thing']: prompts.append(f"Tell me what {name} does.") # Strategy 3: Look for location/institution mentions institution_pattern = r'\b(?:at|in|from)\s+([A-Z][A-Z\s]+|[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*)\b' institutions = re.findall(institution_pattern, text) institutions = [inst.strip() for inst in institutions if inst.strip() not in common_caps and len(inst.strip()) > 2] if institutions: inst = institutions[0] prompts.append(f"Tell me about {inst}.") # Strategy 4: Extract subjects/topics from key sentences # Look for "about X", "X is", "X are" patterns topic_patterns = [ r'(?:about|regarding|concerning)\s+([\w\s]+?)(?:\.|,|\s+(?:and|or|is|are|was|were))', r'^([A-Z][a-z]+(?:\s+[a-z]+){0,3})\s+(?:is|are|was|were)', ] for pattern in topic_patterns: for sent in sents[:3]: matches = re.findall(pattern, sent) for match in matches: topic = match.strip() # Filter out very short or generic topics if len(topic.split()) >= 2 and len(topic) > 10: if not any(generic in topic.lower() for generic in ['person', 'people', 'thing', 'this', 'that']): prompts.append(f"Tell me about {topic}.") break # Strategy 5: Use proper nouns found earlier for noun in proper_nouns[:2]: if len(prompts) < n: prompts.append(f"Tell me about {noun}.") # Strategy 6: Extract main subject from first few sentences for sent in sents[:min(3, len(sents))]: if len(prompts) >= n: break # Simple subject extraction - words before "is/are/was/were/has/have" words = sent.split() for i, word in enumerate(words): if word.lower() in ['is', 'are', 'was', 'were', 'has', 'have'] and i > 0: # Get subject (2-4 words before the verb) start_idx = max(0, i - 3) subject = ' '.join(words[start_idx:i]).strip() # Clean up subject subject = re.sub(r'^(the|a|an)\s+', '', subject, flags=re.IGNORECASE) # Only use if it's meaningful and not too generic if (len(subject.split()) >= 1 and len(subject) > 5 and not any(generic in subject.lower() for generic in ['person', 'people', 'thing', 'this', 'that', 'there'])): prompts.append(f"Tell me about {subject}.") break # Remove duplicates and limit to n questions prompts = list(dict.fromkeys(prompts))[:n] # If we still don't have enough questions, add content-based fallbacks general_content_questions = [ "Tell me about the main topic of this reading.", "Tell me what you learned from this reading.", "Tell me about the important information in the reading.", "Tell me about what the reading is discussing.", "Tell me about the key points from the reading." ] # Fill remaining slots with content-based questions while len(prompts) < n: if general_content_questions: prompts.append(general_content_questions.pop(0)) else: break return prompts[:n] except Exception as e: logger.error(f"Error deriving prompts from text: {e}") # Last-resort content-based "Tell me" questions fallback_prompts = [ "Tell me about the main topic of the reading.", "Tell me what the reading is about.", "Tell me about the information in the reading.", "Tell me what you understand from the reading.", "Tell me about the content you just read." ] return fallback_prompts[:n] def validate_audio_quality(self, audio_path): """ Validate audio quality before processing Args: audio_path (str): Path to the audio file Returns: dict: Validation result with status and issues """ try: # Check if file exists if not os.path.exists(audio_path): return { 'valid': False, 'issues': ['file_not_found'], 'message': 'Audio file not found. Please try recording again.' } # Check file size (minimum 1KB for meaningful audio - very lenient) file_size = os.path.getsize(audio_path) if file_size < 1000: # 1KB - very small, likely empty return { 'valid': False, 'issues': ['file_too_small'], 'message': f'Audio file too small ({file_size} bytes). Please ensure you speak for at least 2-3 seconds.' } logger.info(f"Validating audio file: {audio_path} ({file_size} bytes)") # Try to load audio and check duration try: audio = AudioSegment.from_file(audio_path) duration = len(audio) / 1000.0 # Convert to seconds # Check duration (minimum 1 second, maximum 120 seconds) if duration < 1.0: return { 'valid': False, 'issues': ['duration_too_short'], 'message': f'Audio too short ({duration:.1f}s). Please speak for at least 2-3 seconds.' } if duration > 120.0: return { 'valid': False, 'issues': ['duration_too_long'], 'message': f'Audio too long ({duration:.1f}s). Please keep your response under 2 minutes.' } # Check if audio has sufficient loudness (detect complete silence only) # Use dBFS (decibels relative to full scale) for better detection try: # Get the maximum amplitude in dBFS max_dbfs = audio.max_dBFS # Only reject if audio is extremely quiet (likely complete silence) # -60 dBFS is very quiet but still has some audio # -inf means complete silence (no audio at all) if max_dbfs == float('-inf') or max_dbfs < -60: logger.warning(f"Audio very quiet: {max_dbfs} dBFS") return { 'valid': False, 'issues': ['audio_too_quiet'], 'message': 'No speech detected or audio too quiet. Please speak louder and closer to the microphone.' } # Log the audio level for debugging logger.info(f"Audio validation passed: {max_dbfs} dBFS, duration: {duration}s, size: {file_size} bytes") except Exception as e: # If we can't measure loudness, just check for complete silence using RMS logger.warning(f"Could not measure dBFS, using RMS fallback: {e}") samples = audio.get_array_of_samples() if len(samples) > 0: import math # Calculate RMS from a larger sample (50k samples) for better accuracy sample_size = min(len(samples), 50000) sum_squares = sum(sample ** 2 for sample in samples[:sample_size]) rms = math.sqrt(sum_squares / sample_size) # Much more lenient threshold - only reject complete silence # Normal speech typically has RMS > 10, whispers > 5 if rms < 3: # Only reject if essentially silent logger.warning(f"Audio RMS too low: {rms}") return { 'valid': False, 'issues': ['audio_too_quiet'], 'message': 'No speech detected or audio too quiet. Please speak louder and closer to the microphone.' } logger.info(f"Audio validation passed: RMS={rms}, duration={duration}s") # All checks passed return { 'valid': True, 'duration': duration, 'file_size': file_size, 'message': 'Audio quality validated successfully' } except Exception as e: logger.error(f"Error loading audio file: {str(e)}") return { 'valid': False, 'issues': ['invalid_format'], 'message': 'Invalid audio format. Please try recording again.' } except Exception as e: logger.error(f"Error validating audio quality: {str(e)}") return { 'valid': False, 'issues': ['validation_error'], 'message': 'Error validating audio. Please try again.' } def transcribe_with_whisper(self, audio_path): """ Transcribe audio using OpenAI Whisper API (if available) This provides superior transcription accuracy compared to other methods Args: audio_path (str): Path to the audio file Returns: dict: Transcription result with text and confidence """ try: # Check if OpenAI API key is configured openai_api_key = current_app.config.get('OPENAI_API_KEY') if not openai_api_key or openai_api_key == 'your-openai-api-key-here': logger.info("OpenAI API key not configured, skipping Whisper transcription") return None try: import openai openai.api_key = openai_api_key # Open audio file with open(audio_path, 'rb') as audio_file: # Call Whisper API transcript = openai.Audio.transcribe( model="whisper-1", file=audio_file, language="en", response_format="verbose_json" # Get detailed info including confidence ) # Extract transcription and confidence transcription_text = transcript.get('text', '') # Whisper doesn't provide direct confidence, but we can estimate from other factors # If transcription is long and coherent, confidence is likely high word_count = len(transcription_text.split()) estimated_confidence = min(0.95, 0.7 + (word_count * 0.02)) # Higher for longer responses logger.info(f"Whisper transcription successful: '{transcription_text[:50]}...' (confidence: {estimated_confidence:.2f})") return { 'text': transcription_text, 'confidence': estimated_confidence, 'source': 'whisper' } except ImportError: logger.warning("OpenAI library not installed. Install with: pip install openai") return None except Exception as e: logger.warning(f"Whisper transcription failed: {str(e)}") return None except Exception as e: logger.error(f"Error in Whisper transcription: {str(e)}") return None def preprocess_audio(self, audio_path): """ Preprocess audio to improve transcription accuracy Args: audio_path (str): Path to the audio file Returns: str: Path to the preprocessed audio file """ try: logger.info(f"Preprocessing audio: {audio_path}") # Load audio audio = AudioSegment.from_file(audio_path) # 1. Normalize audio (adjust volume to optimal level) normalized = audio.normalize() logger.debug("Audio normalized") # 2. Remove silence from beginning and end # Find non-silent parts (silence threshold: -40 dBFS) nonsilent_ranges = [] silence_thresh = normalized.dBFS - 14 # Adaptive threshold min_silence_len = 300 # 300ms # Simple trim: remove leading/trailing silence start_trim = 0 end_trim = len(normalized) # Find first non-silent segment for i in range(0, len(normalized), 100): # Check every 100ms chunk = normalized[i:i+100] if chunk.dBFS > silence_thresh: start_trim = max(0, i - 100) # Include 100ms before speech break # Find last non-silent segment for i in range(len(normalized), 0, -100): chunk = normalized[max(0, i-100):i] if chunk.dBFS > silence_thresh: end_trim = min(len(normalized), i + 100) # Include 100ms after speech break trimmed = normalized[start_trim:end_trim] logger.debug(f"Trimmed silence: {start_trim}ms to {end_trim}ms") # 3. Convert to optimal format for transcription # 16kHz mono is ideal for speech recognition optimized = trimmed.set_frame_rate(16000).set_channels(1) logger.debug("Converted to 16kHz mono") # 4. Apply slight compression to reduce dynamic range # This helps with quiet/loud variations compressed = optimized.compress_dynamic_range( threshold=-20.0, ratio=3.0, attack=5.0, release=50.0 ) logger.debug("Applied dynamic range compression") # 5. Save preprocessed audio preprocessed_path = audio_path.replace('.webm', '_preprocessed.wav').replace('.wav', '_preprocessed.wav') compressed.export(preprocessed_path, format='wav') logger.info(f"Preprocessed audio saved: {preprocessed_path}") return preprocessed_path except Exception as e: logger.warning(f"Error preprocessing audio: {str(e)}. Using original file.") return audio_path # Return original if preprocessing fails def assess_speaking(self, audio_path, question_text, topic_content=None): """ Assess a student's speaking response to a question Args: audio_path (str): Path to the audio file question_text (str): The question text that was asked topic_content (str): Optional topic content for context Returns: dict: Assessment result with scores and feedback """ try: logger.info(f"Starting speaking assessment for audio: {audio_path}") # Step 1: Validate audio quality (lenient - only reject obvious problems) validation_result = self.validate_audio_quality(audio_path) if not validation_result['valid']: logger.warning(f"Audio validation failed: {validation_result['message']}") # Don't immediately fail - try to process anyway unless it's a critical error critical_issues = ['file_not_found', 'invalid_format'] issues = validation_result.get('issues', []) if any(issue in critical_issues for issue in issues): # Only fail for critical issues logger.error(f"Critical audio validation error: {issues}") return { 'audio_detected': False, 'transcription': "", 'scores': { 'pronunciation': 0, 'fluency': 0, 'grammar': 0, 'vocabulary': 0, 'relevance': 0, 'overall': 0 }, 'accent': { 'type': 'Unknown', 'confidence': 0 }, 'feedback': { 'general': validation_result['message'], 'pronunciation': "Unable to assess pronunciation. " + validation_result['message'], 'fluency': "Unable to assess fluency. " + validation_result['message'], 'grammar': "Unable to assess grammar. " + validation_result['message'], 'vocabulary': "Unable to assess vocabulary. " + validation_result['message'], 'accent': "Unable to detect accent. " + validation_result['message'] }, 'validation_failed': True, 'validation_issues': validation_result.get('issues', []) } else: # Non-critical issue (e.g., audio_too_quiet, duration_too_short) # Log warning but continue processing - let AI try to transcribe logger.warning(f"Non-critical validation issue: {issues}. Attempting to process anyway.") if validation_result['valid']: logger.info(f"Audio validation passed: {validation_result['message']} (Duration: {validation_result.get('duration', 0):.1f}s, Size: {validation_result.get('file_size', 0)} bytes)") else: logger.info(f"Proceeding with audio processing despite validation warning: {validation_result.get('issues', [])}") # Step 2: Preprocess audio to improve quality try: preprocessed_path = self.preprocess_audio(audio_path) logger.info(f"Using preprocessed audio: {preprocessed_path}") # Use preprocessed audio for evaluation audio_path = preprocessed_path except Exception as e: logger.warning(f"Audio preprocessing failed: {str(e)}. Using original audio.") # Continue with original audio if preprocessing fails # Step 3: Transcription - Try multiple methods in order of accuracy transcription = None transcription_confidence = 0 transcription_source = None # Method 1: Try Whisper API (most accurate if available) whisper_result = self.transcribe_with_whisper(audio_path) if whisper_result: transcription = whisper_result['text'] transcription_confidence = whisper_result['confidence'] transcription_source = 'whisper' self._last_transcription_confidence = transcription_confidence logger.info(f"Using Whisper transcription (confidence: {transcription_confidence:.2f})") # Method 2: Try AI multimodal if Whisper unavailable if not transcription: use_ai = getattr(self, '_use_ai_transcription', False) genai_configured = self._configure_genai() model_available = self.get_model() is not None logger.debug(f"AI transcription decision: use_ai={use_ai}, genai_configured={genai_configured}, model_available={model_available}") if use_ai and genai_configured and model_available: try: logger.info("Using AI multimodal transcription + assessment for audio") ai_result = self._ai_transcribe_and_assess(audio_path, question_text, topic_content) logger.debug(f"AI multimodal result: {ai_result}") if ai_result and isinstance(ai_result, dict): # Ensure required fields and return directly ai_result.setdefault('audio_detected', True) ai_result.setdefault('transcription', ai_result.get('transcription') or "") ai_result.setdefault('scores', ai_result.get('scores') or {}) ai_result.setdefault('feedback', ai_result.get('feedback') or {}) ai_result.setdefault('accent', ai_result.get('accent') or {'type': 'Unknown', 'confidence': 0}) logger.info(f"Returning AI assessment with transcription: '{ai_result.get('transcription', '')[:50]}...'") return ai_result else: logger.warning('AI multimodal assessment returned no usable result; falling back to local transcription') except Exception as e: logger.warning(f"AI multimodal assessment failed: {str(e)}; falling back to local transcription") else: logger.warning(f"AI transcription not used - conditions not met") # Method 3: Fall back to local transcription (least accurate) if not transcription: logger.info("Using local fallback transcription") transcription = self._transcribe_audio(audio_path) transcription_source = 'local' # Local transcription has lower default confidence if transcription: self._last_transcription_confidence = 0.6 # Diagnostic logging: record transcription details for debugging try: size_info = os.path.getsize(audio_path) if os.path.exists(audio_path) else None # build a safe transcription preview to avoid nested f-string parentheses if transcription: preview = (transcription[:120] + '...') if len(transcription) > 120 else transcription tlen = len(transcription) else: preview = '' tlen = 0 logger.debug("Transcription debug - file: %s, size: %s, transcription_len: %s, transcription_preview: '%s', confidence: %s", audio_path, size_info, tlen, preview, getattr(self, '_last_transcription_confidence', None)) except Exception: pass if not transcription or transcription == "": # Check if the audio file exists and has content if not os.path.exists(audio_path) or os.path.getsize(audio_path) < 100: logger.warning(f"Audio file missing or too small: {audio_path}, size: {os.path.getsize(audio_path) if os.path.exists(audio_path) else 'N/A'}") error_reason = "The audio recording was too short or failed to save properly." else: logger.warning(f"No transcription found in audio file: {audio_path}, size: {os.path.getsize(audio_path)}") error_reason = "Unable to process the audio. This might be due to background noise, unclear speech, or audio format issues. Please try recording again in a quiet environment." return { 'audio_detected': False, 'transcription': "", 'scores': { 'pronunciation': 0, 'fluency': 0, 'grammar': 0, 'vocabulary': 0, 'relevance': 0, 'overall': 0 }, 'accent': { 'type': 'Asian', # Default for beginners 'confidence': 0 }, 'feedback': { 'general': f"No speech was detected. {error_reason} Try speaking closer to the microphone and in a quiet environment.", 'pronunciation': "Pronunciation score: 0%. Unable to assess pronunciation without speech.", 'fluency': "Fluency score: 0%. Unable to assess fluency without speech.", 'grammar': "Grammar score: 0%. Unable to assess grammar without speech.", 'vocabulary': "Vocabulary score: 0%. Unable to assess vocabulary without speech.", 'accent': "Accent: Asian. Unable to detect accent without speech." } } logger.info(f"Transcription successful: '{transcription[:50]}...'") # If transcription confidence is low, avoid running AI evaluation and return conservative feedback try: conf = getattr(self, '_last_transcription_confidence', None) # IMPROVED: Increased confidence threshold from 0.5 to 0.7 for better accuracy if conf is not None and conf < 0.7: logger.warning(f"Low transcription confidence ({conf}), returning conservative assessment") return { 'audio_detected': True, 'transcription': transcription, 'scores': { 'pronunciation': 45, 'fluency': 40, 'grammar': 35, 'vocabulary': 40, 'relevance': 30, 'overall': 38 }, 'accent': { 'type': 'Asian', 'confidence': int(conf * 100) if conf is not None else 0 }, 'feedback': { 'general': "We couldn't confidently transcribe your response. Please try again in a quieter environment and speak clearly.", 'pronunciation': "Pronunciation appears to need improvement. Try speaking more slowly and enunciating.", 'fluency': "Fluency could be improved. Pause less between phrases and try to maintain a steady rhythm.", 'grammar': "Grammar could not be fully assessed due to transcription uncertainty.", 'vocabulary': "Try to include specific department names or course-related terms.", 'accent': "Accent detection is inconclusive due to low transcription confidence." } } except Exception: pass # Step 2: Check if AI is available, if not use simpler assessment if not self._configure_genai() or not self.get_model(): logger.warning("Mistral API not available, using fallback speaking assessment") return self._assess_speaking_fallback(transcription, question_text, topic_content) # Step 3: Generate AI assessment model = self.get_model() # Create the prompt with stricter relevance scoring # Validate transcription quality words = transcription.split() if len(words) < 3 or not any(word.lower() in ['department', 'study', 'course', 'faculty', 'major'] for word in words): logger.warning(f"Low quality or irrelevant transcription detected: {transcription}") return { 'audio_detected': True, 'transcription': transcription, 'scores': { 'pronunciation': 50, 'fluency': 40, 'grammar': 30, 'vocabulary': 40, 'relevance': 20, 'overall': 35 }, 'accent': { 'type': 'Asian', 'confidence': 50 }, 'feedback': { 'general': "Your response was unclear or may not directly answer the question. Please try speaking more clearly and directly answer what department you are in.", 'pronunciation': "Pronunciation score: 50%. Some words were unclear. Try speaking more slowly and clearly.", 'fluency': "Fluency score: 40%. The response was difficult to follow. Try organizing your thoughts before speaking.", 'grammar': "Grammar score: 30%. The response had some structural issues. Try using complete sentences.", 'vocabulary': "Vocabulary score: 40%. Try using specific terms related to your department.", 'accent': "Keep practicing clear pronunciation of department and course names." } } prompt = f""" You are an English teacher evaluating a student's speaking response about their academic department. Be accurate but encouraging in your assessment. QUESTION: {question_text} RESPONSE TRANSCRIPTION: {transcription} Evaluation Requirements: 1. Verify if the response actually mentions a department or field of study 2. Check for proper sentence structure (subject + verb + object) 3. Look for specific grammar issues like: - Subject-verb agreement ("I am studying" vs "I am study") - Article usage ("I am in the Department of" vs "I am in Department of") - Verb tense consistency 4. Assess response completeness and relevance to the department question {f'TOPIC CONTENT FOR CONTEXT: {topic_content[:500]}...' if topic_content else ''} STRICT ASSESSMENT GUIDELINES: 1. GRAMMAR ACCURACY: - Base score on actual grammar correctness - Deduct points for: * Wrong verb forms (e.g., "I study" vs "I am studying") * Missing articles * Incorrect subject-verb agreement - Provide specific corrections for each error 2. RELEVANCE TO DEPARTMENT QUESTION: - Must mention specific department/field - Must use proper academic terminology - Should provide some detail about their studies 3. RESPONSE COMPLETENESS: - Check if all parts of question are addressed - Verify response length meets 20-second requirement - Look for supporting details 4. SCORING GUIDELINES: - Grammar: Score based on actual errors (not generosity) - Relevance: Must specifically name department for high score - Fluency: Consider natural flow and hesitation - Pronunciation: Focus on clarity of key terms 5. MINIMUM SCORING GUIDELINES: - Any comprehensible answer that addresses the question: minimum 60 points - Grammatically correct short answer: minimum 70 points - Complete sentence with correct grammar: minimum 80 points - Only give scores below 50 for completely unrelated or unintelligible responses 6. LANGUAGE SKILLS ASSESSMENT (Be encouraging): - Grammar: Focus on what they got right, gently note improvements. IN THE GRAMMAR ANALYSIS SECTION, CLEARLY POINT OUT ANY MISTAKES IN GRAMMAR STRUCTURE FOUND IN THE STUDENT'S RESPONSE, AND PROVIDE THE CORRECT SENTENCE(S) ACCORDING TO ENGLISH GRAMMAR RULES. - Vocabulary: Praise their word choices, suggest expansions positively - Fluency: Encourage natural speech, don't penalize hesitation - Pronunciation: Be supportive, focus on understandable communication Please evaluate the response and provide scores from 0-100 for each: 1. Pronunciation: How clearly can the student communicate? (Be generous - communication is key!) 2. Fluency: How naturally does the student express themselves? (Short answers can still be fluent!) PRONUNCIATION ANALYSIS - Provide specific, actionable feedback: - State the overall clarity level (e.g., "clear", "mostly clear", "needs improvement") - Identify specific sounds or words that were difficult to understand (if any) - Give one concrete tip: e.g., "Focus on pronouncing the 'th' sound in 'the'" or "Slow down when saying technical terms" - Example: "Your pronunciation was mostly clear (80%). The word 'department' was slightly unclear. Practice saying 'de-PART-ment' with emphasis on the second syllable." FLUENCY ANALYSIS - Provide specific, actionable feedback: - Describe the speech rhythm (e.g., "smooth and natural", "some hesitation", "choppy") - Note specific issues: long pauses, filler words (um, uh), repetition - Give one concrete tip: e.g., "Reduce the 2-second pauses between words" or "Try recording your answer first to organize thoughts" - Example: "You spoke smoothly (75%) with only minor hesitation. There were 2 noticeable pauses before 'engineering'. Practice your answer once before recording to reduce hesitation." GRAMMAR ANALYSIS - Provide specific, actionable feedback: - State the grammar accuracy level (e.g., "grammatically correct", "minor errors", "needs work") - IDENTIFY EXACT MISTAKES: Point out specific grammar errors found in the response - PROVIDE CORRECTIONS: Show the correct version of incorrect sentences - Example format: "Good grammar overall (85%). Error found: 'I study in Computer Science' should be 'I study Computer Science' or 'I am in the Computer Science department'. The preposition 'in' is incorrect with 'study'." VOCABULARY ANALYSIS - Provide specific, actionable feedback: - Evaluate word choice appropriateness for the context - Identify if specific terminology was used or missing - Suggest 1-2 better word choices or additional terms to include - Example: "Good vocabulary (80%). You used 'department' correctly. Consider adding terms like 'major', 'specialization', or your specific field name for more precision." RELEVANCE ANALYSIS - Provide specific, actionable feedback: - State how well the response answers the specific question asked - Identify what was included and what was missing - Suggest what to add for a complete answer - Example: "Partially addressed (65%). You mentioned your department but didn't explain what you study there. Add 1-2 sentences about your courses or research focus." IMPROVEMENTS NEEDED - Provide the most important actionable suggestion: - Identify the weakest area based on scores - Give ONE specific, practical action to improve - Make it achievable: e.g., "Practice saying [specific word] 5 times slowly" or "Write out your answer first, then speak it" - Example: "Focus on grammar accuracy. Before speaking, mentally form complete sentences: Subject + Verb + Object. Practice: 'I am studying [your field]' instead of 'I study in [field]'." Format your response as a JSON object with these EXACT fields: { "pronunciation_score": number (0-100, be fair but encouraging), "fluency_score": number (0-100, be fair but encouraging), "grammar_score": number (0-100, be fair but encouraging), "vocabulary_score": number (0-100, be fair but encouraging), "relevance_score": number (0-100, strict scoring - low for off-topic/incomplete), "overall_score": number (average of all 5 scores), "accent_type": "American" | "European" | "Asian", "accent_confidence": number (0-100), "detailed_feedback": { "pronunciation": "2-3 sentences with: clarity level + specific issue + concrete tip", "fluency": "2-3 sentences with: rhythm description + specific issue + concrete tip", "grammar": "2-3 sentences with: accuracy level + exact mistakes + correct versions", "vocabulary": "2-3 sentences with: word choice evaluation + missing terms + suggestions", "relevance": "2-3 sentences with: how well question answered + what's missing + what to add", "accent": "1-2 sentences with positive observation about accent" }, "improvements_needed": "2-3 sentences with: weakest area + one specific action + why it helps", "general_feedback": "2-3 sentences with: overall performance + main strength + key area to work on" } """ # Generate assessment response = model.generate_content(prompt) response_text = response.text.strip() # Extract JSON from response json_match = re.search(r'```json\s*(.*?)\s*```', response_text, re.DOTALL) if json_match: json_str = json_match.group(1) else: json_str = response_text # Clean up any non-JSON text json_str = re.sub(r'^[\s\S]*?\{', '{', json_str) json_str = re.sub(r'\}[\s\S]*?$', '}', json_str) # Parse JSON and return assessment assessment = json.loads(json_str) # Add the transcription to the assessment assessment['transcription'] = transcription assessment['audio_detected'] = True # Restructure the assessment to match the expected format with pronunciation, fluency, and accent formatted_assessment = { 'transcription': transcription, 'audio_detected': True, 'scores': { 'pronunciation': assessment.get('pronunciation_score', 0), 'fluency': assessment.get('fluency_score', 0), 'grammar': assessment.get('grammar_score', 0), 'vocabulary': assessment.get('vocabulary_score', 0), 'relevance': assessment.get('relevance_score', 0), 'overall': assessment.get('overall_score', 0) }, 'accent': { 'type': assessment.get('accent_type', 'Other'), 'confidence': assessment.get('accent_confidence', 0) }, 'feedback': { 'general': assessment.get('general_feedback', 'Assessment completed successfully.'), 'pronunciation': assessment.get('detailed_feedback', {}).get('pronunciation', f"Pronunciation score: {assessment.get('pronunciation_score', 0)}%. Focus on clear articulation."), 'fluency': assessment.get('detailed_feedback', {}).get('fluency', f"Fluency score: {assessment.get('fluency_score', 0)}%. Work on speaking rhythm and pace."), 'grammar': assessment.get('detailed_feedback', {}).get('grammar', f"Grammar score: {assessment.get('grammar_score', 0)}%. Focus on verb tenses and sentence structure."), 'vocabulary': assessment.get('detailed_feedback', {}).get('vocabulary', f"Vocabulary score: {assessment.get('vocabulary_score', 0)}%. Continue expanding your vocabulary range."), 'accent': assessment.get('detailed_feedback', {}).get('accent', f"Detected accent: {assessment.get('accent_type', 'Other')} ({assessment.get('accent_confidence', 0)}% confidence)") } } return formatted_assessment except Exception as e: logger.error(f"Error in speaking assessment: {str(e)}") return self._assess_speaking_fallback( self._transcribe_audio(audio_path) or "No transcription available", question_text, topic_content ) def _assess_speaking_fallback(self, transcription, question_text, topic_content=None): """ Encouraging fallback method for speaking assessment when AI is unavailable. This method is designed to be supportive and give generous scores to build student confidence. Args: transcription (str): Transcribed text from speech question_text (str): The question text that was asked topic_content (str): Optional topic content for context Returns: dict: Assessment result with encouraging scores and positive feedback """ try: # Simple word count-based assessment word_count = len(transcription.split()) # Basic scoring based on word count - BE GENEROUS! if word_count == 0: return { 'audio_detected': False, 'transcription': transcription, 'scores': { 'pronunciation': 0, 'fluency': 0, 'grammar': 0, 'vocabulary': 0, 'relevance': 0, 'overall': 0 }, 'accent': { 'type': 'Asian', # Default for beginners 'confidence': 0 }, 'feedback': { 'general': "No speech was detected. Please ensure your microphone is working and speak clearly.", 'pronunciation': "Pronunciation score: 0%. Unable to assess pronunciation without clear speech.", 'fluency': "Fluency score: 0%. Unable to assess fluency without speech.", 'grammar': "Grammar score: 0%. Unable to assess grammar without speech.", 'vocabulary': "Vocabulary score: 0%. Unable to assess vocabulary without speech.", 'accent': "Accent: Asian. Unable to detect accent without clear speech." } } # Check if this is a personal information question - be extra encouraging! is_personal_question = any(keyword in question_text.lower() for keyword in ['your name', 'about yourself', 'introduce', 'where are you from', 'your city', 'your family', 'tell me about', 'your age', 'your job', 'your work', 'your studies', 'your hobby', 'your interest']) # GENEROUS SCORING: Give high marks for any reasonable attempt! if is_personal_question: # Personal questions get very high scores for any answer if word_count >= 1: # Even one word deserves good marks! task_completion = max(75, min(95, 60 + word_count * 3)) # 75-95 range else: task_completion = 60 # Still decent for trying else: # For other questions, still be very generous question_keywords = set(word.lower() for word in question_text.split() if len(word) > 3 and word.lower() not in {'what', 'how', 'when', 'where', 'why', 'who', 'the', 'and', 'that', 'with', 'this', 'your', 'you', 'about'}) response_words = set(word.lower() for word in transcription.split()) matching_keywords = question_keywords.intersection(response_words) # Calculate generous task completion score base_score = 70 # Start with good score keyword_bonus = min(25, len(matching_keywords) * 5) # Bonus for keywords length_bonus = min(10, word_count) # Small bonus for length task_completion = min(100, base_score + keyword_bonus + length_bonus) # Set encouraging but brief feedback based on response length if word_count < 3: general_feedback = "Great job speaking English! Try adding a bit more detail next time." elif word_count < 8: general_feedback = "Nice work! You gave a clear answer." elif word_count < 15: general_feedback = "Excellent effort! You're expressing yourself well in English." else: general_feedback = "Fantastic! You provided a detailed and thoughtful response." # TextBlob for basic language assessment try: blob = TextBlob(transcription) # Count sentences sentence_count = len(blob.sentences) # Enhanced grammar assessment grammar_errors = [] grammar_score = 80 # Start with a base score tags = blob.tags words = transcription.split() # Check subject-verb agreement and basic sentence structure for i in range(len(tags) - 1): if tags[i][1].startswith('NN') and tags[i+1][1].startswith('VB'): # Subject-verb agreement checks if tags[i][0].lower() in ['i', 'you', 'we', 'they']: if tags[i+1][0].lower() in ['is', 'was', 'has']: grammar_score -= 10 grammar_errors.append({ 'original': f"{tags[i][0]} {tags[i+1][0]}", 'corrected': f"{tags[i][0]} {'am' if tags[i][0].lower() == 'i' else 'are'}" }) elif tags[i][0].lower() in ['he', 'she', 'it']: if tags[i+1][0].lower() in ['am', 'are', 'were']: grammar_score -= 10 grammar_errors.append({ 'original': f"{tags[i][0]} {tags[i+1][0]}", 'corrected': f"{tags[i][0]} is" }) # Check for incorrect verb forms for i, word in enumerate(words): if word.lower() == "am" and i > 0: prev_word = words[i-1].lower() if prev_word not in ['i']: grammar_score -= 10 grammar_errors.append({ 'original': f"{prev_word} am", 'corrected': f"{prev_word} {'is' if prev_word in ['he', 'she', 'it'] else 'are'}" }) elif word.lower() == "study": if i > 0 and words[i-1].lower() in ['i', 'you', 'we', 'they']: grammar_score -= 0 # Correct usage elif i > 0 and words[i-1].lower() in ['he', 'she', 'it']: grammar_score -= 10 grammar_errors.append({ 'original': f"{words[i-1]} study", 'corrected': f"{words[i-1]} studies" }) # Ensure minimum score grammar_score = max(50, grammar_score) except: grammar_score = 60 # Default if TextBlob fails # Enhanced sentence structure analysis sentence_lengths = [len(str(sentence).split()) for sentence in blob.sentences] if sentence_count > 0 else [0] # Generous sentence structure analysis structure_score = 0 if len(sentence_lengths) == 0: structure_score = 65 # Even fragments show effort elif len(sentence_lengths) == 1: # Single sentence - generous scoring based on length if sentence_lengths[0] < 3: structure_score = 70 # Short but complete elif sentence_lengths[0] < 8: structure_score = 80 # Good sentence length else: structure_score = 85 # Great detailed sentence else: # Multiple sentences - excellent work! length_variety = max(sentence_lengths) - min(sentence_lengths) sentence_count_bonus = min(15, sentence_count * 3) # Bonus for multiple sentences structure_score = min(95, 75 + length_variety * 2 + sentence_count_bonus) # Start higher # Enhanced vocabulary analysis unique_words = set([w.lower() for w in transcription.split()]) unique_ratio = len(unique_words) / max(1, word_count) # Detect if there are some longer words (potential indicator of advanced vocabulary) longer_words = [w for w in unique_words if len(w) > 6] advanced_word_ratio = len(longer_words) / max(1, len(unique_words)) # GENEROUS vocabulary scoring - celebrate word usage! vocabulary_score = max(70, min(95, int((unique_ratio * 30) + (advanced_word_ratio * 25) + 40))) # Calculate encouraging overall score overall_score = max(65, int((grammar_score * 0.3) + (structure_score * 0.3) + (vocabulary_score * 0.25) + (task_completion * 0.15))) # Create encouraging feedback messages grammar_feedback = f"Grammar score: {grammar_score}%. " if grammar_score < 70: grammar_feedback += "You're making good progress! Keep practicing sentence formation - you're on the right track." elif grammar_score < 85: grammar_feedback += "Great grammar foundation! You're using sentences well. Keep building on this success." else: grammar_feedback += "Excellent grammar skills! You're expressing yourself clearly and correctly. Well done!" structure_feedback = f"Structure score: {structure_score}%. " if structure_score < 70: structure_feedback += "Good start with sentence structure! Try connecting your ideas - you're doing well." elif structure_score < 85: structure_feedback += "Nice sentence variety! You're creating clear, well-structured responses. Keep it up!" else: structure_feedback += "Outstanding sentence structure! You're showing excellent variety and complexity. Fantastic work!" vocabulary_feedback = f"Vocabulary score: {vocabulary_score}%. " if vocabulary_score < 75: vocabulary_feedback += f"Good word choices with {len(unique_words)} different words used effectively." elif vocabulary_score < 85: vocabulary_feedback += f"Great variety using {len(unique_words)} different words including some advanced terms." else: vocabulary_feedback += f"Excellent vocabulary range with {len(unique_words)} words showing sophisticated usage." task_feedback = f"Relevance score: {task_completion}%. " if is_personal_question: task_feedback += "Perfect! You answered the personal question beautifully. Sharing about yourself takes courage!" elif task_completion < 75: task_feedback += "Good effort addressing the question! You're on the right track - keep practicing." else: task_feedback += "Excellent response to the question! You understood perfectly and answered thoroughly." # GENEROUS scoring for pronunciation and fluency pronunciation_score = max(70, min(95, 65 + word_count * 2)) # Start at 70, generous increase fluency_score = max(70, min(90, 65 + sentence_count * 5)) # Start at 70, good bonus for sentences # Simple accent detection based on transcription patterns (very basic) accent_type = "Asian" # Default for beginners accent_confidence = 50 # Medium confidence for basic detection # Basic accent hints from common word patterns and spelling preferences text_lower = transcription.lower() if any(word in text_lower for word in ['color', 'center', 'program', 'realized', 'analyzed']): accent_type = "American" accent_confidence = 60 elif any(word in text_lower for word in ['colour', 'centre', 'programme', 'realised', 'analysed']): accent_type = "European" accent_confidence = 60 # Enhanced grammar feedback with error detection using TextBlob try: blob = TextBlob(transcription) grammar_errors = [] # Check for common grammar issues words = transcription.split() # Check for basic subject-verb agreement issues (but be gentle!) for i, word in enumerate(words): if word.lower() in ['i', 'he', 'she', 'it'] and i + 1 < len(words): next_word = words[i + 1].lower() if word.lower() == 'i' and next_word in ['are', 'were']: grammar_errors.append(f"Great effort! Small tip: '{word} {next_word}' could be '{word} am/was' - but your meaning was clear!") elif word.lower() in ['he', 'she', 'it'] and next_word in ['are', 'were']: grammar_errors.append(f"Nice work! Quick suggestion: '{word} {next_word}' could be '{word} is/was' - you're doing great!") # Check for missing articles (but stay positive!) for i, word in enumerate(words[:-1]): if word.lower() in ['have', 'has', 'am', 'is', 'are'] and words[i + 1].lower() in ['student', 'teacher', 'doctor', 'engineer']: grammar_errors.append(f"Good sentence! You could add 'a' to make it '{word} a {words[i + 1]}' - but your meaning is perfect!") # Generate grammar feedback based on errors if grammar_errors: grammar_feedback = f"Grammar score: {grammar_score}%. " corrections = [] for error in grammar_errors: corrections.append(f"Consider using '{error['corrected']}' instead of '{error['original']}'") grammar_feedback += " " + corrections[0] if corrections else "" else: grammar_feedback = f"Grammar score: {grammar_score}%. Your grammar is correct!" except: grammar_feedback = f"Grammar score: {grammar_score}%. You're expressing yourself well in English! Keep practicing and you'll keep improving." overall_score = int((pronunciation_score + fluency_score + grammar_score + vocabulary_score + task_completion) / 5) # Generate improvement suggestions based on lowest scores scores = { 'pronunciation': pronunciation_score, 'fluency': fluency_score, 'grammar': grammar_score, 'vocabulary': vocabulary_score, 'relevance': task_completion } # Find the area with the lowest score for improvement suggestions lowest_area = min(scores, key=scores.get) lowest_score = scores[lowest_area] # Generate improvement suggestion based on the lowest scoring area if lowest_score < 75: if lowest_area == 'pronunciation': improvements_needed = "Practice speaking more slowly and clearly. Focus on pronouncing each word distinctly." elif lowest_area == 'fluency': improvements_needed = "Try speaking in complete sentences without long pauses. Practice connecting your ideas smoothly." elif lowest_area == 'grammar': improvements_needed = "Review basic sentence structures. Practice using correct verb tenses in simple sentences." elif lowest_area == 'vocabulary': improvements_needed = "Try using more varied words in your responses. Learn 2-3 new words each day and practice using them." else: # relevance improvements_needed = "Make sure to directly answer the question asked. Take a moment to understand what's being asked." else: improvements_needed = "Great work overall! Keep practicing speaking English daily to maintain your progress." return { 'audio_detected': True, 'transcription': transcription, 'scores': { 'pronunciation': pronunciation_score, 'fluency': fluency_score, 'grammar': grammar_score, 'vocabulary': vocabulary_score, 'relevance': task_completion, 'overall': overall_score }, 'accent': { 'type': accent_type, 'confidence': accent_confidence }, 'feedback': { 'general': general_feedback, 'pronunciation': f"Pronunciation score: {pronunciation_score}%. Your words are clear and easy to understand.", 'fluency': f"Fluency score: {fluency_score}%. Your speech flows naturally with good rhythm.", 'grammar': grammar_feedback, 'vocabulary': vocabulary_feedback, 'accent': f"Detected accent: {accent_type} ({accent_confidence}% confidence). Your accent adds character to your English!", 'improvements_needed': improvements_needed } } except Exception as e: logger.error(f"Error in fallback speaking assessment: {str(e)}") # Return encouraging assessment in case of errors return { 'audio_detected': True if transcription else False, 'transcription': transcription or "No transcription available", 'scores': { 'pronunciation': 75, # Generous default scores 'fluency': 75, 'grammar': 75, 'vocabulary': 75, 'relevance': 75, 'overall': 75 }, 'accent': { 'type': 'Asian', # Default for beginners 'confidence': 50 }, 'feedback': { 'general': "Great effort speaking in English! Keep up the wonderful work!", 'pronunciation': "Pronunciation score: 75%. Your speech is clear and easy to understand.", 'fluency': "Fluency score: 75%. Your speech flows naturally with good rhythm.", 'grammar': "Grammar score: 75%. You use English grammar structures well for clear communication.", 'vocabulary': "Vocabulary score: 75%. You choose words appropriately and effectively.", 'accent': "Accent: Asian (50% confidence). Your accent adds character to your unique English!", 'improvements_needed': "Keep practicing speaking English daily to build confidence and fluency." } } def validate_word_count(self, text, min_words=3, max_words=150): """Allow longer text for feedback - up to 150 words""" if not text: return "" # For longer texts like feedback, we don't need to strictly limit the length # Just ensure it's not empty and remove any excessive whitespace cleaned_text = re.sub(r'\s+', ' ', text).strip() return cleaned_text def generate_comprehension_questions(self, reading_content): """ Generate open-ended comprehension questions based on the reading content using AI Args: reading_content (str): The reading content to generate questions from Returns: list: List of ComprehensionQuestion objects """ try: # First try AI generation if available if self._configure_genai() and self.get_model(): try: model = self.get_model() prompt = f""" You are an English teacher creating very simple comprehension questions for beginner-level English learners. Based on the following reading passage, generate 5 short, open-ended questions. Requirements: 1. Prefer to start questions with "Tell me" when it sounds natural. 2. If "Tell me" makes the sentence sound unnatural, write a simple, natural question instead (e.g., "Where do you live?"). 3. Keep each question between 5 to 7 words. 4. Use very simple and clear English (A1 level). 5. Focus on basic personal details and main ideas from the passage. 6. Avoid complex grammar or difficult vocabulary. 7. Make sure every question sounds natural and meaningful. 8. Return ONLY a JSON array of question strings. Reading Content: {reading_content} Return format: ["Question 1?", "Question 2?", "Question 3?", "Question 4?", "Question 5?"] """ logger.info("Requesting AI-generated comprehension questions") response = model.generate_content(prompt) response_text = getattr(response, 'text', str(response)) # Parse JSON response import json try: # Try to extract JSON from response json_match = re.search(r'\[.*?\]', response_text, re.DOTALL) if json_match: questions_data = json.loads(json_match.group(0)) else: questions_data = json.loads(response_text) # Create ComprehensionQuestion objects questions = [] for q_text in questions_data[:5]: # Limit to 5 questions if q_text and q_text.strip(): question = ComprehensionQuestion(text=q_text.strip()) questions.append(question) if questions: logger.info(f"Generated {len(questions)} AI comprehension questions") return questions except json.JSONDecodeError as e: logger.warning(f"Failed to parse AI response as JSON: {e}") except Exception as e: logger.warning(f"AI comprehension question generation failed: {e}") # Fall back to local generation logger.info("Using fallback comprehension question generation") return self._generate_fallback_comprehension_questions(reading_content) except Exception as e: logger.error(f"Error generating comprehension questions: {e}") return self._generate_fallback_comprehension_questions(reading_content) def _generate_fallback_comprehension_questions(self, reading_content): """ Generate basic comprehension questions when AI is unavailable Args: reading_content (str): The reading content to generate questions from Returns: list: List of ComprehensionQuestion objects """ try: # Extract key information from the content content = reading_content.lower() sentences = sent_tokenize(reading_content) # Create basic question templates questions = [] # Question 1: Main topic/summary questions.append(ComprehensionQuestion( text="What is the main topic of this reading? Summarize it in your own words." )) # Question 2: Personal connection questions.append(ComprehensionQuestion( text="How does this reading relate to your own life or experiences?" )) # Question 3: Key details if len(sentences) > 2: questions.append(ComprehensionQuestion( text="What are the most important details mentioned in this reading?" )) # Question 4: Opinion/analysis questions.append(ComprehensionQuestion( text="What do you think about the ideas presented in this reading? Do you agree or disagree?" )) # Question 5: Application/extension if 'university' in content or 'study' in content: questions.append(ComprehensionQuestion( text="How can you apply what you learned from this reading to your studies?" )) elif 'family' in content: questions.append(ComprehensionQuestion( text="How does this reading help you think about family relationships?" )) elif 'work' in content or 'job' in content: questions.append(ComprehensionQuestion( text="How might this reading be useful for your future career?" )) else: questions.append(ComprehensionQuestion( text="What questions do you still have after reading this? What would you like to know more about?" )) logger.info(f"Generated {len(questions)} fallback comprehension questions") return questions except Exception as e: logger.error(f"Error generating fallback comprehension questions: {e}") # Return basic default questions if all else fails return [ ComprehensionQuestion(text="What is this reading about?"), ComprehensionQuestion(text="What did you learn from this reading?"), ComprehensionQuestion(text="Do you agree with the ideas in this reading? Why?"), ComprehensionQuestion(text="How does this reading relate to your life?"), ComprehensionQuestion(text="What questions do you have about this topic?") ] def evaluate_answer(self, question_text, student_answer, reading_content): ''' Evaluate a student's answer to a comprehension question with detailed analysis Args: question_text (str): The comprehension question student_answer (str): The student's answer reading_content (str): The reading content the question is based on Returns: dict: Detailed evaluation with scores, grammar analysis, and feedback ''' try: model = self.get_model() if not model: return self._evaluate_answer_fallback(question_text, student_answer, reading_content) # Log to verify reading content is being passed logger.info(f"Evaluating answer with reading_content length: {len(reading_content) if reading_content else 0}") logger.debug(f"Reading content preview: {reading_content[:200] if reading_content else 'NO CONTENT'}") # Context-aware, beginner-friendly evaluation prompt prompt = f""" You are a supportive English teacher evaluating a BEGINNER student's answer to a reading comprehension question. ⚠️ CRITICAL: You MUST read the READING CONTENT below FIRST before evaluating. The student's answer should be based on this reading. 📚 READING CONTENT (READ THIS FIRST): {reading_content[:2000] if reading_content else "No reading content available"} ❓ QUESTION ASKED: {question_text} ✍️ STUDENT'S ANSWER: {student_answer} 🎯 YOUR EVALUATION STEPS (FOLLOW IN ORDER): 1. READ the reading content above carefully 2. UNDERSTAND what the question is asking 3. CHECK if the student's answer is based on the reading content 4. VERIFY if the information in the answer is CORRECT according to the reading 5. Only then evaluate grammar/spelling (but be VERY lenient) 6. Be SUPPORTIVE and ENCOURAGING - this is a beginner learner 7. Focus on COMMUNICATION SUCCESS over perfection ⚠️ CRITICAL RULES (MUST FOLLOW): ✅ RELEVANCE FIRST: - Does the answer make sense for this question and reading? - Is the information correct according to the passage? - If YES → give high content score (80-100) - If PARTIALLY → medium score (50-79) - If NO/OFF-TOPIC → low score (0-49) ✅ REAL GRAMMAR/SPELLING ONLY: - Only mark ACTUAL mistakes (wrong tense, wrong word form, obvious typos) - Do NOT mark valid English words as spelling errors - Do NOT mark PROPER NOUNS (names, places, organizations) as spelling errors - Do NOT mark ABBREVIATIONS or ACRONYMS as spelling errors - Examples of what NOT to flag: ❌ "Lahore" → DON'T suggest "labor" (it's a city name!) ❌ "LGU" → DON'T suggest "you" (it's an abbreviation!) ❌ "yes" → DON'T suggest "eyes" (it's correct!) ❌ "students" → DON'T suggest "student" if plural makes sense - Be EXTREMELY LENIENT with beginners - minor issues are OK ✅ SMALL CORRECTIONS ONLY: - Capitalization (start sentences with capital) - Missing punctuation (periods, question marks) - Clear typos only (e.g., "teh" → "the", "studetn" → "student") - Basic grammar (is/are, was/were) - DO NOT "correct" proper nouns or abbreviations! ✅ WHAT NOT TO DO (CRITICAL): - ❌ Don't create nonsense corrections like "Lahore" → "labor" - ❌ Don't mark proper nouns (city names, person names, organizations) as wrong - ❌ Don't mark abbreviations/acronyms (LGU, MIT, USA, etc.) as wrong - ❌ Don't mark correct words as wrong - ❌ Don't be overly strict with beginners - ❌ Don't add information not in the reading - ❌ Don't suggest complex vocabulary for simple answers 🔍 HOW TO IDENTIFY PROPER NOUNS & ABBREVIATIONS: - If a word is capitalized in the middle of a sentence, it's likely a proper noun - LEAVE IT ALONE - If a word is ALL CAPS or mixed case (like LGU, MIT, USA), it's an abbreviation - LEAVE IT ALONE - City names, university names, person names are NEVER spelling errors - Check the READING CONTENT first - if the word appears there, it's correct! ✅ FEEDBACK STYLE: - Warm, encouraging, teacher-like - Start with what's GOOD - Then mention small improvements - Keep it simple and clear 📋 REQUIRED JSON OUTPUT (exact format): {{ "content_score": 0-100 (Is answer relevant and correct? 80-100 if yes), "grammar_score": 0-100 (Real grammar mistakes only, be lenient), "vocabulary_score": 0-100 (Is vocabulary appropriate? Usually 70-100 for beginners), "coherence_score": 0-100 (Does it make sense? Usually 70-100 for simple answers), "overall_score": 0-100 (Average, but favor content score), "is_relevant": true|false (Does answer match the reading and question?), "is_correct": true|false (Is the factual content accurate?), "grammar_errors": [ {{"original": "exact phrase from student", "corrected": "fixed version", "explanation": "simple reason (e.g., 'Start with capital letter')"}} ], "spelling_errors": [ {{"original": "wrong word", "corrected": "correct spelling", "explanation": "brief note"}} ], "content_feedback": "Short, positive feedback about the answer's relevance and accuracy", "grammar_feedback": "What grammar/punctuation to improve (or 'Great job!' if none)", "vocabulary_feedback": "Vocabulary comment (usually positive for beginners)", "coherence_feedback": "Does the answer flow well?", "strengths": "What the student did well (always find something positive!)", "improvements": "ONE small thing to improve (capitalization, punctuation, or 'Keep practicing!')", "corrected_answer": "Student's answer with ONLY small fixes (capitalization, punctuation, real typos). Keep the same words and meaning. If answer is already good, return it unchanged or with minimal fixes." }} 🎓 SCORING GUIDE FOR BEGINNERS (FOLLOW EXACTLY): 📊 CONTENT SCORE (Most Important): - Answer is correct and relevant → 90-100 (be generous!) - Answer is mostly correct, minor details off → 80-89 - Answer is partially correct → 65-79 - Answer is somewhat off-topic but tries → 40-64 - Answer is completely wrong → 0-39 📊 GRAMMAR SCORE (Be VERY Lenient): - Perfect or 1-2 tiny mistakes → 95-100 - Few small errors (missing capital, no period) → 85-94 - Several small errors but understandable → 75-84 - Some grammar issues but communication OK → 65-74 - Major grammar problems → below 65 📊 VOCABULARY & COHERENCE: - Simple but appropriate words → 90-100 - Basic vocabulary, clear meaning → 85-95 - Very simple but gets point across → 75-84 📊 OVERALL SCORE (CRITICAL): - If answer is CORRECT and RELEVANT → Overall MUST be 90-100 (even if grammar isn't perfect!) - If content_score ≥ 90 → Overall should be 90-100 - If content_score ≥ 80 → Overall should be 85-95 - If content_score 65-79 → Overall should be 70-85 - Grammar mistakes should reduce overall by MAX 5-10 points if content is correct ⚠️ CRITICAL RULES (FOLLOW EXACTLY): 🔍 STEP 1 - CHECK THE READING FIRST: - Look at the READING CONTENT above - Find the information related to the question - Does the student's answer match the reading? - If YES → content_score MUST be 90-100, overall_score MUST be 95-100 🔍 STEP 2 - SCORING: 1. If answer is FACTUALLY CORRECT from the reading → content_score = 95-100 2. If content_score ≥ 90 → overall_score MUST be 95-100 (ignore minor grammar) 3. Simple, correct answers deserve 95-100 scores 4. Example: Question "How is Person A?" Answer "Person A is doing well." → If this matches the reading, give 95-100 5. Do NOT give scores below 90 for factually correct answers 🔍 STEP 3 - GRAMMAR (ONLY IF ANSWER IS CORRECT): - Minor grammar issues should NOT reduce score below 90 - Only reduce by 5 points MAX for small grammar issues - Focus: Did they communicate the correct information? YES = 95+ ⚠️ REMEMBER: - This is a BEGINNER - be GENEROUS - CORRECT INFORMATION = 95-100 score (even if simple) - Check READING CONTENT first, then evaluate - Don't be picky about style/detail if facts are right Return ONLY a JSON block: ```json {{ ...your evaluation... }} ``` """ logger.info("Requesting AI evaluation for comprehension answer") response = model.generate_content(prompt) response_text = getattr(response, 'text', str(response)).strip() logger.debug(f"AI comprehension evaluation response: {response_text[:500]}...") # Parse JSON response with safer extraction and schema validation parsed = None try: # Prefer fenced json block fence_match = re.search(r"```json\s*(.*?)\s*```", response_text, re.DOTALL | re.IGNORECASE) json_str = None if fence_match: json_str = fence_match.group(1) else: # Fallback: remove stray fences and try full text json_str = re.sub(r"```json\s*|\s*```", "", response_text, flags=re.IGNORECASE) # As last resort, capture first JSON object if '{' not in json_str: brace_match = re.search(r"\{[\s\S]*\}", response_text) if brace_match: json_str = brace_match.group(0) parsed = json.loads(json_str) except Exception as e: logger.warning(f"Failed to parse AI response as JSON: {e}") return self._evaluate_answer_fallback(question_text, student_answer, reading_content) # Schema validation and coercion def _clamp_int(v, lo=0, hi=100, default=70): try: return max(lo, min(hi, int(v))) except Exception: return default def _float01(v, default=0.2): try: f = float(v) if f < 0: return 0.0 if f > 1: return 1.0 return f except Exception: return default # Defaults if not isinstance(parsed, dict): parsed = {} parsed.setdefault('evidence', []) parsed.setdefault('relevance_notes', '') parsed.setdefault('length_comment', '') parsed.setdefault('uncertainty', 0.2) parsed.setdefault('flags', []) parsed.setdefault('grammar_errors', []) parsed.setdefault('spelling_errors', []) parsed.setdefault('vocabulary_suggestions', []) parsed.setdefault('is_relevant', True) parsed.setdefault('is_correct', True) parsed.setdefault('content_feedback', '') parsed.setdefault('grammar_feedback', '') parsed.setdefault('vocabulary_feedback', '') parsed.setdefault('coherence_feedback', '') parsed.setdefault('strengths', '') parsed.setdefault('improvements', '') parsed.setdefault('corrected_answer', student_answer) content_score = _clamp_int(parsed.get('content_score')) grammar_score = _clamp_int(parsed.get('grammar_score')) vocabulary_score = _clamp_int(parsed.get('vocabulary_score')) coherence_score = _clamp_int(parsed.get('coherence_score')) overall_score = _clamp_int(parsed.get('overall_score')) # Light post-processing: ensure evidence quotes are short evidence = parsed.get('evidence') or [] safe_evidence = [] for item in evidence if isinstance(evidence, list) else []: if isinstance(item, dict): quote = str(item.get('quote', '')) reason = str(item.get('reason', '')) if quote: if len(quote.split()) > 15: quote = ' '.join(quote.split()[:15]) safe_evidence.append({'quote': quote, 'reason': reason}) if not safe_evidence and reading_content: # Provide minimal fallback evidence by selecting a short sentence from reading try: first_sent = sent_tokenize(reading_content)[0] safe_evidence.append({'quote': ' '.join(first_sent.split()[:12]), 'reason': 'Representative idea from the reading'}) except Exception: pass if parsed and isinstance(parsed, dict): # Format grammar errors, spelling errors, and vocabulary suggestions safely grammar_errors = parsed.get('grammar_errors', []) if isinstance(parsed.get('grammar_errors', []), list) else [] spelling_errors = parsed.get('spelling_errors', []) if isinstance(parsed.get('spelling_errors', []), list) else [] vocab_suggestions = parsed.get('vocabulary_suggestions', []) if isinstance(parsed.get('vocabulary_suggestions', []), list) else [] # Build result result = { 'score': overall_score, 'content_score': content_score, 'grammar_score': grammar_score, 'vocabulary_score': vocabulary_score, 'coherence_score': coherence_score, 'overall_score': overall_score, 'evidence': safe_evidence, 'relevance_notes': parsed.get('relevance_notes', ''), 'length_comment': parsed.get('length_comment', ''), 'uncertainty': _float01(parsed.get('uncertainty', 0.2)), 'flags': parsed.get('flags', []) if isinstance(parsed.get('flags', []), list) else [], 'is_relevant': parsed.get('is_relevant', True), 'is_correct': parsed.get('is_correct', True), 'grammar_analysis': { 'score': grammar_score, 'errors': grammar_errors }, 'spelling_errors': spelling_errors, 'vocabulary_analysis': vocab_suggestions, 'feedback': parsed.get('content_feedback', 'Good effort on the comprehension question.'), 'grammar_feedback': parsed.get('grammar_feedback', 'Keep working on grammar accuracy.'), 'vocabulary_feedback': parsed.get('vocabulary_feedback', 'Try to use more varied vocabulary.'), 'coherence_feedback': parsed.get('coherence_feedback', ''), 'strengths': parsed.get('strengths', 'You showed effort in answering the question.'), 'suggestions': parsed.get('improvements', 'Focus on providing more detailed and accurate answers.'), 'corrected_answer': parsed.get('corrected_answer', student_answer) } logger.info(f"AI comprehension evaluation completed: overall={overall_score}, grammar={grammar_score}, content={content_score}") return result # If we couldn't parse a valid response, fall back logger.warning("AI response was not in expected format, using fallback") return self._evaluate_answer_fallback(question_text, student_answer, reading_content) except Exception as e: logger.error(f"Error in AI comprehension evaluation: {str(e)}") return self._evaluate_answer_fallback(question_text, student_answer, reading_content) def _ai_transcribe_and_assess(self, audio_path, question_text, topic_content=None): """Send audio to Gemini multimodal model and request a JSON assessment. Expected return structure (JSON): { "transcription": "...", "confidence": 0.87, "scores": {"pronunciation": 79, "fluency": 70, "grammar": 60, "vocabulary": 65, "relevance": 75, "overall": 70}, "feedback": {"general": "...", ...} } """ try: # Ensure enhanced WAV exists with aggressive enhancement for low volume/whispered audio wav_path = audio_path.replace(os.path.splitext(audio_path)[1], '_enhanced.wav') if not os.path.exists(wav_path): # create enhanced WAV with aggressive audio processing try: logger.debug(f"Creating enhanced audio for AI assessment from: {audio_path}") audio_format = "webm" if audio_path.endswith('.webm') else None audio = AudioSegment.from_file(audio_path, format=audio_format) # Check if audio is silent or very quiet peak_amplitude = audio.max_dBFS logger.info(f"Original audio peak amplitude: {peak_amplitude} dBFS") # AGGRESSIVE ENHANCEMENT FOR LOW VOLUME/WHISPERED AUDIO # Step 1: Remove DC offset and silence audio = audio.strip_silence(silence_thresh=-50, padding=100) # Step 2: Apply multiple passes of normalization for very quiet audio if peak_amplitude < -20: # Very quiet audio logger.info("Detected low volume audio - applying aggressive enhancement") # First pass: aggressive normalization with minimal headroom audio = audio.normalize(headroom=0.1) # Second pass: boost by 10dB audio = audio + 10 # Third pass: normalize again audio = audio.normalize(headroom=0.3) elif peak_amplitude < -10: # Moderately quiet logger.info("Detected moderate volume audio - applying standard enhancement") audio = audio.normalize(headroom=0.3) audio = audio + 6 else: # Normal volume audio = audio.normalize(headroom=0.5) # Step 3: High-pass filter to remove low-frequency noise enhanced_audio = audio.high_pass_filter(80) # Step 4: Apply dynamic range compression to even out volume # More aggressive compression for low volume audio enhanced_audio = enhanced_audio.compress_dynamic_range( threshold=-25.0, # Lower threshold to catch more ratio=6.0, # Higher ratio for more compression attack=5.0, # Quick attack release=50.0 # Moderate release ) # Step 5: Final boost for whispered audio if peak_amplitude < -25: logger.info("Applying whisper boost") enhanced_audio = enhanced_audio + 8 # Additional 8dB for whispers else: enhanced_audio = enhanced_audio + 3 # Standard 3dB boost # Step 6: Apply a subtle noise gate to remove background noise # while preserving quiet speech enhanced_audio = enhanced_audio.strip_silence( silence_thresh=-45, # More lenient threshold padding=200 # More padding to preserve speech edges ) # Step 7: Final normalization to ensure consistent level enhanced_audio = enhanced_audio.normalize(headroom=0.1) # Export with optimal settings for speech recognition enhanced_audio.export( wav_path, format='wav', parameters=[ '-ar', '16000', # 16kHz sample rate (optimal for speech) '-sample_fmt', 's16', # 16-bit samples '-ac', '1' # Mono ] ) final_amplitude = enhanced_audio.max_dBFS logger.info(f"Enhanced audio created: {wav_path} (amplitude: {peak_amplitude} -> {final_amplitude} dBFS)") except Exception as e: logger.warning(f"Could not create enhanced WAV for AI assessment: {e}") wav_path = audio_path # Verify the audio file exists and has content if not os.path.exists(wav_path): logger.error(f"Audio file does not exist: {wav_path}") return None file_size = os.path.getsize(wav_path) if file_size < 50: # Very lenient minimum (50 bytes) to accept even tiny whispers logger.error(f"Audio file too small for processing: {file_size} bytes") return None logger.info(f"Reading audio file for AI assessment: {wav_path} ({file_size} bytes)") with open(wav_path, 'rb') as f: audio_bytes = f.read() model = self.get_model() # Build a detailed prompt asking the model to return strict JSON with specific fields prompt = ( "You are an expert English assessor specializing in transcribing and evaluating BEGINNER students' speech, including LOW VOLUME and WHISPERED audio.\n\n" "🎯 CRITICAL INSTRUCTIONS:\n" "1. This audio may be VERY QUIET, whispered, or have low volume - LISTEN CAREFULLY and try your best to transcribe it\n" "2. Even if the audio is quiet or unclear, attempt to transcribe what you hear\n" "3. DO NOT return empty transcription unless the audio is completely silent\n" "4. For unclear words, use your best guess based on context from the question\n" "5. Be LENIENT with pronunciation and fluency scores for quiet/whispered speech\n" "6. If you detect any speech at all, provide a transcription and reasonable scores\n\n" "📝 CONTEXT:\n" "Question: " + (question_text or 'General speaking question') + "\n" "Topic: " + (topic_content or 'General conversation') + "\n\n" "📋 REQUIRED JSON OUTPUT (no other text):\n" "{\n" ' "transcription": "the spoken text (even if unclear, transcribe what you hear)",\n' ' "confidence": 0.0-1.0 (your confidence in the transcription),\n' ' "scores": {\n' ' "pronunciation": 0-100,\n' ' "fluency": 0-100,\n' ' "grammar": 0-100,\n' ' "vocabulary": 0-100,\n' ' "relevance": 0-100 (how well it answers the question),\n' ' "overall": 0-100 (average of above)\n' ' },\n' ' "accent": {"type": "Pakistani/American/British/Indian/Unknown", "confidence": 0-100},\n' ' "feedback": {\n' ' "general": "brief overall comment",\n' ' "pronunciation": "pronunciation feedback",\n' ' "fluency": "fluency feedback",\n' ' "grammar": "grammar feedback",\n' ' "vocabulary": "vocabulary feedback",\n' ' "relevance": "relevance feedback"\n' ' }\n' '}\n\n' "⚠️ IMPORTANT:\n" "- If audio is quiet/whispered, acknowledge this in feedback but still provide scores\n" "- Minimum score should be 30-40 if any speech is detected (don't give 0 for trying)\n" "- Be ENCOURAGING for beginner students\n" "- Return ONLY the JSON object, no other text\n" ) logger.info("Sending prompt to Mistral API for speaking assessment (text-only)") # Mistral is text-only; we cannot pass raw audio bytes. # The MistralModelWrapper.generate_content() handles this gracefully: # if a list is passed it extracts the text part and ignores binary data. try: response = model.generate_content(prompt) except Exception as api_error: logger.error(f"Error calling Mistral API: {api_error}") return None raw_text = None if hasattr(response, 'text') and response.text: raw_text = response.text logger.debug(f"Received response from Mistral: {raw_text[:200]}...") elif hasattr(response, 'candidates') and response.candidates: # Try to extract text from candidates for candidate in response.candidates: if hasattr(candidate, 'content') and hasattr(candidate.content, 'parts'): for part in candidate.content.parts: if hasattr(part, 'text') and part.text: raw_text = part.text break if raw_text: break if raw_text: logger.debug(f"Extracted text from candidates: {raw_text[:200]}...") else: raw_text = str(response) logger.warning(f"Unexpected response format from Mistral: {raw_text[:200]}...") if not raw_text or raw_text.strip() == "": logger.error("Empty response from Mistral API") return None # Try parsing JSON import json, re parsed = None try: parsed = json.loads(raw_text.strip()) logger.debug("Successfully parsed JSON response from Gemini") except Exception as json_error: logger.warning(f"Failed to parse JSON directly: {json_error}") # Try to extract JSON from response m = re.search(r"\{.*\}", raw_text, re.S) if m: try: parsed = json.loads(m.group(0)) logger.debug("Successfully extracted and parsed JSON from response") except Exception as extract_error: logger.error(f"Failed to parse extracted JSON: {extract_error}") parsed = None else: logger.error("No JSON structure found in response") if parsed and isinstance(parsed, dict): # Normalize fields result = {} result['transcription'] = parsed.get('transcription') or parsed.get('transcript') or '' try: result['confidence'] = float(parsed.get('confidence')) if parsed.get('confidence') is not None else 0.0 except Exception: result['confidence'] = 0.0 # Scores and feedback normalization result['scores'] = parsed.get('scores') if isinstance(parsed.get('scores'), dict) else {} result['feedback'] = parsed.get('feedback') if isinstance(parsed.get('feedback'), dict) else {} # Accent normalization: expect {'type': 'Pakistani', 'confidence': 80} accent_raw = parsed.get('accent') if isinstance(accent_raw, dict): accent_type = accent_raw.get('type') or accent_raw.get('label') or 'Unknown' try: accent_conf = int(accent_raw.get('confidence') or accent_raw.get('score') or 0) except Exception: accent_conf = 0 else: accent_type = 'Unknown' accent_conf = 0 # Normalize common accent labels (simple mapping) accent_map = { 'pakistani': 'Pakistani', 'pakistani-english': 'Pakistani', 'american': 'American', 'us': 'American', 'british': 'British', 'uk': 'British', 'indian': 'Indian' } at_lower = (accent_type or '').strip().lower() accent_type_norm = accent_map.get(at_lower, accent_type.title() if accent_type else 'Unknown') result['accent'] = {'type': accent_type_norm, 'confidence': max(0, min(100, accent_conf))} # Ensure numeric scores exist (fill defaults) and cast to ints scores = result['scores'] for k in ['pronunciation', 'fluency', 'grammar', 'vocabulary', 'relevance', 'overall']: try: scores[k] = int(scores.get(k, 0) or 0) except Exception: scores[k] = 0 # If relevance is missing or zero, compute a very simple fallback relevance score if not scores.get('relevance'): # compute overlap between question keywords and transcription words try: q_words = set(re.findall(r"\w+", (question_text or '').lower())) t_words = set(re.findall(r"\w+", (result.get('transcription') or '').lower())) if q_words: overlap = len(q_words & t_words) relevance_score = min(100, int((overlap / max(1, len(q_words))) * 100)) else: relevance_score = 0 except Exception: relevance_score = 0 scores['relevance'] = relevance_score # Set last transcription confidence try: self._last_transcription_confidence = float(result.get('confidence', 0.0)) except Exception: self._last_transcription_confidence = 0.0 # Check if transcription is empty or very short transcription = result['transcription'].strip() if not transcription or len(transcription) < 3: logger.warning(f"Empty or very short transcription received. Providing fallback response.") # Provide partial credit for attempting the task return { 'audio_detected': True, 'transcription': transcription if transcription else "[Audio was too quiet or unclear to transcribe. Please try speaking a bit louder.]", 'scores': { 'pronunciation': 40, # Partial credit for attempting 'fluency': 35, 'grammar': 40, 'vocabulary': 35, 'relevance': 30, 'overall': 36 }, 'accent': {'type': 'Unknown', 'confidence': 0}, 'feedback': { 'general': 'Your audio was very quiet or unclear. Please try recording again with a slightly louder voice.', 'pronunciation': 'Could not assess due to low audio volume.', 'fluency': 'Could not assess due to unclear audio.', 'grammar': 'Could not assess - please record again.', 'vocabulary': 'Could not assess - please speak louder.', 'relevance': 'Unable to evaluate - audio unclear.' } } logger.info(f"AI assessment parsed successfully: overall={scores.get('overall')} confidence={self._last_transcription_confidence} transcription_length={len(transcription)}") return { 'audio_detected': True, 'transcription': transcription, 'scores': scores, 'accent': result.get('accent', {'type': 'Unknown', 'confidence': 0}), 'feedback': result.get('feedback', {}) } else: logger.warning("AI model did not return valid JSON for transcription+assessment") # Provide fallback response instead of None return { 'audio_detected': True, 'transcription': "[Unable to process audio. Please try recording again.]", 'scores': { 'pronunciation': 35, 'fluency': 35, 'grammar': 35, 'vocabulary': 35, 'relevance': 35, 'overall': 35 }, 'accent': {'type': 'Unknown', 'confidence': 0}, 'feedback': { 'general': 'We encountered an issue processing your audio. Please try recording again.', 'pronunciation': 'Unable to assess - please try again.', 'fluency': 'Unable to assess - please try again.', 'grammar': 'Unable to assess - please try again.', 'vocabulary': 'Unable to assess - please try again.', 'relevance': 'Unable to assess - please try again.' } } except Exception as e: logger.error(f"AI transcription+assessment error: {str(e)}") # Provide fallback result instead of None to avoid 0 scores return { 'audio_detected': True, 'transcription': "[Error processing audio. Please try again with clearer audio.]", 'scores': { 'pronunciation': 40, 'fluency': 40, 'grammar': 40, 'vocabulary': 40, 'relevance': 40, 'overall': 40 }, 'accent': {'type': 'Unknown', 'confidence': 0}, 'feedback': { 'general': 'There was a technical issue processing your audio. You receive partial credit for attempting. Please try recording again.', 'pronunciation': 'Could not assess due to technical error.', 'fluency': 'Could not assess due to technical error.', 'grammar': 'Could not assess due to technical error.', 'vocabulary': 'Could not assess due to technical error.', 'relevance': 'Could not assess due to technical error.' } } # end of _ai_transcribe_and_assess def _evaluate_answer_fallback(self, question_text, student_answer, reading_content): """ Enhanced fallback evaluation with detailed grammar analysis when AI is not available """ # Initialize scores and analysis content_score = 0 grammar_score = 0 vocabulary_score = 0 coherence_score = 0 grammar_errors = [] vocabulary_suggestions = [] # Check if answer is not empty if not student_answer.strip(): return { 'score': 0, 'content_score': 0, 'grammar_score': 0, 'vocabulary_score': 0, 'coherence_score': 0, 'grammar_analysis': {'score': 0, 'errors': []}, 'vocabulary_analysis': [], 'feedback': 'No answer provided. Please respond to the question.', 'grammar_feedback': 'Cannot assess grammar without an answer.', 'vocabulary_feedback': 'Cannot assess vocabulary without an answer.', 'strengths': 'N/A', 'suggestions': 'Please read the question carefully and provide a thoughtful answer based on the reading content.', 'corrected_answer': '' } # Enhanced grammar check using TextBlob try: blob = TextBlob(student_answer) sentences = blob.sentences word_count = len(student_answer.split()) # Grammar Analysis grammar_issues = 0 for sentence in sentences: sentence_text = str(sentence).strip() # Check for basic grammar issues tags = sentence.tags words = sentence.words # Subject-verb agreement check for i in range(len(tags)-1): word, pos = tags[i] next_word, next_pos = tags[i+1] # Check subject-verb agreement if pos.startswith('NN') and next_pos.startswith('VB'): if not self._check_subject_verb_agreement(word, next_word): corrected = self._fix_subject_verb_agreement(sentence_text) grammar_errors.append({ 'original': sentence_text, 'corrected': corrected, 'explanation': f'Subject-verb agreement: "{word}" and "{next_word}" do not agree.' }) grammar_issues += 1 # Check for sentence structure issues if len(words) < 3: grammar_errors.append({ 'original': sentence_text, 'corrected': sentence_text + ' (needs more detail)', 'explanation': 'This sentence is too brief and may be incomplete.' }) grammar_issues += 1 # Check for capitalization if sentence_text and not sentence_text[0].isupper(): corrected = sentence_text[0].upper() + sentence_text[1:] grammar_errors.append({ 'original': sentence_text, 'corrected': corrected, 'explanation': 'Sentences should start with a capital letter.' }) grammar_issues += 1 # Check for ending punctuation if sentence_text and sentence_text[-1] not in '.!?': grammar_errors.append({ 'original': sentence_text, 'corrected': sentence_text + '.', 'explanation': 'Sentences should end with proper punctuation.' }) grammar_issues += 1 # Calculate grammar score max_deduction = min(grammar_issues * 15, 40) # Max 40 points deduction grammar_score = max(60, 100 - max_deduction) # Minimum 60 for attempting # Vocabulary Analysis common_words = {'the', 'and', 'a', 'an', 'is', 'are', 'was', 'were', 'have', 'has', 'do', 'does', 'will', 'would', 'can', 'could', 'should', 'this', 'that', 'these', 'those'} unique_words = set(word.lower() for word in blob.words if word.lower() not in common_words and len(word) > 2) # Check for vocabulary repetition word_freq = {} for word in blob.words: if len(word) > 3 and word.lower() not in common_words: word_freq[word.lower()] = word_freq.get(word.lower(), 0) + 1 # Suggest alternatives for repeated words for word, freq in word_freq.items(): if freq > 2: # Word used more than twice vocabulary_suggestions.append({ 'word': word, 'suggestion': 'Consider using synonyms to avoid repetition', 'context': f'The word "{word}" appears {freq} times' }) # Basic vocabulary score vocab_diversity = len(unique_words) / max(1, word_count) * 100 vocabulary_score = min(100, max(60, 60 + vocab_diversity * 40)) except Exception as e: logger.warning(f"Error in grammar analysis: {e}") grammar_score = 70 # Default score if analysis fails vocabulary_score = 70 # Content Analysis word_count = len(student_answer.split()) # Length assessment if word_count < 5: content_score = 30 length_feedback = 'Your answer is too brief.' length_suggestion = 'Provide more detailed answers that show understanding.' elif word_count < 15: content_score = 50 length_feedback = 'Your answer could be more detailed.' length_suggestion = 'Try to explain your thoughts more thoroughly.' elif word_count > 200: content_score = 60 length_feedback = 'Your answer is quite long.' length_suggestion = 'Focus on the most important points and be more concise.' else: content_score = 75 length_feedback = 'Your answer has appropriate length.' length_suggestion = 'Good job on providing a detailed response.' # Content relevance check question_lower = question_text.lower() answer_lower = student_answer.lower() content_lower = reading_content.lower() # Check for overlap with reading content answer_words = set(answer_lower.split()) content_words = set(content_lower.split()) overlap = len(answer_words.intersection(content_words)) if overlap > 8: # Good connection to reading content_score += 15 relevance_feedback = 'Your answer shows good understanding of the reading content.' elif overlap > 4: # Some connection content_score += 5 relevance_feedback = 'Your answer shows some connection to the reading.' else: relevance_feedback = 'Try to include more specific details from the reading content.' # Coherence score (basic assessment) coherence_score = 70 # Base score if len(sentences) > 1: coherence_score += 10 # Bonus for multiple sentences if word_count > 20: coherence_score += 10 # Bonus for substantial content # Overall score calculation overall_score = (content_score * 0.4 + grammar_score * 0.3 + vocabulary_score * 0.2 + coherence_score * 0.1) overall_score = max(0, min(100, int(overall_score))) # Generate comprehensive feedback strengths = [] if grammar_score >= 80: strengths.append("good grammar usage") if vocabulary_score >= 75: strengths.append("varied vocabulary") if content_score >= 70: strengths.append("relevant content") strengths_text = "You demonstrated " + ", ".join(strengths) if strengths else "You made an effort to answer the question" # Generate improvement suggestions improvements = [] if grammar_score < 75: improvements.append("focus on grammar accuracy and sentence structure") if vocabulary_score < 70: improvements.append("use more varied and sophisticated vocabulary") if content_score < 70: improvements.append("provide more specific details from the reading") improvements_text = "Try to " + ", and ".join(improvements) if improvements else "Keep practicing to improve your writing skills" # Create corrected answer (basic version) corrected_answer = student_answer for error in grammar_errors[:3]: # Apply first 3 corrections corrected_answer = corrected_answer.replace(error['original'], error['corrected']) return { 'score': overall_score, 'content_score': max(0, min(100, int(content_score))), 'grammar_score': max(0, min(100, int(grammar_score))), 'vocabulary_score': max(0, min(100, int(vocabulary_score))), 'coherence_score': max(0, min(100, int(coherence_score))), 'grammar_analysis': { 'score': grammar_score, 'errors': grammar_errors[:5] # Limit to 5 most important errors }, 'vocabulary_analysis': vocabulary_suggestions, 'feedback': f"{length_feedback} {relevance_feedback}", 'grammar_feedback': f"Grammar score: {grammar_score}%. " + ("Good grammar usage!" if grammar_score >= 80 else "Focus on sentence structure and grammar rules."), 'vocabulary_feedback': f"Vocabulary score: {vocabulary_score}%. " + ("Good vocabulary variety!" if vocabulary_score >= 75 else "Try to use more diverse vocabulary."), 'strengths': strengths_text, 'suggestions': improvements_text, 'corrected_answer': corrected_answer } def generate_vocabulary_quiz(self, topic_content, num_questions=5): """ Generate vocabulary quiz questions from difficult words in the topic content Args: topic_content (str): The text content to extract vocabulary from num_questions (int): Number of questions to generate Returns: list: List of QuizQuestion objects """ try: # Personalize the content first - replace placeholders like [name] with student's name topic_content = self.personalize_content(topic_content) # First, check if the API is properly configured if not self._configure_genai(): logger.error("Failed to configure Gemini API. Using fallback vocabulary quiz.") return self._generate_fallback_vocabulary_questions(topic_content, num_questions) # Get the model model = self.get_model() if not model: logger.error("Failed to get AI model. Using fallback vocabulary quiz.") return self._generate_fallback_vocabulary_questions(topic_content, num_questions) # Create the prompt prompt = f""" Create a vocabulary quiz with {num_questions} questions based on the following text. IMPORTANT REQUIREMENTS: 1. Identify {num_questions} difficult or important words from the text 2. For each word, create an MCQ question asking for its meaning 3. Provide 4 options for each question, with one correct answer 4. Include the sentence from the text where the word appears (as context) 5. Format your response as JSON with the following structure: [ {{ "word": "example", "context": "This is an example sentence from the text.", "question": "What does 'example' mean in this context?", "correct_answer": "A representative case or instance", "options": [ "A representative case or instance", "A difficult problem to solve", "A type of examination", "A mathematical equation" ] }}, // more questions... ] Text: {topic_content} """ # Generate questions response = model.generate_content(prompt) response_text = response.text.strip() # Extract JSON from response (in case of markdown code blocks) json_match = re.search(r'```json\s*(.*?)\s*```', response_text, re.DOTALL) if json_match: json_str = json_match.group(1) else: json_str = response_text # Clean up any non-JSON text json_str = re.sub(r'^[\s\S]*?\[', '[', json_str) json_str = re.sub(r'\][\s\S]*?$', ']', json_str) # Parse JSON and create quiz questions questions_data = json.loads(json_str) questions = [] for q_data in questions_data[:num_questions]: # Limit to requested number question = QuizQuestion( word=q_data.get('word', ''), context=q_data.get('context', ''), question_text=q_data.get('question', f"What does '{q_data.get('word', '')}' mean?"), correct_answer=q_data.get('correct_answer', ''), options=json.dumps(q_data.get('options', [])) ) questions.append(question) return questions except Exception as e: logger.error(f"Error generating vocabulary quiz: {str(e)}") return self._generate_fallback_vocabulary_questions(topic_content, num_questions) def _generate_fallback_vocabulary_questions(self, topic_content, num_questions=5): '''Generate vocabulary questions using WordNet when AI is unavailable''' try: # Ensure content is personalized (in case this method is called directly) topic_content = self.personalize_content(topic_content) # Tokenize text into words words = word_tokenize(topic_content.lower()) # Remove common words and duplicates stopwords = {'a', 'an', 'the', 'and', 'or', 'but', 'if', 'because', 'as', 'what', 'when', 'where', 'how', 'why', 'is', 'am', 'are', 'was', 'were', 'be', 'been', 'being', 'have', 'has', 'had', 'do', 'does', 'did', 'to', 'at', 'by', 'for', 'with', 'about', 'against', 'between', 'into', 'through', 'during', 'before', 'after', 'above', 'below', 'from', 'up', 'down', 'in', 'out', 'on', 'off', 'over', 'under'} unique_words = [word for word in words if word.isalpha() and word not in stopwords and len(word) > 4] unique_words = list(set(unique_words)) # Remove duplicates # Extract sentences containing these words sentences = sent_tokenize(topic_content) word_contexts = {} # Lemmatizer to get base form of words lemmatizer = WordNetLemmatizer() # Create a dictionary to store words with their WordNet info wordnet_words = {} # Find words that exist in WordNet with definitions for word in unique_words: lemma = lemmatizer.lemmatize(word) synsets = wordnet.synsets(lemma) if synsets: # Word exists in WordNet # Find a sentence containing this word for sentence in sentences: if word in sentence.lower(): word_contexts[word] = sentence # Store WordNet info wordnet_words[word] = { 'synsets': synsets, 'definition': synsets[0].definition(), # Use first definition 'lemma': lemma } break # Filter words that have WordNet entries and contexts candidate_words = list(wordnet_words.keys()) # If we don't have enough words, add back some other words if len(candidate_words) < num_questions: for word in unique_words: if word not in candidate_words and word in word_contexts: candidate_words.append(word) if len(candidate_words) >= num_questions: break # Select random words if we have more than needed if len(candidate_words) > num_questions: selected_words = random.sample(candidate_words, num_questions) else: selected_words = candidate_words # Extended common definitions for fallback common_definitions = { 'important': 'having great significance or value', 'different': 'not the same as another or each other', 'special': 'better, greater, or otherwise different from what is usual', 'example': 'a thing characteristic of its kind', 'learning': 'the acquisition of knowledge or skills', 'specific': 'clearly defined or identified', 'general': 'affecting or concerning all or most people, places, or things', 'complete': 'having all the necessary or appropriate parts', 'various': 'of different kinds or sorts', 'similar': 'resembling without being identical', 'process': 'a series of actions taken to achieve a particular end', 'student': 'a person who is studying at a school or university', 'teacher': 'a person who teaches, especially in a school', 'language': 'the method of human communication using words', 'english': 'the language of England and many other countries', 'system': 'a set of things working together as parts of a mechanism', 'practice': 'the actual application or use of an idea or method', 'develop': 'grow or cause to grow and become more mature or advanced', 'improve': 'make or become better', 'understand': 'perceive the intended meaning of words or language', 'exercise': 'activity requiring physical effort carried out for health', 'question': 'a sentence worded to elicit information', 'answer': 'a response to a question or situation', 'knowledge': 'facts, information, and skills acquired through experience', 'education': 'the process of receiving or giving instruction', 'lesson': 'a period of learning or teaching', 'course': 'a series of lectures or lessons in a particular subject' } # Create questions questions = [] for word in selected_words: if word in wordnet_words: # Use WordNet definition synsets = wordnet_words[word]['synsets'] correct_definition = wordnet_words[word]['definition'] lemma = wordnet_words[word]['lemma'] # Get alternative options from WordNet options = [correct_definition] # Try to get different definitions for distractors similar_synsets = [] for syn in wordnet.synsets(lemma): if syn.definition() != correct_definition and syn.definition() not in options: similar_synsets.append(syn) # Add some similar word definitions as distractors if len(similar_synsets) > 0: for i in range(min(2, len(similar_synsets))): options.append(similar_synsets[i].definition()) # Add opposite or related concepts antonyms = [] for syn in synsets: for lemma in syn.lemmas(): if lemma.antonyms(): antonym = lemma.antonyms()[0].name().replace('_', ' ') antonym_synsets = wordnet.synsets(antonym) if antonym_synsets: antonyms.append(antonym_synsets[0].definition()) if antonyms: options.append(random.choice(antonyms)) # Fill remaining options with plausible alternatives while len(options) < 4: # Get a random word from WordNet random_word = random.choice(list(wordnet_words.keys())) if random_word != word: random_def = wordnet_words[random_word]['synsets'][0].definition() if random_def not in options: options.append(random_def) # If still not enough options, add some generic ones while len(options) < 4: fake_option = f"Related to the concept of {lemma} but not its definition" if fake_option not in options: options.append(fake_option) elif word in common_definitions: # Use predefined definitions correct_definition = common_definitions[word] # Get other definitions as distractors options = [correct_definition] other_defs = list(common_definitions.values()) other_defs.remove(correct_definition) random.shuffle(other_defs) for def_option in other_defs[:3]: options.append(def_option) else: # Create a plausible definition using word characteristics if len(word) > 6: word_type = "An advanced concept relating to" else: word_type = "A basic term meaning" correct_definition = f"{word_type} something similar to what '{word}' describes in the context" options = [ correct_definition, f"The opposite of what '{word}' typically means", f"A specialized type of action related to '{word}'", f"A category or classification system for '{word}'" ] # Ensure we have exactly 4 options if len(options) > 4: options = options[:4] # Shuffle options to randomize correct answer position random.shuffle(options) # Create the question question = QuizQuestion( word=word, context=word_contexts[word], question_text=f"What does '{word}' mean in this context?", correct_answer=correct_definition, options=json.dumps(options) ) questions.append(question) return questions except Exception as e: logger.error(f"Error generating fallback vocabulary questions: {str(e)}") # Return basic questions if WordNet approach fails questions = [] for i in range(min(num_questions, 5)): # Try to extract some actual words from the content try: words = word_tokenize(topic_content) filtered_words = [w for w in words if w.isalpha() and len(w) > 4] if filtered_words: word = random.choice(filtered_words) else: word = f"vocabulary{i+1}" # Create a basic context sentences = sent_tokenize(topic_content) if sentences: context = random.choice(sentences) else: context = f"This is an example sentence containing the word {word}." correct_answer = f"The definition of '{word}' based on its usage in the text" options = [ correct_answer, f"A meaning of '{word}' not related to this context", f"A concept that contradicts the meaning of '{word}'", f"A specialized usage of '{word}' in technical contexts" ] random.shuffle(options) except Exception: # If everything fails, create a completely generic question word = f"term{i+1}" context = "This is a sample text for vocabulary learning." correct_answer = f"The correct definition of '{word}'" options = [ correct_answer, f"An incorrect definition of '{word}'", f"A meaning unrelated to '{word}'", f"A definition that is opposite to '{word}'" ] question = QuizQuestion( word=word, context=context, question_text=f"What does '{word}' mean in this context?", correct_answer=correct_answer, options=json.dumps(options) ) questions.append(question) return questions