import pandas as pd from io import BytesIO from datetime import datetime, timedelta from flask import current_app, send_file from werkzeug.utils import secure_filename import os import uuid from typing import Dict from sqlalchemy import or_, and_ from src.utils import custom_response, ApiABC, send_email, send_email_with_attachment from src.models.models import ( NotificationPreference, Student, Enrollment, Attendance, AttendanceStatus,Assignment, Course,Supervisor, tutor_course_association, AssignmentSubmission, CourseProgress, Tutor, TeachingSession, SubmissionStatus, AcademicSession ) from src.models import DatabaseContextManager from sqlalchemy import func, extract, and_, or_, case from sqlalchemy.orm import joinedload class EnrollmentManager(ApiABC): def __init__(self): self.table = Enrollment def create(self, payload: Dict) -> Dict: """Enroll a student in a course with validation""" with DatabaseContextManager() as ctx: # Check if student exists and is active student = ctx.session.query(Student).filter( Student.id == payload['student_id'], Student.is_active == True ).first() if not student: return custom_response( success=False, data="Student not found or inactive", status_code=404 ) # Check if course exists and is active course = ctx.session.query(Course).filter( Course.id == payload['course_id'], Course.is_active == True ).first() if not course: return custom_response( success=False, data="Course not found or inactive", status_code=404 ) # Check if already enrolled existing_enrollment = ctx.session.query(Enrollment).filter( Enrollment.student_id == payload['student_id'], Enrollment.course_id == payload['course_id'] ).first() if existing_enrollment: return custom_response( success=False, data="Student is already enrolled in this course", status_code=400 ) # Check course capacity current_enrollments = ctx.session.query(Enrollment).filter( Enrollment.course_id == payload['course_id'], Enrollment.status == 'active' ).count() if course.max_students and current_enrollments >= course.max_students: return custom_response( success=False, data="Course has reached maximum capacity", status_code=400 ) # Create enrollment enrollment = Enrollment( id=str(uuid.uuid4()), student_id=payload['student_id'], course_id=payload['course_id'], enrollment_date=payload.get('enrollment_date', datetime.utcnow().date()), status=payload.get('status', 'active'), enrollment_type=payload.get('enrollment_type', 'regular'), is_audit=payload.get('is_audit', False) ) ctx.session.add(enrollment) # Create initial progress record progress = CourseProgress( id=str(uuid.uuid4()), enrollment_id=enrollment.id, student_id=payload['student_id'], completion_percentage=0.0, attendance_rate=0.0 ) ctx.session.add(progress) ctx.session.commit() # Send notifications self._send_enrollment_notifications(enrollment, ctx) return custom_response( success=True, data={ 'message': 'Enrollment created successfully', 'enrollment_id': enrollment.id }, status_code=201 ) def _send_enrollment_notifications(self, enrollment: Enrollment, ctx) -> None: """Send notifications to student, tutors, and supervisor about new enrollment""" # Get student notification preferences student_prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == enrollment.student_id ).first() # Get course tutors tutors = ctx.session.query(Tutor).join( tutor_course_association, Tutor.id == tutor_course_association.c.tutor_id ).filter( tutor_course_association.c.course_id == enrollment.course_id ).all() # Get course supervisor supervisor = enrollment.course.supervisor # Send email to student if they have email notifications enabled if student_prefs and student_prefs.receive_email: subject = f"Enrollment Confirmation: {enrollment.course.code} - {enrollment.course.title}" tutor_list_html = "".join([ f"""
  • {t.first_name} {t.last_name}
    Email: {t.email}{' · Phone: ' + (t.phone or '') if hasattr(t, 'phone') and t.phone else ''}
    • """ for t in tutors ]) if tutors else "
  • Tutors will be assigned soon.
    • " supervisor_html = ( f"""
    {supervisor.first_name} {supervisor.last_name}
    Email: {supervisor.email}{' · Phone: ' + (supervisor.phone or '') if hasattr(supervisor, 'phone') and supervisor.phone else ''}
    """ ) if supervisor else "
    No supervisor assigned yet.
    " message = f"""
    Enrollment Confirmed
    {current_app.config['APP_NAME']}

    Hello {enrollment.student.first_name},

    You have been enrolled in the following course:

    {enrollment.course.code} — {enrollment.course.title}
    Credits: {enrollment.course.credits} · Department: {enrollment.course.department}
    Type: {enrollment.enrollment_type} · Status: {enrollment.status}
    Assigned Tutor(s)
      {tutor_list_html}
    Supervisor
    {supervisor_html}
    Open My Courses

    If you have any questions, reply to this email to reach your course team.

    © {datetime.utcnow().year} {current_app.config['APP_NAME']}
    """ try: send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=enrollment.student.email, subject=subject, message=message ) except Exception as e: current_app.logger.error(f"Failed to send enrollment email to student {enrollment.student_id}: {str(e)}") # Send notifications to tutors for tutor in tutors: tutor_prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == tutor.id ).first() if tutor_prefs and tutor_prefs.receive_email: subject = f"New Student Enrollment: {enrollment.course.code}" message = f"""
    New Student Enrolled
    {enrollment.course.code} — {enrollment.course.title}

    Hello {tutor.first_name},

    A student has been enrolled in your course:

    {enrollment.student.first_name} {enrollment.student.last_name} • {enrollment.student.student_id}
    Program: {enrollment.student.program} · Year: {enrollment.student.year_of_study}
    View Course
    © {datetime.utcnow().year} {current_app.config['APP_NAME']}
    """ try: send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=tutor.email, subject=subject, message=message ) except Exception as e: current_app.logger.error(f"Failed to send enrollment email to tutor {tutor.id}: {str(e)}") # Send notification to supervisor if exists if supervisor: supervisor_prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == supervisor.id ).first() if supervisor_prefs and supervisor_prefs.receive_email: subject = f"New Enrollment in Supervised Course: {enrollment.course.code}" message = f"""
    New Enrollment
    {enrollment.course.code} — {enrollment.course.title}

    Hello {supervisor.first_name},

    A student was enrolled in a course you supervise:

    {enrollment.student.first_name} {enrollment.student.last_name} • {enrollment.student.student_id}
    Type: {enrollment.enrollment_type} · Status: {enrollment.status}
    © {datetime.utcnow().year} {current_app.config['APP_NAME']}
    """ try: send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=supervisor.email, subject=subject, message=message ) except Exception as e: current_app.logger.error(f"Failed to send enrollment email to supervisor {supervisor.id}: {str(e)}") def get(self, enrollment_id: str) -> Dict: """Get detailed information about a specific enrollment""" with DatabaseContextManager() as ctx: enrollment = ctx.session.query(Enrollment).filter(Enrollment.id == enrollment_id).first() if not enrollment: return custom_response( success=False, data="Enrollment not found", status_code=404 ) # Get progress record progress = ctx.session.query(CourseProgress).filter( CourseProgress.enrollment_id == enrollment_id ).first() # Get attendance statistics attendance = ctx.session.query( func.count(Attendance.id), func.sum(case((Attendance.status == AttendanceStatus.present, 1), else_=0)) ).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( TeachingSession.course_id == enrollment.course_id, Attendance.student_id == enrollment.student_id ).first() # Get assignment statistics assignments = ctx.session.query( func.count(AssignmentSubmission.id), func.sum(case((AssignmentSubmission.status == SubmissionStatus.graded, 1), else_=0)), func.avg(AssignmentSubmission.grade) ).join( Assignment, AssignmentSubmission.assignment_id == Assignment.id ).filter( Assignment.course_id == enrollment.course_id, AssignmentSubmission.student_id == enrollment.student_id ).first() enrollment_data = { 'id': enrollment.id, 'student': { 'id': enrollment.student.id, 'name': f"{enrollment.student.first_name} {enrollment.student.last_name}", 'student_id': enrollment.student.student_id, 'email': enrollment.student.email, 'program': enrollment.student.program, 'year_of_study': enrollment.student.year_of_study }, 'course': { 'id': enrollment.course.id, 'code': enrollment.course.code, 'title': enrollment.course.title, 'credits': enrollment.course.credits, 'department': enrollment.course.department }, 'enrollment_date': str(enrollment.enrollment_date), 'status': enrollment.status, 'enrollment_type': enrollment.enrollment_type, 'is_audit': enrollment.is_audit, 'completion_date': str(enrollment.completion_date) if enrollment.completion_date else None, 'grade': enrollment.grade, 'progress': { 'completion_percentage': progress.completion_percentage if progress else 0, 'current_grade': progress.current_grade if progress else None, 'attendance_rate': progress.attendance_rate if progress else 0, 'last_accessed': progress.last_accessed.isoformat() if progress and progress.last_accessed else None } if progress else None, 'attendance': { 'present': attendance[1] if attendance else 0, 'total': attendance[0] if attendance else 0, 'rate': round((attendance[1] / attendance[0]) * 100, 1) if attendance and attendance[0] else 0 }, 'assignments': { 'completed': assignments[1] if assignments else 0, 'total': assignments[0] if assignments else 0, 'average_grade': round(assignments[2], 1) if assignments and assignments[2] else None } } return custom_response( success=True, data=enrollment_data, status_code=200 ) def update(self, enrollment_id: str, payload: Dict) -> Dict: """Update enrollment information""" with DatabaseContextManager() as ctx: enrollment = ctx.session.query(Enrollment).filter(Enrollment.id == enrollment_id).first() if not enrollment: return custom_response( success=False, data="Enrollment not found", status_code=404 ) # Update allowed fields updatable_fields = [ 'status', 'enrollment_type', 'is_audit', 'grade', 'midterm_grade', 'final_exam_grade', 'project_grade', 'comments', 'withdrawal_date', 'withdrawal_reason' ] for field in updatable_fields: if field in payload: setattr(enrollment, field, payload[field]) # Handle status changes if 'status' in payload: if payload['status'] == 'completed' and not enrollment.completion_date: enrollment.completion_date = datetime.utcnow().date() # If changing from withdrawn back to active, clear withdrawal info if payload['status'] == 'active' and enrollment.status == 'withdrawn': enrollment.withdrawal_date = None enrollment.withdrawal_reason = None ctx.session.commit() return custom_response( success=True, data="Enrollment updated successfully", status_code=200 ) def delete(self, enrollment_id: str) -> Dict: """Withdraw a student from a course (soft delete)""" with DatabaseContextManager() as ctx: enrollment = ctx.session.query(Enrollment).filter(Enrollment.id == enrollment_id).first() if not enrollment: return custom_response( success=False, data="Enrollment not found", status_code=404 ) # Set status to withdrawn and record withdrawal date enrollment.status = 'withdrawn' enrollment.withdrawal_date = datetime.utcnow().date() ctx.session.commit() # Send withdrawal notification self._send_withdrawal_notification(enrollment, ctx) return custom_response( success=True, data="Student withdrawn from course successfully", status_code=200 ) def _send_withdrawal_notification(self, enrollment: Enrollment, ctx) -> None: """Send notifications about course withdrawal""" # Get student notification preferences student_prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == enrollment.student_id ).first() # Get course tutors tutors = ctx.session.query(Tutor).join( tutor_course_association, Tutor.id == tutor_course_association.c.tutor_id ).filter( tutor_course_association.c.course_id == enrollment.course_id ).all() # Get course supervisor supervisor = enrollment.course.supervisor # Send email to student if they have email notifications enabled if student_prefs and student_prefs.receive_email: subject = f"Withdrawal Confirmation: {enrollment.course.code} - {enrollment.course.title}" message = f"""

    Course Withdrawal Confirmation

    Hello {enrollment.student.first_name},

    You have been withdrawn from:

    Course: {enrollment.course.code} - {enrollment.course.title}

    Withdrawal Date: {enrollment.withdrawal_date}

    Reason: {enrollment.withdrawal_reason or 'Not specified'}

    If this was a mistake, please contact your academic advisor immediately.

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

    """ try: send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=enrollment.student.email, subject=subject, message=message ) except Exception as e: current_app.logger.error(f"Failed to send withdrawal email to student {enrollment.student_id}: {str(e)}") # Send notifications to tutors for tutor in tutors: tutor_prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == tutor.id ).first() if tutor_prefs and tutor_prefs.receive_email: subject = f"Student Withdrawal: {enrollment.course.code}" message = f"""

    Student Withdrawn from Course

    Hello {tutor.first_name},

    A student has been withdrawn from your course:

    Course: {enrollment.course.code} - {enrollment.course.title}

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

    Student ID: {enrollment.student.student_id}

    Withdrawal Date: {enrollment.withdrawal_date}

    Reason: {enrollment.withdrawal_reason or 'Not specified'}

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

    """ try: send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=tutor.email, subject=subject, message=message ) except Exception as e: current_app.logger.error(f"Failed to send withdrawal email to tutor {tutor.id}: {str(e)}") # Send notification to supervisor if exists if supervisor: supervisor_prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == supervisor.id ).first() if supervisor_prefs and supervisor_prefs.receive_email: subject = f"Withdrawal in Supervised Course: {enrollment.course.code}" message = f"""

    Course Withdrawal Notification

    Hello {supervisor.first_name},

    A student has been withdrawn from a course you supervise:

    Course: {enrollment.course.code} - {enrollment.course.title}

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

    Student ID: {enrollment.student.student_id}

    Withdrawal Date: {enrollment.withdrawal_date}

    Reason: {enrollment.withdrawal_reason or 'Not specified'}

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

    """ try: send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=supervisor.email, subject=subject, message=message ) except Exception as e: current_app.logger.error(f"Failed to send withdrawal email to supervisor {supervisor.id}: {str(e)}") def fetchAll(self, page: int = 1, per_page: int = 20, filters: Dict = None) -> Dict: """Fetch all enrollments with pagination and optional filters""" with DatabaseContextManager() as ctx: query = ctx.session.query(Enrollment) # Apply filters if filters: if 'student_id' in filters: query = query.filter(Enrollment.student_id == filters['student_id']) if 'course_id' in filters: query = query.filter(Enrollment.course_id == filters['course_id']) if 'status' in filters: query = query.filter(Enrollment.status == filters['status']) if 'enrollment_type' in filters: query = query.filter(Enrollment.enrollment_type == filters['enrollment_type']) if 'is_audit' in filters: query = query.filter(Enrollment.is_audit == filters['is_audit']) if 'department' in filters: query = query.join(Course).filter(Course.department == filters['department']) if 'semester' in filters: query = query.join(Course).filter(Course.semester == filters['semester']) if 'search' in filters: search = f"%{filters['search']}%" query = query.join(Student).join(Course).filter( or_( Student.first_name.ilike(search), Student.last_name.ilike(search), Student.student_id.ilike(search), Course.code.ilike(search), Course.title.ilike(search) ) ) total = query.count() enrollments = query.order_by( Enrollment.enrollment_date.desc() ).offset( (page - 1) * per_page ).limit(per_page).all() enrollment_list = [] for enrollment in enrollments: enrollment_list.append({ 'id': enrollment.id, 'student': { 'id': enrollment.student.id, 'name': f"{enrollment.student.first_name} {enrollment.student.last_name}", 'student_id': enrollment.student.student_id }, 'course': { 'id': enrollment.course.id, 'code': enrollment.course.code, 'title': enrollment.course.title }, 'enrollment_date': str(enrollment.enrollment_date), 'status': enrollment.status, 'enrollment_type': enrollment.enrollment_type, 'is_audit': enrollment.is_audit, 'grade': enrollment.grade, 'withdrawal_date': str(enrollment.withdrawal_date) if enrollment.withdrawal_date else None }) return custom_response( success=True, data={ 'enrollments': enrollment_list, 'total': total, 'page': page, 'per_page': per_page } ) def generate_upload_template(self) -> Dict: """Generate an Excel template for bulk enrollment upload with dropdowns""" try: with DatabaseContextManager() as ctx: # Get active students students = ctx.session.query(Student).filter( Student.is_active == True ).all() student_data = [{ 'student_id': s.student_id, 'name': f"{s.first_name} {s.last_name}", 'email': s.email, 'program': s.program, 'year_of_study': s.year_of_study } for s in students] # Get active courses courses = ctx.session.query(Course).filter( Course.is_active == True ).all() course_data = [{ 'course_id': c.id, 'code': c.code, 'title': c.title, 'department': c.department, 'credits': c.credits, 'semester': c.semester } for c in courses] # Create Excel file with dropdowns output = BytesIO() writer = pd.ExcelWriter(output, engine='xlsxwriter') # Create main data sheet df = pd.DataFrame(columns=[ 'student_id', 'course_id', 'enrollment_type', 'status', 'is_audit' ]) df.to_excel(writer, sheet_name='Enrollments', index=False) # Create reference sheets pd.DataFrame(student_data).to_excel( writer, sheet_name='Students', index=False ) pd.DataFrame(course_data).to_excel( writer, sheet_name='Courses', index=False ) # Get workbook and worksheet objects workbook = writer.book worksheet = writer.sheets['Enrollments'] # Add data validation (dropdowns) # Student ID dropdown student_ids = [s['student_id'] for s in student_data] student_validation = { 'validate': 'list', 'source': student_ids, 'input_title': 'Select Student ID', 'input_message': 'Choose from the list of active students' } worksheet.data_validation( 'A2:A1048576', student_validation ) # Course ID dropdown course_ids = [str(c['course_id']) for c in course_data] course_validation = { 'validate': 'list', 'source': course_ids, 'input_title': 'Select Course ID', 'input_message': 'Choose from the list of active courses' } worksheet.data_validation( 'B2:B1048576', course_validation ) # Enrollment type dropdown enrollment_types = ['regular', 'credit', 'non-credit'] type_validation = { 'validate': 'list', 'source': enrollment_types, 'input_title': 'Select Enrollment Type', 'input_message': 'Choose from regular, credit, or non-credit' } worksheet.data_validation( 'C2:C1048576', type_validation ) # Status dropdown status_types = ['active', 'withdrawn', 'completed', 'failed'] status_validation = { 'validate': 'list', 'source': status_types, 'input_title': 'Select Status', 'input_message': 'Choose from active, withdrawn, completed, or failed' } worksheet.data_validation( 'D2:D1048576', status_validation ) # Is audit dropdown audit_types = ['TRUE', 'FALSE'] audit_validation = { 'validate': 'list', 'source': audit_types, 'input_title': 'Select Audit Status', 'input_message': 'TRUE for audit enrollment, FALSE for regular' } worksheet.data_validation( 'E2:E1048576', audit_validation ) # Add instructions instructions = """ Instructions for Bulk Enrollment Upload: 1. Fill in the rows with student-course combinations 2. Use the dropdowns to select valid values 3. Required columns: - student_id: Must match a student from the Students sheet - course_id: Must match a course from the Courses sheet - enrollment_type: regular, credit, or non-credit - status: active, withdrawn, completed, or failed - is_audit: TRUE or FALSE 4. Do not modify the header row 5. Save the file and upload it through the system """ instructions_df = pd.DataFrame([instructions.split('\n')]) instructions_df.to_excel( writer, sheet_name='Instructions', index=False, header=False ) writer.close() output.seek(0) # Save to temporary file filename = f"enrollment_template_{datetime.now().strftime('%Y%m%d')}.xlsx" filepath = os.path.join(current_app.config['UPLOAD_FOLDER'], filename) with open(filepath, 'wb') as f: f.write(output.getvalue()) # Send as downloadable file return send_file( filepath, as_attachment=True, download_name=filename, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) except Exception as e: current_app.logger.error(f"Failed to generate enrollment template: {str(e)}") return custom_response( success=False, data=f"Failed to generate template: {str(e)}", status_code=500 ) def process_bulk_upload(self, file) -> Dict: """Process bulk enrollment upload from Excel or CSV file""" try: # Save uploaded file temporarily filename = secure_filename(file.filename) filepath = os.path.join(current_app.config['UPLOAD_FOLDER'], filename) file.save(filepath) # Read file based on extension if filename.endswith('.xlsx'): df = pd.read_excel(filepath) elif filename.endswith('.csv'): df = pd.read_csv(filepath) else: return custom_response( success=False, data="Unsupported file format. Please upload Excel (.xlsx) or CSV (.csv) file.", status_code=400 ) # Validate required columns required_columns = ['student_id', 'course_id', 'enrollment_type', 'status', 'is_audit'] if not all(col in df.columns for col in required_columns): return custom_response( success=False, data=f"Missing required columns. File must contain: {', '.join(required_columns)}", status_code=400 ) results = { 'total': len(df), 'success': 0, 'failed': 0, 'details': [] } with DatabaseContextManager() as ctx: for _, row in df.iterrows(): try: # Get student by student_id (not the UUID) student = ctx.session.query(Student).filter( Student.student_id == str(row['student_id']), Student.is_active == True ).first() if not student: raise ValueError(f"Student with ID {row['student_id']} not found or inactive") # Get course by ID course = ctx.session.query(Course).filter( Course.id == str(row['course_id']), Course.is_active == True ).first() if not course: raise ValueError(f"Course with ID {row['course_id']} not found or inactive") # Check if already enrolled existing = ctx.session.query(Enrollment).filter( Enrollment.student_id == student.id, Enrollment.course_id == course.id ).first() if existing: raise ValueError("Student is already enrolled in this course") # Check course capacity current_enrollments = ctx.session.query(Enrollment).filter( Enrollment.course_id == course.id, Enrollment.status == 'active' ).count() if course.max_students and current_enrollments >= course.max_students: raise ValueError("Course has reached maximum capacity") # Create enrollment enrollment = Enrollment( id=str(uuid.uuid4()), student_id=student.id, course_id=course.id, enrollment_date=datetime.utcnow().date(), status=row['status'], enrollment_type=row['enrollment_type'], is_audit=bool(row['is_audit']) ) ctx.session.add(enrollment) # Create initial progress record progress = CourseProgress( id=str(uuid.uuid4()), enrollment_id=enrollment.id, student_id=student.id, completion_percentage=0.0, attendance_rate=0.0 ) ctx.session.add(progress) ctx.session.commit() # Send notifications self._send_enrollment_notifications(enrollment, ctx) results['success'] += 1 results['details'].append({ 'row': _ + 2, # +1 for 0-based index, +1 for header 'student_id': row['student_id'], 'course_id': row['course_id'], 'status': 'success', 'message': 'Enrollment created successfully' }) except Exception as e: ctx.session.rollback() results['failed'] += 1 results['details'].append({ 'row': _ + 2, 'student_id': row.get('student_id', 'N/A'), 'course_id': row.get('course_id', 'N/A'), 'status': 'failed', 'message': str(e) }) continue # Generate report report_filename = f"enrollment_upload_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" report_path = os.path.join(current_app.config['UPLOAD_FOLDER'], report_filename) report_df = pd.DataFrame(results['details']) report_df.to_excel(report_path, index=False) # Send email with report to admin admin_email = current_app.config['ADMIN_EMAIL'] subject = f"Bulk Enrollment Upload Report - {results['success']} success, {results['failed']} failed" message = f"""

    Bulk Enrollment Upload Report

    The bulk enrollment upload has completed with the following results:

    Total Records: {results['total']}

    Successful: {results['success']}

    Failed: {results['failed']}

    Detailed results are attached to this email.

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

    """ send_email_with_attachment( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=admin_email, subject=subject, message=message, attachment_path=report_path ) # Clean up temporary files os.remove(filepath) os.remove(report_path) return custom_response( success=True, data={ 'message': 'Bulk upload processed', 'results': { 'total': results['total'], 'success': results['success'], 'failed': results['failed'] } }, status_code=200 ) except Exception as e: current_app.logger.error(f"Failed to process bulk enrollment upload: {str(e)}") return custom_response( success=False, data=f"Failed to process upload: {str(e)}", status_code=500 ) def get_enrollment_analytics(self, supervisor_id: str, academic_session_id: str = None) -> Dict: """ Generate comprehensive enrollment analytics for courses supervised by the given supervisor. Returns data structured for visualization with shadcn/ui charts, organized by academic sessions. Args: supervisor_id: ID of the supervisor whose courses' analytics to generate academic_session_id: Optional academic session ID to filter analytics for specific session Returns: Dict containing various analytics metrics and datasets for visualization, organized by academic sessions """ with DatabaseContextManager() as ctx: supervisor = ctx.session.query(Supervisor).filter( Supervisor.id == supervisor_id ).first() if not supervisor: return { "error": "Supervisor not found", "status_code": 404 } # Get all courses supervised by this supervisor base_course_query = ctx.session.query(Course).filter( Course.supervisor_id == supervisor_id, Course.is_active == True ) # Filter by academic session if provided if academic_session_id: base_course_query = base_course_query.filter( Course.academic_session_id == academic_session_id ) supervised_courses = base_course_query.options( joinedload(Course.enrollments) .joinedload(Enrollment.student), joinedload(Course.tutors), joinedload(Course.academic_session) ).all() if not supervised_courses: return { "error": "No active courses found for this supervisor", "status_code": 404 } # Get all academic sessions for the supervisor's courses academic_sessions = ctx.session.query(Course.academic_session_id).filter( Course.supervisor_id == supervisor_id, Course.is_active == True ).distinct().all() academic_session_ids = [session[0] for session in academic_sessions if session[0]] # Get academic session details session_details = ctx.session.query( AcademicSession.id, AcademicSession.name, AcademicSession.year, AcademicSession.start_date, AcademicSession.end_date, AcademicSession.status ).filter( AcademicSession.id.in_(academic_session_ids) ).order_by( AcademicSession.start_date.desc() ).all() # Current academic term (assuming semester system) current_date = datetime.now().date() current_year = current_date.year current_month = current_date.month current_semester = "Spring" if 1 <= current_month <= 5 else "Fall" if 8 <= current_month <= 12 else "Summer" # Build base filter for enrollments base_enrollment_filter = [ Course.supervisor_id == supervisor_id, Enrollment.status == 'active' ] if academic_session_id: base_enrollment_filter.append(Course.academic_session_id == academic_session_id) # 1. Overall Enrollment Statistics total_enrollments = ctx.session.query(func.count(Enrollment.id)).join( Course ).filter(*base_enrollment_filter).scalar() # 2. Enrollment Trends Over Time (last 12 months) - SQLite compatible version # For SQLite, we'll use strftime to format the date by month/year enrollment_trends = ctx.session.query( func.strftime('%Y-%m', Enrollment.enrollment_date).label('month'), func.count(Enrollment.id).label('count') ).join( Course ).filter( *base_enrollment_filter, Enrollment.enrollment_date >= (current_date - timedelta(days=365)) ).group_by( func.strftime('%Y-%m', Enrollment.enrollment_date) ).order_by( func.strftime('%Y-%m', Enrollment.enrollment_date) ).all() # Format for line chart - parse the YYYY-MM string into a display format enrollment_trend_data = [{ "name": datetime.strptime(month, "%Y-%m").strftime("%b %Y"), "enrollments": count } for month, count in enrollment_trends] # 3. Enrollment by Department dept_enrollments = ctx.session.query( Course.department, func.count(Enrollment.id).label('count') ).join( Enrollment.course ).filter(*base_enrollment_filter).group_by( Course.department ).all() dept_data = [{ "name": dept, "value": count } for dept, count in dept_enrollments] # 4. Enrollment by Course (Top 10) course_enrollments = ctx.session.query( Course.code, Course.title, func.count(Enrollment.id).label('count') ).join( Enrollment.course ).filter(*base_enrollment_filter).group_by( Course.code, Course.title ).order_by( func.count(Enrollment.id).desc() ).limit(10).all() course_data = [{ "course": f"{code} - {title[:20]}..." if len(title) > 20 else f"{code} - {title}", "enrollments": count } for code, title, count in course_enrollments] # 5. Enrollment Status Distribution status_distribution = ctx.session.query( Enrollment.status, func.count(Enrollment.id).label('count') ).join( Course ).filter(*base_enrollment_filter).group_by( Enrollment.status ).all() status_data = [{ "name": status.capitalize(), "value": count } for status, count in status_distribution] # 6. Enrollment Type Distribution type_distribution = ctx.session.query( Enrollment.enrollment_type, func.count(Enrollment.id).label('count') ).join( Course ).filter(*base_enrollment_filter).group_by( Enrollment.enrollment_type ).all() type_data = [{ "name": type.capitalize(), "value": count } for type, count in type_distribution] # 7. Student Demographics # Age distribution age_distribution = ctx.session.query( (current_year - extract('year', Student.date_of_birth)).label('age'), func.count(Enrollment.id).label('count') ).join( Enrollment.student ).join( Enrollment.course ).filter( *base_enrollment_filter, Student.date_of_birth.isnot(None) ).group_by( (current_year - extract('year', Student.date_of_birth)) ).order_by( (current_year - extract('year', Student.date_of_birth)) ).all() age_data = [{ "name": f"{int(age)} yrs", "value": count } for age, count in age_distribution] # Gender distribution gender_distribution = ctx.session.query( Student.gender, func.count(Enrollment.id).label('count') ).join( Enrollment.student ).join( Enrollment.course ).filter( *base_enrollment_filter, Student.gender.isnot(None) ).group_by( Student.gender ).all() gender_data = [{ "name": gender.capitalize(), "value": count } for gender, count in gender_distribution] # 8. Tutor Workload Analysis tutor_workload = ctx.session.query( Tutor.first_name, Tutor.last_name, func.count(Enrollment.id).label('enrollments'), func.count(func.distinct(Course.id)).label('courses') ).join( tutor_course_association, Tutor.id == tutor_course_association.c.tutor_id ).join( Course, tutor_course_association.c.course_id == Course.id ).join( Enrollment, Enrollment.course_id == Course.id ).filter(*base_enrollment_filter).group_by( Tutor.first_name, Tutor.last_name ).order_by( func.count(Enrollment.id).desc() ).all() tutor_data = [{ "tutor": f"{first_name} {last_name}", "enrollments": enrollments, "courses": courses } for first_name, last_name, enrollments, courses in tutor_workload] # 9. Course Completion Rates completion_rates = ctx.session.query( Course.code, Course.title, func.avg(CourseProgress.completion_percentage).label('avg_completion'), func.avg(CourseProgress.attendance_rate).label('avg_attendance') ).join( Enrollment.course ).join( Enrollment.progress ).filter(*base_enrollment_filter).group_by( Course.code, Course.title ).all() completion_data = [{ "course": f"{code} - {title[:20]}..." if len(title) > 20 else f"{code} - {title}", "completion": float(avg_completion), "attendance": float(avg_attendance) } for code, title, avg_completion, avg_attendance in completion_rates] # 10. Current Semester Focus current_semester_stats = ctx.session.query( func.count(Enrollment.id).label('total'), func.sum(case((Enrollment.is_audit == True, 1), else_=0)).label('audits'), func.sum(case((Enrollment.enrollment_type == 'credit', 1), else_=0)).label('credit'), func.sum(case((Enrollment.enrollment_type == 'non-credit', 1), else_=0)).label('non_credit') ).join( Course ).filter( *base_enrollment_filter, Course.semester == current_semester, extract('year', Enrollment.enrollment_date) == current_year ).first() semester_data = { "total": current_semester_stats[0] or 0, "audits": current_semester_stats[1] or 0, "credit": current_semester_stats[2] or 0, "non_credit": current_semester_stats[3] or 0 } # 11. Academic Session Breakdown session_breakdown = [] for session in session_details: session_id, session_name, session_year, start_date, end_date, status = session # Get enrollments for this specific session session_enrollments = ctx.session.query( func.count(Enrollment.id).label('count') ).join( Course ).filter( Course.supervisor_id == supervisor_id, Course.academic_session_id == session_id, Enrollment.status == 'active' ).scalar() # Get courses count for this session session_courses = ctx.session.query( func.count(Course.id).label('count') ).filter( Course.supervisor_id == supervisor_id, Course.academic_session_id == session_id, Course.is_active == True ).scalar() session_breakdown.append({ "id": session_id, "name": session_name, "year": session_year, "start_date": start_date.isoformat() if start_date else None, "end_date": end_date.isoformat() if end_date else None, "status": status, "enrollments": session_enrollments or 0, "courses": session_courses or 0, "is_current": start_date <= current_date <= end_date if start_date and end_date else False }) # Sort sessions by start date (most recent first) session_breakdown.sort(key=lambda x: x['start_date'] or '', reverse=True) # 12. Enrollment Trends by Academic Session session_enrollment_trends = [] for session in session_details: session_id = session[0] # Get monthly enrollments for this session session_trends = ctx.session.query( func.strftime('%Y-%m', Enrollment.enrollment_date).label('month'), func.count(Enrollment.id).label('count') ).join( Course ).filter( Course.supervisor_id == supervisor_id, Course.academic_session_id == session_id, Enrollment.status == 'active' ).group_by( func.strftime('%Y-%m', Enrollment.enrollment_date) ).order_by( func.strftime('%Y-%m', Enrollment.enrollment_date) ).all() session_enrollment_trends.append({ "session_id": session_id, "session_name": session[1], "data": [{ "name": datetime.strptime(month, "%Y-%m").strftime("%b %Y"), "enrollments": count } for month, count in session_trends] }) # Compile all the data into a structured response analytics = { "summary": { "total_courses": len(supervised_courses), "total_enrollments": total_enrollments, "current_semester": f"{current_semester} {current_year}", "supervisor_name": f"{supervisor.first_name} {supervisor.last_name}", "academic_sessions_count": len(session_details), "current_academic_session": next((s for s in session_breakdown if s['is_current']), None) }, "academic_sessions": { "available_sessions": session_breakdown, "current_session_id": academic_session_id, "session_enrollment_trends": session_enrollment_trends }, "charts": { "enrollment_trend": { "title": "Enrollment Trends (Last 12 Months)", "description": "Monthly enrollment growth across all supervised courses", "data": enrollment_trend_data, "chart_type": "line" # For shadcn line chart }, "department_distribution": { "title": "Enrollments by Department", "description": "Breakdown of active enrollments across departments", "data": dept_data, "chart_type": "pie" # For shadcn pie chart }, "top_courses": { "title": "Top Courses by Enrollment", "description": "Most popular courses under supervision (top 10)", "data": course_data, "chart_type": "bar" # For shadcn bar chart }, "status_distribution": { "title": "Enrollment Status", "description": "Distribution of enrollment statuses", "data": status_data, "chart_type": "pie" }, "type_distribution": { "title": "Enrollment Types", "description": "Breakdown of enrollment types (credit, non-credit, etc.)", "data": type_data, "chart_type": "pie" }, "age_distribution": { "title": "Student Age Distribution", "description": "Age ranges of enrolled students", "data": age_data, "chart_type": "bar" }, "gender_distribution": { "title": "Student Gender Distribution", "description": "Gender breakdown of enrolled students", "data": gender_data, "chart_type": "pie" }, "completion_rates": { "title": "Course Completion Rates", "description": "Average completion and attendance rates by course", "data": completion_data, "chart_type": "dual_bar", # Custom type for two metrics "metrics": ["completion", "attendance"] }, "academic_session_breakdown": { "title": "Enrollments by Academic Session", "description": "Breakdown of enrollments across different academic sessions", "data": [{ "name": f"{s['name']} {s['year']}", "enrollments": s['enrollments'], "courses": s['courses'] } for s in session_breakdown], "chart_type": "bar" } }, "tables": { "tutor_workload": { "title": "Tutor Workload Analysis", "description": "Enrollments and courses per tutor", "headers": ["Tutor", "Enrollments", "Courses"], "data": tutor_data }, "academic_sessions": { "title": "Academic Sessions Overview", "description": "Detailed breakdown of enrollments and courses by academic session", "headers": ["Session", "Year", "Status", "Enrollments", "Courses", "Period"], "data": [{ "session": s['name'], "year": s['year'], "status": s['status'], "enrollments": s['enrollments'], "courses": s['courses'], "period": f"{s['start_date'][:10] if s['start_date'] else 'N/A'} to {s['end_date'][:10] if s['end_date'] else 'N/A'}" } for s in session_breakdown] } }, "current_semester": { "title": f"{current_semester} {current_year} Highlights", "data": semester_data } } return custom_response( data=analytics, success=True, status_code=200 )