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) # speech_recognition removed: using AI-based transcription instead 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 import logging import nltk from nltk.tokenize import sent_tokenize, word_tokenize # Download NLTK data if needed try: nltk.download('punkt', 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 if not hasattr(current_app, 'ai_assessor'): current_app.ai_assessor = AIAssessment() 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 # recognizer removed; using AI-based transcription instead # Replace language tool initialization with TextBlob try: # We don't need to initialize TextBlob specifically 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: # Verify paths exist if os.path.exists(current_app.config['FFMPEG_PATH']) and os.path.exists(current_app.config['FFPROBE_PATH']): logger.info(f"FFmpeg found at: {current_app.config['FFMPEG_PATH']}") logger.info(f"FFprobe found at: {current_app.config['FFPROBE_PATH']}") # Configure pydub paths AudioSegment.converter = current_app.config['FFMPEG_PATH'] AudioSegment.ffmpeg = current_app.config['FFMPEG_PATH'] AudioSegment.ffprobe = current_app.config['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}") logger.info(f"Found FFprobe in PATH: {ffprobe_in_path}") # Configure pydub paths 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 _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', '') 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 = [ 'mistral-small-latest', 'mistral-medium-latest', 'mistral-large-latest', ] 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) test_response = wrapper.generate_content("Hello, world!") logger.info(f"Successfully initialized 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 logger.error("All Mistral models failed to initialize") return None return self.model def get_language_tool(self): """Use TextBlob instead of LanguageTool for grammar checking""" # Return None as we're using TextBlob directly when needed return None 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 """ 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_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: # First, check if the API is properly configured if not self._configure_genai(): logger.error("Failed to configure Mistral 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 Mistral 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 basic vocabulary questions when AI is unavailable""" try: # 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 = {} for word in unique_words: for sentence in sentences: if word in sentence.lower(): word_contexts[word] = sentence break # Select words that have contexts selected_words = list(word_contexts.keys()) if len(selected_words) > num_questions: selected_words = random.sample(selected_words, num_questions) # Create basic questions questions = [] 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', } for word in selected_words: # Generate fake options if word in common_definitions: correct_answer = common_definitions[word] else: correct_answer = f"The meaning of '{word}'" options = [ correct_answer, f"The opposite of '{word}'", f"A type of {word}", f"Related to {word} but different" ] random.shuffle(options) question = QuizQuestion( word=word, context=word_contexts[word], question_text=f"What does '{word}' mean?", correct_answer=correct_answer, options=json.dumps(options) ) questions.append(question) return questions except Exception as e: logger.error(f"Error generating fallback vocabulary questions: {str(e)}") # Return extremely basic questions if all else fails questions = [] for i in range(min(num_questions, 3)): word = f"word{i+1}" question = QuizQuestion( word=word, context=f"This is a sentence with {word}.", question_text=f"What does '{word}' mean?", correct_answer=f"Definition of {word}", options=json.dumps([f"Definition of {word}", f"Not {word}", f"Similar to {word}", f"Opposite of {word}"]) ) questions.append(question) return questions