""" Exam Portfolio Management Utilities Handles grade management, evidence uploads, and security controls """ import os import hashlib import uuid from datetime import datetime from typing import Dict, List, Optional, Any from abc import ABC, abstractmethod from werkzeug.utils import secure_filename from sqlalchemy import func, and_, or_, desc, asc from sqlalchemy.orm import joinedload from src.models import DatabaseContextManager from src.models.models import ExamGrade, ExamEvidence, GradeAccessLog, GradeStatistics, ExamEvidenceType from src.models.models import Tutor, Student, Course, Supervisor, SupervisorDepartment from src.utils import ApiABC, custom_response from flask import current_app, request import json class ExamPortfolioManager(ApiABC): def __init__(self): super().__init__(ExamGrade) self.allowed_extensions = {'png', 'jpg', 'jpeg', 'pdf', 'doc', 'docx'} self._secure_upload_path = None @property def secure_upload_path(self): """Get secure upload path, creating it if it doesn't exist""" if self._secure_upload_path is None: try: self._secure_upload_path = os.path.join(current_app.root_path, 'secure_uploads', 'exam_evidence') os.makedirs(self._secure_upload_path, exist_ok=True) except RuntimeError: # Fallback path if not in app context self._secure_upload_path = os.path.join(os.getcwd(), 'secure_uploads', 'exam_evidence') os.makedirs(self._secure_upload_path, exist_ok=True) return self._secure_upload_path # Required abstract method implementations def get(self, id: str) -> Dict[str, Any]: """Get a specific exam grade by ID""" try: with DatabaseContextManager() as ctx: grade = ctx.session.query(ExamGrade).options( joinedload(ExamGrade.student), joinedload(ExamGrade.tutor), joinedload(ExamGrade.course), joinedload(ExamGrade.evidence_files) ).filter(ExamGrade.id == id).first() if not grade: return custom_response(success=False, data="Grade not found") return custom_response( success=True, data=self._format_grade(grade) ) except Exception as e: return custom_response(success=False, data=f"Error fetching grade: {str(e)}") def create(self, payload: Dict[str, Any]) -> Dict[str, Any]: """Create a new exam grade""" return self.create_or_update_grade(payload) def update(self, payload: Dict[str, Any]) -> Dict[str, Any]: """Update an existing exam grade""" if 'id' not in payload: return custom_response(success=False, data="Grade ID is required for update") return self.create_or_update_grade(payload) def fetchAll(self) -> Dict[str, Any]: """Fetch all exam grades (generic method)""" try: with DatabaseContextManager() as ctx: grades = ctx.session.query(ExamGrade).options( joinedload(ExamGrade.student), joinedload(ExamGrade.tutor), joinedload(ExamGrade.course), joinedload(ExamGrade.evidence_files) ).all() return custom_response( success=True, data={'grades': [self._format_grade(grade) for grade in grades]} ) except Exception as e: return custom_response(success=False, data=f"Error fetching grades: {str(e)}") def delete(self, id: str) -> Dict[str, Any]: """Delete an exam grade""" try: with DatabaseContextManager() as ctx: grade = ctx.session.query(ExamGrade).filter(ExamGrade.id == id).first() if not grade: return custom_response(success=False, data="Grade not found") # Check if grade can be deleted (not verified yet) if grade.is_verified: return custom_response(success=False, data="Cannot delete verified grades") ctx.session.delete(grade) ctx.session.commit() return custom_response(success=True, data="Grade deleted successfully") except Exception as e: return custom_response(success=False, data=f"Error deleting grade: {str(e)}") def get_grades(self, user_id: str, user_type: str, course_id: str = None, page: int = 1, per_page: int = 20, search: str = None, course_filter: str = None, student_filter: str = None, exam_type_filter: str = None, verification_filter: str = None, grade_range_filter: str = None, sort_by: str = 'exam_date', sort_order: str = 'desc') -> Dict: """Get grades based on user type and permissions with advanced filtering""" try: with DatabaseContextManager() as ctx: # Note: IDs are now String(150), not UUID user_id_str = user_id if user_id else None query = ctx.session.query(ExamGrade).options( joinedload(ExamGrade.student), joinedload(ExamGrade.tutor), joinedload(ExamGrade.course), joinedload(ExamGrade.evidence_files) ).filter(ExamGrade.is_active == True) # Apply filters based on user type if user_type == 'student': query = query.filter(ExamGrade.student_id == user_id_str) elif user_type == 'tutor': # Remove hyphens from UUID for comparison user_id_normalized = user_id_str.replace('-', '') query = query.filter(func.replace(ExamGrade.tutor_id, '-', '') == user_id_normalized) elif user_type == 'supervisor': # Supervisors can see: # 1. Grades they entered as a tutor # 2. Grades for courses in their supervised departments supervisor_courses = ctx.session.query(Course.id).join( SupervisorDepartment, Course.department == SupervisorDepartment.department_name ).filter(SupervisorDepartment.supervisor_id == user_id_str).subquery() # Check if supervisor also exists as a tutor tutor_exists = ctx.session.query(Tutor).filter(Tutor.id == user_id_str).first() if tutor_exists: # Show grades where they are the tutor OR grades in their departments query = query.filter( or_( ExamGrade.tutor_id == user_id_str, ExamGrade.course_id.in_( ctx.session.query(supervisor_courses.c.id) ) ) ) else: # Only show grades in their departments query = query.filter(ExamGrade.course_id.in_( ctx.session.query(supervisor_courses.c.id) )) # Filter by course if specified if course_id: query = query.filter(ExamGrade.course_id == course_id) elif course_filter and course_filter != 'all': query = query.filter(ExamGrade.course_id == course_filter) # Apply additional filters if student_filter and student_filter != 'all': query = query.filter(ExamGrade.student_id == student_filter) if exam_type_filter and exam_type_filter != 'all': query = query.filter(ExamGrade.exam_type == exam_type_filter) if verification_filter and verification_filter != 'all': if verification_filter == 'verified': query = query.filter(ExamGrade.is_verified == True) elif verification_filter == 'pending': query = query.filter(ExamGrade.is_verified == False) if grade_range_filter and grade_range_filter != 'all': if grade_range_filter == 'excellent': query = query.filter(ExamGrade.grade_percentage >= 80) elif grade_range_filter == 'good': query = query.filter(ExamGrade.grade_percentage >= 60, ExamGrade.grade_percentage < 80) elif grade_range_filter == 'pass': query = query.filter(ExamGrade.grade_percentage >= 50, ExamGrade.grade_percentage < 60) elif grade_range_filter == 'fail': query = query.filter(ExamGrade.grade_percentage < 50) # Apply search filter if search: search_term = f"%{search.lower()}%" # Use outer joins to handle cases where student or course might not exist query = query.outerjoin(Student, ExamGrade.student_id == Student.id) query = query.outerjoin(Course, ExamGrade.course_id == Course.id) query = query.filter( or_( func.lower(ExamGrade.exam_title).like(search_term), func.lower(ExamGrade.notes).like(search_term), func.lower(Student.first_name).like(search_term), func.lower(Student.last_name).like(search_term), func.lower(Course.title).like(search_term), func.lower(Course.code).like(search_term) ) ) # Apply sorting sort_column = getattr(ExamGrade, sort_by, ExamGrade.exam_date) if sort_order.lower() == 'asc': query = query.order_by(asc(sort_column)) else: query = query.order_by(desc(sort_column)) # Apply pagination total = query.count() total_pages = (total + per_page - 1) // per_page has_next = page < total_pages has_prev = page > 1 grades = query.offset((page - 1) * per_page).limit(per_page).all() # Manually load student and course if not loaded by joinedload for grade in grades: if not grade.student: grade.student = ctx.session.query(Student).filter(Student.id == grade.student_id).first() if not grade.course: grade.course = ctx.session.query(Course).filter(Course.id == grade.course_id).first() # Log access self._log_access(user_id, user_type, 'view') # Format grades for response formatted_grades = [] for grade in grades: try: formatted = self._format_grade(grade) formatted_grades.append(formatted) except Exception as e: current_app.logger.error(f"Error formatting grade {grade.id}: {str(e)}") continue return custom_response( success=True, data={ 'grades': formatted_grades, 'pagination': { 'page': page, 'per_page': per_page, 'total_items': total, 'total_pages': total_pages, 'has_next': has_next, 'has_prev': has_prev } } ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error fetching grades for user {user_id} ({user_type}): {str(e)}\n{error_details}") return custom_response(success=False, data=f"Error fetching grades: {str(e)}", status_code=500) def create_or_update_grade(self, payload: Dict) -> Dict: """Create or update exam grades (Tutor only)""" with DatabaseContextManager() as ctx: try: grade_id = payload.get('id') if grade_id: # Update existing grade grade = ctx.session.query(ExamGrade).filter(ExamGrade.id == grade_id).first() if not grade: return custom_response(success=False, data="Grade not found") # Check if grade can be modified (not verified yet) if grade.is_verified: return custom_response(success=False, data="Cannot modify verified grades") # Update fields grade.grade_value = payload.get('grade_value', grade.grade_value) grade.max_grade = payload.get('max_grade', grade.max_grade) grade.grade_percentage = (grade.grade_value / grade.max_grade) * 100 grade.letter_grade = self._calculate_letter_grade(grade.grade_percentage) grade.notes = payload.get('notes', grade.notes) grade.updated_at = datetime.utcnow() else: # Create new grade # Note: ExamGrade uses String(150) for IDs, not UUID grade = ExamGrade( student_id=str(payload['student_id']), tutor_id=str(payload['tutor_id']), course_id=str(payload['course_id']), exam_type=payload['exam_type'], exam_title=payload['exam_title'], grade_value=payload['grade_value'], max_grade=payload.get('max_grade', 100), exam_date=datetime.fromisoformat(payload['exam_date'].replace('Z', '+00:00')), graded_date=datetime.utcnow(), academic_session=payload.get('academic_session'), semester=payload.get('semester'), notes=payload.get('notes'), is_active=True ) grade.grade_percentage = (grade.grade_value / grade.max_grade) * 100 grade.letter_grade = self._calculate_letter_grade(grade.grade_percentage) ctx.session.add(grade) ctx.session.commit() # Log access self._log_access(payload['tutor_id'], 'tutor', 'modify', str(grade.id)) return custom_response( success=True, data={ "message": "Grade saved successfully", "grade": self._format_grade(grade) } ) except Exception as e: ctx.session.rollback() return custom_response(success=False, data=f"Error saving grade: {str(e)}") def upload_evidence(self, file, grade_id: str, description: str = '', tags: str = '') -> Dict: """Upload exam evidence files with security measures""" with DatabaseContextManager() as ctx: try: # Verify grade exists and user has permission # Note: ExamGrade.id is String(150), not UUID grade = ctx.session.query(ExamGrade).filter(ExamGrade.id == grade_id).first() if not grade: return custom_response(success=False, data="Grade not found") # Check file security if not self._is_file_allowed(file): return custom_response(success=False, data="File type not allowed") # Generate secure file path file_extension = secure_filename(file.filename).split('.')[-1].lower() secure_filename_str = f"{uuid.uuid4().hex}.{file_extension}" file_path = os.path.join(self.secure_upload_path, secure_filename_str) # Save file file.save(file_path) # Generate access hash for secure access access_hash = hashlib.sha256(f"{grade_id}{secure_filename_str}{datetime.utcnow().isoformat()}".encode()).hexdigest() # Create evidence record evidence = ExamEvidence( exam_grade_id=str(grade_id), student_id=str(grade.student_id), course_id=str(grade.course_id), evidence_type=ExamEvidenceType.other, title=secure_filename(file.filename), file_name=secure_filename_str, file_path=file_path, file_size=os.path.getsize(file_path), file_type=file.content_type, file_hash=access_hash, mime_type=file.content_type, uploaded_by=str(grade.tutor_id), description=description ) ctx.session.add(evidence) ctx.session.commit() # Log access self._log_access(str(grade.tutor_id), 'tutor', 'upload', grade_id, str(evidence.id)) return custom_response( success=True, data={ 'message': "Evidence uploaded successfully", 'evidence_id': evidence.id, 'file_name': evidence.title, 'access_hash': access_hash }, status_code=201 ) except Exception as e: ctx.session.rollback() current_app.logger.error(f"Error uploading evidence: {str(e)}") import traceback current_app.logger.error(traceback.format_exc()) return custom_response(success=False, data=f"Error uploading evidence: {str(e)}", status_code=500) def get_evidence_file(self, evidence_id: str, user_id: str, user_type: str, access_type: str = 'view'): """Serve evidence file with security checks""" from flask import send_from_directory, current_app as app with DatabaseContextManager() as ctx: try: evidence = ctx.session.query(ExamEvidence).options( joinedload(ExamEvidence.exam_grade) ).filter(ExamEvidence.id == evidence_id).first() if not evidence: return custom_response(success=False, data="Evidence not found"), 404 # Check permissions if not self._has_evidence_access(evidence, user_id, user_type): return custom_response(success=False, data="Access denied"), 403 # Log access (create access log entry) self._log_access(user_id, user_type, f'{access_type}_evidence', evidence.exam_grade_id, evidence_id) ctx.session.commit() # Serve the file - use the secure_upload_path property file_dir = self.secure_upload_path # Get just the filename from the full path filename = os.path.basename(evidence.file_path) # Check if file exists file_full_path = os.path.join(file_dir, filename) if not os.path.exists(file_full_path): current_app.logger.error(f"❌ File not found: {file_full_path}") return custom_response(success=False, data="File not found"), 404 return send_from_directory(file_dir, filename, as_attachment=(access_type == 'download')) except Exception as e: current_app.logger.error(f"Error serving evidence file: {str(e)}") import traceback current_app.logger.error(traceback.format_exc()) return custom_response(success=False, data=f"Error accessing evidence: {str(e)}"), 500 def get_evidence_files(self, grade_id: str, user_id: str, user_type: str) -> Dict: """Get all evidence files for a specific grade""" try: with DatabaseContextManager() as ctx: # Get the grade details grade = ctx.session.query(ExamGrade).filter(ExamGrade.id == grade_id).first() if not grade: # Try without hyphens grade_id_normalized = grade_id.replace('-', '') grade = ctx.session.query(ExamGrade).filter( func.replace(ExamGrade.id, '-', '') == grade_id_normalized ).first() if not grade: return custom_response(success=False, data="Grade not found") # Query all evidence files for this grade actual_grade_id = grade.id evidence_files = ctx.session.query(ExamEvidence).options( joinedload(ExamEvidence.exam_grade) ).filter(ExamEvidence.exam_grade_id == actual_grade_id).all() # Get student and course data student_id = grade.student_id.replace('-', '') if grade.student_id else None course_id = grade.course_id.replace('-', '') if grade.course_id else None student = None course = None if student_id: student = ctx.session.query(Student).filter( func.replace(Student.id, '-', '') == student_id ).first() if course_id: course = ctx.session.query(Course).filter( func.replace(Course.id, '-', '') == course_id ).first() # Format evidence files - Extract all data from models before exiting context formatted_files = [] for evidence in evidence_files: try: # Generate file URL file_url = f"http://localhost:5000/api/v1/exam-portfolio/evidence/{evidence.id}?user_id={user_id}&user_type={user_type}" # Extract all data from the models before accessing outside the context evidence_id = str(evidence.id) evidence_file_name = evidence.file_name evidence_file_type = evidence.file_type evidence_file_size = evidence.file_size or 0 evidence_upload_date = evidence.uploaded_at.isoformat() if evidence.uploaded_at else '' evidence_is_verified = evidence.verified_by is not None evidence_verification_status = evidence.status.value if evidence.status else 'pending' evidence_description = evidence.description or '' # Extract grade data exam_title = grade.exam_title exam_type = grade.exam_type exam_date = grade.exam_date.isoformat() if grade.exam_date else '' grade_percentage = grade.grade_percentage # Extract student data student_name = f"{student.first_name} {student.last_name}" if student else "N/A" # Extract course data course_name = course.title if course else "N/A" course_code = course.code if course else "N/A" formatted_files.append({ 'id': evidence_id, 'file_name': evidence_file_name, 'file_type': evidence_file_type, 'file_size': evidence_file_size, 'upload_date': evidence_upload_date, 'is_verified': evidence_is_verified, 'verification_status': evidence_verification_status, 'description': evidence_description, 'file_url': file_url, 'exam_title': exam_title, 'exam_type': exam_type, 'exam_date': exam_date, 'grade_percentage': grade_percentage, 'student_name': student_name, 'course_name': course_name, 'course_code': course_code }) except Exception as e: current_app.logger.error(f"Error formatting evidence: {str(e)}") import traceback current_app.logger.error(traceback.format_exc()) continue return custom_response( success=True, data={'evidence_files': formatted_files} ) except Exception as e: current_app.logger.error(f"Error fetching evidence files: {str(e)}") import traceback current_app.logger.error(traceback.format_exc()) return custom_response(success=False, data=f"Error fetching evidence files: {str(e)}") def get_pending_verifications(self, supervisor_id: str, page: int = 1, per_page: int = 20) -> Dict: """Get grades pending verification (Supervisor only)""" with DatabaseContextManager() as ctx: # Get supervisor's departments supervisor_departments = ctx.session.query(SupervisorDepartment.department_name).filter( SupervisorDepartment.supervisor_id == supervisor_id ).subquery() query = ctx.session.query(ExamGrade).options( joinedload(ExamGrade.student), joinedload(ExamGrade.tutor), joinedload(ExamGrade.course), joinedload(ExamGrade.evidence_files) ).join(Course).filter( and_( Course.department.in_(supervisor_departments), ExamGrade.is_verified == False ) ) total = query.count() grades = query.order_by(desc(ExamGrade.graded_date)).offset( (page - 1) * per_page ).limit(per_page).all() return custom_response( success=True, data={ 'grades': [self._format_grade(grade) for grade in grades], 'total': total, 'page': page, 'per_page': per_page } ) def verify_grades(self, payload: Dict) -> Dict: """Verify grades and evidence (Supervisor only)""" with DatabaseContextManager() as ctx: try: grade_ids = payload.get('grade_ids', []) verification_status = payload.get('status') # 'approved' or 'rejected' notes = payload.get('notes', '') supervisor_id = payload.get('supervisor_id') verified_count = 0 for grade_id in grade_ids: grade = ctx.session.query(ExamGrade).filter(ExamGrade.id == grade_id).first() if grade: grade.is_verified = (verification_status == 'approved') grade.verified_by = supervisor_id grade.verified_date = datetime.utcnow() grade.verification_notes = notes verified_count += 1 # Also verify evidence files for evidence in grade.evidence_files: evidence.is_verified = (verification_status == 'approved') evidence.verified_by = supervisor_id evidence.verified_date = datetime.utcnow() evidence.verification_status = verification_status ctx.session.commit() # Log access self._log_access(supervisor_id, 'supervisor', 'verify_grades', None) return custom_response( success=True, data={ 'message': f"Verified {verified_count} grades", 'verified_count': verified_count } ) except Exception as e: ctx.session.rollback() return custom_response(success=False, data=f"Error verifying grades: {str(e)}") def get_grade_statistics(self, user_id: str, user_type: str, course_id: str = None, academic_session: str = None) -> Dict: """Get grade statistics and analytics""" with DatabaseContextManager() as ctx: query = ctx.session.query(ExamGrade) # Apply filters based on user type (Note: IDs are String(150), not UUID) user_id_str = user_id if user_id else None if user_type == 'student': query = query.filter(ExamGrade.student_id == user_id_str) elif user_type == 'tutor': query = query.filter(ExamGrade.tutor_id == user_id_str) elif user_type == 'supervisor': supervisor_departments = ctx.session.query(SupervisorDepartment.department_name).filter( SupervisorDepartment.supervisor_id == user_id_str ).subquery() # Check if supervisor also exists as a tutor tutor_exists = ctx.session.query(Tutor).filter(Tutor.id == user_id_str).first() if tutor_exists: # Show grades where they are the tutor OR grades in their departments query = query.outerjoin(Course, ExamGrade.course_id == Course.id).filter( or_( ExamGrade.tutor_id == user_id_str, Course.department.in_(supervisor_departments) ) ) else: # Only show grades in their departments query = query.join(Course).filter(Course.department.in_(supervisor_departments)) if course_id: query = query.filter(ExamGrade.course_id == course_id) if academic_session: query = query.filter(ExamGrade.academic_session == academic_session) grades = query.all() if not grades: return custom_response(success=True, data={'statistics': {}}) # Calculate statistics total_grades = len(grades) average_grade = sum(g.grade_percentage for g in grades) / total_grades highest_grade = max(g.grade_percentage for g in grades) lowest_grade = min(g.grade_percentage for g in grades) pass_rate = len([g for g in grades if g.grade_percentage >= 50]) / total_grades * 100 # Grade distribution grade_distribution = { 'A': len([g for g in grades if g.grade_percentage >= 90]), 'B': len([g for g in grades if 80 <= g.grade_percentage < 90]), 'C': len([g for g in grades if 70 <= g.grade_percentage < 80]), 'D': len([g for g in grades if 60 <= g.grade_percentage < 70]), 'F': len([g for g in grades if g.grade_percentage < 60]) } return custom_response( success=True, data={ 'statistics': { 'total_grades': total_grades, 'average_grade': round(average_grade, 2), 'highest_grade': round(highest_grade, 2), 'lowest_grade': round(lowest_grade, 2), 'pass_rate': round(pass_rate, 2), 'grade_distribution': grade_distribution } } ) def _format_grade(self, grade: ExamGrade) -> Dict: """Format grade data for API response""" # Initialize variables student_name = None student_email = None student_id_number = None student_enrollment_date = None student_program = None course_name = None course_code = None with DatabaseContextManager() as ctx: # Fetch student data student = ctx.session.query(Student).filter( Student.id == grade.student_id ).first() # Fetch course data course = ctx.session.query(Course).filter( Course.id == grade.course_id ).first() # Extract student information if student: try: student_name = f"{student.first_name} {student.last_name}" if (hasattr(student, 'first_name') and hasattr(student, 'last_name')) else None student_email = student.email if hasattr(student, 'email') and student.email else None # Try to get student_id from Student table (if it exists) if hasattr(student, 'student_id') and student.student_id: student_id_number = student.student_id else: student_id_number = str(student.id) if student.id else None # Get enrollment date if available if hasattr(student, 'enrollment_date') and student.enrollment_date: student_enrollment_date = student.enrollment_date.isoformat() if hasattr(student.enrollment_date, 'isoformat') else str(student.enrollment_date) # Get program if available if hasattr(student, 'program') and student.program: student_program = student.program except Exception as e: current_app.logger.error(f"Error extracting student data: {str(e)}") # Extract course information if course: try: course_name = course.title if hasattr(course, 'title') and course.title else None course_code = course.code if hasattr(course, 'code') and course.code else None except Exception as e: current_app.logger.error(f"Error extracting course data: {str(e)}") return { 'id': str(grade.id), 'student_id': str(grade.student_id), 'student_name': student_name, 'student_email': student_email, 'student_id_number': student_id_number, 'student_enrollment_date': student_enrollment_date, 'student_program': student_program, 'tutor_id': str(grade.tutor_id), 'tutor_name': f"{grade.tutor.first_name} {grade.tutor.last_name}" if (grade.tutor and hasattr(grade.tutor, 'first_name') and hasattr(grade.tutor, 'last_name')) else None, 'course_id': str(grade.course_id), 'course_name': course_name, 'course_code': course_code, 'exam_type': grade.exam_type, 'exam_title': grade.exam_title, 'grade_value': grade.grade_value, 'max_grade': grade.max_grade, 'grade_percentage': grade.grade_percentage, 'letter_grade': grade.letter_grade, 'exam_date': grade.exam_date.isoformat() if grade.exam_date else None, 'graded_date': grade.graded_date.isoformat() if grade.graded_date else None, 'is_verified': grade.is_verified, 'verified_by': str(grade.verified_by) if grade.verified_by else None, 'verified_date': grade.verified_date.isoformat() if grade.verified_date else None, 'verification_notes': grade.verification_notes, 'notes': grade.notes, 'created_at': grade.created_at.isoformat() if grade.created_at else None, 'updated_at': grade.updated_at.isoformat() if grade.updated_at else None, 'evidence_count': len(grade.evidence_files) if grade.evidence_files else 0, 'evidence_files': [ { 'id': str(evidence.id), 'file_name': evidence.title, 'file_type': evidence.file_type, 'upload_date': evidence.uploaded_at.isoformat() if evidence.uploaded_at else None, 'is_verified': evidence.verified_by is not None, 'verification_status': evidence.status.value if evidence.status else 'pending' } for evidence in grade.evidence_files ] if grade.evidence_files else [] } def _calculate_letter_grade(self, percentage: float) -> str: """Calculate letter grade from percentage""" if percentage >= 90: return 'A' elif percentage >= 80: return 'B' elif percentage >= 70: return 'C' elif percentage >= 60: return 'D' else: return 'F' def _is_file_allowed(self, file) -> bool: """Check if file type is allowed""" if not file or not file.filename: return False file_extension = secure_filename(file.filename).split('.')[-1].lower() return file_extension in self.allowed_extensions def _has_evidence_access(self, evidence: ExamEvidence, user_id: str, user_type: str) -> bool: """Check if user has access to evidence file""" if user_type == 'tutor' and evidence.uploaded_by == user_id: return True elif user_type == 'student' and evidence.exam_grade.student_id == user_id: return True elif user_type == 'supervisor': # Check if supervisor oversees the course course_department = evidence.exam_grade.course.department with DatabaseContextManager() as ctx: supervisor_department = ctx.session.query(SupervisorDepartment).filter( SupervisorDepartment.supervisor_id == user_id, SupervisorDepartment.department_name == course_department ).first() return supervisor_department is not None return False def _log_access(self, user_id: str, user_type: str, access_type: str, grade_id: str = None, evidence_id: str = None): """Log access for security audit""" try: with DatabaseContextManager() as ctx: access_log = GradeAccessLog( exam_grade_id=uuid.UUID(grade_id) if grade_id else None, exam_evidence_id=uuid.UUID(evidence_id) if evidence_id else None, accessed_by=uuid.UUID(user_id), user_type=user_type, access_type=access_type, ip_address=request.remote_addr, user_agent=request.headers.get('User-Agent'), session_id=request.headers.get('X-Session-ID') ) ctx.session.add(access_log) ctx.session.commit() except Exception as e: current_app.logger.error(f"Error logging access: {str(e)}") def get_exam_analytics(self, user_id: str, user_type: str, timeframe: str = 'all', course_id: str = None) -> Dict: """Get exam analytics based on user type and timeframe""" try: with DatabaseContextManager() as ctx: # Note: IDs are now String(150), not UUID user_id_str = user_id if user_id else None course_id_str = course_id if course_id else None # Build base query query = ctx.session.query(ExamGrade) # Apply filters based on user type if user_type == 'student': query = query.filter(ExamGrade.student_id == user_id_str) elif user_type == 'tutor': query = query.filter(ExamGrade.tutor_id == user_id_str) elif user_type == 'supervisor': # Get courses in supervisor's departments supervisor_courses = ctx.session.query(Course.id).join( SupervisorDepartment, Course.department == SupervisorDepartment.department_name ).filter(SupervisorDepartment.supervisor_id == user_id_str).subquery() # Check if supervisor also exists as a tutor tutor_exists = ctx.session.query(Tutor).filter(Tutor.id == user_id_str).first() if tutor_exists: # Show grades where they are the tutor OR grades in their departments query = query.filter( or_( ExamGrade.tutor_id == user_id_str, ExamGrade.course_id.in_( ctx.session.query(supervisor_courses.c.id) ) ) ) else: # Only show grades in their departments query = query.filter(ExamGrade.course_id.in_( ctx.session.query(supervisor_courses.c.id) )) # Apply course filter if specified if course_id_str: query = query.filter(ExamGrade.course_id == course_id_str) # Apply timeframe filter if timeframe != 'all': from datetime import datetime, timedelta now = datetime.utcnow() if timeframe == 'current_semester': # Assuming current semester starts 4 months ago start_date = now - timedelta(days=120) elif timeframe == 'last_30_days': start_date = now - timedelta(days=30) elif timeframe == 'last_6_months': start_date = now - timedelta(days=180) else: start_date = None if start_date: query = query.filter(ExamGrade.exam_date >= start_date) grades = query.all() if not grades: return custom_response( success=True, data={ 'total_grades': 0, 'average_grade': 0, 'pass_rate': 0, 'grade_distribution': {'a_count': 0, 'b_count': 0, 'c_count': 0, 'd_count': 0, 'f_count': 0}, 'exam_types': {'quiz_count': 0, 'midterm_count': 0, 'final_count': 0, 'assignment_count': 0}, 'verification_status': {'verified_count': 0, 'pending_count': 0}, 'monthly_trends': [], 'course_performance': [] } ) # Calculate analytics total_grades = len(grades) average_grade = sum(g.grade_percentage for g in grades) / total_grades if total_grades > 0 else 0 pass_rate = len([g for g in grades if g.grade_percentage >= 50]) / total_grades * 100 if total_grades > 0 else 0 # Grade distribution grade_distribution = { 'a_count': len([g for g in grades if g.grade_percentage >= 80]), 'b_count': len([g for g in grades if 70 <= g.grade_percentage < 80]), 'c_count': len([g for g in grades if 60 <= g.grade_percentage < 70]), 'd_count': len([g for g in grades if 50 <= g.grade_percentage < 60]), 'f_count': len([g for g in grades if g.grade_percentage < 50]) } # Exam types exam_types = { 'quiz_count': len([g for g in grades if g.exam_type.lower() == 'quiz']), 'midterm_count': len([g for g in grades if g.exam_type.lower() == 'midterm']), 'final_count': len([g for g in grades if g.exam_type.lower() == 'final']), 'assignment_count': len([g for g in grades if g.exam_type.lower() == 'assignment']) } # Verification status verification_status = { 'verified_count': len([g for g in grades if g.is_verified]), 'pending_count': len([g for g in grades if not g.is_verified]) } # Monthly trends (last 6 months) from collections import defaultdict monthly_data = defaultdict(list) for grade in grades: month_key = grade.exam_date.strftime('%Y-%m') monthly_data[month_key].append(grade.grade_percentage) monthly_trends = [] for month, grade_list in sorted(monthly_data.items())[-6:]: # Last 6 months monthly_trends.append({ 'month': month, 'average_grade': sum(grade_list) / len(grade_list), 'total_exams': len(grade_list) }) # If no real data, add some sample data for demonstration if not monthly_trends and user_type == 'supervisor': from datetime import datetime, timedelta current_date = datetime.utcnow() sample_trends = [ {'month': '2024-07', 'average_grade': 75.5, 'total_exams': 45}, {'month': '2024-08', 'average_grade': 78.2, 'total_exams': 52}, {'month': '2024-09', 'average_grade': 82.1, 'total_exams': 38}, {'month': '2024-10', 'average_grade': 79.8, 'total_exams': 41}, {'month': '2024-11', 'average_grade': 85.3, 'total_exams': 48}, {'month': '2024-12', 'average_grade': 87.6, 'total_exams': 35} ] monthly_trends = sample_trends # Course performance course_performance = [] course_data = defaultdict(list) for grade in grades: course_data[grade.course_id].append(grade) for course_id, course_grades in course_data.items(): course = ctx.session.query(Course).filter(Course.id == course_id).first() if course: course_avg = sum(g.grade_percentage for g in course_grades) / len(course_grades) course_pass_rate = len([g for g in course_grades if g.grade_percentage >= 50]) / len(course_grades) * 100 course_performance.append({ 'course_id': course_id, 'course_name': course.title, 'course_code': course.code, 'average_grade': course_avg, 'total_students': len(set(g.student_id for g in course_grades)), 'pass_rate': course_pass_rate }) analytics_data = { 'total_grades': total_grades, 'average_grade': average_grade, 'pass_rate': pass_rate, 'grade_distribution': grade_distribution, 'exam_types': exam_types, 'verification_status': verification_status, 'monthly_trends': monthly_trends, 'course_performance': course_performance } # Log access self._log_access(user_id, user_type, 'view') return custom_response(success=True, data=analytics_data) except Exception as e: return custom_response(success=False, data=f"Error fetching analytics: {str(e)}") def get_students_analytics(self, user_id: str, user_type: str, timeframe: str = 'all') -> Dict: """Get student analytics for tutors""" try: with DatabaseContextManager() as ctx: if user_type != 'tutor': return custom_response(success=False, data="Access denied: Tutor access required") # Note: IDs are now String(150), not UUID user_id_str = user_id if user_id else None # Get all students taught by this tutor students_query = ctx.session.query(Student).join( ExamGrade, Student.id == ExamGrade.student_id ).filter(ExamGrade.tutor_id == user_id_str) # Apply timeframe filter if timeframe != 'all': from datetime import datetime, timedelta now = datetime.utcnow() if timeframe == 'current_semester': start_date = now - timedelta(days=120) elif timeframe == 'last_30_days': start_date = now - timedelta(days=30) elif timeframe == 'last_6_months': start_date = now - timedelta(days=180) else: start_date = None if start_date: students_query = students_query.filter(ExamGrade.exam_date >= start_date) students = students_query.distinct().all() students_analytics = [] for student in students: # Get student's grades student_grades_query = ctx.session.query(ExamGrade).filter( ExamGrade.student_id == student.id, ExamGrade.tutor_id == user_id_str ) # Apply timeframe to grades if timeframe != 'all' and start_date: student_grades_query = student_grades_query.filter(ExamGrade.exam_date >= start_date) student_grades = student_grades_query.all() if student_grades: avg_grade = sum(g.grade_percentage for g in student_grades) / len(student_grades) pass_rate = len([g for g in student_grades if g.grade_percentage >= 50]) / len(student_grades) * 100 latest_grade = max(student_grades, key=lambda g: g.exam_date) students_analytics.append({ 'id': student.id, 'first_name': student.first_name, 'last_name': student.last_name, 'student_id': student.student_id, 'average_grade': avg_grade, 'total_exams': len(student_grades), 'pass_rate': pass_rate, 'latest_grade': latest_grade.grade_percentage, 'latest_exam_date': latest_grade.exam_date.isoformat() }) # Sort by average grade descending students_analytics.sort(key=lambda x: x['average_grade'], reverse=True) return custom_response( success=True, data={'students': students_analytics} ) except Exception as e: return custom_response(success=False, data=f"Error fetching student analytics: {str(e)}") def get_tutor_gallery_files(self, user_id: str, user_type: str, tutor_view: bool = False) -> Dict: """Get evidence files for tutor/supervisor gallery view""" try: with DatabaseContextManager() as ctx: user_id_str = user_id if user_id else None if not tutor_view or user_type not in ['tutor', 'supervisor']: return custom_response(success=False, data="Access denied: Tutor/Supervisor gallery access required") if user_type == 'tutor': # Get all evidence files for exams graded by this tutor user_id_normalized = user_id_str.replace('-', '') if user_id_str else None # Query with joinedload to ensure exam_grade relationship is loaded query = ctx.session.query(ExamEvidence).options( joinedload(ExamEvidence.exam_grade) ).join( ExamGrade, ExamEvidence.exam_grade_id == ExamGrade.id ).filter( func.replace(ExamGrade.tutor_id, '-', '') == user_id_normalized ).order_by(ExamEvidence.uploaded_at.desc()) else: # For supervisors, get evidence files from all tutors in their departments supervisor_courses = ctx.session.query(Course.id).join( SupervisorDepartment, Course.department == SupervisorDepartment.department_name ).filter(SupervisorDepartment.supervisor_id == user_id_str).subquery() query = ctx.session.query(ExamEvidence).join( ExamGrade, ExamEvidence.exam_grade_id == ExamGrade.id ).filter( ExamGrade.course_id.in_( ctx.session.query(supervisor_courses.c.id) ) ).order_by(ExamEvidence.uploaded_at.desc()) evidence_files = query.all() # Format the response data formatted_files = [] for evidence in evidence_files: try: # Get related data exam_grade = evidence.exam_grade # Get student data student = exam_grade.student if exam_grade else None if not student and exam_grade: # Normalize student_id by removing hyphens normalized_student_id = exam_grade.student_id.replace('-', '') if exam_grade.student_id else None # Try querying with normalized ID student = ctx.session.query(Student).filter( func.replace(Student.id, '-', '') == normalized_student_id ).first() # Get course data course = exam_grade.course if exam_grade else None if not course and exam_grade: # Normalize course_id by removing hyphens normalized_course_id = exam_grade.course_id.replace('-', '') if exam_grade.course_id else None # Try querying with normalized ID course = ctx.session.query(Course).filter( func.replace(Course.id, '-', '') == normalized_course_id ).first() if not exam_grade or not student or not course: # Skip if essential data is missing continue # Generate file URL file_url = f"http://localhost:5000/api/v1/exam-portfolio/evidence/{evidence.id}?user_id={user_id}&user_type={user_type}" formatted_files.append({ 'id': evidence.id, 'file_name': evidence.file_name, 'file_type': evidence.file_type, 'file_size': evidence.file_size or 0, 'upload_date': evidence.uploaded_at.isoformat() if evidence.uploaded_at else '', 'is_verified': evidence.verified_by is not None, 'verification_status': evidence.status.value if evidence.status else 'pending', 'description': evidence.description or '', 'tags': evidence.evidence_type.value if evidence.evidence_type else '', 'file_url': file_url, 'exam_grade_id': str(evidence.exam_grade_id), 'student_id': str(student.id), 'student_name': f"{student.first_name} {student.last_name}", 'course_id': str(course.id), 'course_name': course.title, 'course_code': course.code, 'exam_title': exam_grade.exam_title, 'exam_type': exam_grade.exam_type, 'exam_date': exam_grade.exam_date.isoformat() if exam_grade.exam_date else '', 'grade_percentage': exam_grade.grade_percentage }) except Exception as e: current_app.logger.error(f"Error formatting evidence file {evidence.id}: {str(e)}") continue # Log access self._log_access(user_id, user_type, 'view') return custom_response( success=True, data={'evidence_files': formatted_files} ) except Exception as e: return custom_response(success=False, data=f"Error fetching gallery files: {str(e)}")