from src.utils import ( custom_response, check_password, generate_otp, send_email, hash_password ) from flask import current_app from src.models import DatabaseContextManager from src.models.models import ( NotificationPreference, User, Admin, TutorAvailability, TeachingSession, Tutor, TutorTeachingLog, Attendance, Supervisor, Timetable, Course, AttendanceStatus, TeachingVerificationMethod, Speciality, tutor_course_association, AcademicSession, Enrollment, enrollment_courses, SupervisorDepartment, TutorDepartment, CourseProgress, AssignmentSubmission, Assignment, Student, SubmissionStatus, course_speciality_association, CourseDepartment, CourseModule, CourseResource, AcademicSession, DailyTeachingSession ) from flask import send_file, Response from io import BytesIO from openpyxl.worksheet.datavalidation import DataValidation from flask import current_app, request import uuid import datetime import jwt from sqlalchemy import func, case, or_ from sqlalchemy.orm import joinedload from datetime import datetime, timedelta import pandas as pd import os class AdminManager: def login(self, payload): """Handle user login with enhanced security including device verification""" with DatabaseContextManager() as ctx: user = ctx.session.query(Admin).filter( Admin.email == payload['email'] ).first() if not user: return custom_response( success=False, data="Admin not found", status_code=404 ) # Verify password if not check_password(payload['password'], user.password_hash, salt=current_app.config['SECRET_KEY']): # Log failed login attempt user.failed_login_attempts = (user.failed_login_attempts or 0) + 1 # Lock account after 5 failed attempts if user.failed_login_attempts >= 5: user.account_locked = True user.account_locked_until = datetime.utcnow() + timedelta(minutes=30) ctx.session.commit() return custom_response( success=False, data="Invalid credentials", status_code=401 ) # Check if account is locked if user.account_locked: if user.account_locked_until and user.account_locked_until > datetime.utcnow(): return custom_response( success=False, data="Account temporarily locked. Try again later.", status_code=403 ) else: # Unlock account if lock period has passed user.account_locked = False user.failed_login_attempts = 0 otp = generate_otp() user.otp = otp user.otp_expiry = datetime.utcnow() + timedelta(minutes=10) user.otp_attempts = 0 # Reset failed login attempts user.failed_login_attempts = 0 ctx.session.commit() # Send OTP via email if self.send_otp_email(user=user, otp=otp): return custom_response( success=True, data={ "message": "OTP sent for verification", "two_factor_required": True, "user_id": user.id }, status_code=200 ) else: return custom_response( success=False, data="Failed to send OTP", status_code=500 ) def verify_otp(self, payload): """Verify OTP for 2FA login with enhanced security checks""" with DatabaseContextManager() as ctx: user = ctx.session.query(Admin).filter( Admin.email == payload.get('email') ).first() if not user: return custom_response( success=False, data="Invalid session", status_code=401 ) # Check if OTP fields exist (defensive programming) if not hasattr(user, 'otp_expiry') or not hasattr(user, 'otp') or not hasattr(user, 'otp_attempts'): return custom_response( success=False, data="OTP verification not properly configured", status_code=500 ) # Check OTP expiry if user.otp_expiry and user.otp_expiry < datetime.utcnow(): return custom_response( success=False, data="OTP expired", status_code=401 ) # Increment OTP attempts user.otp_attempts = (user.otp_attempts or 0) + 1 # Lock account after too many failed OTP attempts if user.otp_attempts >= 3: user.account_locked = True user.account_locked_until = datetime.utcnow() + timedelta(minutes=30) ctx.session.commit() return custom_response( success=False, data="Too many failed attempts. Account locked.", status_code=403 ) # Verify OTP if user.otp != payload['otp']: ctx.session.commit() return custom_response( success=False, data="Invalid OTP", status_code=401 ) # OTP is valid - complete login user.otp = None user.otp_expiry = None user.otp_attempts = 0 user.failed_login_attempts = 0 user.account_locked = False user.last_login = datetime.utcnow() ctx.session.commit() token = jwt.encode({ "sub": str(user.id), "name": f"{user.first_name} {user.last_name}", "email": user.email, "exp": datetime.utcnow() + timedelta(days=1), "iat": datetime.utcnow() }, current_app.config['SECRET_KEY'], algorithm='HS256') response = custom_response( data={ "admin": user.to_json(), "token": token }, status_code=200, success=True ) # Set HTTP-only cookie with JWT token response.set_cookie( 'admin_auth_token', value=token, httponly=True, secure=True, samesite='Strict', max_age=86400, # 1 day path='/' ) return response def remote_unlock_account(self, payload): """Remote unlock a temporarily locked account""" with DatabaseContextManager() as ctx: user_id = payload.get('user_id') if not user_id: return custom_response( success=False, data="User ID is required", status_code=400 ) # Find the user user = ctx.session.query(User).filter(User.id == user_id).first() if not user: return custom_response( success=False, data="User not found", status_code=404 ) # Check if account is actually locked if not user.account_locked: return custom_response( success=False, data="Account is not locked", status_code=400 ) # Check if lock period has already passed if user.account_locked_until and user.account_locked_until <= datetime.utcnow(): # Auto-unlock since time has passed user.account_locked = False user.failed_login_attempts = 0 user.account_locked_until = None ctx.session.commit() return custom_response( success=True, data={ "message": "Account automatically unlocked (lock period expired)", "user_id": user.id, "unlocked_at": datetime.utcnow().isoformat() }, status_code=200 ) # Perform remote unlock user.account_locked = False user.failed_login_attempts = 0 user.account_locked_until = None # Log the remote unlock action current_app.logger.info(f"Account {user.id} ({user.email}) remotely unlocked by user request") ctx.session.commit() return custom_response( success=True, data={ "message": "Account successfully unlocked", "user_id": user.id, "unlocked_at": datetime.utcnow().isoformat(), "previous_lock_until": user.account_locked_until.isoformat() if user.account_locked_until else None }, status_code=200 ) def unlock_admin_account(self, payload): """Unlock a locked admin account""" with DatabaseContextManager() as ctx: admin_id = payload.get('admin_id') unlocked_by = payload.get('unlocked_by') # ID of admin performing the unlock if not admin_id: return custom_response( success=False, data="Admin ID is required", status_code=400 ) if not unlocked_by: return custom_response( success=False, data="Unlocked by admin ID is required", status_code=400 ) # Find the admin to unlock first admin_to_unlock = ctx.session.query(Admin).filter(Admin.id == admin_id).first() if not admin_to_unlock: return custom_response( success=False, data="Admin account not found", status_code=404 ) # Verify the admin performing the unlock exists and is active # Special case: if unlocked_by is the same as admin_id, allow emergency unlock if unlocked_by == admin_id: # Emergency unlock case - admin unlocking themselves unlocking_admin = admin_to_unlock current_app.logger.warning( f"Emergency unlock: Admin {admin_to_unlock.id} ({admin_to_unlock.email}) unlocking themselves" ) else: unlocking_admin = ctx.session.query(Admin).filter( Admin.id == unlocked_by, Admin.active == True ).first() if not unlocking_admin: return custom_response( success=False, data="Unauthorized: Admin performing unlock not found or inactive", status_code=403 ) # Check if account is actually locked if not admin_to_unlock.account_locked: return custom_response( success=False, data="Admin account is not locked", status_code=400 ) # Check if lock period has already passed if admin_to_unlock.account_locked_until and admin_to_unlock.account_locked_until <= datetime.utcnow(): # Auto-unlock since time has passed admin_to_unlock.account_locked = False admin_to_unlock.failed_login_attempts = 0 admin_to_unlock.account_locked_until = None ctx.session.commit() return custom_response( success=True, data={ "message": "Admin account automatically unlocked (lock period expired)", "admin_id": admin_to_unlock.id, "admin_email": admin_to_unlock.email, "unlocked_at": datetime.utcnow().isoformat(), "unlocked_by": unlocking_admin.email }, status_code=200 ) # Perform manual unlock previous_lock_until = admin_to_unlock.account_locked_until.isoformat() if admin_to_unlock.account_locked_until else None admin_to_unlock.account_locked = False admin_to_unlock.failed_login_attempts = 0 admin_to_unlock.account_locked_until = None # Log the unlock action current_app.logger.info( f"Admin account {admin_to_unlock.id} ({admin_to_unlock.email}) unlocked by admin {unlocking_admin.id} ({unlocking_admin.email})" ) ctx.session.commit() return custom_response( success=True, data={ "message": "Admin account successfully unlocked", "admin_id": admin_to_unlock.id, "admin_email": admin_to_unlock.email, "admin_name": f"{admin_to_unlock.first_name} {admin_to_unlock.last_name}", "unlocked_at": datetime.utcnow().isoformat(), "unlocked_by": unlocking_admin.email, "unlocked_by_name": f"{unlocking_admin.first_name} {unlocking_admin.last_name}", "previous_lock_until": previous_lock_until, "failed_attempts_reset": True }, status_code=200 ) def get_locked_admin_accounts(self): """Get list of all locked admin accounts""" with DatabaseContextManager() as ctx: locked_admins = ctx.session.query(Admin).filter( Admin.account_locked == True ).all() locked_accounts = [] for admin in locked_admins: locked_accounts.append({ "admin_id": admin.id, "email": admin.email, "first_name": admin.first_name, "last_name": admin.last_name, "full_name": f"{admin.first_name} {admin.last_name}", "failed_login_attempts": admin.failed_login_attempts, "account_locked_until": admin.account_locked_until.isoformat() if admin.account_locked_until else None, "locked_since": admin.account_locked_until.isoformat() if admin.account_locked_until else "Unknown", "is_lock_expired": admin.account_locked_until <= datetime.utcnow() if admin.account_locked_until else False, "phone": admin.phone, "active": admin.active }) return custom_response( success=True, data={ "message": f"Found {len(locked_accounts)} locked admin accounts", "locked_accounts": locked_accounts, "total_count": len(locked_accounts), "retrieved_at": datetime.utcnow().isoformat() }, status_code=200 ) def _complete_login(self, user, device_info=None): """Complete the login process by generating auth token and logging device""" # Generate JWT token token_payload = { "sub": str(user.id), "name": f"{user.first_name} {user.last_name}", "email": user.email, "user_type": user.user_type.value, "exp": datetime.utcnow() + timedelta(days=1), "iat": datetime.utcnow(), "device_id": device_info['device_id'] if device_info else None } token = jwt.encode(token_payload, current_app.config['SECRET_KEY'], algorithm='HS256') return custom_response( data={ "user": { "id": user.id, "first_name": user.first_name, "last_name": user.last_name, "email": user.email, "user_type": user.user_type.value }, "token": token # Add token to response }, status_code=200, success=True ) def verify_auth_token(self): """Verify the authentication token from request payload""" # Get token from Authorization header auth_header = request.headers.get('Authorization') if not auth_header: return custom_response( success=False, data="Authentication required", status_code=401 ) # Check if header is in the format "Bearer " parts = auth_header.split() if len(parts) != 2 or parts[0].title() != 'Bearer': return custom_response( success=False, data="Invalid authorization header format", status_code=401 ) token = parts[1] if not token: return custom_response( success=False, data="Token is missing", status_code=401 ) try: # Decode the token payload = jwt.decode(token, current_app.config['SECRET_KEY'], algorithms=['HS256']) user_id = payload.get('sub') with DatabaseContextManager() as ctx: user = ctx.session.query(Admin).filter(Admin.id == user_id).first() if not user: return custom_response( success=False, data="Invalid token", status_code=401 ) return custom_response( success=True, data={ "admin": user.to_json() }, status_code=200 ) except jwt.ExpiredSignatureError: return custom_response( success=False, data="Session expired", status_code=401 ) except Exception as e: return custom_response( success=False, data=f"Invalid token: {str(e)}", status_code=401 ) def logout(self): """Handle user logout""" auth_token = request.cookies.get('auth_token') if auth_token: try: # Decode token to get user ID payload = jwt.decode(auth_token, current_app.config['SECRET_KEY'], algorithms=['HS256']) user_id = payload.get('sub') except: pass response = custom_response( success=True, data="Logged out successfully", status_code=200 ) # Clear auth token cookie response.delete_cookie('auth_token') return response def send_otp_email(self, user, otp): """Send OTP via email with platform-specific styling""" subject = "Your Verification Code for Educational Platform" login_time = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC') ip_address = request.remote_addr message = f""" Admin Portal Verification

🔐 Admin Portal Access

Two-Factor Authentication Required

Hello {user.first_name} {user.last_name},

A login attempt has been detected on your administrator account. To complete the authentication process and access the admin portal, please use the verification code below:

Verification Code

{otp}

⏰ This code expires in 10 minutes

🛡️ Login Attempt Details

Time: {login_time}
IP Address: {ip_address}
Browser: {request.user_agent.string}

⚠️ Security Notice

If you did not initiate this login attempt, please contact the IT security team immediately and change your password. Never share your verification code with anyone.

Secure access powered by

Mutable Tech Enterprises

LMS

