""" Quiz Management Utilities Handles quiz creation, updating, and question management """ import uuid from datetime import datetime from typing import Dict, List, Optional, Any from sqlalchemy.orm import joinedload from sqlalchemy import func from src.models import DatabaseContextManager from src.models.models import ( ModuleQuiz, QuizQuestion, QuizOption, QuizAttempt, QuizAnswer, Course, Tutor, Supervisor, CourseModule, CourseLevel, Student ) from src.utils import ApiABC, custom_response from flask import current_app class QuizManager(ApiABC): def __init__(self): super().__init__(ModuleQuiz) def get_course_quizzes(self, course_id: str, requester_id: str, requester_type: str) -> Dict: """Get all quizzes for a course with questions and options""" try: with DatabaseContextManager() as ctx: # Verify user has access to the course course = ctx.session.query(Course).filter(Course.id == course_id).first() if not course: return custom_response( success=False, data="Course not found", status_code=404 ) # Get all modules for the course modules = ctx.session.query(CourseModule).filter( CourseModule.course_id == course_id, CourseModule.is_active == True ).all() all_quizzes = [] for module in modules: # Get quizzes with their questions and options eagerly loaded quizzes = ctx.session.query(ModuleQuiz).options( joinedload(ModuleQuiz.questions).joinedload(QuizQuestion.options) ).filter( ModuleQuiz.module_id == module.id, ModuleQuiz.is_active == True ).all() for quiz in quizzes: quiz_data = self._format_quiz(quiz, include_questions=True) quiz_data['module_name'] = module.title if hasattr(module, 'title') else 'Unknown Module' all_quizzes.append(quiz_data) return custom_response( success=True, data={'quizzes': all_quizzes}, status_code=200 ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error fetching course quizzes: {str(e)}\n{error_details}") return custom_response( success=False, data=f"Error fetching quizzes: {str(e)}", status_code=500 ) def create_quiz(self, payload: Dict) -> Dict: """Create a new quiz""" try: # Debug: Log incoming payload current_app.logger.info('📥 ===================================================') current_app.logger.info('📥 QUIZ CREATION - INCOMING PAYLOAD') current_app.logger.info('📥 ===================================================') current_app.logger.info(f'📥 Quiz Title: {payload.get("quiz_title")}') current_app.logger.info(f'📥 Total Questions: {len(payload.get("questions", []))}') # Log each question with its options for q_idx, question_data in enumerate(payload.get('questions', [])): current_app.logger.info(f'📥 Question {q_idx + 1}:') current_app.logger.info(f'📥 Type: {question_data.get("question_type")}') current_app.logger.info(f'📥 Text: {question_data.get("question_text", "")[:50]}...') current_app.logger.info(f'📥 Options count: {len(question_data.get("options", []))}') if question_data.get('options'): for opt_idx, option in enumerate(question_data.get('options', [])): current_app.logger.info(f'📥 Option {opt_idx + 1}: "{option.get("option_text", "")[:30]}" (correct: {option.get("is_correct", False)})') else: current_app.logger.warning(f'📥 ⚠️ WARNING: No options for this question!') current_app.logger.info('📥 ===================================================') with DatabaseContextManager() as ctx: # Verify requester is tutor or supervisor requester_id = payload.get('requester_id') requester_type = payload.get('requester_type') if requester_type not in ['tutor', 'supervisor']: return custom_response( success=False, data="Only tutors and supervisors can create quizzes", status_code=403 ) # Verify course exists course_id = payload.get('course_id') course = ctx.session.query(Course).filter(Course.id == course_id).first() if not course: return custom_response( success=False, data="Course not found", status_code=404 ) # Handle module_id and level_id module_id = payload.get('module_id') level_id = payload.get('level_id') # If level_id is provided but not module_id, find or create a quiz module for that level if level_id and not module_id: level = ctx.session.query(CourseLevel).filter( CourseLevel.id == level_id ).first() if not level: return custom_response( success=False, data="Course level not found", status_code=404 ) # Find or create a "Quizzes" module for this level module_title = f"{level.level_name} - Quizzes" module = ctx.session.query(CourseModule).filter( CourseModule.title == module_title ).first() if not module: # Find the first week in this level, or create one from src.models.models import CourseWeek week = ctx.session.query(CourseWeek).filter( CourseWeek.course_level_id == level_id ).first() if not week: # Create a default week for this level week = CourseWeek( id=str(uuid.uuid4()), course_level_id=level_id, week_number=1, week_title=f"{level.level_name} - Week 1", description="Default week for quizzes", is_active=True ) ctx.session.add(week) ctx.session.flush() # Create the quiz module module = CourseModule( id=str(uuid.uuid4()), course_id=course_id, course_week_id=week.id, title=module_title, description=f"Quizzes for {level.level_name} level", sequence=999, is_published=True, is_active=True ) ctx.session.add(module) ctx.session.flush() module_id = module.id # If neither module_id nor level_id is provided, create a course-level quiz module elif not module_id and not level_id: module = ctx.session.query(CourseModule).filter( CourseModule.course_id == course_id, CourseModule.title == "Course Quizzes" ).first() if not module: module = CourseModule( id=str(uuid.uuid4()), course_id=course_id, title="Course Quizzes", description="General course quizzes not associated with a specific module or level", sequence=999, is_published=True, is_active=True ) ctx.session.add(module) ctx.session.flush() module_id = module.id # Verify module exists if module_id is directly provided elif module_id: module = ctx.session.query(CourseModule).filter( CourseModule.id == module_id ).first() if not module: return custom_response( success=False, data="Module not found", status_code=404 ) # Create the quiz quiz = ModuleQuiz( id=str(uuid.uuid4()), module_id=module_id, quiz_title=payload.get('quiz_title', 'Untitled Quiz'), quiz_description=payload.get('quiz_description'), quiz_type=payload.get('quiz_type', 'mixed'), # Default to mixed since it can have multiple question types max_attempts=payload.get('max_attempts', 3), time_limit_minutes=payload.get('time_limit_minutes'), max_points=payload.get('max_points', 100.0), passing_score=payload.get('passing_score', 60.0), is_randomized=payload.get('is_randomized', False), show_correct_answers=payload.get('show_correct_answers', True), show_answers_after=payload.get('show_answers_after', 'immediate'), due_date=datetime.fromisoformat(payload['due_date'].replace('Z', '+00:00')) if payload.get('due_date') else None, is_published=payload.get('is_published', False), created_by=requester_id, is_active=True ) ctx.session.add(quiz) ctx.session.flush() # Get quiz ID for questions # Add questions if provided questions_data = payload.get('questions', []) current_app.logger.info(f"📝 Creating quiz with {len(questions_data)} questions") for question_data in questions_data: question = QuizQuestion( id=str(uuid.uuid4()), quiz_id=quiz.id, question_text=question_data.get('question_text', ''), question_type=question_data.get('question_type', 'multiple_choice'), points=question_data.get('points', 1.0), question_order=question_data.get('question_order', len(questions_data)), explanation=question_data.get('explanation'), is_active=True ) ctx.session.add(question) ctx.session.flush() # Get question ID for options # Add options for multiple choice or true/false questions if question.question_type in ['multiple_choice', 'true_false']: options_data = question_data.get('options', []) current_app.logger.info(f"📝 Question {question.question_type} has {len(options_data)} options") current_app.logger.info(f"📝 Options data: {options_data}") for idx, option_data in enumerate(options_data): try: option_text = option_data.get('option_text', '').strip() # Skip empty options if not option_text: current_app.logger.warning(f"⚠️ Skipping empty option {idx + 1}") continue option = QuizOption( id=str(uuid.uuid4()), question_id=question.id, option_text=option_text, is_correct=option_data.get('is_correct', False), option_order=option_data.get('option_order', idx + 1) ) current_app.logger.info(f"📝 Creating option {idx + 1}: '{option.option_text}' (correct: {option.is_correct}, order: {option.option_order})") ctx.session.add(option) ctx.session.flush() # Force immediate insert to catch constraint errors current_app.logger.info(f"✅ Option {idx + 1} added successfully") except Exception as option_error: current_app.logger.error(f"❌ Error creating option {idx + 1}: {str(option_error)}") import traceback current_app.logger.error(traceback.format_exc()) raise option_error try: ctx.session.commit() current_app.logger.info(f"✅ Quiz created successfully with {len(questions_data)} questions") except Exception as commit_error: current_app.logger.error(f"❌ Error committing quiz: {str(commit_error)}") import traceback current_app.logger.error(traceback.format_exc()) ctx.session.rollback() raise commit_error return custom_response( success=True, data={ 'message': 'Quiz created successfully', 'quiz': self._format_quiz(quiz, include_questions=True) }, status_code=201 ) except Exception as e: current_app.logger.error(f"Error creating quiz: {str(e)}") return custom_response( success=False, data=f"Error creating quiz: {str(e)}", status_code=500 ) def get_quiz(self, quiz_id: str, requester_id: str, requester_type: str) -> Dict: """Get quiz details""" try: with DatabaseContextManager() as ctx: # Load quiz with questions and their options quiz = ctx.session.query(ModuleQuiz).options( joinedload(ModuleQuiz.questions).joinedload(QuizQuestion.options) ).filter(ModuleQuiz.id == quiz_id).first() if not quiz: return custom_response( success=False, data="Quiz not found", status_code=404 ) return custom_response( success=True, data={'quiz': self._format_quiz(quiz, include_questions=True)}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching quiz: {str(e)}") return custom_response( success=False, data=f"Error fetching quiz: {str(e)}", status_code=500 ) def update_quiz(self, quiz_id: str, payload: Dict) -> Dict: """Update quiz details""" try: with DatabaseContextManager() as ctx: quiz = ctx.session.query(ModuleQuiz).filter(ModuleQuiz.id == quiz_id).first() if not quiz: return custom_response( success=False, data="Quiz not found", status_code=404 ) # Update fields if 'quiz_title' in payload: quiz.quiz_title = payload['quiz_title'] if 'quiz_description' in payload: quiz.quiz_description = payload['quiz_description'] if 'max_attempts' in payload: quiz.max_attempts = payload['max_attempts'] if 'time_limit_minutes' in payload: quiz.time_limit_minutes = payload['time_limit_minutes'] if 'max_points' in payload: quiz.max_points = payload['max_points'] if 'passing_score' in payload: quiz.passing_score = payload['passing_score'] if 'is_published' in payload: quiz.is_published = payload['is_published'] if 'due_date' in payload: quiz.due_date = datetime.fromisoformat(payload['due_date'].replace('Z', '+00:00')) if payload['due_date'] else None quiz.updated_at = datetime.utcnow() ctx.session.commit() return custom_response( success=True, data={ 'message': 'Quiz updated successfully', 'quiz': self._format_quiz(quiz) }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error updating quiz: {str(e)}") return custom_response( success=False, data=f"Error updating quiz: {str(e)}", status_code=500 ) def delete_quiz(self, quiz_id: str, requester_id: str, requester_type: str) -> Dict: """Delete a quiz""" try: with DatabaseContextManager() as ctx: quiz = ctx.session.query(ModuleQuiz).filter(ModuleQuiz.id == quiz_id).first() if not quiz: return custom_response( success=False, data="Quiz not found", status_code=404 ) # Soft delete quiz.is_active = False quiz.updated_at = datetime.utcnow() ctx.session.commit() return custom_response( success=True, data={'message': 'Quiz deleted successfully'}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error deleting quiz: {str(e)}") return custom_response( success=False, data=f"Error deleting quiz: {str(e)}", status_code=500 ) def get_quiz_questions(self, quiz_id: str, requester_id: str, requester_type: str) -> Dict: """Get all questions for a quiz""" try: with DatabaseContextManager() as ctx: quiz = ctx.session.query(ModuleQuiz).filter(ModuleQuiz.id == quiz_id).first() if not quiz: return custom_response( success=False, data="Quiz not found", status_code=404 ) # Load questions with their options questions = ctx.session.query(QuizQuestion).options( joinedload(QuizQuestion.options) ).filter( QuizQuestion.quiz_id == quiz_id, QuizQuestion.is_active == True ).order_by(QuizQuestion.question_order).all() return custom_response( success=True, data={'questions': [self._format_question(q) for q in questions]}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching quiz questions: {str(e)}") return custom_response( success=False, data=f"Error fetching questions: {str(e)}", status_code=500 ) def add_question(self, payload: Dict) -> Dict: """Add a question to a quiz with options""" try: with DatabaseContextManager() as ctx: quiz_id = payload.get('quiz_id') quiz = ctx.session.query(ModuleQuiz).filter(ModuleQuiz.id == quiz_id).first() if not quiz: return custom_response( success=False, data="Quiz not found", status_code=404 ) # Get the next question order max_order = ctx.session.query(func.max(QuizQuestion.question_order)).filter( QuizQuestion.quiz_id == quiz_id ).scalar() or 0 question = QuizQuestion( id=str(uuid.uuid4()), quiz_id=quiz_id, question_text=payload['question_text'], question_type=payload.get('question_type', 'multiple_choice'), points=payload.get('points', 1.0), question_order=max_order + 1, explanation=payload.get('explanation'), is_active=True ) ctx.session.add(question) ctx.session.flush() # Get question ID for options # Add options for multiple choice or true/false questions if question.question_type in ['multiple_choice', 'true_false']: options_data = payload.get('options', []) for idx, option_data in enumerate(options_data): option = QuizOption( id=str(uuid.uuid4()), question_id=question.id, option_text=option_data.get('option_text', ''), is_correct=option_data.get('is_correct', False), option_order=idx + 1 ) ctx.session.add(option) ctx.session.commit() return custom_response( success=True, data={ 'message': 'Question added successfully', 'question': self._format_question(question, include_options=True) }, status_code=201 ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error adding question: {str(e)}\n{error_details}") return custom_response( success=False, data=f"Error adding question: {str(e)}", status_code=500 ) def update_question(self, question_id: str, payload: Dict) -> Dict: """Update a question""" try: with DatabaseContextManager() as ctx: question = ctx.session.query(QuizQuestion).filter( QuizQuestion.id == question_id ).first() if not question: return custom_response( success=False, data="Question not found", status_code=404 ) # Update fields if 'question_text' in payload: question.question_text = payload['question_text'] if 'question_type' in payload: question.question_type = payload['question_type'] if 'points' in payload: question.points = payload['points'] if 'explanation' in payload: question.explanation = payload['explanation'] # Handle options update for multiple_choice and true_false questions if 'options' in payload and question.question_type in ['multiple_choice', 'true_false']: # Delete existing options existing_options = ctx.session.query(QuizOption).filter( QuizOption.question_id == question_id ).all() for option in existing_options: ctx.session.delete(option) ctx.session.flush() # Add new options options_data = payload.get('options', []) current_app.logger.info(f"📝 Updating question with {len(options_data)} options") for idx, option_data in enumerate(options_data): try: option_text = option_data.get('option_text', '').strip() # Skip empty options if not option_text: current_app.logger.warning(f"⚠️ Skipping empty option {idx + 1}") continue option = QuizOption( id=str(uuid.uuid4()), question_id=question.id, option_text=option_text, is_correct=option_data.get('is_correct', False), option_order=option_data.get('option_order', idx + 1) ) current_app.logger.info(f"📝 Updating option {idx + 1}: '{option.option_text}' (correct: {option.is_correct}, order: {option.option_order})") ctx.session.add(option) except Exception as option_error: current_app.logger.error(f"❌ Error updating option {idx + 1}: {str(option_error)}") import traceback current_app.logger.error(traceback.format_exc()) raise option_error question.updated_at = datetime.utcnow() try: ctx.session.commit() current_app.logger.info(f"✅ Question updated successfully with options") except Exception as commit_error: current_app.logger.error(f"❌ Error committing question update: {str(commit_error)}") import traceback current_app.logger.error(traceback.format_exc()) ctx.session.rollback() raise commit_error return custom_response( success=True, data={ 'message': 'Question updated successfully', 'question': self._format_question(question) }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error updating question: {str(e)}") return custom_response( success=False, data=f"Error updating question: {str(e)}", status_code=500 ) def delete_question(self, question_id: str, requester_id: str, requester_type: str) -> Dict: """Delete a question""" try: with DatabaseContextManager() as ctx: question = ctx.session.query(QuizQuestion).filter( QuizQuestion.id == question_id ).first() if not question: return custom_response( success=False, data="Question not found", status_code=404 ) # Soft delete question.is_active = False question.updated_at = datetime.utcnow() ctx.session.commit() return custom_response( success=True, data={'message': 'Question deleted successfully'}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error deleting question: {str(e)}") return custom_response( success=False, data=f"Error deleting question: {str(e)}", status_code=500 ) def _notify_tutor_quiz_submission(self, ctx, quiz: ModuleQuiz, attempt: QuizAttempt, student_id: str, quiz_id: str) -> None: """Notify tutors about a quiz submission""" try: # Get the student student = ctx.session.query(Student).filter(Student.id == student_id).first() if not student: return # Get the course from the module module = ctx.session.query(CourseModule).filter(CourseModule.id == quiz.module_id).first() if not module: return # Get course with tutors from src.models.models import Course course = ctx.session.query(Course).options( joinedload(Course.tutors) ).filter(Course.id == module.course_id).first() if not course or not course.tutors: return # Get notification preferences for tutors from src.models.models import NotificationPreference tutors_to_notify = [] for tutor in course.tutors: prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == tutor.id ).first() if prefs and prefs.receive_email: tutors_to_notify.append(tutor) if not tutors_to_notify: return # Prepare email content subject = f"Quiz Submission: {quiz.quiz_title}" for tutor in tutors_to_notify: message = f"""

