from datetime import datetime, timedelta from typing import Dict, List, Optional from sqlalchemy import func, and_, or_, desc from sqlalchemy.orm import joinedload, aliased import pandas as pd import io import csv import os from fpdf import FPDF from src.models import DatabaseContextManager from src.models.models import ( Tutor, Supervisor, Course, TeachingSession, Attendance, Assignment, AssignmentSubmission, Enrollment, Student, TutorDepartment, SupervisorDepartment ) from src.utils import custom_response class PerformanceReportsManager: def __init__(self): pass def get_tutor_performance_analytics(self, period: str = '30d', start_date: str = None, end_date: str = None, department: str = None, supervisor_id: str = None) -> Dict: """ Get comprehensive tutor performance analytics with detailed metrics """ try: with DatabaseContextManager() as ctx: # Calculate date range date_range = self._calculate_date_range(period, start_date, end_date) # Get tutor performance data tutor_performance = self._get_tutor_performance_data( ctx, date_range, department, supervisor_id ) # Get overall statistics overall_stats = self._calculate_tutor_overall_stats(ctx, tutor_performance) # Get performance trends performance_trends = self._get_tutor_performance_trends(ctx, date_range) return custom_response( success=True, data={ 'overall_stats': overall_stats, 'tutor_performance': tutor_performance, 'performance_trends': performance_trends, 'date_range': { 'start': date_range['start'].isoformat(), 'end': date_range['end'].isoformat(), 'period': period }, 'filters': { 'department': department, 'supervisor_id': supervisor_id }, 'message': "Tutor performance analytics retrieved successfully" } ) except Exception as e: return custom_response( success=False, data={'error': f"Error retrieving tutor performance data: {str(e)}"}, status_code=500 ) def _get_tutor_performance_data(self, ctx, date_range: Dict, department: str = None, supervisor_id: str = None) -> List[Dict]: """ Get detailed tutor performance data with comprehensive metrics """ try: # Base query for tutors with their courses, sessions, and departments query = ctx.session.query(Tutor).options( joinedload(Tutor.courses).joinedload(Course.supervisor), joinedload(Tutor.scheduled_sessions), joinedload(Tutor.departments) ) # Apply filters if department: # Filter by tutor's departments instead of course department query = query.join(Tutor.departments).filter( and_( TutorDepartment.department_name == department, TutorDepartment.is_active == True ) ) if supervisor_id: query = query.join(Tutor.courses).filter(Course.supervisor_id == supervisor_id) tutors = query.all() tutor_performance = [] for tutor in tutors: # Get sessions in date range (using created_at for now since scheduled_date doesn't exist) sessions = ctx.session.query(TeachingSession).filter( and_( TeachingSession.tutor_id == tutor.id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).all() # Calculate attendance metrics attendance_metrics = self._calculate_tutor_attendance_metrics(ctx, tutor.id, date_range) # Calculate student performance metrics student_metrics = self._calculate_tutor_student_metrics(ctx, tutor.id, date_range) # Calculate session metrics session_metrics = self._calculate_tutor_session_metrics(sessions) # Calculate overall performance score performance_score = self._calculate_tutor_performance_score( attendance_metrics, student_metrics, session_metrics ) # Get trend data trend = self._get_tutor_trend(ctx, tutor.id, date_range) # Get primary department or first active department primary_department = None if tutor.departments: primary_dept = next((d for d in tutor.departments if d.is_primary and d.is_active), None) if primary_dept: primary_department = primary_dept.department_name else: # Fallback to first active department active_dept = next((d for d in tutor.departments if d.is_active), None) if active_dept: primary_department = active_dept.department_name tutor_data = { 'id': tutor.id, 'name': f"{tutor.first_name} {tutor.last_name}", 'email': tutor.email, 'department': primary_department or 'Unknown', 'phone': tutor.phone, 'courses_count': len(tutor.courses), 'total_students': student_metrics['total_students'], 'sessions_completed': session_metrics['completed_sessions'], 'sessions_scheduled': session_metrics['total_sessions'], 'attendance_rate': attendance_metrics['attendance_rate'], 'average_rating': student_metrics['average_rating'], 'performance_score': performance_score, 'trend': trend, 'status': self._get_performance_status(performance_score), 'courses': [ { 'id': course.id, 'title': course.title, 'code': course.code, 'students_count': len(course.enrollments), 'completion_rate': self._get_course_completion_rate(ctx, course.id) } for course in tutor.courses ], 'detailed_metrics': { 'attendance': attendance_metrics, 'student_performance': student_metrics, 'session_metrics': session_metrics }, 'created_at': tutor.created_at.isoformat() if tutor.created_at else None, 'last_active': self._get_tutor_last_active(ctx, tutor.id) } tutor_performance.append(tutor_data) # Sort by performance score tutor_performance.sort(key=lambda x: x['performance_score'], reverse=True) return tutor_performance except Exception as e: raise Exception(f"Error calculating tutor performance data: {str(e)}") def _calculate_tutor_attendance_metrics(self, ctx, tutor_id: str, date_range: Dict) -> Dict: """ Calculate detailed attendance metrics for a tutor """ try: # Get all sessions for the tutor in date range sessions = ctx.session.query(TeachingSession).filter( and_( TeachingSession.tutor_id == tutor_id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).all() if not sessions: return { 'total_sessions': 0, 'attended_sessions': 0, 'attendance_rate': 0.0, 'late_sessions': 0, 'early_departures': 0, 'cancelled_sessions': 0 } total_sessions = len(sessions) attended_sessions = sum(1 for session in sessions if session.attendance_taken) late_sessions = sum(1 for session in sessions if session.status == 'late') cancelled_sessions = sum(1 for session in sessions if session.status == 'cancelled') attendance_rate = (attended_sessions / total_sessions * 100) if total_sessions > 0 else 0 return { 'total_sessions': total_sessions, 'attended_sessions': attended_sessions, 'attendance_rate': round(attendance_rate, 2), 'late_sessions': late_sessions, 'early_departures': 0, # This would need to be tracked in the session model 'cancelled_sessions': cancelled_sessions } except Exception as e: raise Exception(f"Error calculating attendance metrics: {str(e)}") def _calculate_tutor_student_metrics(self, ctx, tutor_id: str, date_range: Dict) -> Dict: """ Calculate student performance metrics for a tutor's courses """ try: # Get tutor first to access their courses tutor = ctx.session.query(Tutor).options(joinedload(Tutor.courses)).filter(Tutor.id == tutor_id).first() if not tutor or not tutor.courses: return { 'total_students': 0, 'average_rating': 0.0, 'completion_rate': 0.0, 'assignment_completion': 0.0, 'active_students': 0 } # Get course IDs course_ids = [course.id for course in tutor.courses] # Get all students enrolled in tutor's courses students = ctx.session.query(Student).join(Enrollment).filter( Enrollment.course_id.in_(course_ids) ).distinct().all() if not students: return { 'total_students': 0, 'average_rating': 0.0, 'completion_rate': 0.0, 'assignment_completion': 0.0, 'active_students': 0 } # Calculate assignment completion rate total_assignments = ctx.session.query(Assignment).filter( Assignment.course_id.in_(course_ids) ).count() completed_assignments = ctx.session.query(AssignmentSubmission).join(Assignment).filter( and_( Assignment.course_id.in_(course_ids), AssignmentSubmission.status == 'submitted' ) ).count() assignment_completion = (completed_assignments / total_assignments * 100) if total_assignments > 0 else 0 # Mock rating calculation (in real implementation, this would come from student feedback) average_rating = 4.2 + (hash(tutor_id) % 100) / 100 # Mock rating between 4.2-5.2 return { 'total_students': len(students), 'average_rating': round(average_rating, 1), 'completion_rate': round(assignment_completion, 2), 'assignment_completion': round(assignment_completion, 2), 'active_students': len([s for s in students if self._is_student_active(ctx, s.id, date_range)]) } except Exception as e: raise Exception(f"Error calculating student metrics: {str(e)}") def _calculate_tutor_session_metrics(self, sessions: List[TeachingSession]) -> Dict: """ Calculate session-related metrics for a tutor """ if not sessions: return { 'total_sessions': 0, 'completed_sessions': 0, 'completion_rate': 0.0, 'average_duration': 0, 'punctuality_rate': 0.0 } total_sessions = len(sessions) completed_sessions = sum(1 for session in sessions if session.status == 'completed') completion_rate = (completed_sessions / total_sessions * 100) if total_sessions > 0 else 0 # Calculate average duration (mock calculation) average_duration = 90 # minutes # Calculate punctuality rate on_time_sessions = sum(1 for session in sessions if session.status in ['completed', 'in_progress']) punctuality_rate = (on_time_sessions / total_sessions * 100) if total_sessions > 0 else 0 return { 'total_sessions': total_sessions, 'completed_sessions': completed_sessions, 'completion_rate': round(completion_rate, 2), 'average_duration': average_duration, 'punctuality_rate': round(punctuality_rate, 2) } def _calculate_tutor_performance_score(self, attendance_metrics: Dict, student_metrics: Dict, session_metrics: Dict) -> float: """ Calculate overall performance score for a tutor """ # Weighted scoring system attendance_weight = 0.3 student_weight = 0.4 session_weight = 0.3 attendance_score = attendance_metrics['attendance_rate'] student_score = (student_metrics['average_rating'] / 5.0) * 100 session_score = session_metrics['completion_rate'] performance_score = ( (attendance_score * attendance_weight) + (student_score * student_weight) + (session_score * session_weight) ) return round(performance_score, 2) def _get_performance_status(self, score: float) -> str: """ Determine performance status based on score """ if score >= 90: return 'excellent' elif score >= 80: return 'good' elif score >= 70: return 'average' else: return 'needs_improvement' def _calculate_date_range(self, period: str, start_date: str = None, end_date: str = None) -> Dict: """ Calculate date range based on period or custom dates """ if start_date and end_date: return { 'start': datetime.fromisoformat(start_date), 'end': datetime.fromisoformat(end_date) } end = datetime.now() if period == '7d': start = end - timedelta(days=7) elif period == '30d': start = end - timedelta(days=30) elif period == '90d': start = end - timedelta(days=90) elif period == '1y': start = end - timedelta(days=365) else: start = end - timedelta(days=30) # Default to 30 days return { 'start': start, 'end': end } def _get_tutor_trend(self, ctx, tutor_id: str, date_range: Dict) -> str: """ Determine performance trend for a tutor """ try: # Get performance data for previous period period_duration = (date_range['end'] - date_range['start']).days previous_start = date_range['start'] - timedelta(days=period_duration) previous_end = date_range['start'] # Calculate current period performance current_sessions = ctx.session.query(TeachingSession).filter( and_( TeachingSession.tutor_id == tutor_id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).count() # Calculate previous period performance previous_sessions = ctx.session.query(TeachingSession).filter( and_( TeachingSession.tutor_id == tutor_id, TeachingSession.created_at >= previous_start, TeachingSession.created_at <= previous_end ) ).count() if current_sessions > previous_sessions: return 'up' elif current_sessions < previous_sessions: return 'down' else: return 'stable' except Exception: return 'stable' # Default to stable if calculation fails def _get_course_completion_rate(self, ctx, course_id: str) -> float: """ Calculate course completion rate """ try: enrollments = ctx.session.query(Enrollment).filter( Enrollment.course_id == course_id ).all() if not enrollments: return 0.0 # Mock completion rate calculation return round(75.0 + (hash(course_id) % 25), 1) except Exception: return 0.0 def _get_tutor_last_active(self, ctx, tutor_id: str) -> str: """ Get tutor's last active date """ try: last_session = ctx.session.query(TeachingSession).filter( TeachingSession.tutor_id == tutor_id ).order_by(desc(TeachingSession.created_at)).first() if last_session: return last_session.created_at.isoformat() return None except Exception: return None def _is_student_active(self, ctx, student_id: str, date_range: Dict) -> bool: """ Check if student is active in the given date range """ try: # Check if student has any attendance records in the date range attendance = ctx.session.query(Attendance).join(TeachingSession).filter( and_( Attendance.student_id == student_id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).first() return attendance is not None except Exception: return False def _calculate_tutor_overall_stats(self, ctx, tutor_performance: List[Dict]) -> Dict: """ Calculate overall statistics for all tutors """ if not tutor_performance: return { 'total_tutors': 0, 'average_performance': 0.0, 'excellent_performers': 0, 'needs_improvement': 0, 'average_attendance': 0.0, 'average_rating': 0.0 } total_tutors = len(tutor_performance) total_performance = sum(tutor['performance_score'] for tutor in tutor_performance) average_performance = total_performance / total_tutors excellent_performers = len([t for t in tutor_performance if t['status'] == 'excellent']) needs_improvement = len([t for t in tutor_performance if t['status'] == 'needs_improvement']) total_attendance = sum(tutor['attendance_rate'] for tutor in tutor_performance) average_attendance = total_attendance / total_tutors total_rating = sum(tutor['average_rating'] for tutor in tutor_performance) average_rating = total_rating / total_tutors return { 'total_tutors': total_tutors, 'average_performance': round(average_performance, 2), 'excellent_performers': excellent_performers, 'needs_improvement': needs_improvement, 'average_attendance': round(average_attendance, 2), 'average_rating': round(average_rating, 1), 'improvement_rate': round((excellent_performers / total_tutors * 100), 1) if total_tutors > 0 else 0 } def _get_tutor_performance_trends(self, ctx, date_range: Dict) -> List[Dict]: """ Get performance trends over time """ try: # Get weekly performance data for the period trends = [] current_date = date_range['start'] while current_date <= date_range['end']: week_start = current_date week_end = current_date + timedelta(days=7) # Get tutors active in this week active_tutors = ctx.session.query(Tutor).join(TeachingSession).filter( and_( TeachingSession.created_at >= week_start, TeachingSession.created_at <= week_end ) ).distinct().count() trends.append({ 'week': week_start.isoformat(), 'active_tutors': active_tutors, 'average_sessions': 5.2, # Mock data 'average_attendance': 88.5 # Mock data }) current_date += timedelta(days=7) return trends except Exception as e: return [] def get_top_performers_analytics(self, date_range: Dict = None, limit: int = 10) -> Dict: """ Get top 10 performing tutors and supervisors based on attendance and credit hours """ try: if not date_range: date_range = self._calculate_date_range('30d') with DatabaseContextManager() as ctx: # Get top performing tutors top_tutors = self._get_top_performing_tutors(ctx, date_range, limit) # Get top performing supervisors top_supervisors = self._get_top_performing_supervisors(ctx, date_range, limit) return custom_response( success=True, data={ 'top_tutors': top_tutors, 'top_supervisors': top_supervisors, 'date_range': { 'start': date_range['start'].isoformat(), 'end': date_range['end'].isoformat() }, 'message': "Top performers analytics retrieved successfully" } ) except Exception as e: return custom_response( success=False, data={'error': f"Error retrieving top performers: {str(e)}"}, status_code=500 ) def _get_top_performing_tutors(self, ctx, date_range: Dict, limit: int) -> List[Dict]: """ Get top performing tutors based on attendance and credit hours """ try: # Get tutors with their sessions and courses tutors = ctx.session.query(Tutor).options( joinedload(Tutor.scheduled_sessions), joinedload(Tutor.courses), joinedload(Tutor.departments) ).all() tutor_performance = [] for tutor in tutors: # Get sessions in date range sessions = ctx.session.query(TeachingSession).filter( and_( TeachingSession.tutor_id == tutor.id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).all() if not sessions: continue # Calculate attendance metrics total_sessions = len(sessions) attended_sessions = sum(1 for session in sessions if session.attendance_taken) attendance_rate = (attended_sessions / total_sessions * 100) if total_sessions > 0 else 0 # Calculate total credit hours from courses (using credits field) total_credit_hours = sum(course.credits or 0 for course in tutor.courses) # Calculate performance score based on attendance and credit hours # Weight: 70% attendance, 30% credit hours (normalized) attendance_score = attendance_rate credit_hours_score = min(100, (total_credit_hours / 20) * 100) # Normalize to 20 credit hours max performance_score = (attendance_score * 0.7) + (credit_hours_score * 0.3) # Get primary department primary_department = self._get_tutor_primary_department(tutor) tutor_data = { 'id': tutor.id, 'name': f"{tutor.first_name} {tutor.last_name}", 'department': primary_department, 'attendance_rate': round(attendance_rate, 2), 'total_sessions': total_sessions, 'attended_sessions': attended_sessions, 'total_credit_hours': total_credit_hours, 'performance_score': round(performance_score, 2), 'courses_count': len(tutor.courses), 'student_count': sum(len(course.enrollments) for course in tutor.courses) } tutor_performance.append(tutor_data) # Sort by performance score and return top performers tutor_performance.sort(key=lambda x: x['performance_score'], reverse=True) return tutor_performance[:limit] except Exception as e: raise Exception(f"Error calculating top performing tutors: {str(e)}") def _get_top_performing_supervisors(self, ctx, date_range: Dict, limit: int) -> List[Dict]: """ Get top performing supervisors based on tutor performance and course oversight """ try: # Get supervisors with their courses and tutors supervisors = ctx.session.query(Supervisor).options( joinedload(Supervisor.courses).joinedload(Course.tutors), joinedload(Supervisor.departments) ).all() supervisor_performance = [] for supervisor in supervisors: if not supervisor.courses: continue # Calculate metrics for supervised tutors total_tutors = 0 total_sessions = 0 total_attended_sessions = 0 total_credit_hours = 0 total_students = 0 for course in supervisor.courses: total_credit_hours += course.credits or 0 total_students += len(course.enrollments) for tutor in course.tutors: total_tutors += 1 # Get tutor sessions in date range tutor_sessions = ctx.session.query(TeachingSession).filter( and_( TeachingSession.tutor_id == tutor.id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).all() total_sessions += len(tutor_sessions) total_attended_sessions += sum(1 for session in tutor_sessions if session.attendance_taken) if total_sessions == 0: continue # Calculate supervisor performance metrics overall_attendance_rate = (total_attended_sessions / total_sessions * 100) if total_sessions > 0 else 0 average_credit_hours_per_tutor = total_credit_hours / total_tutors if total_tutors > 0 else 0 # Performance score: 50% tutor attendance, 30% credit hours management, 20% course oversight attendance_score = overall_attendance_rate credit_management_score = min(100, (average_credit_hours_per_tutor / 15) * 100) # Normalize to 15 credit hours oversight_score = min(100, (len(supervisor.courses) / 10) * 100) # Normalize to 10 courses performance_score = (attendance_score * 0.5) + (credit_management_score * 0.3) + (oversight_score * 0.2) # Get primary department primary_department = self._get_supervisor_primary_department(supervisor) supervisor_data = { 'id': supervisor.id, 'name': f"{supervisor.first_name} {supervisor.last_name}", 'department': primary_department, 'overall_attendance_rate': round(overall_attendance_rate, 2), 'total_sessions_overseeing': total_sessions, 'total_attended_sessions': total_attended_sessions, 'total_credit_hours_managed': total_credit_hours, 'average_credit_hours_per_tutor': round(average_credit_hours_per_tutor, 2), 'performance_score': round(performance_score, 2), 'tutors_supervised': total_tutors, 'courses_managed': len(supervisor.courses), 'total_students': total_students } supervisor_performance.append(supervisor_data) # Sort by performance score and return top performers supervisor_performance.sort(key=lambda x: x['performance_score'], reverse=True) return supervisor_performance[:limit] except Exception as e: raise Exception(f"Error calculating top performing supervisors: {str(e)}") def get_supervisor_performance_analytics(self, period: str = '30d', start_date: str = None, end_date: str = None, department: str = None) -> Dict: """ Get comprehensive supervisor performance analytics """ try: with DatabaseContextManager() as ctx: # Calculate date range date_range = self._calculate_date_range(period, start_date, end_date) # Get supervisor performance data supervisor_performance = self._get_supervisor_performance_data( ctx, date_range, department ) # Get overall statistics overall_stats = self._calculate_supervisor_overall_stats(ctx, supervisor_performance) return custom_response( success=True, data={ 'overall_stats': overall_stats, 'supervisor_performance': supervisor_performance, 'date_range': { 'start': date_range['start'].isoformat(), 'end': date_range['end'].isoformat(), 'period': period }, 'filters': { 'department': department }, 'message': "Supervisor performance analytics retrieved successfully" } ) except Exception as e: return custom_response( success=False, data={'error': f"Error retrieving supervisor performance data: {str(e)}"}, status_code=500 ) def _get_supervisor_performance_data(self, ctx, date_range: Dict, department: str = None) -> List[Dict]: """ Get detailed supervisor performance data """ try: # Base query for supervisors query = ctx.session.query(Supervisor).options( joinedload(Supervisor.courses).joinedload(Course.tutors).joinedload(Tutor.departments), joinedload(Supervisor.departments) ) # Apply filters if department: # Filter by supervisor's departments instead of course department query = query.join(Supervisor.departments).filter( and_( SupervisorDepartment.department_name == department, SupervisorDepartment.is_active == True ) ) supervisors = query.all() supervisor_performance = [] for supervisor in supervisors: # Get tutors supervised tutors = [] for course in supervisor.courses: tutors.extend(course.tutors) # Remove duplicates tutors = list(set(tutors)) # Calculate metrics tutor_metrics = self._calculate_supervisor_tutor_metrics(ctx, supervisor.id, date_range) course_metrics = self._calculate_supervisor_course_metrics(ctx, supervisor.id, date_range) retention_metrics = self._calculate_supervisor_retention_metrics(ctx, supervisor.id, date_range) # Calculate overall performance score performance_score = self._calculate_supervisor_performance_score( tutor_metrics, course_metrics, retention_metrics ) # Get trend data trend = self._get_supervisor_trend(ctx, supervisor.id, date_range) # Get primary department or first active department primary_department = self._get_supervisor_primary_department(supervisor) supervisor_data = { 'id': supervisor.id, 'name': f"{supervisor.first_name} {supervisor.last_name}", 'email': supervisor.email, 'department': primary_department, 'phone': supervisor.phone, 'tutors_supervised': len(tutors), 'courses_overseeing': len(supervisor.courses), 'average_tutor_rating': tutor_metrics['average_rating'], 'tutor_retention_rate': retention_metrics['retention_rate'], 'sessions_overseeing': course_metrics['total_sessions'], 'performance_score': performance_score, 'trend': trend, 'status': self._get_performance_status(performance_score), 'tutors': [ { 'id': tutor.id, 'name': f"{tutor.first_name} {tutor.last_name}", 'department': self._get_tutor_primary_department(tutor), 'performance_score': self._calculate_tutor_performance_score( self._calculate_tutor_attendance_metrics(ctx, tutor.id, date_range), self._calculate_tutor_student_metrics(ctx, tutor.id, date_range), self._calculate_tutor_session_metrics( ctx.session.query(TeachingSession).filter( and_( TeachingSession.tutor_id == tutor.id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).all() ) ) } for tutor in tutors[:5] # Limit to top 5 tutors ], 'detailed_metrics': { 'tutor_metrics': tutor_metrics, 'course_metrics': course_metrics, 'retention_metrics': retention_metrics }, 'created_at': supervisor.created_at.isoformat() if supervisor.created_at else None, 'last_active': self._get_supervisor_last_active(ctx, supervisor.id) } supervisor_performance.append(supervisor_data) # Sort by performance score supervisor_performance.sort(key=lambda x: x['performance_score'], reverse=True) return supervisor_performance except Exception as e: raise Exception(f"Error calculating supervisor performance data: {str(e)}") def _calculate_supervisor_tutor_metrics(self, ctx, supervisor_id: str, date_range: Dict) -> Dict: """ Calculate tutor-related metrics for a supervisor """ try: # Get tutors supervised by this supervisor through courses tutors = ctx.session.query(Tutor).join( Tutor.courses ).filter( Course.supervisor_id == supervisor_id ).distinct().all() if not tutors: return { 'total_tutors': 0, 'average_rating': 0.0, 'active_tutors': 0, 'tutors_needing_improvement': 0 } # Calculate average rating of supervised tutors total_rating = 0 active_tutors = 0 tutors_needing_improvement = 0 for tutor in tutors: # Mock rating calculation tutor_rating = 4.0 + (hash(tutor.id) % 100) / 100 total_rating += tutor_rating # Check if tutor is active if self._is_tutor_active(ctx, tutor.id, date_range): active_tutors += 1 # Check if tutor needs improvement (mock calculation) if tutor_rating < 4.0: tutors_needing_improvement += 1 average_rating = total_rating / len(tutors) return { 'total_tutors': len(tutors), 'average_rating': round(average_rating, 1), 'active_tutors': active_tutors, 'tutors_needing_improvement': tutors_needing_improvement } except Exception as e: raise Exception(f"Error calculating supervisor tutor metrics: {str(e)}") def _calculate_supervisor_course_metrics(self, ctx, supervisor_id: str, date_range: Dict) -> Dict: """ Calculate course-related metrics for a supervisor """ try: # Get courses supervised courses = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id ).all() if not courses: return { 'total_courses': 0, 'total_sessions': 0, 'average_enrollment': 0.0, 'courses_active': 0 } # Calculate total sessions across all courses total_sessions = 0 total_enrollments = 0 active_courses = 0 for course in courses: course_sessions = ctx.session.query(TeachingSession).filter( and_( TeachingSession.course_id == course.id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).count() total_sessions += course_sessions total_enrollments += len(course.enrollments) if course_sessions > 0: active_courses += 1 average_enrollment = total_enrollments / len(courses) if courses else 0 return { 'total_courses': len(courses), 'total_sessions': total_sessions, 'average_enrollment': round(average_enrollment, 1), 'courses_active': active_courses } except Exception as e: raise Exception(f"Error calculating supervisor course metrics: {str(e)}") def _calculate_supervisor_retention_metrics(self, ctx, supervisor_id: str, date_range: Dict) -> Dict: """ Calculate retention metrics for a supervisor """ try: # Get tutors that were supervised at the beginning of the period tutors_at_start = ctx.session.query(Tutor).join( Tutor.courses ).filter( Course.supervisor_id == supervisor_id ).distinct().all() if not tutors_at_start: return { 'retention_rate': 0.0, 'tutors_retained': 0, 'tutors_lost': 0 } # Get tutors still active at the end of the period tutors_at_end = [] for tutor in tutors_at_start: if self._is_tutor_active(ctx, tutor.id, date_range): tutors_at_end.append(tutor) tutors_retained = len(tutors_at_end) tutors_lost = len(tutors_at_start) - tutors_retained retention_rate = (tutors_retained / len(tutors_at_start) * 100) if tutors_at_start else 0 return { 'retention_rate': round(retention_rate, 2), 'tutors_retained': tutors_retained, 'tutors_lost': tutors_lost } except Exception as e: raise Exception(f"Error calculating supervisor retention metrics: {str(e)}") def _calculate_supervisor_performance_score(self, tutor_metrics: Dict, course_metrics: Dict, retention_metrics: Dict) -> float: """ Calculate overall performance score for a supervisor """ # Weighted scoring system tutor_weight = 0.4 course_weight = 0.3 retention_weight = 0.3 # Tutor performance score (based on average rating) tutor_score = (tutor_metrics['average_rating'] / 5.0) * 100 # Course performance score (based on activity) course_score = (course_metrics['courses_active'] / course_metrics['total_courses'] * 100) if course_metrics['total_courses'] > 0 else 0 # Retention score retention_score = retention_metrics['retention_rate'] performance_score = ( (tutor_score * tutor_weight) + (course_score * course_weight) + (retention_score * retention_weight) ) return round(performance_score, 2) def _get_supervisor_trend(self, ctx, supervisor_id: str, date_range: Dict) -> str: """ Determine performance trend for a supervisor """ try: # Get performance data for previous period period_duration = (date_range['end'] - date_range['start']).days previous_start = date_range['start'] - timedelta(days=period_duration) previous_end = date_range['start'] # Calculate current period performance (number of active courses) current_courses = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id ).count() # Calculate previous period performance (mock calculation) previous_courses = current_courses + (hash(supervisor_id) % 3) - 1 if current_courses > previous_courses: return 'up' elif current_courses < previous_courses: return 'down' else: return 'stable' except Exception: return 'stable' def _get_supervisor_last_active(self, ctx, supervisor_id: str) -> str: """ Get supervisor's last active date """ try: # Get the most recent course activity last_course = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id ).order_by(desc(Course.updated_at)).first() if last_course and last_course.updated_at: return last_course.updated_at.isoformat() return None except Exception: return None def _is_tutor_active(self, ctx, tutor_id: str, date_range: Dict) -> bool: """ Check if tutor is active in the given date range """ try: # Check if tutor has any sessions in the date range session = ctx.session.query(TeachingSession).filter( and_( TeachingSession.tutor_id == tutor_id, TeachingSession.created_at >= date_range['start'], TeachingSession.created_at <= date_range['end'] ) ).first() return session is not None except Exception: return False def _calculate_supervisor_overall_stats(self, ctx, supervisor_performance: List[Dict]) -> Dict: """ Calculate overall statistics for all supervisors """ if not supervisor_performance: return { 'total_supervisors': 0, 'average_performance': 0.0, 'excellent_performers': 0, 'needs_improvement': 0, 'average_retention': 0.0, 'average_tutor_rating': 0.0 } total_supervisors = len(supervisor_performance) total_performance = sum(supervisor['performance_score'] for supervisor in supervisor_performance) average_performance = total_performance / total_supervisors excellent_performers = len([s for s in supervisor_performance if s['status'] == 'excellent']) needs_improvement = len([s for s in supervisor_performance if s['status'] == 'needs_improvement']) total_retention = sum(supervisor['tutor_retention_rate'] for supervisor in supervisor_performance) average_retention = total_retention / total_supervisors total_rating = sum(supervisor['average_tutor_rating'] for supervisor in supervisor_performance) average_rating = total_rating / total_supervisors return { 'total_supervisors': total_supervisors, 'average_performance': round(average_performance, 2), 'excellent_performers': excellent_performers, 'needs_improvement': needs_improvement, 'average_retention': round(average_retention, 2), 'average_tutor_rating': round(average_rating, 1), 'improvement_rate': round((excellent_performers / total_supervisors * 100), 1) if total_supervisors > 0 else 0 } def export_performance_data(self, export_type: str = 'tutors', period: str = '30d', format_type: str = 'csv') -> Dict: """ Export performance data to CSV or Excel format """ try: with DatabaseContextManager() as ctx: # Calculate date range date_range = self._calculate_date_range(period) if export_type == 'tutors': # Get tutor performance data data = self._get_tutor_performance_data(ctx, date_range) filename = f"tutor_performance_{period}_{datetime.now().strftime('%Y%m%d')}" else: # Get supervisor performance data data = self._get_supervisor_performance_data(ctx, date_range) filename = f"supervisor_performance_{period}_{datetime.now().strftime('%Y%m%d')}" if format_type == 'csv': return self._export_to_csv(data, filename) elif format_type == 'pdf': return self._export_to_pdf(data, filename, export_type) else: return self._export_to_excel(data, filename) except Exception as e: return custom_response( success=False, data={'error': f"Error exporting performance data: {str(e)}"}, status_code=500 ) def _export_to_csv(self, data: List[Dict], filename: str) -> Dict: """ Export data to CSV format """ try: if not data: return custom_response( success=False, data={'error': "No data available for export"}, status_code=404 ) # Prepare data for export (remove nested structures) export_data = self._prepare_data_for_export(data) # Create CSV content output = io.StringIO() fieldnames = list(export_data[0].keys()) writer = csv.DictWriter(output, fieldnames=fieldnames) writer.writeheader() for row in export_data: writer.writerow(row) csv_content = output.getvalue() output.close() return custom_response( success=True, data={ 'content': csv_content, 'filename': f"{filename}.csv", 'content_type': 'text/csv', 'message': "Data exported successfully" } ) except Exception as e: return custom_response( success=False, data={'error': f"Error creating CSV export: {str(e)}"}, status_code=500 ) def _export_to_excel(self, data: List[Dict], filename: str) -> Dict: """ Export data to Excel format """ try: if not data: return custom_response( success=False, data={'error': "No data available for export"}, status_code=404 ) # Create DataFrame df = pd.DataFrame(data) # Create Excel file in memory output = io.BytesIO() with pd.ExcelWriter(output, engine='openpyxl') as writer: df.to_excel(writer, sheet_name='Performance Data', index=False) excel_content = output.getvalue() output.close() return custom_response( success=True, data={ 'content': excel_content.decode('latin-1'), # For binary content 'filename': f"{filename}.xlsx", 'content_type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'message': "Data exported successfully" } ) except Exception as e: return custom_response( success=False, data={'error': f"Error creating Excel export: {str(e)}"}, status_code=500 ) def _export_to_pdf(self, data: List[Dict], filename: str, export_type: str) -> Dict: """ Export data to PDF format with top-notch styling """ try: if not data: return custom_response( success=False, data={'error': "No data available for export"}, status_code=404 ) # Create PDF with enhanced styling pdf = PerformanceReportPDF() pdf.add_page() # Set document title report_title = f"{export_type.title()} Performance Report" pdf.set_document_title(report_title, f"Generated on {datetime.now().strftime('%B %d, %Y at %I:%M %p')}") # Create professional performance report pdf.create_performance_report(data, export_type) # Generate PDF content pdf_output = pdf.output(dest='S') # Ensure PDF content is bytes if isinstance(pdf_output, str): pdf_content = pdf_output.encode('latin1') else: pdf_content = bytes(pdf_output) return custom_response( success=True, data={ 'content': pdf_content.decode('latin-1'), 'filename': f"{filename}.pdf", 'content_type': 'application/pdf', 'message': "PDF report generated successfully" } ) except Exception as e: return custom_response( success=False, data={'error': f"Error creating PDF export: {str(e)}"}, status_code=500 ) def _get_tutor_primary_department(self, tutor) -> str: """ Get the primary department for a tutor """ try: if tutor.departments: primary_dept = next((d for d in tutor.departments if d.is_primary and d.is_active), None) if primary_dept: return primary_dept.department_name else: # Fallback to first active department active_dept = next((d for d in tutor.departments if d.is_active), None) if active_dept: return active_dept.department_name return 'Unknown' except Exception: return 'Unknown' def _get_supervisor_primary_department(self, supervisor) -> str: """ Get the primary department for a supervisor """ try: if supervisor.departments: primary_dept = next((d for d in supervisor.departments if d.is_primary and d.is_active), None) if primary_dept: return primary_dept.department_name else: # Fallback to first active department active_dept = next((d for d in supervisor.departments if d.is_active), None) if active_dept: return active_dept.department_name return 'Unknown' except Exception: return 'Unknown' def _flatten_dict(self, d: Dict, parent_key: str = '', sep: str = '_') -> Dict: """ Flatten nested dictionary for CSV export """ items = [] for k, v in d.items(): new_key = f"{parent_key}{sep}{k}" if parent_key else k if isinstance(v, dict): items.extend(self._flatten_dict(v, new_key, sep=sep).items()) elif isinstance(v, list): # Convert list to string representation items.append((new_key, str(v))) else: items.append((new_key, v)) return dict(items) def _prepare_data_for_export(self, data: List[Dict]) -> List[Dict]: """ Prepare data for export by removing nested structures that cause CSV issues """ export_data = [] for item in data: # Create a clean copy without nested detailed_metrics clean_item = {k: v for k, v in item.items() if k != 'detailed_metrics'} # Add key metrics from detailed_metrics as separate fields if 'detailed_metrics' in item and item['detailed_metrics']: detailed = item['detailed_metrics'] # Attendance metrics if 'attendance' in detailed: attendance = detailed['attendance'] clean_item.update({ 'attendance_total_sessions': attendance.get('total_sessions', 0), 'attendance_attended_sessions': attendance.get('attended_sessions', 0), 'attendance_rate': attendance.get('attendance_rate', 0), 'attendance_late_sessions': attendance.get('late_sessions', 0), 'attendance_cancelled_sessions': attendance.get('cancelled_sessions', 0) }) # Student performance metrics if 'student_performance' in detailed: student_perf = detailed['student_performance'] clean_item.update({ 'student_total_students': student_perf.get('total_students', 0), 'student_average_rating': student_perf.get('average_rating', 0), 'student_completion_rate': student_perf.get('completion_rate', 0), 'student_active_students': student_perf.get('active_students', 0) }) # Session metrics if 'session_metrics' in detailed: session_metrics = detailed['session_metrics'] clean_item.update({ 'session_total_sessions': session_metrics.get('total_sessions', 0), 'session_completed_sessions': session_metrics.get('completed_sessions', 0), 'session_completion_rate': session_metrics.get('completion_rate', 0), 'session_average_duration': session_metrics.get('average_duration', 0), 'session_punctuality_rate': session_metrics.get('punctuality_rate', 0) }) export_data.append(clean_item) return export_data class PerformanceReportPDF(FPDF): def __init__(self): super().__init__(orientation='P', unit='mm', format='A4') self.set_auto_page_break(auto=True, margin=15) self.set_margins(15, 15, 15) # Professional color palette self.primary_color = (59, 130, 246) # Blue-500 self.secondary_color = (16, 185, 129) # Emerald-500 self.accent_color = (245, 158, 11) # Amber-500 self.danger_color = (239, 68, 68) # Red-500 self.success_color = (34, 197, 94) # Green-500 self.gray_light = (243, 244, 246) # Gray-100 self.gray_medium = (156, 163, 175) # Gray-400 self.gray_dark = (75, 85, 99) # Gray-600 self.white = (255, 255, 255) self.black = (0, 0, 0) def header(self): # No header - we'll handle it manually pass def footer(self): # Professional footer self.set_y(-15) self.set_font('Arial', 'I', 8) self.set_text_color(self.gray_medium[0], self.gray_medium[1], self.gray_medium[2]) self.cell(0, 10, f'Page {self.page_no()}', 0, 0, 'C') def set_document_title(self, title, subtitle=None): """Set the document title with professional styling""" # Title with blue background self.set_fill_color(self.primary_color[0], self.primary_color[1], self.primary_color[2]) self.set_text_color(self.white[0], self.white[1], self.white[2]) self.set_font('Arial', 'B', 18) self.cell(0, 15, title, 0, 1, 'C', True) if subtitle: # Subtitle self.set_text_color(self.gray_dark[0], self.gray_dark[1], self.gray_dark[2]) self.set_font('Arial', '', 12) self.cell(0, 8, subtitle, 0, 1, 'C') self.ln(10) def create_performance_report(self, data, export_type): """Create a comprehensive performance report""" # Summary statistics self._create_summary_section(data, export_type) self.ln(10) # Individual performance details self._create_detailed_section(data, export_type) def _create_summary_section(self, data, export_type): """Create summary statistics section""" # Section header self.set_fill_color(self.gray_light[0], self.gray_light[1], self.gray_light[2]) self.set_text_color(self.black[0], self.black[1], self.black[2]) self.set_font('Arial', 'B', 14) self.cell(0, 10, f'{export_type.title()} Performance Summary', 0, 1, 'L', True) self.ln(5) # Calculate summary statistics total_count = len(data) avg_performance = sum(item.get('performance_score', 0) for item in data) / total_count if total_count > 0 else 0 excellent_count = len([item for item in data if item.get('status') == 'excellent']) needs_improvement_count = len([item for item in data if item.get('status') == 'needs_improvement']) # Summary cards self._create_summary_card('Total Count', str(total_count), self.primary_color) self._create_summary_card('Average Performance', f"{avg_performance:.1f}%", self.secondary_color) self._create_summary_card('Excellent Performers', str(excellent_count), self.success_color) self._create_summary_card('Needs Improvement', str(needs_improvement_count), self.danger_color) def _create_summary_card(self, title, value, color): """Create a summary card""" # Card background self.set_fill_color(color[0], color[1], color[2]) self.set_text_color(self.white[0], self.white[1], self.white[2]) # Title self.set_font('Arial', 'B', 10) self.cell(40, 8, title, 0, 0, 'C', True) # Value self.set_font('Arial', 'B', 14) self.cell(20, 8, value, 0, 0, 'C', True) # Spacing self.cell(5, 8, '', 0, 0, 'C') def _create_detailed_section(self, data, export_type): """Create detailed performance section for each person""" # Section header self.set_fill_color(self.gray_light[0], self.gray_light[1], self.gray_light[2]) self.set_text_color(self.black[0], self.black[1], self.black[2]) self.set_font('Arial', 'B', 14) self.cell(0, 10, f'Detailed {export_type.title()} Performance Analysis', 0, 1, 'L', True) self.ln(5) for i, person in enumerate(data): # Check if we need a new page if self.get_y() > 250: self.add_page() self._create_person_detail_card(person, export_type, i + 1) self.ln(8) def _create_person_detail_card(self, person, export_type, index): """Create a detailed card for each person""" # Card border self.set_draw_color(self.gray_medium[0], self.gray_medium[1], self.gray_medium[2]) self.set_line_width(0.5) # Header section self.set_fill_color(self.primary_color[0], self.primary_color[1], self.primary_color[2]) self.set_text_color(self.white[0], self.white[1], self.white[2]) self.set_font('Arial', 'B', 12) # Name and status name = person.get('name', 'Unknown') status = person.get('status', 'unknown').replace('_', ' ').title() status_color = self._get_status_color(person.get('status', 'unknown')) # Draw header rectangle self.rect(self.get_x(), self.get_y(), 180, 12, 'F') self.cell(120, 12, f"{index}. {name}", 0, 0, 'L') self.set_fill_color(status_color[0], status_color[1], status_color[2]) self.rect(self.get_x(), self.get_y(), 60, 12, 'F') self.cell(60, 12, status, 0, 1, 'C') # Details section self.set_fill_color(self.white[0], self.white[1], self.white[2]) self.set_text_color(self.black[0], self.black[1], self.black[2]) # Performance score with progress bar self.set_font('Arial', 'B', 11) self.cell(40, 8, 'Performance Score:', 0, 0, 'L') score = person.get('performance_score', 0) self.cell(20, 8, f"{score}%", 0, 0, 'R') # Progress bar bar_width = 80 bar_height = 4 progress_width = (score / 100) * bar_width # Background bar self.set_fill_color(self.gray_light[0], self.gray_light[1], self.gray_light[2]) self.rect(self.get_x(), self.get_y() + 2, bar_width, bar_height, 'F') # Progress bar progress_color = self._get_performance_color(score) self.set_fill_color(progress_color[0], progress_color[1], progress_color[2]) self.rect(self.get_x(), self.get_y() + 2, progress_width, bar_height, 'F') self.set_x(self.get_x() + bar_width + 5) self.ln(8) # Key metrics in a grid metrics = self._get_key_metrics(person, export_type) self._create_metrics_grid(metrics) # Detailed explanation self._create_detailed_explanation(person, export_type) def _get_status_color(self, status): """Get color for status""" status_colors = { 'excellent': self.success_color, 'good': self.secondary_color, 'average': self.accent_color, 'needs_improvement': self.danger_color } return status_colors.get(status, self.gray_medium) def _get_performance_color(self, score): """Get color for performance score""" if score >= 90: return self.success_color elif score >= 80: return self.secondary_color elif score >= 70: return self.accent_color else: return self.danger_color def _get_key_metrics(self, person, export_type): """Get key metrics based on export type""" if export_type == 'tutor': return { 'Students': person.get('total_students', 0), 'Rating': person.get('average_rating', 0), 'Attendance': f"{person.get('attendance_rate', 0)}%", 'Sessions': person.get('sessions_completed', 0) } else: # supervisor return { 'Tutors Supervised': person.get('tutors_supervised', 0), 'Courses': person.get('courses_overseeing', 0), 'Avg Tutor Rating': person.get('average_tutor_rating', 0), 'Retention Rate': f"{person.get('tutor_retention_rate', 0)}%" } def _create_metrics_grid(self, metrics): """Create a metrics grid""" self.set_font('Arial', '', 9) self.set_fill_color(self.gray_light[0], self.gray_light[1], self.gray_light[2]) col_width = 45 for i, (label, value) in enumerate(metrics.items()): # Label self.set_text_color(self.gray_dark[0], self.gray_dark[1], self.gray_dark[2]) self.cell(col_width, 6, f"{label}:", 0, 0, 'L') # Value self.set_text_color(self.black[0], self.black[1], self.black[2]) self.set_font('Arial', 'B', 9) self.cell(col_width, 6, str(value), 0, 0, 'R') self.set_font('Arial', '', 9) if (i + 1) % 2 == 0: self.ln(6) if len(metrics) % 2 != 0: self.ln(6) def _create_detailed_explanation(self, person, export_type): """Create detailed explanation for the person's performance""" self.ln(3) self.set_font('Arial', 'B', 10) self.set_text_color(self.gray_dark[0], self.gray_dark[1], self.gray_dark[2]) self.cell(0, 6, 'Performance Analysis:', 0, 1, 'L') self.set_font('Arial', '', 9) self.set_text_color(self.black[0], self.black[1], self.black[2]) # Generate detailed explanation explanation = self._generate_performance_explanation(person, export_type) # Multi-cell for explanation text self.multi_cell(180, 4, explanation, 0, 'J') def _generate_performance_explanation(self, person, export_type): """Generate detailed performance explanation""" name = person.get('name', 'This person') score = person.get('performance_score', 0) status = person.get('status', 'unknown').replace('_', ' ') if export_type == 'tutor': students = person.get('total_students', 0) rating = person.get('average_rating', 0) attendance = person.get('attendance_rate', 0) sessions = person.get('sessions_completed', 0) explanation = f"{name} demonstrates {status} performance with an overall score of {score}%. " if score >= 90: explanation += f"This exceptional performance is reflected in their outstanding student rating of {rating}/5.0 and excellent attendance rate of {attendance}%. " explanation += f"With {students} students under their guidance and {sessions} completed sessions, they consistently deliver high-quality education. " explanation += "This tutor serves as an excellent role model and should be considered for leadership opportunities or advanced responsibilities." elif score >= 80: explanation += f"This strong performance is supported by a solid student rating of {rating}/5.0 and good attendance rate of {attendance}%. " explanation += f"Managing {students} students with {sessions} completed sessions shows effective workload management. " explanation += "With continued development and support, this tutor has the potential to achieve excellent performance levels." elif score >= 70: explanation += f"While showing average performance, there are positive indicators including a student rating of {rating}/5.0 and attendance rate of {attendance}%. " explanation += f"With {students} students and {sessions} sessions, there's room for improvement in teaching effectiveness. " explanation += "Targeted professional development and mentoring could help elevate their performance to the next level." else: explanation += f"This performance level requires immediate attention. Despite managing {students} students and completing {sessions} sessions, " explanation += f"the low score of {score}% indicates significant areas for improvement. " explanation += "A comprehensive development plan, additional training, and close supervision are recommended to address performance gaps." else: # supervisor tutors = person.get('tutors_supervised', 0) courses = person.get('courses_overseeing', 0) avg_rating = person.get('average_tutor_rating', 0) retention = person.get('tutor_retention_rate', 0) explanation = f"{name} shows {status} supervisory performance with an overall score of {score}%. " if score >= 90: explanation += f"This exceptional leadership is demonstrated through their supervision of {tutors} tutors across {courses} courses. " explanation += f"The high average tutor rating of {avg_rating}/5.0 and retention rate of {retention}% indicate excellent management skills. " explanation += "This supervisor effectively develops and retains talent while maintaining high standards across their department." elif score >= 80: explanation += f"This strong supervisory performance is evidenced by managing {tutors} tutors across {courses} courses. " explanation += f"With an average tutor rating of {avg_rating}/5.0 and retention rate of {retention}%, they demonstrate solid leadership capabilities. " explanation += "Continued focus on team development and performance management will help achieve excellent supervisory standards." elif score >= 70: explanation += f"While showing average supervisory performance, they oversee {tutors} tutors across {courses} courses. " explanation += f"The average tutor rating of {avg_rating}/5.0 and retention rate of {retention}% suggest areas for improvement in leadership effectiveness. " explanation += "Enhanced supervisory training and mentorship programs could help improve team performance and retention." else: explanation += f"This performance level requires immediate supervisory intervention. Despite overseeing {tutors} tutors across {courses} courses, " explanation += f"the low score of {score}% indicates significant leadership challenges. " explanation += "Urgent action including supervisory training, performance management support, and potential restructuring may be necessary." return explanation