""" try: send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=user.email, subject=subject, message=message ) return True except Exception as e: current_app.logger.error(f"Failed to send OTP email: {str(e)}") return False def get_tutor_analysis(self): """Get comprehensive tutor analysis with the specified format""" with DatabaseContextManager() as ctx: # Get total tutors total_tutors = ctx.session.query(Tutor).count() # Get active tutors (not on leave and is_active) active_tutors = ctx.session.query(Tutor).filter( Tutor.is_active == True, Tutor.is_on_leave == False ).count() # Get tutors on leave on_leave_tutors = ctx.session.query(Tutor).filter( Tutor.is_on_leave == True ).count() # Get full-time vs part-time tutors full_time_tutors = ctx.session.query(Tutor).filter( Tutor.is_full_time == True ).count() part_time_tutors = ctx.session.query(Tutor).filter( Tutor.is_full_time == False ).count() # Calculate average hourly rate avg_hourly_rate_result = ctx.session.query( func.avg(Tutor.hourly_rate).label('avg_rate') ).filter(Tutor.hourly_rate.isnot(None)).first() average_hourly_rate = float(avg_hourly_rate_result.avg_rate or 0) # Calculate total teaching hours (from teaching logs) total_teaching_hours_result = ctx.session.query( func.sum(TutorTeachingLog.session_duration).label('total_hours') ).first() total_teaching_hours = float(total_teaching_hours_result.total_hours or 0) / 60 # Convert minutes to hours # Calculate average attendance rate (from attendance records) avg_attendance_result = ctx.session.query( func.avg(Attendance.status).label('avg_attendance') ).filter( Attendance.status == AttendanceStatus.present ).first() average_attendance_rate = float(avg_attendance_result.avg_attendance or 0) * 100 if avg_attendance_result.avg_attendance else 0 # Calculate average verification rate (from teaching logs) verified_sessions_count = ctx.session.query(TutorTeachingLog).filter( TutorTeachingLog.verification_method.isnot(None) ).count() total_sessions_count = ctx.session.query(TutorTeachingLog).count() average_verification_rate = (verified_sessions_count / total_sessions_count * 100) if total_sessions_count > 0 else 0 # Count upcoming sessions (sessions with status 'scheduled' and future start time) upcoming_sessions = ctx.session.query(TeachingSession).filter( TeachingSession.status == 'scheduled', TeachingSession.start_time > datetime.utcnow() ).count() # Count pending approvals (tutor availabilities not approved) pending_approvals = ctx.session.query(TutorAvailability).filter( TutorAvailability.is_approved == False ).count() # Performance metrics (placeholder - you might want to implement actual performance calculation) performance_metrics = { "excellent": ctx.session.query(Tutor).filter( Tutor.verification_success_rate >= 90 ).count() if hasattr(Tutor, 'verification_success_rate') else 0, "good": ctx.session.query(Tutor).filter( Tutor.verification_success_rate >= 75, Tutor.verification_success_rate < 90 ).count() if hasattr(Tutor, 'verification_success_rate') else 0, "average": ctx.session.query(Tutor).filter( Tutor.verification_success_rate >= 60, Tutor.verification_success_rate < 75 ).count() if hasattr(Tutor, 'verification_success_rate') else 0, "needsImprovement": ctx.session.query(Tutor).filter( Tutor.verification_success_rate < 60 ).count() if hasattr(Tutor, 'verification_success_rate') else 0 } return { "totalTutors": total_tutors, "activeTutors": active_tutors, "onLeaveTutors": on_leave_tutors, "fullTimeTutors": full_time_tutors, "partTimeTutors": part_time_tutors, "averageHourlyRate": round(average_hourly_rate, 2), "totalTeachingHours": round(total_teaching_hours, 2), "averageAttendanceRate": round(average_attendance_rate, 2), "averageVerificationRate": round(average_verification_rate, 2), "upcomingSessions": upcoming_sessions, "pendingApprovals": pending_approvals, "performanceMetrics": performance_metrics } def get_supervisor_analysis(self): """Get comprehensive supervisor analysis with the specified format""" with DatabaseContextManager() as ctx: # Get total supervisors total_supervisors = ctx.session.query(Supervisor).count() # Get active supervisors (not on leave and is_active) active_supervisors = ctx.session.query(Supervisor).filter( Supervisor.is_active == True, Supervisor.is_on_leave == False ).count() # Get supervisors on leave on_leave_supervisors = ctx.session.query(Supervisor).filter( Supervisor.is_on_leave == True ).count() # Get head of department count head_of_department_count = ctx.session.query(Supervisor).filter( Supervisor.is_head_of_department == True ).count() # Calculate average years of experience avg_years_experience_result = ctx.session.query( func.avg(Supervisor.years_of_experience).label('avg_experience') ).filter(Supervisor.years_of_experience.isnot(None)).first() average_years_experience = float(avg_years_experience_result.avg_experience or 0) # Calculate total managed tutors total_managed_tutors = ctx.session.query( func.count(Tutor.id) ).join(Supervisor, Tutor.supervisor_id == Supervisor.id).scalar() or 0 # Calculate total managed courses total_managed_courses = ctx.session.query( func.count(Speciality.id) ).scalar() or 0 # Calculate timetable approval rate approved_timetables_count = ctx.session.query(Timetable).filter( Timetable.approval_status == 'approved' ).count() total_timetables_count = ctx.session.query(Timetable).count() timetable_approval_rate = (approved_timetables_count / total_timetables_count * 100) if total_timetables_count > 0 else 0 # Calculate session verification rate verified_sessions_count = ctx.session.query(TeachingSession).filter( TeachingSession.is_verified == True ).count() total_sessions_count = ctx.session.query(TeachingSession).count() session_verification_rate = (verified_sessions_count / total_sessions_count * 100) if total_sessions_count > 0 else 0 # Count upcoming sessions for supervisors upcoming_sessions = ctx.session.query(TeachingSession).filter( TeachingSession.status == 'scheduled', TeachingSession.start_time > datetime.utcnow(), TeachingSession.supervisor_tutor_id.isnot(None) ).count() # Count pending approvals (tutor availabilities not approved) pending_approvals = ctx.session.query(TutorAvailability).filter( TutorAvailability.is_approved == False ).count() return { "totalSupervisors": total_supervisors, "activeSupervisors": active_supervisors, "onLeaveSupervisors": on_leave_supervisors, "headOfDepartmentCount": head_of_department_count, "averageYearsExperience": round(average_years_experience, 1), "totalManagedTutors": total_managed_tutors, "totalManagedCourses": total_managed_courses, "timetableApprovalRate": round(timetable_approval_rate, 2), "sessionVerificationRate": round(session_verification_rate, 2), "upcomingSessions": upcoming_sessions, "pendingApprovals": pending_approvals } def get_student_analysis(self): """Get comprehensive student analysis""" with DatabaseContextManager() as ctx: # Get total students total_students = ctx.session.query(Student).count() # Get active students active_students = ctx.session.query(Student).filter( Student.is_active == True ).count() # Get inactive students inactive_students = total_students - active_students # Get students on probation on_probation = ctx.session.query(Student).filter( Student.is_on_probation == True ).count() # Get students with scholarships with_scholarships = ctx.session.query(Student).filter( Student.has_scholarship == True ).count() # Calculate average GPA avg_gpa_result = ctx.session.query( func.avg(Student.cumulative_gpa).label('avg_gpa') ).filter(Student.cumulative_gpa.isnot(None)).first() average_gpa = float(avg_gpa_result.avg_gpa or 0) # Get new enrollments this month current_month_start = datetime.utcnow().replace(day=1, hour=0, minute=0, second=0, microsecond=0) new_enrollments_this_month = ctx.session.query(Student).filter( Student.enrollment_date >= current_month_start, Student.enrollment_date < datetime.utcnow() ).count() # Get graduating students this year current_year = datetime.utcnow().year graduating_this_year = ctx.session.query(Student).filter( func.extract('year', Student.graduation_date) == current_year ).count() return { "totalStudents": total_students, "activeStudents": active_students, "inactiveStudents": inactive_students, "onProbation": on_probation, "withScholarships": with_scholarships, "averageGPA": round(average_gpa, 2), "newEnrollmentsThisMonth": new_enrollments_this_month, "graduatingThisYear": graduating_this_year } def get_all_supervisors_data(self): """Get comprehensive data for all supervisors with the specified format""" with DatabaseContextManager() as ctx: supervisors = ctx.session.query(Supervisor).all() result = [] for supervisor in supervisors: # Get managed tutors count managed_tutors_count = ctx.session.query(Tutor).filter( Tutor.supervisor_id == supervisor.id ).count() # Get managed courses count managed_courses_count = ctx.session.query(Course).filter( Course.supervisor_id == supervisor.id ).count() # Get approval stats timetables_approved = ctx.session.query(Timetable).filter( Timetable.approved_by == supervisor.id, Timetable.approval_status == 'approved' ).count() timetables_pending = ctx.session.query(Timetable).filter( Timetable.created_by == supervisor.id, Timetable.approval_status == 'pending' ).count() sessions_verified = ctx.session.query(TeachingSession).filter( TeachingSession.verified_by == supervisor.id, TeachingSession.is_verified == True ).count() # Get upcoming sessions for this supervisor upcoming_sessions = ctx.session.query(TeachingSession).filter( TeachingSession.supervisor_tutor_id == supervisor.id, TeachingSession.status == 'scheduled', TeachingSession.start_time > datetime.utcnow() ).count() # Get departments departments = [] for dept in supervisor.departments: if dept.is_active: departments.append({ "department_name": dept.department_name, "is_primary": dept.is_primary }) # Get leave information if applicable leave_data = {} if supervisor.is_on_leave: leave_data = { "leave_start_date": supervisor.leave_start_date.isoformat() if supervisor.leave_start_date else None, "leave_end_date": supervisor.leave_end_date.isoformat() if supervisor.leave_end_date else None, "leave_reason": supervisor.leave_reason } supervisor_data = { "id": supervisor.id, "first_name": supervisor.first_name, "last_name": supervisor.last_name, "email": supervisor.email, "phone": supervisor.phone, "office_location": supervisor.office_location, "office_hours": supervisor.office_hours, "is_head_of_department": supervisor.is_head_of_department, "years_of_experience": supervisor.years_of_experience, "max_tutors": supervisor.max_tutors, "qualification": supervisor.qualification, "bio": supervisor.bio, "specialization": supervisor.specialization, "is_on_leave": supervisor.is_on_leave, "profile_picture": supervisor.profile_picture or "", "departments": departments, "managed_tutors_count": managed_tutors_count, "managed_courses_count": managed_courses_count, "approval_stats": { "timetables_approved": timetables_approved, "timetables_pending": timetables_pending, "sessions_verified": sessions_verified }, "upcoming_sessions": upcoming_sessions } # Add leave data if supervisor is on leave if supervisor.is_on_leave: supervisor_data.update(leave_data) result.append(supervisor_data) return result def get_dashboard_data(self): """Get comprehensive dashboard data for School Management System""" with DatabaseContextManager() as ctx: # Current date for filtering current_date = datetime.utcnow().date() current_week_start = current_date - timedelta(days=current_date.weekday()) current_month_start = current_date.replace(day=1) # 1. OVERVIEW STATISTICS total_students = ctx.session.query(Student).count() total_tutors = ctx.session.query(Tutor).count() total_supervisors = ctx.session.query(Supervisor).count() total_specialities = ctx.session.query(Speciality).filter(Speciality.is_active == True).count() total_courses = ctx.session.query(Course).filter(Course.is_active == True).count() # Active sessions today (DailyTeachingSession count for current date) from src.models.models import DailyTeachingSession active_sessions_today = ctx.session.query(DailyTeachingSession).filter( func.date(DailyTeachingSession.session_date) == current_date ).count() # 2. ATTENDANCE ANALYTICS # Today's attendance today_attendance = ctx.session.query(Attendance).filter( func.date(Attendance.timestamp) == current_date ).all() present_today = len([a for a in today_attendance if a.status == AttendanceStatus.present]) absent_today = len([a for a in today_attendance if a.status == AttendanceStatus.absent]) late_today = len([a for a in today_attendance if a.status == AttendanceStatus.late]) # Best performing course today (highest attendance rate) best_performing_course = None course_attendance_today = ctx.session.query( Course.id, Course.code, Course.title, func.count(Attendance.id).label('total_attendance'), func.sum(case((Attendance.status == AttendanceStatus.present, 1), else_=0)).label('present_count') ).join( TeachingSession, TeachingSession.course_id == Course.id ).join( Attendance, Attendance.session_id == TeachingSession.id ).filter( func.date(Attendance.timestamp) == current_date ).group_by( Course.id, Course.code, Course.title ).all() if course_attendance_today: # Calculate attendance rates and find best course_rates = [] for course_data in course_attendance_today: if course_data.total_attendance > 0: rate = (course_data.present_count / course_data.total_attendance) * 100 course_rates.append({ 'id': course_data.id, 'code': course_data.code, 'title': course_data.title, 'attendance_rate': round(rate, 1), 'present': course_data.present_count, 'total': course_data.total_attendance }) if course_rates: best_performing_course = max(course_rates, key=lambda x: x['attendance_rate']) # Weekly attendance trend (last 7 days) attendance_trend = [] for i in range(7): date = current_date - timedelta(days=i) day_attendance = ctx.session.query(Attendance).filter( func.date(Attendance.timestamp) == date ).all() present_count = len([a for a in day_attendance if a.status == AttendanceStatus.present]) attendance_trend.append({ "date": date.isoformat(), "present": present_count, "total": len(day_attendance) if day_attendance else 0 }) attendance_trend.reverse() # 3. ACADEMIC PERFORMANCE # Course completion rates course_progress = ctx.session.query( Course.code, Course.title, func.avg(CourseProgress.completion_percentage).label('avg_completion') ).select_from(Course)\ .join(enrollment_courses, enrollment_courses.c.course_id == Course.id)\ .join(Enrollment, enrollment_courses.c.enrollment_id == Enrollment.id)\ .join(CourseProgress, CourseProgress.enrollment_id == Enrollment.id)\ .group_by(Course.id, Course.code, Course.title).all() # Assignment submission rates assignment_stats = ctx.session.query( func.count(AssignmentSubmission.id).label('total_submissions'), func.avg(AssignmentSubmission.grade).label('avg_grade'), func.count(case((AssignmentSubmission.status == SubmissionStatus.late, 1))).label('late_submissions') ).first() # 4. TUTOR PERFORMANCE tutor_analytics = self.get_tutor_analysis() supervisor_analytics = self.get_supervisor_analysis() # 5. UPCOMING EVENTS AND SESSIONS # Next 7 days sessions upcoming_sessions = ctx.session.query(TeachingSession).filter( TeachingSession.status == 'scheduled', TeachingSession.start_time >= datetime.utcnow(), TeachingSession.start_time <= datetime.utcnow() + timedelta(days=7) ).order_by(TeachingSession.start_time).limit(10).all() # Upcoming assignments due upcoming_assignments = ctx.session.query(Assignment).filter( Assignment.due_date >= current_date, Assignment.due_date <= current_date + timedelta(days=14), Assignment.is_published == True ).order_by(Assignment.due_date).limit(10).all() # 6. FINANCIAL OVERVIEW (if applicable) # Tutor payments (assuming hourly_rate * teaching_hours) tutor_payments = ctx.session.query( Tutor.id, Tutor.first_name, Tutor.last_name, Tutor.hourly_rate, func.sum(TutorTeachingLog.session_duration).label('total_minutes') ).join(TutorTeachingLog, TutorTeachingLog.tutor_id == Tutor.id)\ .filter(TutorTeachingLog.checkin_time >= current_month_start)\ .group_by(Tutor.id).all() total_payments = sum((payment.hourly_rate or 0) * (payment.total_minutes or 0) / 60 for payment in tutor_payments) # 7. DEPARTMENT-WISE STATISTICS departments_stats = [] departments = ctx.session.query(SupervisorDepartment.department_name).distinct().all() for dept in departments: dept_name = dept.department_name dept_tutors = ctx.session.query(Tutor).join(TutorDepartment).filter( TutorDepartment.department_name == dept_name, TutorDepartment.is_active == True ).count() dept_courses = ctx.session.query(Course).filter( Course.department == dept_name, Course.is_active == True ).count() dept_students = ctx.session.query(Student).join( Enrollment, Enrollment.student_id == Student.id ).join( enrollment_courses, enrollment_courses.c.enrollment_id == Enrollment.id ).join( Course, enrollment_courses.c.course_id == Course.id ).filter( Course.department == dept_name ).distinct().count() departments_stats.append({ "name": dept_name, "tutors": dept_tutors, "courses": dept_courses, "students": dept_students }) # 8. RECENT ACTIVITIES # Recent attendance records recent_attendance = ctx.session.query(Attendance).order_by( Attendance.timestamp.desc() ).limit(20).all() # Recent assignment submissions recent_submissions = ctx.session.query(AssignmentSubmission).order_by( AssignmentSubmission.submitted_at.desc() ).limit(15).all() # 9. SYSTEM HEALTH # Pending approvals pending_availabilities = ctx.session.query(TutorAvailability).filter( TutorAvailability.is_approved == False ).count() pending_timetables = ctx.session.query(Timetable).filter( Timetable.approval_status == 'pending' ).count() # Unverified sessions unverified_sessions = ctx.session.query(TeachingSession).filter( TeachingSession.is_verified == False, TeachingSession.status == 'completed' ).count() # 10. GRAPHING DATA # Monthly enrollment trend (last 6 months) enrollment_trend = [] for i in range(6): month = (current_date.replace(day=1) - timedelta(days=30*i)) month_enrollments = ctx.session.query(Enrollment).filter( func.extract('year', Enrollment.enrollment_date) == month.year, func.extract('month', Enrollment.enrollment_date) == month.month ).count() enrollment_trend.append({ "month": month.strftime("%Y-%m"), "enrollments": month_enrollments }) enrollment_trend.reverse() # Speciality enrollment distribution (all specialities) speciality_enrollment = ctx.session.query( Speciality.id, Speciality.name, Speciality.department, func.count(func.distinct(Student.id)).label('student_count') ).outerjoin( Student, Student.speciality_id == Speciality.id ).filter( Speciality.is_active == True ).group_by( Speciality.id, Speciality.name, Speciality.department ).order_by( func.count(func.distinct(Student.id)).desc() ).all() # Tutor performance distribution tutor_performance = ctx.session.query( case( (Tutor.verification_success_rate >= 90, "excellent"), (Tutor.verification_success_rate >= 75, "good"), (Tutor.verification_success_rate >= 60, "average"), else_="needs_improvement" ).label('performance_category'), func.count(Tutor.id).label('count') ).filter(Tutor.verification_success_rate.isnot(None))\ .group_by('performance_category').all() # Format the complete response dashboard_data = { "overview": { "totalStudents": total_students, "totalTutors": total_tutors, "totalSupervisors": total_supervisors, "totalSpecialities": total_specialities, "totalCourses": total_courses, "activeSessionsToday": active_sessions_today, "bestPerformingCourse": best_performing_course }, "attendance": { "today": { "present": present_today, "absent": absent_today, "late": late_today, "total": len(today_attendance) }, "weeklyTrend": attendance_trend, "overallRate": round((present_today / len(today_attendance) * 100) if today_attendance else 0, 1) }, "academic": { "courseCompletionRates": [ {"course": f"{course.code} - {course.title}", "completionRate": round(course.avg_completion or 0, 1)} for course in course_progress ], "assignmentStats": { "totalSubmissions": assignment_stats.total_submissions or 0, "averageGrade": round(assignment_stats.avg_grade or 0, 1), "lateSubmissionRate": round((assignment_stats.late_submissions or 0) / (assignment_stats.total_submissions or 1) * 100, 1) } }, "staffPerformance": { "tutors": tutor_analytics, "supervisors": supervisor_analytics, "performanceDistribution": { category: count for category, count in tutor_performance } }, "upcomingEvents": { "sessions": [ { "id": session.id, "title": session.title or f"{session.course.code} Session", "course": session.course.title, "startTime": session.start_time.isoformat(), "tutor": f"{session.tutor.first_name} {session.tutor.last_name}", "room": session.room } for session in upcoming_sessions ], "assignments": [ { "id": assignment.id, "title": assignment.title, "course": assignment.course.title, "dueDate": assignment.due_date.isoformat() if assignment.due_date else None, "points": assignment.total_points } for assignment in upcoming_assignments ] }, "departmentStats": departments_stats, "systemHealth": { "pendingApprovals": { "tutorAvailabilities": pending_availabilities, "timetables": pending_timetables }, "unverifiedSessions": unverified_sessions, "systemUptime": "99.9%", # This would come from system monitoring "storageUsage": "75%" # This would come from system monitoring }, "graphingData": { "enrollmentTrend": enrollment_trend, "specialityEnrollment": [ { "id": spec.id, "name": spec.name, "department": spec.department, "students": spec.student_count } for spec in speciality_enrollment ], "attendanceByDay": attendance_trend, "departmentDistribution": departments_stats }, "recentActivity": { "attendanceRecords": [ { "student": f"{att.student.first_name} {att.student.last_name}", "course": att.session.course.title, "status": att.status.value, "time": att.timestamp.isoformat() } for att in recent_attendance ], "submissions": [ { "student": f"{sub.student.first_name} {sub.student.last_name}", "assignment": sub.assignment.title, "course": sub.assignment.course.title, "grade": sub.grade, "submittedAt": sub.submitted_at.isoformat() if sub.submitted_at else None } for sub in recent_submissions ] }, "timestamp": datetime.utcnow().isoformat(), "academicYear": current_date.year, "currentTerm": "Spring 2024" # This would be dynamically determined } return dashboard_data def get_departments_data(self): """Get comprehensive department data in the specified format""" with DatabaseContextManager() as ctx: # Get all departments from SupervisorDepartment (unique department names) department_names = ctx.session.query( SupervisorDepartment.department_name ).distinct().all() result = [] for dept_name in department_names: dept_name = dept_name.department_name # Get department statistics with explicit joins # Tutor counts tutor_count = ctx.session.query(Tutor).join( TutorDepartment, Tutor.id == TutorDepartment.tutor_id ).filter( TutorDepartment.department_name == dept_name, TutorDepartment.is_active == True ).count() # Supervisor counts - use the exact foreign key column supervisor_count = ctx.session.query(Supervisor).join( SupervisorDepartment, Supervisor.id == SupervisorDepartment.supervisor_id ).filter( SupervisorDepartment.department_name == dept_name, SupervisorDepartment.is_active == True ).count() # Student counts (approximate - students in courses of this department) # Get students through their speciality and course relationship student_count = ctx.session.query(Student).join( Speciality, Student.speciality_id == Speciality.id ).join( Course, Course.speciality_id == Speciality.id ).filter( Course.department == dept_name ).distinct().count() # Course counts course_count = ctx.session.query(Course).filter( Course.department == dept_name, Course.is_active == True ).count() # Get head of department - use the exact foreign key column hod = ctx.session.query(Supervisor).join( SupervisorDepartment, Supervisor.id == SupervisorDepartment.supervisor_id ).filter( SupervisorDepartment.department_name == dept_name, SupervisorDepartment.is_primary == True, Supervisor.is_head_of_department == True, SupervisorDepartment.is_active == True ).first() if hod: user = ctx.session.query(User).filter( User.id == hod.id ).first() # Calculate average grade for the department avg_grade_result = ctx.session.query( func.avg(AssignmentSubmission.grade).label('avg_grade') ).join(Assignment, AssignmentSubmission.assignment_id == Assignment.id ).join(Course, Assignment.course_id == Course.id ).filter( Course.department == dept_name, AssignmentSubmission.grade.isnot(None) ).first() average_grade = float(avg_grade_result.avg_grade or 0) if avg_grade_result else 0 # Calculate enrollment statistics by summing up all courses in the department dept_courses = ctx.session.query(Course).filter( Course.department == dept_name, Course.is_active == True ).all() total_enrolled = 0 total_capacity = 0 for course in dept_courses: # Count enrolled students through enrollments enrolled_count = ctx.session.query(Enrollment).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).filter( enrollment_courses.c.course_id == course.id ).distinct(Enrollment.id).count() total_enrolled += enrolled_count total_capacity += course.max_students if course.max_students else 0 # Calculate enrollment percentage enrollment_percentage = round((total_enrolled / total_capacity * 100), 1) if total_capacity > 0 else 0 # Calculate graduation rate - students who completed courses vs total students completed_students = ctx.session.query(Student).join( Enrollment, Student.id == Enrollment.student_id ).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).join( Course, Course.id == enrollment_courses.c.course_id ).filter( Course.department == dept_name, Enrollment.status == 'completed' ).distinct(Student.id).count() total_students_in_dept = student_count if student_count > 0 else 1 graduation_rate = round((completed_students / total_students_in_dept * 100), 1) if total_students_in_dept > 0 else 0 # Calculate student growth (compare current vs previous academic session) # Get previous academic session previous_session = ctx.session.query(AcademicSession).filter( AcademicSession.is_active == False ).order_by(AcademicSession.end_date.desc()).first() if previous_session: # Count students enrolled in previous session for this department prev_enrolled = ctx.session.query(Enrollment).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).join( Course, Course.id == enrollment_courses.c.course_id ).filter( Course.department == dept_name, Enrollment.academic_session_id == previous_session.id ).distinct(Enrollment.id).count() if prev_enrolled > 0: student_growth = round(((total_enrolled - prev_enrolled) / prev_enrolled * 100), 1) else: student_growth = 0.0 else: student_growth = 0.0 # Get active courses count active_courses_count = ctx.session.query(Course).filter( Course.department == dept_name, Course.is_active == True ).count() # Get completed courses count completed_courses_count = ctx.session.query(Course).filter( Course.department == dept_name, Course.is_active == False ).count() # Get total all courses (active + inactive) total_courses_count = ctx.session.query(Course).filter( Course.department == dept_name ).count() # Count specialities in this department speciality_count = ctx.session.query(Speciality).filter( Speciality.department == dept_name, Speciality.is_active == True ).count() # Calculate average rating from student feedback/grades if average_grade > 0: rating = round((average_grade / 100) * 5, 1) # Convert grade to 5-point scale else: rating = 0.0 # Get location from supervisor department dept_info = ctx.session.query(SupervisorDepartment).filter( SupervisorDepartment.department_name == dept_name ).first() location = dept_info.location if (dept_info and hasattr(dept_info, 'location') and dept_info.location) else f"{dept_name} Building" # Calculate workload hours (sum of all course hours taught by tutors) total_workload = ctx.session.query( func.sum(Course.total_hours).label('total_hours') ).filter( Course.department == dept_name, Course.is_active == True ).first() average_workload = int(total_workload.total_hours or 0) if total_workload else 0 academic_session = ctx.session.query(AcademicSession).filter( AcademicSession.is_active == True ).first() # Generate department code (first 3-4 letters uppercase) dept_code = ''.join([word[0].upper() for word in dept_name.split()])[:4] # Create department data with real values department_data = { "id": f"{dept_code.lower()}-{str(uuid.uuid4())[:8]}", "name": dept_name, "department_code": dept_code, "description": f"The {dept_name} department focuses on delivering quality education and research.", "established_year": 1980, # Default, can be added to database if needed "location": location, "rating": rating, "student_count": total_enrolled, "student_growth": student_growth, "total_faculty": tutor_count + supervisor_count, "professor_count": supervisor_count, "tutor_count": tutor_count, "supervisor_count": supervisor_count, "contact_phone": user.phone if (hod and user) else '', "contact_email": user.email if (hod and user) else '', "total_courses": total_courses_count, "active_courses": active_courses_count, "completed_courses": completed_courses_count, "program_count": speciality_count, "undergraduate_programs": speciality_count, "graduate_programs": 0, # Can be enhanced with speciality level data "enrollment_progress": enrollment_percentage, "max_capacity": total_capacity if total_capacity > 0 else 0, "average_grade": round(average_grade, 1), "grade_letter": self._get_grade_letter(average_grade) if average_grade > 0 else "N/A", "average_workload": average_workload, "graduation_rate": graduation_rate, "graduates_last_year": completed_students, "head_of_department": f"{hod.first_name} {hod.last_name}" if hod else "Not Assigned", "academic_calendar": academic_session.name if academic_session else "Not Set", "student_faculty_ratio": round((total_enrolled) / max(1, (tutor_count + supervisor_count)), 1), "alumni_count": student_count, "popular_programs": self._generate_popular_programs(ctx, dept_name) } result.append(department_data) return custom_response( data=result, status_code=200, success=True ) def _generate_popular_programs(self, session_ctx, department_name): """Generate popular programs by querying specialities from multiple sources""" # Method 1: Direct department matching in Speciality table specialities = session_ctx.session.query(Speciality).filter( Speciality.department == department_name, Speciality.is_active == True ).order_by(Speciality.name).limit(8).all() if specialities: return [speciality.name for speciality in specialities] # Method 2: Partial department name matching specialities_partial = session_ctx.session.query(Speciality).filter( Speciality.department.ilike(f"%{department_name}%"), Speciality.is_active == True ).order_by(Speciality.name).limit(8).all() if specialities_partial: return [speciality.name for speciality in specialities_partial] # Method 3: Get specialities from courses in this department course_specialities = session_ctx.session.query(Speciality).join( Course, Course.speciality_id == Speciality.id ).filter( Course.department == department_name, Course.is_active == True, Speciality.is_active == True ).distinct().order_by(Speciality.name).limit(8).all() if course_specialities: return [speciality.name for speciality in course_specialities] # Method 4: Get specialities from shared course departments shared_specialities = session_ctx.session.query(Speciality).join( course_speciality_association, course_speciality_association.c.speciality_id == Speciality.id ).join( CourseDepartment, CourseDepartment.course_id == course_speciality_association.c.course_id ).filter( CourseDepartment.department_name == department_name, CourseDepartment.is_active == True, Speciality.is_active == True ).distinct().order_by(Speciality.name).limit(8).all() if shared_specialities: return [speciality.name for speciality in shared_specialities] def fetchSpecialitiesStats(self): with DatabaseContextManager() as ctx: # Count total specialities total_specialities = ctx.session.query(Speciality).count() # Count active specialities active_specialities = ctx.session.query(Speciality).filter( Speciality.is_active == True ).count() # Count total courses total_courses = ctx.session.query(Course).count() # Count distinct departments total_departments = ctx.session.query( SupervisorDepartment.department_name ).distinct().count() data = { "totalSpecialities": total_specialities, "activeSpecialities": active_specialities, "totalCourses": total_courses, "totalDepartments": total_departments } return custom_response( data=data, status_code=200, success=True ) def fetchUnits(self, page=1, per_page=10, search="", status_filter="all"): """ Get list of courses with role-specific filtering. """ with DatabaseContextManager() as ctx: query = ctx.session.query(Course).filter(Course.is_active == True) # Apply search filter if search: search_term = f"%{search}%" query = query.filter( (Course.title.ilike(search_term)) | (Course.code.ilike(search_term)) | (Course.department.ilike(search_term)) ) # Apply status filter if status_filter != "all": if status_filter == "active": query = query.filter(Course.is_active == True, Course.is_archived == False) elif status_filter == "archived": query = query.filter(Course.is_archived == True) elif status_filter == "inactive": query = query.filter(Course.is_active == False, Course.is_archived == False) # Pagination total = query.count() courses = query.order_by(Course.code).offset((page - 1) * per_page).limit(per_page).all() # Calculate department-wide summary statistics (for supervisors) department_summary = { 'total_courses': total, 'active_courses': query.filter(Course.is_active == True, Course.is_archived == False).count(), 'total_students': 0, 'average_rating': 0 } return custom_response( success=True, data={ "units": [{ 'id': c.id, 'code': c.code, 'title': c.title, 'department': c.department, 'semester': c.semester, 'enrolled_students': len(c.enrollments), 'tutors': [f"{t.first_name} {t.last_name}" for t in c.tutors], "description" : c.description, "credits" : c.credits, "is_active" : c.is_active, "max_students" : c.max_students, "created_at" : c.created_at, "learning_outcomes" : c.learning_outcomes, "required_materials" : c.required_materials, "assessment_method" : c.assessment_method, "syllabus" : c.syllabus, "total_hours" : c.total_hours, "is_archived" : c.is_archived, "archive_date" : c.archive_date, "course_level" : c.course_level, "language" : c.language, "certification_available" : c.certification_available, "has_practical" : c.has_practical, "is_shared_course": c.is_shared_course, "shared_course_type": c.shared_course_type, "sharing_level": c.sharing_level, "shared_specialities": [{ 'id': s.id, 'name': s.name, 'department': s.department, 'description': s.description, 'code': s.code, 'abbreviation': s.abbreviation } for s in c.shared_specialities], "shared_departments": [{ 'department_name': cd.department_name, 'is_primary_department': cd.is_primary_department, 'assigned_date': str(cd.assigned_date), 'assigned_by': cd.assigned_by, 'notes': cd.notes, 'is_active': cd.is_active } for cd in c.shared_departments if cd.is_active], "sharing_capable_specialities": [{ 'id': s.id, 'name': s.name, 'department': s.department, 'description': s.description, 'code': s.code, 'abbreviation': s.abbreviation } for s in c.shared_specialities], # Note: Currently using shared_specialities, but this should be a separate relationship for sharing-capable specialities "academic_session_id": c.academic_session_id, "academic_session": { 'id': c.academic_session.id, 'name': c.academic_session.name, 'year': c.academic_session.year, 'start_date': c.academic_session.start_date.isoformat() if c.academic_session.start_date else None, 'end_date': c.academic_session.end_date.isoformat() if c.academic_session.end_date else None, 'status': c.academic_session.status } if c.academic_session else None, # Computed status field for frontend compatibility "status": "active" if c.is_active and not c.is_archived else "archived" if c.is_archived else "inactive", # Enhanced supervisor information 'supervisor': self._get_comprehensive_supervisor_info(ctx, c.supervisor_id) if c.supervisor_id else None, 'supervisor_status': 'assigned' if c.supervisor_id else 'unassigned', 'supervisor_availability': self._get_supervisor_availability_status(ctx, c.supervisor_id) if c.supervisor_id else None } for c in courses], "total": total, "page": page, "per_page": per_page, "department_summary": department_summary }, status_code=200 ) def _get_comprehensive_supervisor_info(self, ctx, supervisor_id): """ Get comprehensive supervisor information for display purposes. This provides rich data that can be used to create appealing UI components. Args: ctx: Database context supervisor_id: ID of the supervisor Returns: Dictionary with comprehensive supervisor information """ if not supervisor_id: return None try: supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first() if not supervisor: return None # Get supervisor's current course count current_courses = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id, Course.is_active == True ).count() # Get supervisor's total managed courses (including inactive) total_managed_courses = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id ).count() # Get supervisor's department statistics department_courses = ctx.session.query(Course).filter( Course.department == supervisor.get_primary_department(), Course.is_active == True ).count() # Get supervisor's recent activity (last 30 days) thirty_days_ago = datetime.utcnow() - timedelta(days=30) recent_courses = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id, Course.created_at >= thirty_days_ago ).count() # Calculate supervisor's workload percentage max_courses = getattr(supervisor, 'max_courses', 10) # Default to 10 if not set workload_percentage = round((current_courses / max_courses) * 100, 1) if max_courses > 0 else 0 # Determine availability status if workload_percentage >= 90: availability_status = "High Workload" availability_color = "red" elif workload_percentage >= 70: availability_status = "Moderate Workload" availability_color = "orange" elif workload_percentage >= 50: availability_status = "Available" availability_color = "yellow" else: availability_status = "Highly Available" availability_color = "green" # Get supervisor's expertise areas expertise_areas = [] if supervisor.specialization: expertise_areas.append(supervisor.specialization) if supervisor.qualification: expertise_areas.append(supervisor.qualification) # Get supervisor's contact information with availability hints departments_list = self._get_supervisor_departments_safe(supervisor) contact_info = { 'email': supervisor.email, 'office_location': supervisor.office_location, 'staff_id': supervisor.staff_id, 'departments': departments_list } # Add office hours if available if hasattr(supervisor, 'office_hours') and supervisor.office_hours: contact_info['office_hours'] = supervisor.office_hours # Create comprehensive supervisor info supervisor_info = { # Basic Information 'id': supervisor.id, 'name': f"{supervisor.first_name} {supervisor.last_name}", 'first_name': supervisor.first_name, 'last_name': supervisor.last_name, 'full_name': f"{supervisor.first_name} {supervisor.last_name}", 'display_name': f"{supervisor.first_name} {supervisor.last_name} ({supervisor.staff_id})", # Professional Details 'qualification': supervisor.qualification, 'specialization': supervisor.specialization, 'years_of_experience': supervisor.years_of_experience, 'is_head_of_department': supervisor.is_head_of_department, # Contact Information 'contact': contact_info, # Workload & Availability 'workload': { 'current_courses': current_courses, 'total_managed_courses': total_managed_courses, 'max_courses': max_courses, 'workload_percentage': workload_percentage, 'availability_status': availability_status, 'availability_color': availability_color, 'recent_activity': recent_courses }, # Department Context 'department_context': { 'departments': departments_list, 'department_courses': department_courses, 'department_leadership': supervisor.is_head_of_department }, # Professional Summary 'professional_summary': { 'expertise_areas': expertise_areas, 'leadership_role': "Department Head" if supervisor.is_head_of_department else "Faculty Member", 'specialization_focus': supervisor.specialization or "General Education" }, # UI Display Properties 'display': { 'avatar_initials': f"{supervisor.first_name[0]}{supervisor.last_name[0]}".upper(), 'status_badge': availability_status, 'status_color': availability_color, 'workload_indicator': workload_percentage, 'is_available_for_new_courses': workload_percentage < 80, 'recommended_course_limit': max(1, max_courses - current_courses) }, # Assignment Information 'assignment_details': { 'assigned_at': datetime.utcnow().isoformat(), # This will be updated when we track assignment dates 'assignment_type': 'primary_supervisor', 'responsibilities': [ 'Course oversight and quality assurance', 'Student progress monitoring', 'Tutor coordination and support', 'Academic standards maintenance' ] } } return supervisor_info except Exception as e: current_app.logger.error(f"Error getting comprehensive supervisor info: {str(e)}", exc_info=True) # Return basic info if comprehensive info fails supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first() if supervisor: return { 'id': supervisor.id, 'name': f"{supervisor.first_name} {supervisor.last_name}", 'email': supervisor.email, 'staff_id': supervisor.staff_id, 'departments': supervisor.get_all_departments(), 'error': 'Limited information available' } return None def _get_supervisor_availability_status(self, ctx, supervisor_id): """ Get a quick availability status for supervisor display in course lists. This provides a lightweight status indicator for UI components. Args: ctx: Database context supervisor_id: ID of the supervisor Returns: Dictionary with availability status information """ if not supervisor_id: return None try: supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first() if not supervisor: return None # Get current course count current_courses = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id, Course.is_active == True ).count() # Get max courses (default to 10 if not set) max_courses = getattr(supervisor, 'max_courses', 10) workload_percentage = round((current_courses / max_courses) * 100, 1) if max_courses > 0 else 0 # Determine status if workload_percentage >= 90: status = "High Workload" color = "red" icon = "⚠️" elif workload_percentage >= 70: status = "Moderate" color = "orange" icon = "🟡" elif workload_percentage >= 50: status = "Available" color = "yellow" icon = "🟢" else: status = "Highly Available" color = "green" icon = "✅" return { 'status': status, 'color': color, 'icon': icon, 'workload_percentage': workload_percentage, 'current_courses': current_courses, 'max_courses': max_courses, 'can_take_more': workload_percentage < 80 } except Exception as e: current_app.logger.error(f"Error getting supervisor availability status: {str(e)}", exc_info=True) return None def _get_supervisor_departments_safe(self, supervisor): """ Safely extract departments from supervisor object, handling different data types. Returns a list of department names as strings. """ try: departments_list = [] if hasattr(supervisor, 'departments') and supervisor.departments: if isinstance(supervisor.departments, list): for dept in supervisor.departments: if hasattr(dept, 'name'): departments_list.append(dept.name) elif hasattr(dept, 'department_name'): departments_list.append(dept.department_name) elif isinstance(dept, str): departments_list.append(dept) else: # Try to convert to string departments_list.append(str(dept)) elif hasattr(supervisor.departments, 'name'): departments_list.append(supervisor.departments.name) elif hasattr(supervisor.departments, 'department_name'): departments_list.append(supervisor.departments.department_name) elif isinstance(supervisor.departments, str): departments_list.append(supervisor.departments) else: departments_list.append(str(supervisor.departments)) # If no departments found, try to get from get_primary_department method if not departments_list and hasattr(supervisor, 'get_primary_department'): primary_dept = supervisor.get_primary_department() if primary_dept: departments_list.append(primary_dept) return departments_list except Exception as e: current_app.logger.warning(f"Error extracting departments from supervisor {supervisor.id}: {str(e)}") return [] def fetchModules(self, page=1, per_page=20, search="", course_filter="all", status_filter="all"): """ Get all course modules with pagination, search, and filtering. """ with DatabaseContextManager() as ctx: # Base query with joins query = ctx.session.query(CourseModule).join(Course).filter(Course.is_active == True) # Apply search filter if search: search_term = f"%{search}%" query = query.filter( (CourseModule.title.ilike(search_term)) | (CourseModule.description.ilike(search_term)) | (Course.title.ilike(search_term)) | (Course.code.ilike(search_term)) | (Course.department.ilike(search_term)) ) # Apply course filter if course_filter != "all": query = query.filter(Course.id == course_filter) # Apply status filter if status_filter != "all": if status_filter == "published": query = query.filter(CourseModule.is_published == True) elif status_filter == "unpublished": query = query.filter(CourseModule.is_published == False) elif status_filter == "assessment": query = query.filter(CourseModule.is_assessment_module == True) # Get total count total_modules = query.count() # Apply pagination offset = (page - 1) * per_page modules = query.order_by(CourseModule.created_at.desc()).offset(offset).limit(per_page).all() # Format modules data modules_data = [] for module in modules: # Get resource count resource_count = ctx.session.query(CourseResource).filter( CourseResource.module_id == module.id ).count() # Get assignment count assignment_count = ctx.session.query(Assignment).filter( Assignment.module_id == module.id ).count() modules_data.append({ 'id': module.id, 'title': module.title, 'description': module.description, 'sequence': module.sequence, 'start_date': module.start_date.isoformat() if module.start_date else None, 'end_date': module.end_date.isoformat() if module.end_date else None, 'is_published': module.is_published, 'objectives': module.objectives, 'estimated_study_time': module.estimated_study_time, 'is_assessment_module': module.is_assessment_module, 'created_at': module.created_at.isoformat(), 'resource_count': resource_count, 'assignment_count': assignment_count, 'course': { 'id': module.course.id, 'title': module.course.title, 'code': module.course.code, 'department': module.course.department, 'supervisor': { 'id': module.course.supervisor.id if module.course.supervisor else None, 'name': f"{module.course.supervisor.first_name} {module.course.supervisor.last_name}" if module.course.supervisor else None, 'email': module.course.supervisor.email if module.course.supervisor else None } if module.course.supervisor else None } }) # Calculate pagination info total_pages = (total_modules + per_page - 1) // per_page has_next = page < total_pages has_prev = page > 1 return custom_response( data={ 'modules': modules_data, 'pagination': { 'page': page, 'per_page': per_page, 'total_modules': total_modules, 'total_pages': total_pages, 'has_next': has_next, 'has_prev': has_prev } }, status_code=200, success=True ) def fetchEnrollments(self): return def fetchLearningResources(self, page=1, per_page=20, search="", type_filter="all", module_filter="all", status_filter="all"): """ Get all learning resources with pagination, search, and filtering. """ with DatabaseContextManager() as ctx: # Base query with joins query = ctx.session.query(CourseResource).join(CourseModule).join(Course).filter(Course.is_active == True) # Apply search filter if search: search_term = f"%{search}%" query = query.filter( (CourseResource.title.ilike(search_term)) | (CourseResource.description.ilike(search_term)) | (CourseModule.title.ilike(search_term)) | (Course.title.ilike(search_term)) | (Course.code.ilike(search_term)) ) # Apply type filter if type_filter != "all": query = query.filter(CourseResource.resource_type == type_filter) # Apply module filter if module_filter != "all": query = query.filter(CourseResource.module_id == module_filter) # Apply status filter if status_filter != "all": if status_filter == "published": query = query.filter(CourseResource.is_published == True) elif status_filter == "unpublished": query = query.filter(CourseResource.is_published == False) # Get total count total_resources = query.count() # Apply pagination offset = (page - 1) * per_page resources = query.order_by(CourseResource.created_at.desc()).offset(offset).limit(per_page).all() # Format resources data resources_data = [] for resource in resources: # Get creator info creator = ctx.session.query(Tutor).filter(Tutor.id == resource.created_by).first() resources_data.append({ 'id': resource.id, 'title': resource.title, 'description': resource.description, 'resource_type': resource.resource_type.value, 'file_url': resource.file_path, 'file_size': resource.file_size, 'file_extension': resource.file_path.split('.')[-1] if resource.file_path else None, 'external_url': resource.url, 'is_published': resource.is_published, 'created_at': resource.created_at.isoformat(), 'updated_at': resource.created_at.isoformat(), # Using created_at as updated_at for now 'download_count': resource.downloads, 'view_count': resource.views, 'tags': [], # TODO: Add tags field to model if needed 'module': { 'id': resource.module.id if resource.module else None, 'title': resource.module.title if resource.module else None, 'code': resource.module.course.code if resource.module and resource.module.course else None, 'course': { 'id': resource.course.id, 'title': resource.course.title, 'code': resource.course.code, 'department': resource.course.department } }, 'created_by': { 'id': creator.id if creator else resource.created_by, 'first_name': creator.first_name if creator else 'Unknown', 'last_name': creator.last_name if creator else 'User', 'email': creator.email if creator else 'unknown@example.com' } }) # Calculate pagination info total_pages = (total_resources + per_page - 1) // per_page has_next = page < total_pages has_prev = page > 1 return custom_response( data={ 'resources': resources_data, 'pagination': { 'page': page, 'per_page': per_page, 'total_resources': total_resources, 'total_pages': total_pages, 'has_next': has_next, 'has_prev': has_prev } }, status_code=200, success=True ) def createResource(self, resource_data): """ Create a new learning resource. """ with DatabaseContextManager() as ctx: try: # Create new resource new_resource = CourseResource( id=str(uuid.uuid4()), course_id=resource_data.get('course_id'), module_id=resource_data.get('module_id'), title=resource_data.get('title'), description=resource_data.get('description'), resource_type=resource_data.get('resource_type'), url=resource_data.get('external_url'), file_path=resource_data.get('file_url'), file_size=resource_data.get('file_size'), is_published=resource_data.get('is_published', False), created_by=resource_data.get('created_by'), views=0, downloads=0 ) ctx.session.add(new_resource) ctx.session.commit() return custom_response( data={'message': 'Resource created successfully', 'resource_id': new_resource.id}, status_code=201, success=True ) except Exception as e: ctx.session.rollback() return custom_response( data={'message': f'Error creating resource: {str(e)}'}, status_code=400, success=False ) def updateResource(self, resource_id, resource_data): """ Update an existing learning resource. """ with DatabaseContextManager() as ctx: try: resource = ctx.session.query(CourseResource).filter(CourseResource.id == resource_id).first() if not resource: return custom_response( data={'message': 'Resource not found'}, status_code=404, success=False ) # Update resource fields if 'title' in resource_data: resource.title = resource_data['title'] if 'description' in resource_data: resource.description = resource_data['description'] if 'resource_type' in resource_data: resource.resource_type = resource_data['resource_type'] if 'external_url' in resource_data: resource.url = resource_data['external_url'] if 'file_url' in resource_data: resource.file_path = resource_data['file_url'] if 'file_size' in resource_data: resource.file_size = resource_data['file_size'] if 'is_published' in resource_data: resource.is_published = resource_data['is_published'] ctx.session.commit() return custom_response( data={'message': 'Resource updated successfully'}, status_code=200, success=True ) except Exception as e: ctx.session.rollback() return custom_response( data={'message': f'Error updating resource: {str(e)}'}, status_code=400, success=False ) def deleteResource(self, resource_id): """ Delete a learning resource. """ with DatabaseContextManager() as ctx: try: resource = ctx.session.query(CourseResource).filter(CourseResource.id == resource_id).first() if not resource: return custom_response( data={'message': 'Resource not found'}, status_code=404, success=False ) ctx.session.delete(resource) ctx.session.commit() return custom_response( data={'message': 'Resource deleted successfully'}, status_code=200, success=True ) except Exception as e: ctx.session.rollback() return custom_response( data={'message': f'Error deleting resource: {str(e)}'}, status_code=400, success=False ) def toggleResourceStatus(self, resource_id, is_published): """ Toggle the published status of a resource. """ with DatabaseContextManager() as ctx: try: resource = ctx.session.query(CourseResource).filter(CourseResource.id == resource_id).first() if not resource: return custom_response( data={'message': 'Resource not found'}, status_code=404, success=False ) resource.is_published = is_published ctx.session.commit() status_text = 'published' if is_published else 'unpublished' return custom_response( data={'message': f'Resource {status_text} successfully'}, status_code=200, success=True ) except Exception as e: ctx.session.rollback() return custom_response( data={'message': f'Error updating resource status: {str(e)}'}, status_code=400, success=False ) def uploadResourceFile(self, file_data): """ Handle file upload for resources. """ try: # This is a placeholder for file upload logic # In a real implementation, you would save the file to a storage service # and return the file URL import os from werkzeug.utils import secure_filename if 'file' not in file_data: return custom_response( data={'message': 'No file provided'}, status_code=400, success=False ) file = file_data['file'] filename = secure_filename(file.filename) # Create upload directory if it doesn't exist upload_dir = 'uploads/resources' os.makedirs(upload_dir, exist_ok=True) # Generate unique filename file_id = str(uuid.uuid4()) file_extension = filename.split('.')[-1] if '.' in filename else '' unique_filename = f"{file_id}.{file_extension}" if file_extension else file_id # Save file file_path = os.path.join(upload_dir, unique_filename) file.save(file_path) # Return file URL (in production, this would be a CDN URL) file_url = f"/uploads/resources/{unique_filename}" return custom_response( data={ 'message': 'File uploaded successfully', 'file_url': file_url, 'file_size': os.path.getsize(file_path) }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error uploading file: {str(e)}'}, status_code=400, success=False ) def getModulesForResources(self): """ Get all modules for resource creation dropdown. """ with DatabaseContextManager() as ctx: try: modules = ctx.session.query(CourseModule).join(Course).filter(Course.is_active == True).all() modules_data = [] for module in modules: modules_data.append({ 'id': module.id, 'title': module.title, 'course': { 'id': module.course.id, 'title': module.course.title, 'code': module.course.code, 'department': module.course.department } }) return custom_response( data={'modules': modules_data}, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching modules: {str(e)}'}, status_code=400, success=False ) def fetchAttendances(self): return def fetchEnrollments(self, page=1, per_page=20, search="", status_filter="all", course_filter="all", semester_filter="all"): """ Get all enrollments with pagination, search, and filtering. """ with DatabaseContextManager() as ctx: # Base query with joins query = ctx.session.query(Enrollment).join(Student).filter(Student.is_active == True) # Apply search filter if search: search_term = f"%{search}%" # For course search, we need to join through enrollment_courses query = query.join(enrollment_courses, enrollment_courses.c.enrollment_id == Enrollment.id)\ .join(Course, Course.id == enrollment_courses.c.course_id)\ .filter( (Student.first_name.ilike(search_term)) | (Student.last_name.ilike(search_term)) | (Student.email.ilike(search_term)) | (Student.student_id.ilike(search_term)) | (Course.title.ilike(search_term)) | (Course.code.ilike(search_term)) ) # Apply status filter if status_filter != "all": query = query.filter(Enrollment.status == status_filter) # Apply course filter if course_filter != "all": query = query.join(enrollment_courses, enrollment_courses.c.enrollment_id == Enrollment.id)\ .filter(enrollment_courses.c.course_id == course_filter) # Apply term filter if semester_filter != "all": query = query.filter(Enrollment.term == semester_filter) # Get total count total_enrollments = query.count() # Apply pagination offset = (page - 1) * per_page enrollments = query.order_by(Enrollment.enrollment_date.desc()).offset(offset).limit(per_page).all() # Format enrollments data enrollments_data = [] for enrollment in enrollments: # Get course IDs for this enrollment course_ids = [ec.course_id for ec in enrollment.courses] enrollments_data.append({ 'id': enrollment.id, 'student_id': enrollment.student_id, 'course_ids': course_ids, 'enrollment_date': enrollment.enrollment_date.isoformat(), 'status': enrollment.status, 'grade': getattr(enrollment, 'grade', None), 'term': enrollment.term, 'enrollment_type': enrollment.enrollment_type, 'enrollment_date': enrollment.enrollment_date.isoformat() if enrollment.enrollment_date else None, 'completion_date': enrollment.completion_date.isoformat() if enrollment.completion_date else None, 'student': { 'id': enrollment.student.id, 'first_name': enrollment.student.first_name, 'last_name': enrollment.student.last_name, 'email': enrollment.student.email, 'student_id': enrollment.student.student_id, 'phone': enrollment.student.phone, 'department': enrollment.student.department, 'year_of_study': enrollment.student.year_of_study, 'is_active': enrollment.student.is_active }, 'course': { 'id': enrollment.course.id, 'title': enrollment.course.title, 'code': enrollment.course.code, 'department': enrollment.course.department, 'credits': enrollment.course.credits, 'is_active': enrollment.course.is_active, 'supervisor': { 'id': enrollment.course.supervisor.id if enrollment.course.supervisor else None, 'first_name': enrollment.course.supervisor.first_name if enrollment.course.supervisor else None, 'last_name': enrollment.course.supervisor.last_name if enrollment.course.supervisor else None, 'email': enrollment.course.supervisor.email if enrollment.course.supervisor else None } if enrollment.course.supervisor else None } }) # Calculate pagination info total_pages = (total_enrollments + per_page - 1) // per_page has_next = page < total_pages has_prev = page > 1 return custom_response( data={ 'enrollments': enrollments_data, 'pagination': { 'page': page, 'per_page': per_page, 'total_enrollments': total_enrollments, 'total_pages': total_pages, 'has_next': has_next, 'has_prev': has_prev } }, status_code=200, success=True ) def createEnrollment(self, enrollment_data): """ Create a new enrollment. """ with DatabaseContextManager() as ctx: try: # Check if student exists student = ctx.session.query(Student).filter(Student.id == enrollment_data.get('student_id')).first() if not student: return custom_response( data={'message': 'Student not found'}, status_code=404, success=False ) # Check if course exists course = ctx.session.query(Course).filter(Course.id == enrollment_data.get('course_id')).first() if not course: return custom_response( data={'message': 'Course not found'}, status_code=404, success=False ) # Check if enrollment already exists existing_enrollment = ctx.session.query(Enrollment).filter( Enrollment.student_id == enrollment_data.get('student_id'), Enrollment.speciality_id == enrollment_data.get('speciality_id'), Enrollment.term == enrollment_data.get('term') ).first() if existing_enrollment: return custom_response( data={'message': 'Student is already enrolled in this course for this semester'}, status_code=400, success=False ) # Create new enrollment enrollment = Enrollment( student_id=enrollment_data.get('student_id'), course_id=enrollment_data.get('course_id'), semester=enrollment_data.get('semester'), academic_year=enrollment_data.get('academic_year'), status=enrollment_data.get('status', 'active'), enrollment_date=datetime.now() ) ctx.session.add(enrollment) ctx.session.commit() return custom_response( data={'message': 'Enrollment created successfully', 'enrollment_id': enrollment.id}, status_code=201, success=True ) except Exception as e: ctx.session.rollback() return custom_response( data={'message': f'Error creating enrollment: {str(e)}'}, status_code=400, success=False ) def updateEnrollment(self, enrollment_id, enrollment_data): """ Update an existing enrollment. """ with DatabaseContextManager() as ctx: try: enrollment = ctx.session.query(Enrollment).filter(Enrollment.id == enrollment_id).first() if not enrollment: return custom_response( data={'message': 'Enrollment not found'}, status_code=404, success=False ) # Update enrollment fields if 'status' in enrollment_data: enrollment.status = enrollment_data['status'] if 'grade' in enrollment_data: enrollment.grade = enrollment_data['grade'] if 'term' in enrollment_data: enrollment.term = enrollment_data['term'] if 'enrollment_type' in enrollment_data: enrollment.enrollment_type = enrollment_data['enrollment_type'] if 'comments' in enrollment_data: enrollment.comments = enrollment_data['comments'] ctx.session.commit() return custom_response( data={'message': 'Enrollment updated successfully'}, status_code=200, success=True ) except Exception as e: ctx.session.rollback() return custom_response( data={'message': f'Error updating enrollment: {str(e)}'}, status_code=400, success=False ) def deleteEnrollment(self, enrollment_id): """ Delete an enrollment. """ with DatabaseContextManager() as ctx: try: enrollment = ctx.session.query(Enrollment).filter(Enrollment.id == enrollment_id).first() if not enrollment: return custom_response( data={'message': 'Enrollment not found'}, status_code=404, success=False ) ctx.session.delete(enrollment) ctx.session.commit() return custom_response( data={'message': 'Enrollment deleted successfully'}, status_code=200, success=True ) except Exception as e: ctx.session.rollback() return custom_response( data={'message': f'Error deleting enrollment: {str(e)}'}, status_code=400, success=False ) def getStudentsForEnrollment(self): """ Get all active students for enrollment dropdown. """ with DatabaseContextManager() as ctx: try: students = ctx.session.query(Student).filter(Student.is_active == True).all() students_data = [] for student in students: students_data.append({ 'id': student.id, 'first_name': student.first_name, 'last_name': student.last_name, 'email': student.email, 'student_id': student.student_id, 'phone': student.phone, 'department': student.department, 'year_of_study': student.year_of_study, 'is_active': student.is_active }) return custom_response( data={'students': students_data}, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching students: {str(e)}'}, status_code=400, success=False ) def getCoursesForEnrollment(self): """ Get all active courses for enrollment dropdown. """ with DatabaseContextManager() as ctx: try: courses = ctx.session.query(Course).filter(Course.is_active == True).all() courses_data = [] for course in courses: courses_data.append({ 'id': course.id, 'title': course.title, 'code': course.code, 'department': course.department, 'credits': course.credits, 'is_active': course.is_active, 'supervisor': { 'id': course.supervisor.id if course.supervisor else None, 'first_name': course.supervisor.first_name if course.supervisor else None, 'last_name': course.supervisor.last_name if course.supervisor else None, 'email': course.supervisor.email if course.supervisor else None } if course.supervisor else None }) return custom_response( data={'courses': courses_data}, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching courses: {str(e)}'}, status_code=400, success=False ) def downloadEnrollmentTemplate(self): """ Generate and return enrollment template file. """ with DatabaseContextManager() as ctx: try: # Get students and courses for template students = ctx.session.query(Student).filter(Student.is_active == True).all() courses = ctx.session.query(Course).filter(Course.is_active == True).all() # Create template data template_data = { 'Student ID': [], 'Course Code': [], 'Semester': [], 'Academic Year': [], 'Status': [] } # Add sample data template_data['Student ID'].extend([student.student_id for student in students[:10]]) template_data['Course Code'].extend([course.code for course in courses[:10]]) template_data['Semester'].extend(['Fall 2024'] * 10) template_data['Academic Year'].extend(['2024-2025'] * 10) template_data['Status'].extend(['active'] * 10) # Create DataFrame df = pd.DataFrame(template_data) # Save to Excel file filename = f"enrollment_template_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" filepath = os.path.join('uploads', 'templates', filename) # Create directory if it doesn't exist os.makedirs(os.path.dirname(filepath), exist_ok=True) df.to_excel(filepath, index=False, sheet_name='Enrollment Template') return custom_response( data={ 'message': 'Template generated successfully', 'filename': filename, 'filepath': filepath }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error generating template: {str(e)}'}, status_code=400, success=False ) def bulkEnrollStudents(self, file_data): """ Process bulk enrollment from uploaded file. """ with DatabaseContextManager() as ctx: try: # Parse the uploaded file df = pd.read_excel(file_data) if file_data.endswith('.xlsx') else pd.read_csv(file_data) # Validate required columns required_columns = ['Student ID', 'Course Code', 'Semester', 'Academic Year'] if not all(col in df.columns for col in required_columns): return custom_response( data={'message': f'Missing required columns. Required: {required_columns}'}, status_code=400, success=False ) enrollments_created = 0 errors = [] for index, row in df.iterrows(): try: # Find student by student_id student = ctx.session.query(Student).filter( Student.student_id == row['Student ID'], Student.is_active == True ).first() if not student: errors.append(f"Row {index + 1}: Student with ID {row['Student ID']} not found") continue # Find course by code course = ctx.session.query(Course).filter( Course.code == row['Course Code'], Course.is_active == True ).first() if not course: errors.append(f"Row {index + 1}: Course with code {row['Course Code']} not found") continue # Check if enrollment already exists existing_enrollment = ctx.session.query(Enrollment).filter( Enrollment.student_id == student.id, Enrollment.speciality_id == speciality.id, Enrollment.term == row.get('Term', None) ).first() if existing_enrollment: errors.append(f"Row {index + 1}: Student already enrolled in this course for this semester") continue # Create enrollment enrollment = Enrollment( student_id=student.id, course_id=course.id, semester=row['Semester'], academic_year=row['Academic Year'], status=row.get('Status', 'active'), enrollment_date=datetime.now() ) ctx.session.add(enrollment) enrollments_created += 1 except Exception as e: errors.append(f"Row {index + 1}: {str(e)}") ctx.session.commit() return custom_response( data={ 'message': f'Bulk enrollment completed. {enrollments_created} enrollments created.', 'enrollments_created': enrollments_created, 'errors': errors }, status_code=200, success=True ) except Exception as e: ctx.session.rollback() return custom_response( data={'message': f'Error processing bulk enrollment: {str(e)}'}, status_code=400, success=False ) def getEnrollmentStats(self): """ Get enrollment statistics. """ with DatabaseContextManager() as ctx: try: total_enrollments = ctx.session.query(Enrollment).count() active_enrollments = ctx.session.query(Enrollment).filter(Enrollment.status == 'active').count() completed_enrollments = ctx.session.query(Enrollment).filter(Enrollment.status == 'completed').count() dropped_enrollments = ctx.session.query(Enrollment).filter(Enrollment.status == 'dropped').count() return custom_response( data={ 'total_enrollments': total_enrollments, 'active_enrollments': active_enrollments, 'completed_enrollments': completed_enrollments, 'dropped_enrollments': dropped_enrollments }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching enrollment stats: {str(e)}'}, status_code=400, success=False ) def fetchTeachingSessions(self, page=1, per_page=20, search='', status='all', course_id='all', tutor_id='all', session_type='all', date_filter='all'): """Fetch teaching sessions with pagination and filters""" try: with DatabaseContextManager() as ctx: query = ctx.session.query(TeachingSession).join(User, TeachingSession.tutor_id == User.id).join(Course, TeachingSession.course_id == Course.id) # Apply filters if search: search_term = f"%{search}%" query = query.filter( or_( TeachingSession.title.ilike(search_term), Course.title.ilike(search_term), Course.code.ilike(search_term), User.first_name.ilike(search_term), User.last_name.ilike(search_term), TeachingSession.room.ilike(search_term) ) ) if status != 'all': query = query.filter(TeachingSession.status == status) if course_id != 'all': query = query.filter(TeachingSession.course_id == course_id) if tutor_id != 'all': query = query.filter(TeachingSession.tutor_id == tutor_id) if session_type != 'all': query = query.filter(TeachingSession.session_type == session_type) if date_filter != 'all': # Since TeachingSession only has Time fields, we'll filter by created_at for date filtering today = datetime.now().date() if date_filter == 'today': query = query.filter(TeachingSession.created_at >= today) elif date_filter == 'week': week_start = today - timedelta(days=today.weekday()) query = query.filter(TeachingSession.created_at >= week_start) elif date_filter == 'month': month_start = today.replace(day=1) query = query.filter(TeachingSession.created_at >= month_start) # Get total count total_items = query.count() # Apply pagination offset = (page - 1) * per_page sessions = query.offset(offset).limit(per_page).all() # Calculate pagination info total_pages = (total_items + per_page - 1) // per_page has_next = page < total_pages has_prev = page > 1 # Format sessions data sessions_data = [] for session in sessions: # Get attendance count attendance_count = ctx.session.query(Attendance).filter(Attendance.session_id == session.id).count() # Get venue information if available venue_info = None if session.room: # Try to get venue details from room name venue_info = { 'id': session.room, 'name': session.room, 'building': 'Main Building', # Default building 'floor': 'Ground Floor', # Default floor 'capacity': 50, # Default capacity 'equipment': ['Projector', 'Whiteboard', 'Chairs', 'Tables'], 'location': session.room } session_data = { 'id': session.id, 'title': session.title, 'room': session.room, 'day_of_week': session.day_of_week, 'start_time': session.start_time.strftime('%H:%M:%S') if session.start_time else None, 'end_time': session.end_time.strftime('%H:%M:%S') if session.end_time else None, 'session_type': session.session_type, 'recurring': session.recurring, 'status': session.status, 'notes': session.notes, 'is_verified': session.is_verified, 'created_at': session.created_at.isoformat() if session.created_at else None, 'updated_at': session.updated_at.isoformat() if session.updated_at else None, 'tutor': { 'id': session.tutor.id, 'first_name': session.tutor.first_name, 'last_name': session.tutor.last_name, 'email': session.tutor.email, 'phone': getattr(session.tutor, 'phone', None) }, 'course': { 'id': session.course.id, 'code': session.course.code, 'title': session.course.title, 'department': session.course.department }, 'attendance_count': attendance_count, 'venue': venue_info, 'verifier': { 'id': session.verifier.id, 'first_name': session.verifier.first_name, 'last_name': session.verifier.last_name, 'email': session.verifier.email } if session.verifier else None } sessions_data.append(session_data) return custom_response( data={ 'sessions': sessions_data, 'pagination': { 'current_page': page, 'per_page': per_page, 'total_pages': total_pages, 'total_items': total_items, 'has_next': has_next, 'has_prev': has_prev } }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching teaching sessions: {str(e)}'}, status_code=400, success=False ) def getTeachingSessionDetail(self, session_id): """Get detailed teaching session information""" try: with DatabaseContextManager() as ctx: session = ctx.session.query(TeachingSession).filter(TeachingSession.id == session_id).first() if not session: return custom_response( data={'message': 'Teaching session not found'}, status_code=404, success=False ) # Get attendance records attendance_records = ctx.session.query(Attendance).filter(Attendance.session_id == session_id).all() # Get venue information venue_info = None if session.room: venue_info = { 'id': session.room, 'name': session.room, 'building': 'Main Building', 'floor': 'Ground Floor', 'capacity': 50, 'equipment': ['Projector', 'Whiteboard', 'Chairs', 'Tables'], 'location': session.room } session_detail = { 'id': session.id, 'title': session.title, 'room': session.room, 'day_of_week': session.day_of_week, 'start_time': session.start_time.strftime('%H:%M:%S') if session.start_time else None, 'end_time': session.end_time.strftime('%H:%M:%S') if session.end_time else None, 'session_type': session.session_type, 'recurring': session.recurring, 'status': session.status, 'notes': session.notes, 'is_verified': session.is_verified, 'created_at': session.created_at.isoformat() if session.created_at else None, 'updated_at': session.updated_at.isoformat() if session.updated_at else None, 'tutor': { 'id': session.tutor.id, 'first_name': session.tutor.first_name, 'last_name': session.tutor.last_name, 'email': session.tutor.email, 'phone': getattr(session.tutor, 'phone', None) }, 'course': { 'id': session.course.id, 'code': session.course.code, 'title': session.course.title, 'department': session.course.department }, 'venue': venue_info, 'verifier': { 'id': session.verifier.id, 'first_name': session.verifier.first_name, 'last_name': session.verifier.last_name, 'email': session.verifier.email } if session.verifier else None, 'attendance_records': [ { 'id': record.id, 'student_id': record.student_id, 'status': record.status, 'check_in_time': record.check_in_time.isoformat() if record.check_in_time else None, 'check_out_time': record.check_out_time.isoformat() if record.check_out_time else None, 'notes': record.notes } for record in attendance_records ] } return custom_response( data=session_detail, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching teaching session detail: {str(e)}'}, status_code=400, success=False ) def getTeachingSessionStats(self): """Get teaching session statistics""" try: with DatabaseContextManager() as ctx: # Total sessions total_sessions = ctx.session.query(TeachingSession).count() # Sessions by status ongoing_sessions = ctx.session.query(TeachingSession).filter(TeachingSession.status == 'ongoing').count() completed_sessions = ctx.session.query(TeachingSession).filter(TeachingSession.status == 'completed').count() cancelled_sessions = ctx.session.query(TeachingSession).filter(TeachingSession.status == 'cancelled').count() scheduled_sessions = ctx.session.query(TeachingSession).filter(TeachingSession.status == 'scheduled').count() # Verified sessions verified_sessions = ctx.session.query(TeachingSession).filter(TeachingSession.is_verified == True).count() # Total attendance total_attendance = ctx.session.query(Attendance).count() return custom_response( data={ 'total_sessions': total_sessions, 'ongoing_sessions': ongoing_sessions, 'completed_sessions': completed_sessions, 'cancelled_sessions': cancelled_sessions, 'scheduled_sessions': scheduled_sessions, 'verified_sessions': verified_sessions, 'total_attendance': total_attendance }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching teaching session stats: {str(e)}'}, status_code=400, success=False ) # Attendance Records Management Functions def fetchAttendances(self, page=1, per_page=20, search=None, status=None, tutor_id=None, date_filter=None, verified=None): """Fetch attendance records with pagination and filters""" try: with DatabaseContextManager() as ctx: query = ctx.session.query(Attendance).join( Tutor, Attendance.tutor_id == Tutor.id ).join( TeachingSession, Attendance.session_id == TeachingSession.id ).join( Course, TeachingSession.course_id == Course.id ).join( Student, Attendance.student_id == Student.id ) # Apply filters if search: search_term = f"%{search}%" query = query.filter( or_( Tutor.first_name.ilike(search_term), Tutor.last_name.ilike(search_term), Tutor.employee_id.ilike(search_term), Student.first_name.ilike(search_term), Student.last_name.ilike(search_term), Student.student_id.ilike(search_term), TeachingSession.title.ilike(search_term), Course.title.ilike(search_term), Course.code.ilike(search_term) ) ) if status and status != 'all': query = query.filter(Attendance.status == status) if tutor_id and tutor_id != 'all': query = query.filter(Attendance.tutor_id == tutor_id) if date_filter and date_filter != 'all': today = datetime.now().date() if date_filter == 'today': query = query.filter(func.date(Attendance.timestamp) == today) elif date_filter == 'week': week_ago = today - timedelta(days=7) query = query.filter(func.date(Attendance.timestamp) >= week_ago) elif date_filter == 'month': month_ago = today - timedelta(days=30) query = query.filter(func.date(Attendance.timestamp) >= month_ago) if verified is not None and verified != 'all': if verified == 'verified': query = query.filter(Attendance.verification_method.isnot(None)) elif verified == 'unverified': query = query.filter(Attendance.verification_method.is_(None)) # Get total count total_items = query.count() # Apply pagination offset = (page - 1) * per_page records = query.offset(offset).limit(per_page).all() # Calculate pagination info total_pages = (total_items + per_page - 1) // per_page has_next = page < total_pages has_prev = page > 1 # Format records formatted_records = [] for record in records: # Calculate duration based on session times and attendance timestamp duration_minutes = 0 if record.session.start_time and record.session.end_time: start_time = datetime.combine(record.timestamp.date(), record.session.start_time) end_time = datetime.combine(record.timestamp.date(), record.session.end_time) duration_minutes = int((end_time - start_time).total_seconds() / 60) formatted_record = { 'id': record.id, 'tutor_id': record.tutor_id, 'session_id': record.session_id, 'student_id': record.student_id, 'date': record.timestamp.date().isoformat(), 'check_in_time': record.timestamp.time().isoformat(), 'check_out_time': None, # Not available in current model 'status': record.status.value if hasattr(record.status, 'value') else str(record.status), 'notes': record.notes, 'verified': record.verification_method is not None, 'verified_by': record.supervisor_tutor_id, 'verified_at': record.timestamp.isoformat() if record.verification_method else None, 'created_at': record.created_at.isoformat(), 'updated_at': record.updated_at.isoformat(), 'duration_minutes': duration_minutes, 'punctuality_score': 100 if record.status.value == 'present' else (50 if record.late_minutes > 0 else 0), 'attendance_streak': 1, # This would need to be calculated based on historical data 'late_minutes': record.late_minutes, 'verification_method': record.verification_method.value if record.verification_method else None, 'device_used': record.device_used, 'ip_address': record.ip_address, 'tutor': { 'id': record.tutor.id, 'first_name': record.tutor.first_name, 'last_name': record.tutor.last_name, 'email': record.tutor.email, 'department': getattr(record.tutor, 'department', 'N/A'), 'employee_id': getattr(record.tutor, 'employee_id', 'N/A') }, 'student': { 'id': record.student.id, 'first_name': record.student.first_name, 'last_name': record.student.last_name, 'email': record.student.email, 'student_id': getattr(record.student, 'student_id', 'N/A') }, 'session': { 'id': record.session.id, 'title': record.session.title, 'course': { 'id': record.session.course.id, 'title': record.session.course.title, 'code': record.session.course.code }, 'room': record.session.room, 'start_time': record.session.start_time.isoformat(), 'end_time': record.session.end_time.isoformat() } } if record.supervisor_tutor: formatted_record['verifier'] = { 'id': record.supervisor_tutor.id, 'first_name': record.supervisor_tutor.first_name, 'last_name': record.supervisor_tutor.last_name, 'email': record.supervisor_tutor.email } formatted_records.append(formatted_record) return custom_response( data={ 'records': formatted_records, 'pagination': { 'current_page': page, 'per_page': per_page, 'total_items': total_items, 'total_pages': total_pages, 'has_next': has_next, 'has_prev': has_prev } }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching attendance records: {str(e)}'}, status_code=400, success=False ) def getAttendanceDetail(self, record_id): """Get detailed information for a specific attendance record""" try: with DatabaseContextManager() as ctx: record = ctx.session.query(Attendance).join( Tutor, Attendance.tutor_id == Tutor.id ).join( TeachingSession, Attendance.session_id == TeachingSession.id ).join( Course, TeachingSession.course_id == Course.id ).join( Student, Attendance.student_id == Student.id ).filter(Attendance.id == record_id).first() if not record: return custom_response( data={'message': 'Attendance record not found'}, status_code=404, success=False ) # Calculate duration duration_minutes = 0 if record.session.start_time and record.session.end_time: start_time = datetime.combine(record.timestamp.date(), record.session.start_time) end_time = datetime.combine(record.timestamp.date(), record.session.end_time) duration_minutes = int((end_time - start_time).total_seconds() / 60) formatted_record = { 'id': record.id, 'tutor_id': record.tutor_id, 'session_id': record.session_id, 'student_id': record.student_id, 'date': record.timestamp.date().isoformat(), 'time': record.timestamp.time().isoformat(), 'status': record.status.value, 'notes': record.notes, 'late_minutes': record.late_minutes, 'verification_method': record.verification_method.value if record.verification_method else None, 'device_used': record.device_used, 'ip_address': record.ip_address, 'supervisor_tutor_id': record.supervisor_tutor_id, 'created_at': record.created_at.isoformat(), 'updated_at': record.updated_at.isoformat(), 'duration_minutes': duration_minutes, 'punctuality_score': 100 if record.status == AttendanceStatus.present else (50 if record.status == AttendanceStatus.late else 0), 'tutor': { 'id': record.tutor.id, 'first_name': record.tutor.first_name, 'last_name': record.tutor.last_name, 'email': record.tutor.email, 'employee_id': record.tutor.employee_id }, 'session': { 'id': record.session.id, 'title': record.session.title, 'course': { 'id': record.session.course.id, 'title': record.session.course.title, 'code': record.session.course.code }, 'room': record.session.room, 'start_time': record.session.start_time.isoformat(), 'end_time': record.session.end_time.isoformat() } } if record.supervisor_tutor: formatted_record['supervisor_tutor'] = { 'id': record.supervisor_tutor.id, 'first_name': record.supervisor_tutor.first_name, 'last_name': record.supervisor_tutor.last_name, 'email': record.supervisor_tutor.email } return custom_response( data=formatted_record, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching attendance record detail: {str(e)}'}, status_code=400, success=False ) def getAttendanceStats(self): """Get attendance statistics""" try: with DatabaseContextManager() as ctx: # Get total records total_records = ctx.session.query(Attendance).count() # Get records by status present_records = ctx.session.query(Attendance).filter(Attendance.status == AttendanceStatus.present).count() absent_records = ctx.session.query(Attendance).filter(Attendance.status == AttendanceStatus.absent).count() late_records = ctx.session.query(Attendance).filter(Attendance.status == AttendanceStatus.late).count() excused_records = ctx.session.query(Attendance).filter(Attendance.status == AttendanceStatus.excused).count() # Calculate average attendance percentage average_attendance = (present_records / total_records * 100) if total_records > 0 else 0 # Calculate total hours based on session duration total_hours = 0 records_with_sessions = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).all() for record in records_with_sessions: if record.session.start_time and record.session.end_time: start_time = datetime.combine(record.timestamp.date(), record.session.start_time) end_time = datetime.combine(record.timestamp.date(), record.session.end_time) duration_minutes = int((end_time - start_time).total_seconds() / 60) total_hours += duration_minutes / 60 # Get verified records count verified_records = ctx.session.query(Attendance).filter(Attendance.verification_method.isnot(None)).count() return custom_response( data={ 'totalRecords': total_records, 'presentRecords': present_records, 'absentRecords': absent_records, 'lateRecords': late_records, 'excusedRecords': excused_records, 'averageAttendance': round(average_attendance, 2), 'totalHours': round(total_hours, 2), 'verifiedRecords': verified_records }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching attendance stats: {str(e)}'}, status_code=400, success=False ) def createAttendance(self, tutor_id, session_id, student_id, status='present', notes=None, late_minutes=0, verification_method=None, device_used=None, ip_address=None): """Create a new attendance record""" try: with DatabaseContextManager() as ctx: # Check if record already exists for this tutor, session, and student existing_record = ctx.session.query(Attendance).filter( Attendance.tutor_id == tutor_id, Attendance.session_id == session_id, Attendance.student_id == student_id ).first() if existing_record: return custom_response( data={'message': 'Attendance record already exists for this tutor, session, and student'}, status_code=400, success=False ) # Create new record new_record = Attendance( id=str(uuid.uuid4()), tutor_id=tutor_id, session_id=session_id, student_id=student_id, status=AttendanceStatus.present if status == 'present' else AttendanceStatus.absent if status == 'absent' else AttendanceStatus.late if status == 'late' else AttendanceStatus.excused, notes=notes, late_minutes=late_minutes, verification_method=verification_method, device_used=device_used, ip_address=ip_address, created_at=datetime.now(), updated_at=datetime.now() ) ctx.session.add(new_record) ctx.session.commit() return custom_response( data={'message': 'Attendance record created successfully', 'record_id': new_record.id}, status_code=201, success=True ) except Exception as e: return custom_response( data={'message': f'Error creating attendance record: {str(e)}'}, status_code=400, success=False ) def updateAttendance(self, record_id, status=None, notes=None, late_minutes=None, verification_method=None, supervisor_tutor_id=None): """Update an existing attendance record""" try: with DatabaseContextManager() as ctx: record = ctx.session.query(Attendance).filter(Attendance.id == record_id).first() if not record: return custom_response( data={'message': 'Attendance record not found'}, status_code=404, success=False ) # Update fields if status is not None: record.status = AttendanceStatus.present if status == 'present' else AttendanceStatus.absent if status == 'absent' else AttendanceStatus.late if status == 'late' else AttendanceStatus.excused if notes is not None: record.notes = notes if late_minutes is not None: record.late_minutes = late_minutes if verification_method is not None: record.verification_method = verification_method if supervisor_tutor_id is not None: record.supervisor_tutor_id = supervisor_tutor_id record.updated_at = datetime.now() ctx.session.commit() return custom_response( data={'message': 'Attendance record updated successfully'}, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error updating attendance record: {str(e)}'}, status_code=400, success=False ) def deleteAttendance(self, record_id): """Delete an attendance record""" try: with DatabaseContextManager() as ctx: record = ctx.session.query(Attendance).filter(Attendance.id == record_id).first() if not record: return custom_response( data={'message': 'Attendance record not found'}, status_code=404, success=False ) ctx.session.delete(record) ctx.session.commit() return custom_response( data={'message': 'Attendance record deleted successfully'}, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error deleting attendance record: {str(e)}'}, status_code=400, success=False ) def fetchVerificationLogs(self, page=1, per_page=20, search=None, verification_method=None, tutor_id=None, student_id=None, date_filter=None, status=None): """Fetch paginated verification logs with filters""" try: with DatabaseContextManager() as ctx: # Base query with joins query = ctx.session.query(Attendance)\ .join(Tutor, Attendance.tutor_id == Tutor.id)\ .join(TeachingSession, Attendance.session_id == TeachingSession.id)\ .join(Course, TeachingSession.course_id == Course.id)\ .join(Student, Attendance.student_id == Student.id)\ .filter(Attendance.verification_method.isnot(None)) # Apply search filter if search: search_term = f"%{search}%" query = query.filter( or_( Tutor.first_name.ilike(search_term), Tutor.last_name.ilike(search_term), Student.first_name.ilike(search_term), Student.last_name.ilike(search_term), Student.student_id.ilike(search_term), TeachingSession.title.ilike(search_term), Course.title.ilike(search_term), Course.code.ilike(search_term), Attendance.device_used.ilike(search_term), Attendance.ip_address.ilike(search_term) ) ) # Apply verification method filter if verification_method and verification_method != 'all': query = query.filter(Attendance.verification_method == verification_method) # Apply tutor filter if tutor_id and tutor_id != 'all': query = query.filter(Attendance.tutor_id == tutor_id) # Apply student filter if student_id and student_id != 'all': query = query.filter(Attendance.student_id == student_id) # Apply status filter if status and status != 'all': query = query.filter(Attendance.status == status) # Apply date filter if date_filter and date_filter != 'all': if date_filter == 'today': today = datetime.now().date() query = query.filter(func.date(Attendance.timestamp) == today) elif date_filter == 'week': week_ago = datetime.now() - timedelta(days=7) query = query.filter(Attendance.timestamp >= week_ago) elif date_filter == 'month': month_ago = datetime.now() - timedelta(days=30) query = query.filter(Attendance.timestamp >= month_ago) # Get total count before pagination total_count = query.count() # Apply pagination offset = (page - 1) * per_page records = query.order_by(Attendance.timestamp.desc()).offset(offset).limit(per_page).all() # Format records formatted_records = [] for record in records: formatted_records.append({ 'id': record.id, 'timestamp': record.timestamp.isoformat(), 'date': record.timestamp.date().isoformat(), 'time': record.timestamp.time().isoformat(), 'status': record.status.value, 'verification_method': record.verification_method.value, 'device_used': record.device_used, 'ip_address': record.ip_address, 'late_minutes': record.late_minutes, 'notes': record.notes, 'tutor': { 'id': record.tutor.id, 'first_name': record.tutor.first_name, 'last_name': record.tutor.last_name, 'email': record.tutor.email }, 'student': { 'id': record.student.id, 'first_name': record.student.first_name, 'last_name': record.student.last_name, 'student_id': record.student.student_id, 'email': record.student.email }, 'session': { 'id': record.session.id, 'title': record.session.title, 'room': record.session.room, 'start_time': record.session.start_time.isoformat(), 'end_time': record.session.end_time.isoformat(), 'course': { 'id': record.session.course.id, 'title': record.session.course.title, 'code': record.session.course.code } }, 'supervisor_tutor': { 'id': record.supervisor_tutor.id, 'first_name': record.supervisor_tutor.first_name, 'last_name': record.supervisor_tutor.last_name, 'email': record.supervisor_tutor.email } if record.supervisor_tutor else None }) # Calculate pagination info total_pages = (total_count + per_page - 1) // per_page has_next = page < total_pages has_prev = page > 1 return custom_response( data={ 'verification_logs': formatted_records, 'pagination': { 'current_page': page, 'per_page': per_page, 'total_pages': total_pages, 'total_items': total_count, 'has_next': has_next, 'has_prev': has_prev } }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching verification logs: {str(e)}'}, status_code=400, success=False ) def getVerificationLogStats(self): """Get verification log statistics""" try: with DatabaseContextManager() as ctx: # Total verification logs total_logs = ctx.session.query(Attendance).filter(Attendance.verification_method.isnot(None)).count() # Verification methods breakdown verification_methods = ctx.session.query( Attendance.verification_method, func.count(Attendance.id).label('count') ).filter(Attendance.verification_method.isnot(None))\ .group_by(Attendance.verification_method).all() method_breakdown = {method.value: count for method, count in verification_methods} # Status breakdown for verified records status_breakdown = ctx.session.query( Attendance.status, func.count(Attendance.id).label('count') ).filter(Attendance.verification_method.isnot(None))\ .group_by(Attendance.status).all() status_counts = {status.value: count for status, count in status_breakdown} # Recent verifications (last 7 days) week_ago = datetime.now() - timedelta(days=7) recent_verifications = ctx.session.query(Attendance)\ .filter(Attendance.verification_method.isnot(None))\ .filter(Attendance.timestamp >= week_ago).count() # Today's verifications today = datetime.now().date() today_verifications = ctx.session.query(Attendance)\ .filter(Attendance.verification_method.isnot(None))\ .filter(func.date(Attendance.timestamp) == today).count() # Average verifications per day (last 30 days) month_ago = datetime.now() - timedelta(days=30) monthly_verifications = ctx.session.query(Attendance)\ .filter(Attendance.verification_method.isnot(None))\ .filter(Attendance.timestamp >= month_ago).count() avg_daily_verifications = monthly_verifications / 30 if monthly_verifications > 0 else 0 # Device usage breakdown device_usage = ctx.session.query( Attendance.device_used, func.count(Attendance.id).label('count') ).filter(Attendance.verification_method.isnot(None))\ .filter(Attendance.device_used.isnot(None))\ .group_by(Attendance.device_used).all() device_breakdown = {device: count for device, count in device_usage if device} return custom_response( data={ 'totalLogs': total_logs, 'methodBreakdown': method_breakdown, 'statusBreakdown': status_counts, 'recentVerifications': recent_verifications, 'todayVerifications': today_verifications, 'avgDailyVerifications': round(avg_daily_verifications, 2), 'deviceBreakdown': device_breakdown }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching verification log stats: {str(e)}'}, status_code=400, success=False ) def getVerificationLogDetail(self, log_id): """Get detailed information for a specific verification log""" try: with DatabaseContextManager() as ctx: record = ctx.session.query(Attendance)\ .join(Tutor, Attendance.tutor_id == Tutor.id)\ .join(TeachingSession, Attendance.session_id == TeachingSession.id)\ .join(Course, TeachingSession.course_id == Course.id)\ .join(Student, Attendance.student_id == Student.id)\ .filter(Attendance.id == log_id)\ .filter(Attendance.verification_method.isnot(None)).first() if not record: return custom_response( data={'message': 'Verification log not found'}, status_code=404, success=False ) # Calculate duration if available duration_minutes = None if record.session.start_time and record.session.end_time: start_time = datetime.combine(record.timestamp.date(), record.session.start_time) end_time = datetime.combine(record.timestamp.date(), record.session.end_time) duration_minutes = int((end_time - start_time).total_seconds() / 60) # Calculate punctuality score punctuality_score = 100 if record.late_minutes and record.late_minutes > 0: punctuality_score = max(0, 100 - (record.late_minutes * 2)) # 2 points per minute late detailed_log = { 'id': record.id, 'timestamp': record.timestamp.isoformat(), 'date': record.timestamp.date().isoformat(), 'time': record.timestamp.time().isoformat(), 'status': record.status.value, 'verification_method': record.verification_method.value, 'device_used': record.device_used, 'ip_address': record.ip_address, 'late_minutes': record.late_minutes, 'notes': record.notes, 'duration_minutes': duration_minutes, 'punctuality_score': punctuality_score, 'tutor': { 'id': record.tutor.id, 'first_name': record.tutor.first_name, 'last_name': record.tutor.last_name, 'email': record.tutor.email, 'phone': record.tutor.phone }, 'student': { 'id': record.student.id, 'first_name': record.student.first_name, 'last_name': record.student.last_name, 'student_id': record.student.student_id, 'email': record.student.email, 'program': record.student.program, 'year_of_study': record.student.year_of_study }, 'session': { 'id': record.session.id, 'title': record.session.title, 'room': record.session.room, 'start_time': record.session.start_time.isoformat(), 'end_time': record.session.end_time.isoformat(), 'session_type': record.session.session_type, 'course': { 'id': record.session.course.id, 'title': record.session.course.title, 'code': record.session.course.code, 'credits': record.session.course.credits } }, 'supervisor_tutor': { 'id': record.supervisor_tutor.id, 'first_name': record.supervisor_tutor.first_name, 'last_name': record.supervisor_tutor.last_name, 'email': record.supervisor_tutor.email } if record.supervisor_tutor else None, 'created_at': record.created_at.isoformat(), 'updated_at': record.updated_at.isoformat() } return custom_response( data={'verification_log': detailed_log}, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching verification log detail: {str(e)}'}, status_code=400, success=False ) def getAttendanceAnalytics(self, start_date=None, end_date=None, tutor_id=None, course_id=None, department=None): """ Get comprehensive attendance analytics and statistics """ try: with DatabaseContextManager() as ctx: # Base query with joins query = ctx.session.query(Attendance).join( Tutor, Attendance.tutor_id == Tutor.id ).join( TeachingSession, Attendance.session_id == TeachingSession.id ).join( Course, TeachingSession.course_id == Course.id ).join( Student, Attendance.student_id == Student.id ) # Apply filters if start_date: query = query.filter(func.date(Attendance.timestamp) >= start_date) if end_date: query = query.filter(func.date(Attendance.timestamp) <= end_date) if tutor_id: query = query.filter(Attendance.tutor_id == tutor_id) if course_id: query = query.filter(TeachingSession.course_id == course_id) if department: query = query.filter(Course.department == department) records = query.all() if not records: return { 'total_records': 0, 'attendance_rate': 0, 'punctuality_rate': 0, 'status_breakdown': {}, 'daily_attendance': [], 'tutor_performance': [], 'course_performance': [], 'department_performance': [], 'time_analysis': {}, 'trends': {} } # Calculate basic statistics total_records = len(records) present_count = sum(1 for r in records if r.status.value == 'present') attendance_rate = (present_count / total_records * 100) if total_records > 0 else 0 # Punctuality analysis on_time_count = sum(1 for r in records if r.late_minutes == 0) punctuality_rate = (on_time_count / total_records * 100) if total_records > 0 else 0 # Status breakdown status_breakdown = {} for record in records: status = record.status.value status_breakdown[status] = status_breakdown.get(status, 0) + 1 # Daily attendance trends daily_data = {} for record in records: date_str = record.timestamp.date().isoformat() if date_str not in daily_data: daily_data[date_str] = {'total': 0, 'present': 0, 'late': 0} daily_data[date_str]['total'] += 1 if record.status.value == 'present': daily_data[date_str]['present'] += 1 if record.late_minutes > 0: daily_data[date_str]['late'] += 1 daily_attendance = [ { 'date': date, 'total': data['total'], 'present': data['present'], 'late': data['late'], 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 } for date, data in sorted(daily_data.items()) ] # Tutor performance tutor_performance = {} for record in records: tutor_id = record.tutor_id if tutor_id not in tutor_performance: tutor_performance[tutor_id] = { 'tutor_name': f"{record.tutor.first_name} {record.tutor.last_name}", 'total_sessions': 0, 'present_count': 0, 'late_count': 0, 'avg_late_minutes': 0 } tutor_performance[tutor_id]['total_sessions'] += 1 if record.status.value == 'present': tutor_performance[tutor_id]['present_count'] += 1 if record.late_minutes > 0: tutor_performance[tutor_id]['late_count'] += 1 tutor_performance[tutor_id]['avg_late_minutes'] += record.late_minutes # Calculate averages and rates tutor_performance_list = [] for tutor_id, data in tutor_performance.items(): if data['total_sessions'] > 0: data['attendance_rate'] = data['present_count'] / data['total_sessions'] * 100 data['late_rate'] = data['late_count'] / data['total_sessions'] * 100 if data['late_count'] > 0: data['avg_late_minutes'] = data['avg_late_minutes'] / data['late_count'] else: data['avg_late_minutes'] = 0 tutor_performance_list.append(data) # Course performance course_performance = {} for record in records: course_id = record.session.course_id if course_id not in course_performance: course_performance[course_id] = { 'course_code': record.session.course.code, 'course_title': record.session.course.title, 'total_sessions': 0, 'present_count': 0, 'avg_attendance': 0 } course_performance[course_id]['total_sessions'] += 1 if record.status.value == 'present': course_performance[course_id]['present_count'] += 1 course_performance_list = [] for course_id, data in course_performance.items(): if data['total_sessions'] > 0: data['attendance_rate'] = data['present_count'] / data['total_sessions'] * 100 course_performance_list.append(data) # Department performance department_performance = {} for record in records: dept = record.session.course.department if dept not in department_performance: department_performance[dept] = { 'department': dept, 'total_sessions': 0, 'present_count': 0, 'courses_count': 0 } department_performance[dept]['total_sessions'] += 1 if record.status.value == 'present': department_performance[dept]['present_count'] += 1 # Count unique courses per department dept_courses = {} for record in records: dept = record.session.course.department course_id = record.session.course_id if dept not in dept_courses: dept_courses[dept] = set() dept_courses[dept].add(course_id) department_performance_list = [] for dept, data in department_performance.items(): if data['total_sessions'] > 0: data['attendance_rate'] = data['present_count'] / data['total_sessions'] * 100 data['courses_count'] = len(dept_courses.get(dept, set())) department_performance_list.append(data) # Time analysis time_analysis = { 'hourly_distribution': {}, 'day_of_week_distribution': {}, 'monthly_distribution': {} } for record in records: # Hourly distribution hour = record.timestamp.hour time_analysis['hourly_distribution'][hour] = time_analysis['hourly_distribution'].get(hour, 0) + 1 # Day of week distribution day_of_week = record.timestamp.weekday() time_analysis['day_of_week_distribution'][day_of_week] = time_analysis['day_of_week_distribution'].get(day_of_week, 0) + 1 # Monthly distribution month = record.timestamp.month time_analysis['monthly_distribution'][month] = time_analysis['monthly_distribution'].get(month, 0) + 1 # Trends analysis trends = { 'attendance_trend': 'stable', # Could be calculated based on daily data 'punctuality_trend': 'stable', 'peak_hours': sorted(time_analysis['hourly_distribution'].items(), key=lambda x: x[1], reverse=True)[:3], 'peak_days': sorted(time_analysis['day_of_week_distribution'].items(), key=lambda x: x[1], reverse=True)[:3] } return custom_response( data={ 'total_records': total_records, 'attendance_rate': round(attendance_rate, 2), 'punctuality_rate': round(punctuality_rate, 2), 'status_breakdown': status_breakdown, 'daily_attendance': daily_attendance, 'tutor_performance': tutor_performance_list, 'course_performance': course_performance_list, 'department_performance': department_performance_list, 'time_analysis': time_analysis, 'trends': trends }, status_code=200, success=True ) except Exception as e: return custom_response( data={ 'error': f'Error generating attendance analytics: {str(e)}', 'total_records': 0, 'attendance_rate': 0, 'punctuality_rate': 0, 'status_breakdown': {}, 'daily_attendance': [], 'tutor_performance': [], 'course_performance': [], 'department_performance': [], 'time_analysis': {}, 'trends': {} }, status_code=500, success=False ) def fetchAcademicSessions(self, page=1, per_page=20, search=None, status=None, speciality_id=None, year=None): """ Fetch academic sessions with pagination and filtering """ try: with DatabaseContextManager() as ctx: query = ctx.session.query(AcademicSession).join( Speciality, AcademicSession.speciality_id == Speciality.id ) # Apply filters if search: search_filter = or_( AcademicSession.name.ilike(f'%{search}%'), AcademicSession.year.ilike(f'%{search}%'), AcademicSession.description.ilike(f'%{search}%'), Speciality.name.ilike(f'%{search}%') ) query = query.filter(search_filter) if status and status != 'all': query = query.filter(AcademicSession.status == status) if speciality_id and speciality_id != 'all': query = query.filter(AcademicSession.speciality_id == speciality_id) if year and year != 'all': query = query.filter(AcademicSession.year == year) # Get total count total_count = query.count() # Apply pagination offset = (page - 1) * per_page sessions = query.offset(offset).limit(per_page).all() # Format sessions formatted_sessions = [] for session in sessions: formatted_sessions.append({ 'id': session.id, 'name': session.name, 'year': session.year, 'start_date': session.start_date.isoformat() if session.start_date else None, 'end_date': session.end_date.isoformat() if session.end_date else None, 'is_active': session.is_active, 'status': session.status, 'description': session.description, 'created_at': session.created_at.isoformat() if session.created_at else None, 'updated_at': session.updated_at.isoformat() if session.updated_at else None, 'speciality': { 'id': session.speciality.id, 'name': session.speciality.name, 'department': session.speciality.department, 'code': session.speciality.code } }) # Calculate pagination info total_pages = (total_count + per_page - 1) // per_page has_next = page < total_pages has_prev = page > 1 return custom_response( data={ 'sessions': formatted_sessions, 'pagination': { 'current_page': page, 'per_page': per_page, 'total_items': total_count, 'total_pages': total_pages, 'has_next': has_next, 'has_prev': has_prev } }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching academic sessions: {str(e)}'}, status_code=500, success=False ) def createAcademicSession(self, session_data): """ Create a new academic session """ try: with DatabaseContextManager() as ctx: # Validate required fields required_fields = ['name', 'year', 'speciality_id', 'start_date', 'end_date'] for field in required_fields: if not session_data.get(field): return custom_response( data={'message': f'{field} is required'}, status_code=400, success=False ) # Check if speciality exists speciality = ctx.session.query(Speciality).filter( Speciality.id == session_data['speciality_id'] ).first() if not speciality: return custom_response( data={'message': 'Speciality not found'}, status_code=404, success=False ) # Check for duplicate session name in the same year existing_session = ctx.session.query(AcademicSession).filter( AcademicSession.name == session_data['name'], AcademicSession.year == session_data['year'], AcademicSession.speciality_id == session_data['speciality_id'] ).first() if existing_session: return custom_response( data={'message': 'Academic session with this name already exists for this year and speciality'}, status_code=400, success=False ) # Create new academic session new_session = AcademicSession( name=session_data['name'], year=session_data['year'], speciality_id=session_data['speciality_id'], start_date=datetime.strptime(session_data['start_date'], '%Y-%m-%d').date(), end_date=datetime.strptime(session_data['end_date'], '%Y-%m-%d').date(), description=session_data.get('description', ''), status=session_data.get('status', 'active'), is_active=session_data.get('is_active', True) ) ctx.session.add(new_session) ctx.session.commit() # Return created session return custom_response( data={ 'session': { 'id': new_session.id, 'name': new_session.name, 'year': new_session.year, 'start_date': new_session.start_date.isoformat() if new_session.start_date else None, 'end_date': new_session.end_date.isoformat() if new_session.end_date else None, 'is_active': new_session.is_active, 'status': new_session.status, 'description': new_session.description, 'created_at': new_session.created_at.isoformat() if new_session.created_at else None, 'speciality': { 'id': speciality.id, 'name': speciality.name, 'department': speciality.department, 'code': speciality.code } } }, status_code=201, success=True ) except Exception as e: return custom_response( data={'message': f'Error creating academic session: {str(e)}'}, status_code=500, success=False ) def updateAcademicSession(self, session_id, session_data): """ Update an existing academic session """ try: with DatabaseContextManager() as ctx: session = ctx.session.query(AcademicSession).filter( AcademicSession.id == session_id ).first() if not session: return custom_response( data={'message': 'Academic session not found'}, status_code=404, success=False ) # Update fields if 'name' in session_data: session.name = session_data['name'] if 'year' in session_data: session.year = session_data['year'] if 'speciality_id' in session_data: # Validate speciality exists speciality = ctx.session.query(Speciality).filter( Speciality.id == session_data['speciality_id'] ).first() if not speciality: return custom_response( data={'message': 'Speciality not found'}, status_code=404, success=False ) session.speciality_id = session_data['speciality_id'] if 'start_date' in session_data: session.start_date = datetime.strptime(session_data['start_date'], '%Y-%m-%d').date() if 'end_date' in session_data: session.end_date = datetime.strptime(session_data['end_date'], '%Y-%m-%d').date() if 'description' in session_data: session.description = session_data['description'] if 'status' in session_data: session.status = session_data['status'] if 'is_active' in session_data: session.is_active = session_data['is_active'] session.updated_at = datetime.utcnow() ctx.session.commit() # Return updated session return custom_response( data={ 'session': { 'id': session.id, 'name': session.name, 'year': session.year, 'start_date': session.start_date.isoformat() if session.start_date else None, 'end_date': session.end_date.isoformat() if session.end_date else None, 'is_active': session.is_active, 'status': session.status, 'description': session.description, 'updated_at': session.updated_at.isoformat() if session.updated_at else None, 'speciality': { 'id': session.speciality.id, 'name': session.speciality.name, 'department': session.speciality.department, 'code': session.speciality.code } } }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error updating academic session: {str(e)}'}, status_code=500, success=False ) def deleteAcademicSession(self, session_id): """ Delete an academic session """ try: with DatabaseContextManager() as ctx: session = ctx.session.query(AcademicSession).filter( AcademicSession.id == session_id ).first() if not session: return custom_response( data={'message': 'Academic session not found'}, status_code=404, success=False ) # Check if session has associated courses or students course_count = ctx.session.query(Course).filter( Course.academic_session_id == session_id ).count() if course_count > 0: return custom_response( data={'message': 'Cannot delete academic session with associated courses'}, status_code=400, success=False ) ctx.session.delete(session) ctx.session.commit() return custom_response( data={'message': 'Academic session deleted successfully'}, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error deleting academic session: {str(e)}'}, status_code=500, success=False ) def getSpecialitiesForSessions(self): """ Get all specialities for academic session creation """ try: with DatabaseContextManager() as ctx: specialities = ctx.session.query(Speciality).filter( Speciality.is_active == True ).all() formatted_specialities = [] for speciality in specialities: formatted_specialities.append({ 'id': speciality.id, 'name': speciality.name, 'department': speciality.department, 'code': speciality.code, 'description': speciality.description }) return custom_response( data={'specialities': formatted_specialities}, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching specialities: {str(e)}'}, status_code=500, success=False ) def getAcademicSessionStats(self): """ Get academic session statistics """ try: with DatabaseContextManager() as ctx: total_sessions = ctx.session.query(AcademicSession).count() active_sessions = ctx.session.query(AcademicSession).filter( AcademicSession.is_active == True ).count() completed_sessions = ctx.session.query(AcademicSession).filter( AcademicSession.status == 'completed' ).count() current_year_sessions = ctx.session.query(AcademicSession).filter( AcademicSession.year == str(datetime.now().year) ).count() return custom_response( data={ 'total_sessions': total_sessions, 'active_sessions': active_sessions, 'completed_sessions': completed_sessions, 'current_year_sessions': current_year_sessions }, status_code=200, success=True ) except Exception as e: return custom_response( data={'message': f'Error fetching academic session stats: {str(e)}'}, status_code=500, success=False ) def update_admin_password(self, admin_email, new_password, current_password=None, require_current_password=True): """Update admin password with optional current password verification""" try: with DatabaseContextManager() as ctx: # Find the admin user admin_user = ctx.session.query(Admin).filter( Admin.email == admin_email ).first() if not admin_user: return custom_response( success=False, data="Admin user not found", status_code=404 ) # Verify current password if required if require_current_password and current_password: if not check_password(current_password, admin_user.password_hash, salt=current_app.config['SECRET_KEY']): return custom_response( success=False, data="Current password is incorrect", status_code=400 ) # Validate new password strength if len(new_password) < 8: return custom_response( success=False, data="Password must be at least 8 characters long", status_code=400 ) # Hash the new password using the same method as login password_hash = hash_password(new_password, salt=current_app.config['SECRET_KEY']) # Update the password admin_user.password_hash = password_hash # Commit the changes ctx.session.commit() # Log the password change current_app.logger.info(f"Admin password updated for user: {admin_user.email}") return custom_response( success=True, data={ "message": "Password updated successfully", "admin_id": admin_user.id, "email": admin_user.email }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error updating admin password: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to update password due to server error", status_code=500 ) def reset_admin_password(self, admin_email, new_password): """Reset admin password without requiring current password (for system use)""" try: with DatabaseContextManager() as ctx: # Find the admin user admin_user = ctx.session.query(Admin).filter( Admin.email == admin_email ).first() if not admin_user: return custom_response( success=False, data="Admin user not found", status_code=404 ) # Validate new password strength if len(new_password) < 8: return custom_response( success=False, data="Password must be at least 8 characters long", status_code=400 ) # Hash the new password using the same method as login password_hash = hash_password(new_password, current_app.config['SECRET_KEY']) # Update the password admin_user.password_hash = password_hash # Commit the changes ctx.session.commit() # Log the password reset current_app.logger.warning(f"Admin password reset for user: {admin_user.email}") return custom_response( success=True, data={ "message": "Password reset successfully", "admin_id": admin_user.id, "email": admin_user.email }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error resetting admin password: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to reset password due to server error", status_code=500 ) def get_tutor_workload_analysis(self): """Get comprehensive tutor workload analysis based on course credits""" try: with DatabaseContextManager() as ctx: # Simple query to get all tutors first tutors = ctx.session.query(Tutor).all() # For now, return mock data with proper structure workload_analysis = [] for tutor in tutors: # Get courses for this tutor tutor_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, Course.is_archived == False ).all() total_credits = sum(course.credits or 0 for course in tutor_courses) # Determine workload status based on credit hours if total_credits < 24: workload_status = 'underloaded' efficiency_score = min(95, 70 + (total_credits * 1.5)) elif total_credits > 28: workload_status = 'overloaded' efficiency_score = max(75, 95 - ((total_credits - 28) * 2)) else: workload_status = 'balanced' efficiency_score = 90 # Calculate student count (mock data for now) estimated_students = len(tutor_courses) * 25 workload_analysis.append({ 'tutor_id': tutor.id, 'name': f"{tutor.first_name} {tutor.last_name}", 'email': tutor.email, 'department': tutor.department or 'Unknown', 'courses': [ { 'course_id': course.id, 'course_code': course.code, 'course_title': course.title, 'credits': course.credits or 0, 'is_primary': True # Simplified for now } for course in tutor_courses ], 'total_credits': total_credits, 'total_courses': len(tutor_courses), 'primary_courses': len(tutor_courses), 'secondary_courses': 0, 'estimated_students': estimated_students, 'workload_status': workload_status, 'efficiency_score': round(efficiency_score, 1) }) # Sort by total credits (descending) workload_analysis.sort(key=lambda x: x['total_credits'], reverse=True) # Calculate summary statistics summary = { 'total_tutors': len(workload_analysis), 'underloaded_count': len([t for t in workload_analysis if t['workload_status'] == 'underloaded']), 'balanced_count': len([t for t in workload_analysis if t['workload_status'] == 'balanced']), 'overloaded_count': len([t for t in workload_analysis if t['workload_status'] == 'overloaded']), 'average_efficiency': round( sum(t['efficiency_score'] for t in workload_analysis) / len(workload_analysis), 1 ) if workload_analysis else 0 } return custom_response( success=True, data={ 'tutor_workloads': workload_analysis, 'department_stats': {}, # Simplified for now 'summary': summary }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching tutor workload analysis: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to fetch tutor workload analysis due to server error", status_code=500 ) def get_course_analytics(self): """Get comprehensive course analytics including enrollments and speciality data""" try: with DatabaseContextManager() as ctx: # Get all courses with their relationships courses = ctx.session.query(Course).options( joinedload(Course.enrollments), joinedload(Course.tutors), joinedload(Course.supervisor), joinedload(Course.academic_session) ).all() course_analytics = [] for course in courses: try: # Calculate enrollment metrics enrolled_students = len(course.enrollments) enrollment_rate = (enrolled_students / course.max_students * 100) if course.max_students > 0 else 0 # Mock completion rate and average grade (in real implementation, these would come from actual data) completion_rate = min(95, max(60, 80 + (enrollment_rate - 50) * 0.3)) average_grade = min(95, max(65, 75 + (enrollment_rate - 50) * 0.2)) # Generate enrollment trend (last 6 months) enrollment_trend = [] for i in range(6): month_name = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun'][i] # Simulate gradual enrollment growth enrollments = max(1, int(enrolled_students * (0.3 + (i + 1) * 0.1))) completions = max(0, int(enrollments * 0.1)) if i >= 3 else 0 enrollment_trend.append({ 'month': month_name, 'enrollments': enrollments, 'completions': completions }) # Generate grade distribution grade_distribution = [ {'grade': 'A', 'count': int(enrolled_students * 0.25), 'percentage': 25.0}, {'grade': 'B', 'count': int(enrolled_students * 0.35), 'percentage': 35.0}, {'grade': 'C', 'count': int(enrolled_students * 0.25), 'percentage': 25.0}, {'grade': 'D', 'count': int(enrolled_students * 0.10), 'percentage': 10.0}, {'grade': 'F', 'count': int(enrolled_students * 0.05), 'percentage': 5.0} ] # Get speciality enrollments speciality_enrollments = [] speciality_map = {} for enrollment in course.enrollments: if enrollment.speciality: spec_id = enrollment.speciality.id if spec_id not in speciality_map: speciality_map[spec_id] = { 'speciality_id': spec_id, 'speciality_name': enrollment.speciality.name, 'department': enrollment.speciality.department, 'enrollment_count': 0, 'completion_rate': completion_rate } speciality_map[spec_id]['enrollment_count'] += 1 speciality_enrollments = list(speciality_map.values()) # If no speciality enrollments, create a default one if not speciality_enrollments: speciality_enrollments = [{ 'speciality_id': 'general', 'speciality_name': 'General Studies', 'department': course.department or 'Unknown', 'enrollment_count': enrolled_students, 'completion_rate': completion_rate }] course_data = { 'id': course.id, 'code': course.code, 'title': course.title, 'department': course.department or 'Unknown', 'description': course.description or '', 'credits': course.credits or 0, 'max_students': course.max_students or 0, 'enrolled_students': enrolled_students, 'enrollment_rate': round(enrollment_rate, 1), 'completion_rate': round(completion_rate, 1), 'average_grade': round(average_grade, 1), 'total_hours': course.total_hours or 0, 'semester': course.semester or 'Not Specified', 'course_level': course.course_level or 'Not Specified', 'is_active': course.is_active, 'created_at': course.created_at.isoformat() if course.created_at else '', 'tutors': [f"{tutor.first_name} {tutor.last_name}" for tutor in course.tutors], 'supervisor': { 'id': course.supervisor.id, 'name': f"{course.supervisor.first_name} {course.supervisor.last_name}", 'email': course.supervisor.email, 'staff_id': course.supervisor.staff_id } if course.supervisor else None, 'enrollment_trend': enrollment_trend, 'grade_distribution': grade_distribution, 'speciality_enrollments': speciality_enrollments } course_analytics.append(course_data) except Exception as e: current_app.logger.error(f"Error processing course {course.id}: {str(e)}") continue return custom_response( success=True, data=course_analytics, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching course analytics: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to fetch course analytics due to server error", status_code=500 ) def generate_admin_password_reset_token(self, admin_email): """Generate a password reset token for admin""" try: with DatabaseContextManager() as ctx: # Find the admin user admin_user = ctx.session.query(Admin).filter( Admin.email == admin_email ).first() if not admin_user: return custom_response( success=False, data="Admin user not found", status_code=404 ) # Send password reset notification email try: send_email( to_email=admin_user.email, subject="Password Reset Request", body=f""" Hello {admin_user.first_name}, You have requested a password reset for your admin account. Please contact your system administrator to reset your password. If you did not request this reset, please ignore this email. Best regards, Tutor Management System """ ) except Exception as email_error: current_app.logger.error(f"Failed to send reset email: {str(email_error)}") return custom_response( success=True, data={ "message": "Password reset notification sent", "admin_id": admin_user.id, "email": admin_user.email }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error generating reset token: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to generate reset token due to server error", status_code=500 ) def bulk_upload_students(self, file): """ Bulk upload students from an Excel file. Expected columns: first_name, last_name, email, phone, student_id, speciality_id, department, year_of_study, program, current_semester, student_category Optional columns: gender, date_of_birth, nationality, guardian_name, guardian_contact, emergency_contact, address, city, country, postal_code The method will: 1. Parse the Excel file 2. Validate required fields 3. Generate random passwords for each student 4. Create student accounts 5. Send credentials via email Args: file: Excel file object from request.files Returns: Response with upload results and statistics """ import secrets import string from io import BytesIO with DatabaseContextManager() as ctx: try: # Read Excel file file_content = file.read() df = pd.read_excel(BytesIO(file_content)) # Clean column names (remove extra spaces, lowercase) df.columns = df.columns.str.strip().str.lower().str.replace(' ', '_') # Define required columns required_columns = [ 'first_name', 'last_name', 'email', 'student_id', 'speciality_id', 'department', 'year_of_study', 'program' ] # Check for required columns missing_columns = [col for col in required_columns if col not in df.columns] if missing_columns: return custom_response( success=False, data=f"Missing required columns: {', '.join(missing_columns)}", status_code=400 ) # Track results successful_uploads = [] failed_uploads = [] email_results = [] # Process each row for index, row in df.iterrows(): try: row_number = index + 2 # +2 because index starts at 0 and Excel has header row # Validate required fields validation_errors = [] for col in required_columns: if pd.isna(row[col]) or str(row[col]).strip() == '': validation_errors.append(f"{col} is required") # Validate email format - must be @kisiwa.ac.ke email_str = str(row.get('email', '')).strip().lower() # Additional email format validation if email_str and not email_str.count('@') == 1: validation_errors.append('Invalid email format') if validation_errors: failed_uploads.append({ 'row': row_number, 'student_id': row.get('student_id', 'N/A'), 'name': f"{row.get('first_name', '')} {row.get('last_name', '')}", 'errors': validation_errors }) continue # Check if email already exists existing_user = ctx.session.query(User).filter( User.email == email_str ).first() if existing_user: failed_uploads.append({ 'row': row_number, 'student_id': str(row['student_id']), 'name': f"{row['first_name']} {row['last_name']}", 'errors': ['Email already exists in the system'] }) continue # Check if student_id already exists existing_student = ctx.session.query(Student).filter( Student.student_id == str(row['student_id']).strip() ).first() if existing_student: failed_uploads.append({ 'row': row_number, 'student_id': str(row['student_id']), 'name': f"{row['first_name']} {row['last_name']}", 'errors': ['Student ID already exists'] }) continue # Validate speciality exists speciality = ctx.session.query(Speciality).filter( Speciality.id == str(row['speciality_id']).strip() ).first() if not speciality: failed_uploads.append({ 'row': row_number, 'student_id': str(row['student_id']), 'name': f"{row['first_name']} {row['last_name']}", 'errors': ['Invalid speciality_id'] }) continue # Generate random password (12 characters with mix of letters, numbers, symbols) password_chars = string.ascii_letters + string.digits + "!@#$%^&*" random_password = ''.join(secrets.choice(password_chars) for _ in range(12)) # Hash password password_hash = hash_password(random_password, salt=current_app.config['SECRET_KEY']) # Create student account student_data = { 'id': str(uuid.uuid4()), 'email': str(row['email']).strip().lower(), 'password_hash': password_hash, 'first_name': str(row['first_name']).strip(), 'last_name': str(row['last_name']).strip(), 'phone': str(row.get('phone', '')).strip() if not pd.isna(row.get('phone')) else None, 'user_type': 'student', 'is_active': True, 'student_id': str(row['student_id']).strip(), 'speciality_id': str(row['speciality_id']).strip(), 'department': str(row['department']).strip(), 'year_of_study': int(row['year_of_study']), 'program': str(row['program']).strip(), 'current_semester': str(row.get('current_semester', '')).strip() if not pd.isna(row.get('current_semester')) else None, 'student_category': str(row.get('student_category', 'regular')).strip() if not pd.isna(row.get('student_category')) else 'regular', 'must_change_password': True # Force password change on first login } # Add optional fields if present optional_fields = { 'gender': 'gender', 'date_of_birth': 'date_of_birth', 'nationality': 'nationality', 'guardian_name': 'guardian_name', 'guardian_contact': 'guardian_contact', 'emergency_contact': 'emergency_contact', 'address': 'address', 'city': 'city', 'country': 'country', 'postal_code': 'postal_code' } for excel_col, db_field in optional_fields.items(): if excel_col in df.columns and not pd.isna(row.get(excel_col)): student_data[db_field] = str(row[excel_col]).strip() # Create student student = Student(**student_data) ctx.session.add(student) # Create notification preferences notification_pref = NotificationPreference( id=str(uuid.uuid4()), user_id=student.id, receive_email=True, receive_sms=False, receive_push=True ) ctx.session.add(notification_pref) # Commit to save the student ctx.session.flush() successful_uploads.append({ 'row': row_number, 'student_id': str(row['student_id']), 'name': f"{row['first_name']} {row['last_name']}", 'email': str(row['email']).strip().lower(), 'password': random_password }) # Try to send email with credentials try: email_subject = "Welcome to the Tutor Management System - Your Account Details" email_body = f"""

