from src.models.models import ( Student, Course, Enrollment, Assignment, AssignmentSubmission, Attendance, TeachingSession, tutor_course_association, Tutor, Supervisor, TutorTeachingLog, SupervisorTeachingLog, TutorAvailability, SupervisorAvailability, DailyTeachingSession, Speciality ) from src.utils import custom_response from src.models import DatabaseContextManager from sqlalchemy import func, case from datetime import datetime, timedelta import random class ReportsManager: def __init__(self): pass def get_academic_report(self, user_id: str, user_type: str, period: str = 'current'): """ Generate comprehensive academic performance report. Args: user_id: ID of the requesting user user_type: Type of user (student, tutor, supervisor) period: Report period (current, previous, year) Returns: Dict containing academic report data """ with DatabaseContextManager() as ctx: try: # Determine date range based on period end_date = datetime.now() if period == 'current': start_date = end_date - timedelta(days=30) elif period == 'previous': start_date = end_date - timedelta(days=60) end_date = end_date - timedelta(days=30) elif period == 'year': start_date = end_date - timedelta(days=365) else: start_date = end_date - timedelta(days=30) # Get user-specific data based on user type if user_type == 'supervisor': return self._get_supervisor_report(ctx, user_id, start_date, end_date) elif user_type == 'tutor': return self._get_tutor_report(ctx, user_id, start_date, end_date) elif user_type == 'student': return self._get_student_report(ctx, user_id, start_date, end_date) else: return custom_response( success=False, data="Invalid user type", status_code=400 ) except Exception as e: return custom_response( success=False, data=f"Error generating report: {str(e)}", status_code=500 ) def _get_supervisor_report(self, ctx, supervisor_id: str, start_date: datetime, end_date: datetime): """Generate report for supervisor""" # Get courses supervised by this supervisor courses = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id, Course.is_active == True ).all() if not courses: return custom_response( success=False, data="No courses found for this supervisor", status_code=404 ) course_ids = [course.id for course in courses] # Get summary statistics summary = self._get_summary_stats(ctx, course_ids, start_date, end_date) # Get trend data trend = self._get_trend_data(ctx, course_ids, start_date, end_date) # Get grade distribution grades = self._get_grade_distribution(ctx, course_ids) # Get student performance students = self._get_student_performance(ctx, course_ids) # Get subject performance subjects = self._get_subject_performance(ctx, course_ids) # Get recent activity recent_activity = self._get_recent_activity(ctx, course_ids, start_date, end_date) return custom_response( success=True, data={ 'summary': summary, 'trend': trend, 'grades': grades, 'students': students, 'subjects': subjects, 'recentActivity': recent_activity }, status_code=200 ) def _get_tutor_report(self, ctx, tutor_id: str, start_date: datetime, end_date: datetime): """Generate report for tutor""" # Get courses taught by this tutor using the association table courses = ctx.session.query(Course).join( tutor_course_association, Course.id == tutor_course_association.c.course_id ).filter( tutor_course_association.c.tutor_id == tutor_id, Course.is_active == True ).all() if not courses: return custom_response( success=False, data="No courses found for this tutor", status_code=404 ) course_ids = [course.id for course in courses] # Get summary statistics summary = self._get_summary_stats(ctx, course_ids, start_date, end_date) # Get trend data trend = self._get_trend_data(ctx, course_ids, start_date, end_date) # Get grade distribution grades = self._get_grade_distribution(ctx, course_ids) # Get student performance students = self._get_student_performance(ctx, course_ids) # Get subject performance subjects = self._get_subject_performance(ctx, course_ids) # Get recent activity recent_activity = self._get_recent_activity(ctx, course_ids, start_date, end_date) return custom_response( success=True, data={ 'summary': summary, 'trend': trend, 'grades': grades, 'students': students, 'subjects': subjects, 'recentActivity': recent_activity }, status_code=200 ) def _get_student_report(self, ctx, student_id: str, start_date: datetime, end_date: datetime): """Generate report for student""" # Get enrollments for this student enrollments = ctx.session.query(Enrollment).filter( Enrollment.student_id == student_id, Enrollment.status == 'active' ).all() if not enrollments: return custom_response( success=False, data="No enrollments found for this student", status_code=404 ) course_ids = [enrollment.course_id for enrollment in enrollments] # Get summary statistics summary = self._get_summary_stats(ctx, course_ids, start_date, end_date) # Get trend data trend = self._get_trend_data(ctx, course_ids, start_date, end_date) # Get grade distribution grades = self._get_grade_distribution(ctx, course_ids) # Get student performance (only for this student) students = self._get_student_performance(ctx, course_ids, student_id) # Get subject performance subjects = self._get_subject_performance(ctx, course_ids) # Get recent activity recent_activity = self._get_recent_activity(ctx, course_ids, start_date, end_date, student_id) return custom_response( success=True, data={ 'summary': summary, 'trend': trend, 'grades': grades, 'students': students, 'subjects': subjects, 'recentActivity': recent_activity }, status_code=200 ) def _get_summary_stats(self, ctx, course_ids: list, start_date: datetime, end_date: datetime): """Get summary statistics""" # Calculate average score avg_score = ctx.session.query( func.avg(Enrollment.grade) ).filter( Enrollment.course_id.in_(course_ids), Enrollment.grade.isnot(None) ).scalar() # Calculate pass rate total_enrollments = ctx.session.query(Enrollment).filter( Enrollment.course_id.in_(course_ids) ).count() pass_count = ctx.session.query(Enrollment).filter( Enrollment.course_id.in_(course_ids), Enrollment.grade >= 60 ).count() pass_rate = (pass_count / total_enrollments * 100) if total_enrollments > 0 else 0 fail_rate = 100 - pass_rate # Get top performer top_performer = ctx.session.query( Student, Enrollment ).join( Enrollment, Enrollment.student_id == Student.id ).filter( Enrollment.course_id.in_(course_ids), Enrollment.grade.isnot(None) ).order_by( Enrollment.grade.desc() ).first() # Calculate total students total_students = ctx.session.query( func.count(func.distinct(Enrollment.student_id)) ).filter( Enrollment.course_id.in_(course_ids) ).scalar() # Calculate improvement rate (mock data for now) improvement_rate = random.uniform(2.0, 8.0) # Calculate attendance rate attendance_records = ctx.session.query( func.count(Attendance.id), func.sum(case((Attendance.status == 'present', 1), else_=0)) ).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( TeachingSession.course_id.in_(course_ids), TeachingSession.start_time.between(start_date, end_date) ).first() attendance_rate = 0 if attendance_records[0] and attendance_records[0] > 0: attendance_rate = (attendance_records[1] / attendance_records[0]) * 100 return { 'averageScore': round(avg_score, 1) if avg_score else 0, 'passRate': round(pass_rate, 1), 'failRate': round(fail_rate, 1), 'topPerformer': f"{top_performer[0].first_name} {top_performer[0].last_name}" if top_performer else "N/A", 'topScore': round(top_performer[1].grade, 1) if top_performer and top_performer[1].grade else 0, 'totalStudents': total_students or 0, 'improvementRate': round(improvement_rate, 1), 'attendanceRate': round(attendance_rate, 1) } def _get_trend_data(self, ctx, course_ids: list, start_date: datetime, end_date: datetime): """Get trend data for the last 6 months""" trend_data = [] current_date = start_date for i in range(6): month_start = current_date.replace(day=1) month_end = (month_start + timedelta(days=32)).replace(day=1) - timedelta(days=1) # Calculate average score for this month avg_score = ctx.session.query( func.avg(Enrollment.grade) ).filter( Enrollment.course_id.in_(course_ids), Enrollment.grade.isnot(None), Enrollment.enrollment_date.between(month_start, month_end) ).scalar() trend_data.append({ 'label': month_start.strftime('%b'), 'value': round(avg_score, 1) if avg_score else 0, 'change': random.uniform(-3.0, 5.0), 'period': month_start.strftime('%Y-%m') }) current_date = (current_date + timedelta(days=32)).replace(day=1) return trend_data def _get_grade_distribution(self, ctx, course_ids: list): """Get grade distribution""" grade_ranges = [ ('A', 90, 100, '#10b981'), ('B', 80, 89, '#3b82f6'), ('C', 70, 79, '#f59e0b'), ('D', 60, 69, '#ef4444'), ('F', 0, 59, '#8b5cf6') ] grades = [] total_students = ctx.session.query(Enrollment).filter( Enrollment.course_id.in_(course_ids) ).count() for grade, min_score, max_score, color in grade_ranges: count = ctx.session.query(Enrollment).filter( Enrollment.course_id.in_(course_ids), Enrollment.grade.between(min_score, max_score) ).count() percentage = (count / total_students * 100) if total_students > 0 else 0 grades.append({ 'grade': grade, 'value': count, 'color': color, 'percentage': round(percentage, 1) }) return grades def _get_student_performance(self, ctx, course_ids: list, student_id: str = None): """Get student performance data""" query = ctx.session.query( Student, Enrollment, func.avg(AssignmentSubmission.grade).label('avg_score'), func.sum(case((Attendance.status == 'present', 1), else_=0)).label('present_count'), func.count(Attendance.id).label('attendance_count') ).join( Enrollment, Enrollment.student_id == Student.id ).outerjoin( AssignmentSubmission, AssignmentSubmission.student_id == Student.id ).outerjoin( Attendance, Attendance.student_id == Student.id ).filter( Enrollment.course_id.in_(course_ids), Enrollment.status == 'active' ) if student_id: query = query.filter(Student.id == student_id) results = query.group_by( Student.id, Enrollment.id ).all() students = [] for i, (student, enrollment, avg_score, present_count, attendance_count) in enumerate(results, 1): attendance_percentage = round((present_count / attendance_count * 100), 1) if attendance_count else 0 # Determine performance status if avg_score is None: status = "pass" # Default to pass if no assignments elif avg_score >= 60: status = "pass" else: status = "fail" students.append({ 'id': student.id, 'name': f"{student.first_name} {student.last_name}", 'score': round(avg_score, 1) if avg_score else 0, 'grade': self._get_grade_letter(avg_score) if avg_score else 'N/A', 'status': status, 'attendance': attendance_percentage, 'improvement': random.uniform(-5.0, 8.0), 'rank': i }) return students def _get_subject_performance(self, ctx, course_ids: list): """Get subject performance data""" subjects = [] colors = ['#10b981', '#3b82f6', '#f59e0b', '#ef4444', '#8b5cf6'] for i, course_id in enumerate(course_ids[:5]): # Limit to 5 subjects course = ctx.session.query(Course).filter(Course.id == course_id).first() if not course: continue # Calculate average score for this course avg_score = ctx.session.query( func.avg(Enrollment.grade) ).filter( Enrollment.course_id == course_id, Enrollment.grade.isnot(None) ).scalar() # Calculate pass rate total_enrollments = ctx.session.query(Enrollment).filter( Enrollment.course_id == course_id ).count() pass_count = ctx.session.query(Enrollment).filter( Enrollment.course_id == course_id, Enrollment.grade >= 60 ).count() pass_rate = (pass_count / total_enrollments * 100) if total_enrollments > 0 else 0 subjects.append({ 'name': course.title, 'averageScore': round(avg_score, 1) if avg_score else 0, 'passRate': round(pass_rate, 1), 'totalStudents': total_enrollments, 'color': colors[i % len(colors)] }) return subjects def _get_recent_activity(self, ctx, course_ids: list, start_date: datetime, end_date: datetime, student_id: str = None): """Get recent activity data""" activities = [] # Get recent assignment submissions submissions = ctx.session.query( AssignmentSubmission, Student, Assignment ).join( Student, AssignmentSubmission.student_id == Student.id ).join( Assignment, AssignmentSubmission.assignment_id == Assignment.id ).filter( Assignment.course_id.in_(course_ids), AssignmentSubmission.submitted_at.between(start_date, end_date) ) if student_id: submissions = submissions.filter(Student.id == student_id) submissions = submissions.order_by( AssignmentSubmission.submitted_at.desc() ).limit(5).all() for submission, student, assignment in submissions: activity_type = 'achievement' if submission.grade >= 90 else 'improvement' activities.append({ 'id': submission.id, 'type': activity_type, 'student': f"{student.first_name} {student.last_name}", 'description': f"Submitted assignment: {assignment.title}", 'date': submission.submitted_at.strftime('%Y-%m-%d'), 'score': submission.grade }) # Add some mock activities if not enough real ones if len(activities) < 4: mock_activities = [ { 'id': 'mock1', 'type': 'award', 'student': 'Sarah Johnson', 'description': 'Awarded for English Excellence', 'date': '2024-05-20' }, { 'id': 'mock2', 'type': 'warning', 'student': 'Lisa Thompson', 'description': 'Attendance below 70%', 'date': '2024-05-18' } ] activities.extend(mock_activities) return activities[:5] # Return top 5 activities def _get_grade_letter(self, score: float) -> str: """Convert numeric score to letter grade""" if score is None: return 'N/A' elif score >= 90: return 'A' elif score >= 80: return 'B' elif score >= 70: return 'C' elif score >= 60: return 'D' else: return 'F' def get_tutor_performance_reports(self, period: str = '30d', department: str = None): """ Get comprehensive tutor performance reports Args: period: Time period (7d, 30d, 90d, 1y) department: Optional department filter Returns: Dict containing tutor performance data """ with DatabaseContextManager() as ctx: try: # Calculate date range end_date = datetime.now() if period == '7d': start_date = end_date - timedelta(days=7) elif period == '30d': start_date = end_date - timedelta(days=30) elif period == '90d': start_date = end_date - timedelta(days=90) elif period == '1y': start_date = end_date - timedelta(days=365) else: start_date = end_date - timedelta(days=30) # Get tutors with performance data tutors_query = ctx.session.query(Tutor) if department: tutors_query = tutors_query.filter(Tutor.department == department) tutors = tutors_query.all() if not tutors: return custom_response( success=False, data="No tutors found", status_code=404 ) # Get overall stats overall_stats = self._get_tutor_overall_stats(ctx, tutors, start_date, end_date) # Get individual tutor performance tutor_performance = self._get_individual_tutor_performance(ctx, tutors, start_date, end_date) # Get performance trends trends = self._get_tutor_performance_trends(ctx, tutors, start_date, end_date) # Get department breakdown department_breakdown = self._get_tutor_department_breakdown(ctx, tutors, start_date, end_date) return custom_response( success=True, data={ 'overall_stats': overall_stats, 'tutors': tutor_performance, 'trends': trends, 'department_breakdown': department_breakdown, 'period': period, 'date_range': { 'start': start_date.strftime('%Y-%m-%d'), 'end': end_date.strftime('%Y-%m-%d') } }, status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error generating tutor performance report: {str(e)}", status_code=500 ) def get_supervisor_performance_reports(self, period: str = '30d', department: str = None): """ Get comprehensive supervisor performance reports Args: period: Time period (7d, 30d, 90d, 1y) department: Optional department filter Returns: Dict containing supervisor performance data """ with DatabaseContextManager() as ctx: try: # Calculate date range end_date = datetime.now() if period == '7d': start_date = end_date - timedelta(days=7) elif period == '30d': start_date = end_date - timedelta(days=30) elif period == '90d': start_date = end_date - timedelta(days=90) elif period == '1y': start_date = end_date - timedelta(days=365) else: start_date = end_date - timedelta(days=30) # Get supervisors with performance data supervisors_query = ctx.session.query(Supervisor) if department: supervisors_query = supervisors_query.filter(Supervisor.department == department) supervisors = supervisors_query.all() if not supervisors: return custom_response( success=False, data="No supervisors found", status_code=404 ) # Get overall stats overall_stats = self._get_supervisor_overall_stats(ctx, supervisors, start_date, end_date) # Get individual supervisor performance supervisor_performance = self._get_individual_supervisor_performance(ctx, supervisors, start_date, end_date) # Get performance trends trends = self._get_supervisor_performance_trends(ctx, supervisors, start_date, end_date) # Get department breakdown department_breakdown = self._get_supervisor_department_breakdown(ctx, supervisors, start_date, end_date) return custom_response( success=True, data={ 'overall_stats': overall_stats, 'supervisors': supervisor_performance, 'trends': trends, 'department_breakdown': department_breakdown, 'period': period, 'date_range': { 'start': start_date.strftime('%Y-%m-%d'), 'end': end_date.strftime('%Y-%m-%d') } }, status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error generating supervisor performance report: {str(e)}", status_code=500 ) def get_performance_overview(self, period: str = '30d'): """ Get overall performance overview for all tutors and supervisors Args: period: Time period (7d, 30d, 90d, 1y) Returns: Dict containing overall performance data """ with DatabaseContextManager() as ctx: try: # Calculate date range end_date = datetime.now() if period == '7d': start_date = end_date - timedelta(days=7) elif period == '30d': start_date = end_date - timedelta(days=30) elif period == '90d': start_date = end_date - timedelta(days=90) elif period == '1y': start_date = end_date - timedelta(days=365) else: start_date = end_date - timedelta(days=30) # Get all tutors and supervisors tutors = ctx.session.query(Tutor).all() supervisors = ctx.session.query(Supervisor).all() # Get overview stats overview_stats = self._get_overall_performance_stats(ctx, tutors, supervisors, start_date, end_date) # Get performance comparison comparison = self._get_performance_comparison(ctx, tutors, supervisors, start_date, end_date) # Get top performers top_performers = self._get_top_performers(ctx, tutors, supervisors, start_date, end_date) # Get performance alerts alerts = self._get_performance_alerts(ctx, tutors, supervisors, start_date, end_date) return custom_response( success=True, data={ 'overview_stats': overview_stats, 'comparison': comparison, 'top_performers': top_performers, 'alerts': alerts, 'period': period, 'date_range': { 'start': start_date.strftime('%Y-%m-%d'), 'end': end_date.strftime('%Y-%m-%d') } }, status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error generating performance overview: {str(e)}", status_code=500 ) def get_attendance_reports(self, start_date: str = None, end_date: str = None, speciality: str = None, course_id: str = None, include_course: bool = False, include_tutor: bool = False): """ Generate comprehensive attendance reports for dashboard Args: start_date: Start date for filtering (YYYY-MM-DD) end_date: End date for filtering (YYYY-MM-DD) speciality: Filter by speciality course_id: Filter by course ID include_course: Include course details include_tutor: Include tutor details Returns: Dict containing attendance report data """ with DatabaseContextManager() as ctx: try: # Parse dates if start_date: start_date = datetime.strptime(start_date, '%Y-%m-%d') else: start_date = datetime.now() - timedelta(days=30) if end_date: end_date = datetime.strptime(end_date, '%Y-%m-%d') else: end_date = datetime.now() # Get attendance records with filters attendance_query = ctx.session.query(Attendance).join( DailyTeachingSession, Attendance.daily_session_id == DailyTeachingSession.id ).join( Course, DailyTeachingSession.course_id == Course.id ).filter( DailyTeachingSession.session_date.between(start_date.date(), end_date.date()) ) # Apply filters if speciality and speciality != 'all': attendance_query = attendance_query.filter(Course.department == speciality) if course_id and course_id != 'all': attendance_query = attendance_query.filter(Course.id == course_id) attendance_records = attendance_query.all() # Get summary statistics summary = self._get_attendance_summary(ctx, attendance_records) # Get trend data trends = self._get_attendance_trends(ctx, start_date, end_date, speciality, course_id) # Get course statistics courses = self._get_course_attendance_stats(ctx, start_date, end_date, speciality, include_course) # Get speciality statistics specialities = self._get_speciality_attendance_stats(ctx, start_date, end_date, include_course) # Format attendance records formatted_records = self._format_attendance_records(ctx, attendance_records, include_course, include_tutor) return custom_response( success=True, data={ 'summary': summary, 'trends': trends, 'records': formatted_records, 'courses': courses, 'specialities': specialities }, status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error generating attendance reports: {str(e)}", status_code=500 ) def _get_attendance_summary(self, ctx, attendance_records): """Get attendance summary statistics""" total_records = len(attendance_records) present_count = sum(1 for record in attendance_records if record.status == 'present') absent_count = sum(1 for record in attendance_records if record.status == 'absent') late_count = sum(1 for record in attendance_records if record.status == 'late') excused_count = sum(1 for record in attendance_records if record.status == 'excused') # Get unique students student_ids = set(record.student_id for record in attendance_records) total_students = len(student_ids) # Calculate overall rate overall_rate = (present_count / total_records * 100) if total_records > 0 else 0 return { 'total_students': total_students, 'present_count': present_count, 'absent_count': absent_count, 'late_count': late_count, 'excused_count': excused_count, 'overall_rate': round(overall_rate, 1) } def _get_attendance_trends(self, ctx, start_date, end_date, speciality=None, course_id=None): """Get attendance trends over time""" trends = [] current_date = start_date while current_date <= end_date: # Get attendance for this date daily_query = ctx.session.query(Attendance).join( DailyTeachingSession, Attendance.daily_session_id == DailyTeachingSession.id ).join( Course, DailyTeachingSession.course_id == Course.id ).filter( DailyTeachingSession.session_date == current_date.date() ) if speciality and speciality != 'all': daily_query = daily_query.filter(Course.department == speciality) if course_id and course_id != 'all': daily_query = daily_query.filter(Course.id == course_id) daily_records = daily_query.all() if daily_records: total = len(daily_records) present = sum(1 for record in daily_records if record.status == 'present') absent = sum(1 for record in daily_records if record.status == 'absent') late = sum(1 for record in daily_records if record.status == 'late') excused = sum(1 for record in daily_records if record.status == 'excused') rate = (present / total * 100) if total > 0 else 0 trends.append({ 'date': current_date.strftime('%Y-%m-%d'), 'rate': round(rate, 1), 'present': present, 'absent': absent, 'late': late, 'excused': excused }) current_date += timedelta(days=1) return trends def _get_course_attendance_stats(self, ctx, start_date, end_date, speciality=None, include_course=False): """Get attendance statistics by course""" course_query = ctx.session.query(Course).join( DailyTeachingSession, Course.id == DailyTeachingSession.course_id ).join( Attendance, DailyTeachingSession.id == Attendance.daily_session_id ).filter( DailyTeachingSession.session_date.between(start_date.date(), end_date.date()) ) if speciality and speciality != 'all': course_query = course_query.filter(Course.department == speciality) courses = course_query.distinct().all() course_stats = [] for course in courses: # Get attendance records for this course attendance_records = ctx.session.query(Attendance).join( DailyTeachingSession, Attendance.daily_session_id == DailyTeachingSession.id ).filter( DailyTeachingSession.course_id == course.id, DailyTeachingSession.session_date.between(start_date.date(), end_date.date()) ).all() if attendance_records: total = len(attendance_records) present = sum(1 for record in attendance_records if record.status == 'present') rate = (present / total * 100) if total > 0 else 0 # Get unique students for this course student_ids = set(record.student_id for record in attendance_records) course_stats.append({ 'id': course.id, 'code': course.code, 'title': course.title, 'department': course.department, 'total_students': len(student_ids), 'attendance_rate': round(rate, 1) }) return course_stats def _get_speciality_attendance_stats(self, ctx, start_date, end_date, include_course=False): """Get attendance statistics by speciality""" # Get all departments/specialities departments = ctx.session.query(Course.department).distinct().all() speciality_stats = [] for (department,) in departments: if not department: continue # Get attendance records for this department attendance_records = ctx.session.query(Attendance).join( DailyTeachingSession, Attendance.daily_session_id == DailyTeachingSession.id ).join( Course, DailyTeachingSession.course_id == Course.id ).filter( Course.department == department, DailyTeachingSession.session_date.between(start_date.date(), end_date.date()) ).all() if attendance_records: total = len(attendance_records) present = sum(1 for record in attendance_records if record.status == 'present') rate = (present / total * 100) if total > 0 else 0 # Get unique students for this department student_ids = set(record.student_id for record in attendance_records) speciality_stats.append({ 'id': department.lower().replace(' ', '_'), 'name': department, 'total_students': len(student_ids), 'attendance_rate': round(rate, 1) }) return speciality_stats def _format_attendance_records(self, ctx, attendance_records, include_course=False, include_tutor=False): """Format attendance records for frontend""" formatted_records = [] for record in attendance_records: # Get student info student = ctx.session.query(Student).filter(Student.id == record.student_id).first() # Get daily session info daily_session = ctx.session.query(DailyTeachingSession).filter( DailyTeachingSession.id == record.daily_session_id ).first() if not student or not daily_session: continue # Get course info course = ctx.session.query(Course).filter(Course.id == daily_session.course_id).first() formatted_record = { 'id': record.id, 'student_id': record.student_id, 'student_name': f"{student.first_name} {student.last_name}", 'date': daily_session.session_date.strftime('%Y-%m-%d'), 'status': record.status, 'course_code': course.code if course else 'N/A', 'course_name': course.title if course else 'N/A', 'speciality': course.department if course else 'N/A', 'minutes_late': record.minutes_late, 'notes': record.notes } if include_course and course: formatted_record['course'] = { 'id': course.id, 'code': course.code, 'title': course.title, 'department': course.department } if include_tutor and daily_session.tutor_id: tutor = ctx.session.query(Tutor).filter(Tutor.id == daily_session.tutor_id).first() if tutor: formatted_record['tutor'] = { 'id': tutor.id, 'name': f"{tutor.first_name} {tutor.last_name}", 'email': tutor.email } formatted_records.append(formatted_record) return formatted_records def get_attendance_report(self, session_id: str = None, course_id: str = None, start_date: str = None, end_date: str = None, include_course: bool = False, include_tutor: bool = False, include_student: bool = False, user_id: str = None, user_type: str = None, period: str = None): """ Generate attendance report for a specific session or course Args: session_id: Specific session ID to get attendance for course_id: Course ID to get attendance for start_date: Start date for filtering (YYYY-MM-DD) end_date: End date for filtering (YYYY-MM-DD) include_course: Include course details include_tutor: Include tutor details include_student: Include student details user_id: User ID requesting the report (for authorization) user_type: Type of user requesting the report (student, tutor, supervisor) period: Time period for the report (7d, 30d, 90d, 1y, current, previous, year) Returns: Dict containing attendance report data """ with DatabaseContextManager() as ctx: try: # Parse dates if start_date: start_date = datetime.strptime(start_date, '%Y-%m-%d') else: start_date = datetime.now() - timedelta(days=30) if end_date: end_date = datetime.strptime(end_date, '%Y-%m-%d') else: end_date = datetime.now() # Override dates based on period if provided if period: end_date = datetime.now() if period == '7d': start_date = end_date - timedelta(days=7) elif period == '30d': start_date = end_date - timedelta(days=30) elif period == '90d': start_date = end_date - timedelta(days=90) elif period == '1y': start_date = end_date - timedelta(days=365) elif period == 'current': start_date = end_date - timedelta(days=30) elif period == 'previous': start_date = end_date - timedelta(days=60) end_date = end_date - timedelta(days=30) elif period == 'year': start_date = end_date - timedelta(days=365) # Build attendance query attendance_query = ctx.session.query(Attendance) if session_id: # Get attendance for specific session attendance_query = attendance_query.filter( Attendance.daily_session_id == session_id ) elif course_id: # Get attendance for specific course attendance_query = attendance_query.join( DailyTeachingSession, Attendance.daily_session_id == DailyTeachingSession.id ).filter( DailyTeachingSession.course_id == course_id, DailyTeachingSession.session_date.between(start_date.date(), end_date.date()) ) else: # Get attendance for all sessions in date range attendance_query = attendance_query.join( DailyTeachingSession, Attendance.daily_session_id == DailyTeachingSession.id ).filter( DailyTeachingSession.session_date.between(start_date.date(), end_date.date()) ) attendance_records = attendance_query.all() if not attendance_records: return custom_response( success=True, data={ 'summary': { 'total_students': 0, 'present_count': 0, 'absent_count': 0, 'late_count': 0, 'excused_count': 0, 'overall_rate': 0 }, 'records': [], 'session_info': None, 'course_info': None }, status_code=200 ) # Get summary statistics summary = self._get_attendance_summary(ctx, attendance_records) # Format attendance records with additional info formatted_records = [] for record in attendance_records: # Get student info student = ctx.session.query(Student).filter(Student.id == record.student_id).first() # Get daily session info daily_session = ctx.session.query(DailyTeachingSession).filter( DailyTeachingSession.id == record.daily_session_id ).first() if not student or not daily_session: continue # Get course info course = ctx.session.query(Course).filter(Course.id == daily_session.course_id).first() formatted_record = { 'id': record.id, 'student_id': record.student_id, 'student_name': f"{student.first_name} {student.last_name}", 'date': daily_session.session_date.strftime('%Y-%m-%d'), 'status': record.status, 'minutes_late': record.minutes_late, 'notes': record.notes, 'marked_at': record.marked_at.isoformat() if record.marked_at else None } if include_student: formatted_record['student'] = { 'id': student.id, 'first_name': student.first_name, 'last_name': student.last_name, 'email': student.email, 'student_id': student.student_id } if include_course and course: formatted_record['course'] = { 'id': course.id, 'code': course.code, 'title': course.title, 'department': course.department } if include_tutor and daily_session.tutor_id: tutor = ctx.session.query(Tutor).filter(Tutor.id == daily_session.tutor_id).first() if tutor: formatted_record['tutor'] = { 'id': tutor.id, 'first_name': tutor.first_name, 'last_name': tutor.last_name, 'email': tutor.email } formatted_records.append(formatted_record) # Get session info if session_id provided session_info = None if session_id: daily_session = ctx.session.query(DailyTeachingSession).filter( DailyTeachingSession.id == session_id ).first() if daily_session: course = ctx.session.query(Course).filter(Course.id == daily_session.course_id).first() tutor = ctx.session.query(Tutor).filter(Tutor.id == daily_session.tutor_id).first() if daily_session.tutor_id else None session_info = { 'id': daily_session.id, 'session_date': daily_session.session_date.strftime('%Y-%m-%d'), 'start_time': daily_session.start_time.strftime('%H:%M') if daily_session.start_time else None, 'end_time': daily_session.end_time.strftime('%H:%M') if daily_session.end_time else None, 'room': daily_session.room, 'session_type': daily_session.session_type, 'status': daily_session.status, 'course': { 'id': course.id, 'code': course.code, 'title': course.title, 'department': course.department } if course else None, 'tutor': { 'id': tutor.id, 'name': f"{tutor.first_name} {tutor.last_name}", 'email': tutor.email } if tutor else None } # Get course info if course_id provided course_info = None if course_id: course = ctx.session.query(Course).filter(Course.id == course_id).first() if course: course_info = { 'id': course.id, 'code': course.code, 'title': course.title, 'department': course.department, 'credits': course.credits, 'description': course.description } return custom_response( success=True, data={ 'summary': summary, 'records': formatted_records, 'session_info': session_info, 'course_info': course_info }, status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error generating attendance report: {str(e)}", status_code=500 )