Quiz Submission Notification

A student has submitted a quiz

Quiz Information

Quiz Title: {quiz.quiz_title}

Course: {course.title}

Course Code: {course.code}

Student Information

Student Name: {student.first_name} {student.last_name}

Student ID: {student.id_number or 'N/A'}

Email: {student.email}

Submission Results

Score: {attempt.total_score:.1f} / {attempt.max_possible_score:.1f}

Percentage: {attempt.percentage_score:.1f}%

Passing Score: {quiz.passing_score}%

Status: {"PASSED" if attempt.is_passed else "FAILED"}

Attempt Number: {attempt.attempt_number}

View Submission Details

Best regards,
{current_app.config['APP_NAME']} Team

This is an automated notification. Please do not reply to this email.

""" try: from src.utils import send_email send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=tutor.email, subject=subject, message=message ) current_app.logger.info(f"✅ Quiz submission notification sent to tutor {tutor.email}") except Exception as e: current_app.logger.error(f"❌ Failed to send quiz notification to tutor {tutor.email}: {str(e)}") except Exception as e: current_app.logger.error(f"❌ Error notifying tutor about quiz submission: {str(e)}") def _format_quiz(self, quiz: ModuleQuiz, include_questions: bool = False) -> Dict: """Format quiz data for API response""" formatted = { 'id': quiz.id, 'module_id': quiz.module_id, 'quiz_title': quiz.quiz_title, 'quiz_description': quiz.quiz_description, 'quiz_type': quiz.quiz_type, 'max_attempts': quiz.max_attempts, 'time_limit_minutes': quiz.time_limit_minutes, 'max_points': quiz.max_points, 'passing_score': quiz.passing_score, 'is_randomized': quiz.is_randomized, 'show_correct_answers': quiz.show_correct_answers, 'show_answers_after': quiz.show_answers_after, 'due_date': quiz.due_date.isoformat() if quiz.due_date else None, 'is_published': quiz.is_published, 'is_active': quiz.is_active, 'created_at': quiz.created_at.isoformat() if quiz.created_at else None, 'updated_at': quiz.updated_at.isoformat() if quiz.updated_at else None } if include_questions and hasattr(quiz, 'questions'): formatted['questions'] = [self._format_question(q) for q in quiz.questions if q.is_active] return formatted def _format_question(self, question: QuizQuestion, include_options: bool = True) -> Dict: """Format question data for API response""" formatted = { 'id': question.id, 'quiz_id': question.quiz_id, 'question_text': question.question_text, 'question_type': question.question_type, 'points': question.points, 'question_order': question.question_order, 'explanation': question.explanation, 'is_active': question.is_active, 'created_at': question.created_at.isoformat() if question.created_at else None, 'updated_at': question.updated_at.isoformat() if question.updated_at else None } # Include options for multiple choice and true/false questions if include_options: if hasattr(question, 'options') and question.options is not None: # Sort options by option_order to ensure correct sequence sorted_options = sorted(question.options, key=lambda opt: opt.option_order) formatted['options'] = [self._format_option(opt) for opt in sorted_options] current_app.logger.info(f"📋 Formatted question {question.id} with {len(formatted['options'])} options") else: # Always include options array, even if empty formatted['options'] = [] current_app.logger.warning(f"⚠️ Question {question.id} has no options attribute or options is None") return formatted def _format_option(self, option: QuizOption) -> Dict: """Format quiz option data for API response""" return { 'id': option.id, 'question_id': option.question_id, 'option_text': option.option_text, 'is_correct': option.is_correct, 'option_order': option.option_order, 'created_at': option.created_at.isoformat() if option.created_at else None } def start_quiz_attempt(self, quiz_id: str, student_id: str) -> Dict: """Start a new quiz attempt without submitting""" try: with DatabaseContextManager() as ctx: quiz = ctx.session.query(ModuleQuiz).filter( ModuleQuiz.id == quiz_id ).first() if not quiz: return custom_response( success=False, data="Quiz not found", status_code=404 ) # Check attempt limit attempts_count = ctx.session.query(QuizAttempt).filter( QuizAttempt.quiz_id == quiz_id, QuizAttempt.student_id == student_id ).count() if attempts_count >= quiz.max_attempts: return custom_response( success=False, data=f"Maximum attempts ({quiz.max_attempts}) reached", status_code=400 ) # Calculate max possible score from quiz questions max_possible_score = sum(q.points for q in quiz.questions) if hasattr(quiz, 'questions') and quiz.questions else 100.0 # Create attempt (not completed yet) attempt = QuizAttempt( id=str(uuid.uuid4()), quiz_id=quiz_id, student_id=student_id, attempt_number=attempts_count + 1, started_at=datetime.utcnow(), is_completed=False, is_submitted=False, max_possible_score=max_possible_score, # Set actual max score total_score=0.0, percentage_score=0.0 ) ctx.session.add(attempt) ctx.session.commit() return custom_response( success=True, data={ 'attempt_id': attempt.id, 'id': attempt.id, 'attempt_number': attempt.attempt_number, 'started_at': attempt.started_at.isoformat() }, status_code=201 ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error starting quiz attempt: {str(e)}\n{error_details}") return custom_response( success=False, data=f"Error starting quiz attempt: {str(e)}", status_code=500 ) def submit_quiz(self, quiz_id: str, student_id: str, answers: List[Dict]) -> Dict: """Submit quiz answers and auto-grade""" try: with DatabaseContextManager() as ctx: quiz = ctx.session.query(ModuleQuiz).options( joinedload(ModuleQuiz.questions).joinedload(QuizQuestion.options) ).filter(ModuleQuiz.id == quiz_id).first() if not quiz: return custom_response( success=False, data="Quiz not found", status_code=404 ) # Check for existing in-progress attempt or create new one attempt = ctx.session.query(QuizAttempt).filter( QuizAttempt.quiz_id == quiz_id, QuizAttempt.student_id == student_id, QuizAttempt.is_completed == False ).order_by(QuizAttempt.started_at.desc()).first() if not attempt: # Check attempt limit attempts_count = ctx.session.query(QuizAttempt).filter( QuizAttempt.quiz_id == quiz_id, QuizAttempt.student_id == student_id ).count() if attempts_count >= quiz.max_attempts: return custom_response( success=False, data=f"Maximum attempts ({quiz.max_attempts}) reached", status_code=400 ) # Create new attempt attempt = QuizAttempt( id=str(uuid.uuid4()), quiz_id=quiz_id, student_id=student_id, attempt_number=attempts_count + 1, started_at=datetime.utcnow(), is_completed=False, is_submitted=False ) ctx.session.add(attempt) ctx.session.flush() # Mark attempt as completed attempt.is_completed = True attempt.is_submitted = True attempt.completed_at = datetime.utcnow() # Grade each answer total_points = 0.0 max_points = 0.0 for answer_data in answers: question_id = answer_data.get('question_id') question = next((q for q in quiz.questions if q.id == question_id), None) if not question: continue max_points += question.points # Determine if answer is correct is_correct = False points_earned = 0.0 if question.question_type in ['multiple_choice', 'true_false']: # Handle both selected_option_id and selected_option_ids selected_option_id = answer_data.get('selected_option_id') if not selected_option_id: # Frontend sends selected_option_ids as array selected_option_ids = answer_data.get('selected_option_ids', []) selected_option_id = selected_option_ids[0] if selected_option_ids else None selected_option = next((opt for opt in question.options if opt.id == selected_option_id), None) if selected_option and selected_option.is_correct: is_correct = True points_earned = question.points # Create answer record answer = QuizAnswer( id=str(uuid.uuid4()), attempt_id=attempt.id, question_id=question_id, selected_option_id=selected_option_id, is_correct=is_correct, points_earned=points_earned ) else: # For short answer or essay, store the text (manual grading needed) answer = QuizAnswer( id=str(uuid.uuid4()), attempt_id=attempt.id, question_id=question_id, answer_text=answer_data.get('answer_text', ''), is_correct=False, # Requires manual grading points_earned=0.0 ) ctx.session.add(answer) total_points += points_earned # Update attempt with scores attempt.total_score = total_points attempt.max_possible_score = max_points attempt.percentage_score = (total_points / max_points * 100) if max_points > 0 else 0 attempt.is_passed = attempt.percentage_score >= quiz.passing_score ctx.session.commit() # Notify tutors about the submission self._notify_tutor_quiz_submission(ctx, quiz, attempt, student_id, quiz_id) return custom_response( success=True, data={ 'id': attempt.id, 'total_score': attempt.total_score, 'max_possible_score': attempt.max_possible_score, 'percentage_score': attempt.percentage_score, 'is_passed': attempt.is_passed, 'attempt_number': attempt.attempt_number, 'message': 'Quiz submitted and graded successfully', 'graded_instantly': True }, status_code=200 ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error submitting quiz: {str(e)}\n{error_details}") return custom_response( success=False, data=f"Error submitting quiz: {str(e)}", status_code=500 ) def get_student_quiz_attempts(self, quiz_id: str, student_id: str) -> Dict: """Get all attempts for a specific quiz by a specific student""" try: with DatabaseContextManager() as ctx: attempts = ctx.session.query(QuizAttempt).filter( QuizAttempt.quiz_id == quiz_id, QuizAttempt.student_id == student_id ).order_by(QuizAttempt.attempt_number.desc()).all() attempts_data = [] for attempt in attempts: attempts_data.append({ 'id': attempt.id, 'quiz_id': attempt.quiz_id, 'student_id': attempt.student_id, 'attempt_number': attempt.attempt_number, 'started_at': attempt.started_at.isoformat() if attempt.started_at else None, 'completed_at': attempt.completed_at.isoformat() if attempt.completed_at else None, 'time_spent_minutes': attempt.time_spent_minutes, 'total_score': attempt.total_score, 'max_possible_score': attempt.max_possible_score, 'percentage_score': attempt.percentage_score, 'is_passed': attempt.is_passed, 'is_completed': attempt.is_completed, 'is_submitted': attempt.is_submitted, 'is_graded': attempt.is_graded }) current_app.logger.info(f"Found {len(attempts_data)} attempts for quiz {quiz_id} by student {student_id}") return custom_response( success=True, data={ 'quiz_id': quiz_id, 'student_id': student_id, 'total_attempts': len(attempts_data), 'attempts': attempts_data }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching student quiz attempts: {str(e)}") return custom_response( success=False, data=f"Error fetching attempts: {str(e)}", status_code=500 ) # Required abstract method implementations def create(self, payload: Dict[str, Any]) -> Dict[str, Any]: """Create method required by ApiABC""" return self.create_quiz(payload) def get(self, id: str) -> Dict[str, Any]: """Get method required by ApiABC""" return self.get_quiz(id, None, None) def update(self, payload: Dict[str, Any]) -> Dict[str, Any]: """Update method required by ApiABC""" if 'id' not in payload: return custom_response(success=False, data="Quiz ID is required for update") return self.update_quiz(payload['id'], payload) def fetchAll(self) -> Dict[str, Any]: """Fetch all method required by ApiABC""" return custom_response(success=False, data="Use get_course_quizzes instead") def delete(self, id: str) -> Dict[str, Any]: """Delete method required by ApiABC""" return self.delete_quiz(id, None, None)