Welcome to Tutor Management System

Dear {row['first_name']} {row['last_name']},

Your student account has been successfully created. Below are your login credentials:

Student ID: {row['student_id']}
Email: {row['email']}
Temporary Password: {random_password}
Program: {row['program']}

Important Security Notice:

  • You will be required to change your password on first login
  • Please keep your password secure and do not share it with anyone
  • If you did not request this account, please contact the administration immediately

You can access the system at: {current_app.config.get('FRONTEND_URL', 'https://lms.mutabletech.co.ke')}

Best regards,
Tutor Management System Administration

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

""" # Get email configuration sender_email = os.environ.get('NO_REPLY_EMAIL') or current_app.config.get('MAIL_SENDER') sender_password = os.environ.get('NO_REPLY_PASSWORD') or current_app.config.get('MAIL_PASSWORD') send_email( sender_email=sender_email, sender_password=sender_password, receiver_email=str(row['email']).strip().lower(), subject=email_subject, message=email_body ) email_results.append({ 'student_id': str(row['student_id']), 'email': str(row['email']).strip().lower(), 'status': 'sent' }) except Exception as email_error: current_app.logger.error(f"Error sending email to {row['email']}: {str(email_error)}") email_results.append({ 'student_id': str(row['student_id']), 'email': str(row['email']).strip().lower(), 'status': 'failed', 'error': str(email_error) }) except Exception as row_error: failed_uploads.append({ 'row': row_number, 'student_id': row.get('student_id', 'N/A'), 'name': f"{row.get('first_name', '')} {row.get('last_name', '')}", 'errors': [str(row_error)] }) # Commit all changes ctx.session.commit() # Calculate statistics total_rows = len(df) successful_count = len(successful_uploads) failed_count = len(failed_uploads) emails_sent = len([e for e in email_results if e['status'] == 'sent']) emails_failed = len([e for e in email_results if e['status'] == 'failed']) return custom_response( success=True, data={ 'summary': { 'total_rows': total_rows, 'successful_uploads': successful_count, 'failed_uploads': failed_count, 'success_rate': round((successful_count / total_rows) * 100, 1) if total_rows > 0 else 0, 'emails_sent': emails_sent, 'emails_failed': emails_failed }, 'successful_students': successful_uploads, 'failed_students': failed_uploads, 'email_results': email_results, 'message': f"Successfully uploaded {successful_count} out of {total_rows} students. {emails_sent} emails sent successfully." }, status_code=200 ) except Exception as e: ctx.session.rollback() current_app.logger.error(f"Error in bulk student upload: {str(e)}", exc_info=True) return custom_response( success=False, data=f"Failed to process Excel file: {str(e)}", status_code=500 ) def generate_student_upload_template(self): """ Generate an Excel template for bulk student upload with data validation. This template includes dropdown menus for departments and email validation. Returns: Excel file download response """ with DatabaseContextManager() as ctx: try: # Get all active specialities for reference specialities = ctx.session.query(Speciality).filter( Speciality.is_active == True ).all() # Get all unique departments departments = list(set([s.department for s in specialities if s.department])) departments.sort() # Create sample data with instructions template_data = { 'first_name': ['John', 'Jane', 'Michael'], 'last_name': ['Doe', 'Smith', 'Johnson'], 'email': ['john.doe@kisiwa.ac.ke', 'jane.smith@kisiwa.ac.ke', 'michael.johnson@kisiwa.ac.ke'], 'phone': ['+254712345678', '+254723456789', '+254734567890'], 'student_id': ['STU001', 'STU002', 'STU003'], 'speciality_id': ['', '', ''], 'department': ['Business', 'Hospitality', 'IT'], 'year_of_study': [1, 2, 3], 'program': ['Diploma', 'Certificate', 'Craft'], 'current_semester': ['Semester 1', 'Semester 2', 'Semester 1'], 'student_category': ['regular', 'part-time', 'regular'], 'gender': ['Male', 'Female', 'Male'], 'nationality': ['Kenyan', 'Kenyan', 'Kenyan'], 'guardian_name': ['Parent Name', 'Guardian Name', 'Parent Name'], 'guardian_contact': ['+254700000001', '+254700000002', '+254700000003'], 'emergency_contact': ['+254700000004', '+254700000005', '+254700000006'], 'address': ['123 Main St', '456 Oak Ave', '789 Pine Rd'], 'city': ['Nairobi', 'Mombasa', 'Kisumu'], 'country': ['Kenya', 'Kenya', 'Kenya'], 'postal_code': ['00100', '80100', '40100'] } # Create DataFrame df = pd.DataFrame(template_data) # Create Excel file in memory output = BytesIO() with pd.ExcelWriter(output, engine='openpyxl') as writer: df.to_excel(writer, index=False, sheet_name='Students') # Get the workbook and worksheet workbook = writer.book worksheet = writer.sheets['Students'] # Add department dropdown validation (column G) if departments: # Create a comma-separated list of departments dept_list = ','.join(departments) # Add data validation for department column (G2 to G1000) dept_validation = DataValidation( type="list", formula1=f'"{dept_list}"', allow_blank=False ) dept_validation.error = 'Please select a department from the dropdown list' dept_validation.errorTitle = 'Invalid Department' dept_validation.prompt = 'Select a department from the list' dept_validation.promptTitle = 'Department Selection' # Apply validation to department column worksheet.add_data_validation(dept_validation) dept_validation.add('G2:G1000') # Add email validation pattern (column C) email_validation = DataValidation( type="custom", formula1='AND(ISNUMBER(FIND("@kisiwa.ac.ke",C2)),LEN(C2)-LEN(SUBSTITUTE(C2,"@",""))=1)', allow_blank=False ) email_validation.error = 'Email must be in format: name@kisiwa.ac.ke' email_validation.errorTitle = 'Invalid Email Format' email_validation.prompt = 'Enter email in format: firstname.lastname@kisiwa.ac.ke' email_validation.promptTitle = 'Email Format' worksheet.add_data_validation(email_validation) email_validation.add('C2:C1000') # Add instructions sheet instructions_data = { 'Column Name': [ 'first_name', 'last_name', 'email', 'phone', 'student_id', 'speciality_id', 'department', 'year_of_study', 'program', 'current_semester', 'student_category', 'gender', 'nationality', 'guardian_name', 'guardian_contact', 'emergency_contact', 'address', 'city', 'country', 'postal_code' ], 'Required': [ 'Yes', 'Yes', 'Yes', 'No', 'Yes', 'Yes', 'Yes', 'Yes', 'Yes', 'No', 'No', 'No', 'No', 'No', 'No', 'No', 'No', 'No', 'No', 'No' ], 'Description': [ 'Student first name', 'Student last name', 'Student email (MUST be @kisiwa.ac.ke format)', 'Student phone number (format: +254...)', 'Student ID (must be unique)', 'UUID of the speciality - See Specialities sheet', 'Department name - SELECT from dropdown', 'Year of study (1, 2, 3, 4, etc.)', 'Program name (e.g., Diploma in Business)', 'Current semester (e.g., Semester 1, Semester 2)', 'Student category (regular, part-time, international)', 'Student gender (Male, Female, Other)', 'Student nationality', 'Guardian or parent name', 'Guardian contact number', 'Emergency contact number', 'Student residential address', 'City of residence', 'Country of residence', 'Postal code' ], 'Example': [ 'John', 'Doe', 'john.doe@kisiwa.ac.ke', '+254712345678', 'STU001', 'Copy from Specialities sheet', 'Select from dropdown →', '1', 'Diploma in Business Management', 'Semester 1', 'regular', 'Male', 'Kenyan', 'Parent Name', '+254700000000', '+254700000001', '123 Main Street', 'Nairobi', 'Kenya', '00100' ] } df_instructions = pd.DataFrame(instructions_data) df_instructions.to_excel(writer, index=False, sheet_name='Instructions') # Add Specialities reference sheet specialities_data = { 'Speciality ID': [s.id for s in specialities], 'Speciality Name': [s.name for s in specialities], 'Department': [s.department for s in specialities], 'Code': [s.code or '' for s in specialities], 'Level': [s.speciality_level or '' for s in specialities], 'Description': [s.description or '' for s in specialities] } df_specialities = pd.DataFrame(specialities_data) df_specialities.to_excel(writer, index=False, sheet_name='Specialities') # Format the Specialities sheet spec_worksheet = writer.sheets['Specialities'] # Auto-adjust column widths for column in spec_worksheet.columns: max_length = 0 column_letter = column[0].column_letter for cell in column: try: if len(str(cell.value)) > max_length: max_length = len(str(cell.value)) except: pass adjusted_width = min(max_length + 2, 50) spec_worksheet.column_dimensions[column_letter].width = adjusted_width output.seek(0) # Get the bytes from BytesIO buffer excel_bytes = output.getvalue() output.close() # Create a Response object with the Excel file bytes response = Response( excel_bytes, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers={ 'Content-Disposition': 'attachment; filename=student_upload_template.xlsx', 'Content-Length': str(len(excel_bytes)) } ) return response except Exception as e: current_app.logger.error(f"Error generating template: {str(e)}", exc_info=True) return custom_response( success=False, data=f"Failed to generate template: {str(e)}", status_code=500 ) def get_attendance_analytics(self): """Get comprehensive attendance analytics with visualizations and insights""" try: with DatabaseContextManager() as ctx: # Get all attendance records with related data attendance_records = ctx.session.query(Attendance).options( joinedload(Attendance.session), joinedload(Attendance.student) ).all() # Get all daily teaching sessions daily_sessions = ctx.session.query(DailyTeachingSession).options( joinedload(DailyTeachingSession.teaching_session), joinedload(DailyTeachingSession.course) ).all() # Calculate comprehensive analytics analytics_data = self._calculate_attendance_analytics(attendance_records, daily_sessions) return custom_response( success=True, data=analytics_data, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching attendance analytics: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to fetch attendance analytics due to server error", status_code=500 ) def _calculate_attendance_analytics(self, attendance_records, daily_sessions): """Calculate comprehensive attendance analytics""" from datetime import datetime, timedelta from collections import defaultdict import statistics # Initialize data structures total_attendance_records = len(attendance_records) total_sessions = len(daily_sessions) # Calculate overall attendance rate present_count = sum(1 for record in attendance_records if record.status == 'present') overall_attendance_rate = (present_count / total_attendance_records * 100) if total_attendance_records > 0 else 0 # Daily attendance trends (last 30 days) thirty_days_ago = datetime.now() - timedelta(days=30) daily_trends = defaultdict(lambda: {'total': 0, 'present': 0, 'absent': 0, 'late': 0}) for record in attendance_records: if record.session and record.session.session_date >= thirty_days_ago.date(): date_key = record.session.session_date.isoformat() daily_trends[date_key]['total'] += 1 daily_trends[date_key][record.status] += 1 # Convert to list format for charts daily_trend_data = [] for date_str in sorted(daily_trends.keys()): data = daily_trends[date_str] daily_trend_data.append({ 'date': date_str, 'total': data['total'], 'present': data['present'], 'absent': data['absent'], 'late': data['late'], 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 }) # Weekly attendance patterns weekly_patterns = defaultdict(lambda: {'total': 0, 'present': 0}) for record in attendance_records: if record.session and record.session.session_date: weekday = record.session.session_date.strftime('%A') weekly_patterns[weekday]['total'] += 1 if record.status == 'present': weekly_patterns[weekday]['present'] += 1 weekly_data = [] weekdays = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] for day in weekdays: if day in weekly_patterns: data = weekly_patterns[day] weekly_data.append({ 'day': day, 'total': data['total'], 'present': data['present'], 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 }) # Course-wise attendance analysis course_attendance = defaultdict(lambda: {'total': 0, 'present': 0, 'students': set()}) for record in attendance_records: if record.session and record.session.course: course_id = record.session.course.id course_attendance[course_id]['total'] += 1 course_attendance[course_id]['students'].add(record.student_id) if record.status == 'present': course_attendance[course_id]['present'] += 1 course_data = [] for course_id, data in course_attendance.items(): # Find course details course = next((s.course for s in daily_sessions if s.course and s.course.id == course_id), None) if course: course_data.append({ 'course_id': course_id, 'course_title': course.title, 'course_code': course.code, 'department': course.department, 'total_records': data['total'], 'present_records': data['present'], 'unique_students': len(data['students']), 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 }) # Student attendance performance student_attendance = defaultdict(lambda: {'total': 0, 'present': 0, 'courses': set()}) for record in attendance_records: student_id = record.student_id student_attendance[student_id]['total'] += 1 if record.status == 'present': student_attendance[student_id]['present'] += 1 if record.session and record.session.course: student_attendance[student_id]['courses'].add(record.session.course.id) # Top and bottom performing students student_performance = [] for student_id, data in student_attendance.items(): if data['total'] >= 5: # Only include students with at least 5 attendance records student_performance.append({ 'student_id': student_id, 'total_records': data['total'], 'present_records': data['present'], 'courses_count': len(data['courses']), 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 }) # Sort by attendance rate student_performance.sort(key=lambda x: x['attendance_rate'], reverse=True) top_students = student_performance[:10] bottom_students = student_performance[-10:] if len(student_performance) >= 10 else [] # Monthly attendance trends monthly_trends = defaultdict(lambda: {'total': 0, 'present': 0}) for record in attendance_records: if record.session and record.session.session_date: month_key = record.session.session_date.strftime('%Y-%m') monthly_trends[month_key]['total'] += 1 if record.status == 'present': monthly_trends[month_key]['present'] += 1 monthly_data = [] for month_str in sorted(monthly_trends.keys()): data = monthly_trends[month_str] monthly_data.append({ 'month': month_str, 'total': data['total'], 'present': data['present'], 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 }) # Department-wise analysis department_attendance = defaultdict(lambda: {'total': 0, 'present': 0, 'courses': set()}) for record in attendance_records: if record.session and record.session.course: dept = record.session.course.department department_attendance[dept]['total'] += 1 department_attendance[dept]['courses'].add(record.session.course.id) if record.status == 'present': department_attendance[dept]['present'] += 1 department_data = [] for dept, data in department_attendance.items(): department_data.append({ 'department': dept, 'total_records': data['total'], 'present_records': data['present'], 'courses_count': len(data['courses']), 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 }) # Attendance status distribution status_distribution = defaultdict(int) for record in attendance_records: status_distribution[record.status] += 1 status_data = [ {'status': 'Present', 'count': status_distribution['present'], 'color': '#10b981'}, {'status': 'Absent', 'count': status_distribution['absent'], 'color': '#ef4444'}, {'status': 'Late', 'count': status_distribution['late'], 'color': '#f59e0b'}, {'status': 'Excused', 'count': status_distribution['excused'], 'color': '#6b7280'} ] # Time-based patterns (hourly attendance) hourly_patterns = defaultdict(lambda: {'total': 0, 'present': 0}) for session in daily_sessions: if session.start_time: hour = session.start_time.hour # Count attendance for this session session_attendance = [r for r in attendance_records if r.session_id == session.id] hourly_patterns[hour]['total'] += len(session_attendance) hourly_patterns[hour]['present'] += sum(1 for r in session_attendance if r.status == 'present') hourly_data = [] for hour in range(24): if hour in hourly_patterns: data = hourly_patterns[hour] hourly_data.append({ 'hour': f"{hour:02d}:00", 'total': data['total'], 'present': data['present'], 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 }) # Tutor-wise attendance analysis tutor_attendance = defaultdict(lambda: {'total': 0, 'present': 0, 'courses': set(), 'sessions': 0}) for session in daily_sessions: if session.teaching_session and session.teaching_session.tutor: tutor_id = session.teaching_session.tutor.id tutor = session.teaching_session.tutor # Count attendance for this session session_attendance = [r for r in attendance_records if r.session_id == session.id] tutor_attendance[tutor_id]['total'] += len(session_attendance) tutor_attendance[tutor_id]['present'] += sum(1 for r in session_attendance if r.status == 'present') tutor_attendance[tutor_id]['sessions'] += 1 tutor_attendance[tutor_id]['tutor_name'] = f"{tutor.first_name} {tutor.last_name}" tutor_attendance[tutor_id]['staff_id'] = tutor.staff_id tutor_attendance[tutor_id]['department'] = tutor.departments[0].department_name if tutor.departments else 'Unknown' if session.course: tutor_attendance[tutor_id]['courses'].add(session.course.id) tutor_data = [] for tutor_id, data in tutor_attendance.items(): tutor_data.append({ 'tutor_id': tutor_id, 'tutor_name': data['tutor_name'], 'staff_id': data['staff_id'], 'department': data['department'], 'total_records': data['total'], 'present_records': data['present'], 'sessions_count': data['sessions'], 'courses_count': len(data['courses']), 'attendance_rate': (data['present'] / data['total'] * 100) if data['total'] > 0 else 0 }) # Sort by attendance rate tutor_data.sort(key=lambda x: x['attendance_rate'], reverse=True) return { 'summary': { 'total_attendance_records': total_attendance_records, 'total_sessions': total_sessions, 'overall_attendance_rate': round(overall_attendance_rate, 2), 'present_count': present_count, 'absent_count': status_distribution['absent'], 'late_count': status_distribution['late'], 'excused_count': status_distribution['excused'], 'unique_students': len(student_attendance), 'unique_courses': len(course_attendance), 'departments_count': len(department_attendance) }, 'daily_trends': daily_trend_data, 'weekly_patterns': weekly_data, 'monthly_trends': monthly_data, 'course_analysis': course_data, 'department_analysis': department_data, 'student_performance': { 'top_performers': top_students, 'bottom_performers': bottom_students }, 'status_distribution': status_data, 'hourly_patterns': hourly_data, 'tutor_analysis': tutor_data, 'insights': { 'best_day': max(weekly_data, key=lambda x: x['attendance_rate'])['day'] if weekly_data else None, 'worst_day': min(weekly_data, key=lambda x: x['attendance_rate'])['day'] if weekly_data else None, 'best_course': max(course_data, key=lambda x: x['attendance_rate']) if course_data else None, 'worst_course': min(course_data, key=lambda x: x['attendance_rate']) if course_data else None, 'best_department': max(department_data, key=lambda x: x['attendance_rate']) if department_data else None, 'worst_department': min(department_data, key=lambda x: x['attendance_rate']) if department_data else None } }