from flask import current_app, request from src.models import DatabaseContextManager from src.models.models import ( User, Student, Tutor, Supervisor, NotificationPreference, TrustedDevice, AttendanceStatus, Attendance, Assignment, TeachingSession, TutorAvailability, UserType, TwoFactorMethod, tutor_course_association, Course, AssignmentSubmission, TimetableBlock, SubmissionStatus, Speciality, TutorDepartment, SupervisorDepartment, Enrollment, DailyTeachingSession, enrollment_courses, ExamEvidence, Notification ) from sqlalchemy import func import uuid from src.utils import ( ApiABC, hash_password, check_password, custom_response, send_email, generate_otp ) import os import jwt import datetime from datetime import date from sqlalchemy import inspect import secrets import hashlib class UserManager(ApiABC): def __init__(self): self.table = User def _parse_date(self, date_input): """Helper to parse date from string or return date object""" if date_input is None: return None if isinstance(date_input, datetime.date): return date_input if isinstance(date_input, datetime.datetime): return date_input.date() try: return datetime.datetime.strptime(date_input, '%Y-%m-%d').date() except (ValueError, TypeError): return None def get(self, id): """Get a user by ID with their specific role details""" with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter(User.id == id).first() if not user: return custom_response( success=False, data="User not found", status_code=404 ) # Get base user info user_data = self._to_dict(user) if user.user_type.value == "student": student = ctx.session.query(Student).filter(Student.id == id).first() if student: user_data.update(self._student_to_dict(student)) elif user.user_type.value == "tutor": tutor = ctx.session.query(Tutor).filter(Tutor.id == id).first() if tutor: user_data.update(self._tutor_to_dict(tutor)) elif user.user_type.value == "supervisor": supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == id).first() if supervisor: user_data.update(self._supervisor_to_dict(supervisor)) return custom_response( success=True, data=user_data, status_code=200 ) def _to_dict(self, user): """Convert base User model to dictionary""" return { 'id': user.id, 'email': user.email, 'first_name': user.first_name, 'last_name': user.last_name, 'phone': user.phone, 'user_type': user.user_type.value, 'is_active': user.is_active, 'profile_picture': user.profile_picture, 'gender': user.gender, 'date_of_birth': str(user.date_of_birth) if user.date_of_birth else None, 'nationality': user.nationality, 'created_at': str(user.created_at) if user.created_at else None, 'last_login': str(user.last_login) if user.last_login else None, 'two_factor_enabled': user.two_factor_enabled, 'two_factor_method': user.two_factor_method.value if user.two_factor_method else None } def _student_to_dict(self, student): """Convert Student model to dictionary""" return { 'student_details': { 'student_id': student.student_id, 'year_of_study': student.year_of_study, 'program': student.program, 'enrollment_date': str(student.enrollment_date) if student.enrollment_date else None, 'current_semester': student.current_semester, 'student_category': student.student_category, 'guardian_name': student.guardian_name, 'guardian_contact': student.guardian_contact, 'emergency_contact': student.emergency_contact, 'address': { 'street': student.address, 'city': student.city, 'country': student.country, 'postal_code': student.postal_code }, 'academic_advisor': student.academic_advisor, 'has_scholarship': student.has_scholarship, 'scholarship_details': student.scholarship_details, 'is_on_probation': student.is_on_probation, 'probation_details': { 'reason': student.probation_reason, 'start_date': str(student.probation_start_date) if student.probation_start_date else None, 'end_date': str(student.probation_end_date) if student.probation_end_date else None } if student.is_on_probation else None } } def _tutor_to_dict(self, tutor): """Convert Tutor model to dictionary""" return { 'tutor_details': { 'staff_id': tutor.staff_id, 'department': tutor.get_primary_department() if hasattr(tutor, 'get_primary_department') else 'Not assigned', 'departments': [item.to_json() for item in tutor.departments] if hasattr(tutor, 'departments') and tutor.departments else [], 'office_location': tutor.office_location, 'qualification': tutor.qualification, 'bio': tutor.bio, 'hourly_rate': tutor.hourly_rate, 'max_teaching_hours': tutor.max_teaching_hours, 'supervisor': tutor.supervisor_id, 'employment_type': 'Full-time' if tutor.is_full_time else 'Part-time', 'specialization': tutor.specialization, 'years_of_teaching': tutor.years_of_teaching, 'leave_status': { 'on_leave': tutor.is_on_leave, 'start_date': str(tutor.leave_start_date) if tutor.leave_start_date else None, 'end_date': str(tutor.leave_end_date) if tutor.leave_end_date else None, 'reason': tutor.leave_reason } if tutor.is_on_leave else None } } def _supervisor_to_dict(self, supervisor): """Convert Supervisor model to dictionary""" return { 'supervisor_details': { 'staff_id': supervisor.staff_id, 'department': supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Not assigned', 'departments': [item.to_json() for item in supervisor.departments] if hasattr(supervisor, 'departments') and supervisor.departments else [], 'office_location': supervisor.office_location, 'office_hours': supervisor.office_hours, 'max_tutors': supervisor.max_tutors, 'is_head_of_department': supervisor.is_head_of_department, 'years_of_experience': supervisor.years_of_experience, 'permissions': { 'approve_timetables': supervisor.can_approve_timetables, 'approve_leave': supervisor.can_approve_leave, 'manage_courses': supervisor.can_manage_courses } } } def create(self, payload): """Create a new user with role-specific details and send welcome email""" with DatabaseContextManager() as ctx: required_fields = { 'email': (str, lambda x: x and '@' in x), 'password': (str, lambda x: len(x) >= 8), 'first_name': (str, lambda x: len(x.strip()) > 0), 'last_name': (str, lambda x: len(x.strip()) > 0), 'user_type': (str, lambda x: x in [ut.value for ut in UserType]) } validation_errors = [] for field, (type_check, validator) in required_fields.items(): if field not in payload: validation_errors.append(f"Missing required field: {field}") elif not isinstance(payload[field], type_check): validation_errors.append(f"Invalid type for {field}, expected {type_check.__name__}") elif not validator(payload[field]): validation_errors.append(f"Invalid value for {field}") if validation_errors: return custom_response( success=False, data={"errors": validation_errors}, status_code=400 ) try: # Generate ID once user_id = str(uuid.uuid4()) # Helper function to parse dates def parse_date(date_str): if isinstance(date_str, str): try: return datetime.datetime.strptime(date_str, '%Y-%m-%d').date() except ValueError: return None elif isinstance(date_str, datetime.date): return date_str return None # Create base user data user_data = { 'id': user_id, 'email': payload['email'], 'password_hash': hash_password(payload['password'], salt=current_app.config['SECRET_KEY']), 'first_name': payload['first_name'], 'last_name': payload['last_name'], 'user_type': payload['user_type'], 'is_active': True, 'created_at': datetime.datetime.utcnow(), 'two_factor_enabled': False, 'two_factor_verified_at': datetime.datetime.utcnow() } # Add optional fields with proper type conversion optional_fields = { 'phone': str, 'gender': str, 'date_of_birth': parse_date, 'nationality': str, 'profile_picture': str } for field, converter in optional_fields.items(): if field in payload and payload[field] is not None: user_data[field] = converter(payload[field]) # Create the appropriate user type if payload['user_type'] == UserType.student.value: # Handle date fields properly enrollment_date = parse_date(payload.get('enrollment_date')) or datetime.date.today() student_data = { 'student_id': payload.get('student_id'), 'year_of_study': int(payload.get('year_of_study', 1)), 'program': payload.get('program'), 'department': payload.get('department'), 'speciality_id': payload.get('speciality_id'), 'enrollment_date': enrollment_date, 'current_semester': str(payload.get('current_semester', '1')), 'student_category': payload.get('student_category', 'regular'), } # Add optional student fields with proper type conversion optional_student_fields = { 'guardian_name': str, 'guardian_contact': str, 'emergency_contact': str, 'address': str, 'city': str, 'country': str, 'postal_code': str, 'academic_advisor': str, 'scholarship_details': str, 'graduation_date': parse_date, 'last_attendance_date': parse_date, 'probation_start_date': parse_date, 'probation_end_date': parse_date } for field, converter in optional_student_fields.items(): if field in payload and payload[field] is not None: student_data[field] = converter(payload[field]) # Add boolean fields boolean_fields = ['has_scholarship', 'is_on_probation'] for field in boolean_fields: if field in payload: student_data[field] = bool(payload[field]) # Add probation reason if on probation if student_data.get('is_on_probation'): student_data['probation_reason'] = payload.get('probation_reason') # Create Student with combined data user = Student(**user_data, **student_data) elif payload['user_type'] == UserType.tutor.value: tutor_data = { 'staff_id': payload.get('staff_id'), 'office_location': payload.get('office_location'), 'qualification': payload.get('qualification'), 'bio': payload.get('bio'), 'hourly_rate': payload.get('hourly_rate', 0), 'max_teaching_hours': payload.get('max_teaching_hours', 28), 'supervisor_id': payload.get('supervisor_id'), 'is_full_time': payload.get('is_full_time', True), 'specialization': payload.get('specialization'), 'years_of_teaching': payload.get('years_of_teaching', 0), } # Add optional tutor fields optional_tutor_fields = [ 'is_on_leave', 'leave_start_date', 'leave_end_date', 'leave_reason' ] for field in optional_tutor_fields: if field in payload: tutor_data[field] = payload[field] user = Tutor(**user_data, **tutor_data) # Handle department assignment through the new many-to-many relationship if 'department' in payload and payload['department']: from src.models.models import TutorDepartment department_assoc = TutorDepartment( tutor_id=user_id, department_name=payload['department'], is_primary=True, assigned_by=payload.get('assigned_by'), notes=f"Department assigned during user creation" ) ctx.session.add(department_assoc) # Assign courses if provided if 'course_ids' in payload and payload['course_ids']: for course_id in payload['course_ids']: stmt = tutor_course_association.insert().values( tutor_id=user_id, course_id=course_id, is_primary=payload.get('is_primary', False) ) ctx.session.execute(stmt) elif payload['user_type'] == UserType.supervisor.value: # Remove department from supervisor_data since it's now handled separately supervisor_data = { 'staff_id': payload.get('staff_id'), 'office_location': payload.get('office_location'), 'office_hours': payload.get('office_hours'), 'max_tutors': payload.get('max_tutors', 10), 'is_head_of_department': payload.get('is_head_of_department', False), 'years_of_experience': payload.get('years_of_experience', 0), 'can_approve_timetables': payload.get('can_approve_timetables', True), 'can_approve_leave': payload.get('can_approve_leave', True), 'can_manage_courses': payload.get('can_manage_courses', True) } user = Supervisor(**user_data, **supervisor_data) # Handle department assignment through the new many-to-many relationship if 'department' in payload and payload['department']: from src.models.models import SupervisorDepartment department_assoc = SupervisorDepartment( supervisor_id=user_id, department_name=payload['department'], is_primary=True, assigned_by=payload.get('assigned_by'), notes=f"Department assigned during user creation" ) ctx.session.add(department_assoc) # Assign managed courses if provided if 'managed_course_ids' in payload and payload['managed_course_ids']: for course_id in payload['managed_course_ids']: course = ctx.session.query(Course).filter(Course.id == course_id).first() if course: course.supervisor_id = user_id ctx.session.add(user) # Create default notification preferences notification_pref = NotificationPreference( id=str(uuid.uuid4()), user_id=user_id, receive_email=True, receive_sms=False, receive_push=True, email_notification_time=24, sms_notification_time=2, push_notification_time=1, notify_attendance_updates=True, notify_grade_updates=True, notify_schedule_changes=True, notify_system_messages=True ) ctx.session.add(notification_pref) ctx.session.commit() self._send_welcome_email(user, payload['user_type']) return custom_response( success=True, data={ "id": user.id, "message": "User created successfully", "user_type": payload['user_type'] }, status_code=201 ) except Exception as e: ctx.session.rollback() current_app.logger.error(f"Error creating user: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to create user due to server error", status_code=500 ) def _send_welcome_email(self, user, user_type): """Send a welcome email to the newly created user""" subject = f"Welcome to {current_app.config['APP_NAME']}!" # Customize message based on user type if user_type == UserType.student: role_description = "student account" next_steps = "You can now log in to access your courses, assignments, and academic resources." elif user_type == UserType.tutor: role_description = "tutor account" next_steps = "You can now log in to manage your courses, schedule sessions, and track student progress." else: role_description = "supervisor account" next_steps = "You can now log in to manage tutors, approve schedules, and oversee courses." # Security information (2FA disabled) security_note = """

Welcome to LMS!

Your account has been created successfully. You can now log in directly with your email and password.

""" # Disable 2FA for existing users (migration) self._disable_2fa_for_user(user.id) message = f"""

Welcome to {current_app.config['APP_NAME']}!

Hello {user.first_name},

Your {role_description} has been successfully created.

{next_steps}

{security_note}

Login Details:

Email: {user.email}

Account Type: {str(user_type).split('.')[-1].capitalize()}

For security reasons, we recommend changing your password after first login.

Login to Your Account

Best regards,

Mutable Tech Enterprises Team

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 ) except Exception as e: current_app.logger.error(f"Failed to send welcome email: {str(e)}") def _disable_2fa_for_user(self, user_id): """Disable 2FA for a user (migration helper)""" try: with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter(User.id == user_id).first() if user and user.two_factor_enabled: user.two_factor_enabled = False user.two_factor_method = TwoFactorMethod.NONE.value user.two_factor_secret = None user.two_factor_backup_codes = None ctx.session.commit() current_app.logger.info(f"Disabled 2FA for user {user_id}") except Exception as e: current_app.logger.error(f"Failed to disable 2FA for user {user_id}: {str(e)}") def disable_2fa_for_all_users(self): """Disable 2FA for all users (migration)""" try: with DatabaseContextManager() as ctx: users_with_2fa = ctx.session.query(User).filter(User.two_factor_enabled == True).all() count = 0 for user in users_with_2fa: user.two_factor_enabled = False user.two_factor_method = TwoFactorMethod.NONE.value user.two_factor_secret = None user.two_factor_backup_codes = None count += 1 ctx.session.commit() current_app.logger.info(f"Disabled 2FA for {count} users") return custom_response( success=True, data=f"Disabled 2FA for {count} users", status_code=200 ) except Exception as e: current_app.logger.error(f"Failed to disable 2FA for all users: {str(e)}") return custom_response( success=False, data=f"Failed to disable 2FA: {str(e)}", status_code=500 ) def update(self, id, payload): """Update user information with proper date handling""" with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter(User.id == id).first() if not user: return custom_response( success=False, data="User not found", status_code=404 ) # Update base user fields with date handling if 'email' in payload: user.email = payload['email'] if 'first_name' in payload: user.first_name = payload['first_name'] if 'last_name' in payload: user.last_name = payload['last_name'] if 'phone' in payload: user.phone = payload['phone'] if 'gender' in payload: user.gender = payload['gender'] if 'date_of_birth' in payload: del payload['date_of_birth'] if 'nationality' in payload: user.nationality = payload['nationality'] if 'profile_picture' in payload: user.profile_picture = payload['profile_picture'] # Update role-specific fields if user.user_type == UserType.student: student = ctx.session.query(Student).filter(Student.id == id).first() if student: if 'student_id' in payload: student.student_id = payload['student_id'] if 'year_of_study' in payload: student.year_of_study = payload['year_of_study'] if 'program' in payload: student.program = payload['program'] # Add date handling for student-specific dates if needed elif user.user_type == UserType.tutor: tutor = ctx.session.query(Tutor).filter(Tutor.id == id).first() if tutor: if 'staff_id' in payload: tutor.staff_id = payload['staff_id'] if 'department' in payload: # Handle department assignment through the new many-to-many relationship from src.models.models import TutorDepartment # Remove existing department assignments existing_depts = ctx.session.query(TutorDepartment).filter( TutorDepartment.tutor_id == id, TutorDepartment.is_active == True ).all() for dept in existing_depts: dept.is_active = False # Add new department assignment if payload['department']: new_dept = TutorDepartment( tutor_id=id, department_name=payload['department'], is_primary=True, assigned_by=payload.get('assigned_by'), notes=f"Department updated during user update" ) ctx.session.add(new_dept) if 'qualification' in payload: tutor.qualification = payload['qualification'] # Add date handling for tutor-specific dates if needed elif user.user_type == UserType.supervisor: supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == id).first() if supervisor: if 'staff_id' in payload: supervisor.staff_id = payload['staff_id'] if 'department' in payload: # Handle department assignment through the new many-to-many relationship from src.models.models import SupervisorDepartment # Remove existing department assignments existing_depts = ctx.session.query(SupervisorDepartment).filter( SupervisorDepartment.supervisor_id == id, SupervisorDepartment.is_active == True ).all() for dept in existing_depts: dept.is_active = False # Add new department assignment if payload['department']: new_dept = SupervisorDepartment( supervisor_id=id, department_name=payload['department'], is_primary=True, assigned_by=payload.get('assigned_by'), notes=f"Department updated during user update" ) ctx.session.add(new_dept) ctx.session.commit() return custom_response( success=True, data="User updated successfully", status_code=200 ) def delete(self, id): """Delete a user (soft delete by setting is_active=False)""" with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter(User.id == id).first() if not user: return custom_response( success=False, data="User not found", status_code=404 ) user.is_active = False ctx.session.commit() return custom_response( success=True, data="User deactivated successfully", status_code=200 ) 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.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC') ip_address = request.remote_addr message = f"""

Secure Login Verification

Hello {user.first_name},

You are attempting to access your account. Please use the verification code below:

{otp}

This code expires in 10 minutes

Login Details:

Time: {login_time}

Location: {ip_address}

Device: {request.user_agent.string}

Best regards,

Mutable Tech Enterprises Team

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 login(self, payload): """Handle user login with enhanced security including device verification""" with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter( User.email == payload['email'] ).first() if not user: return custom_response( success=False, data="Invalid credentials", status_code=401 ) # Check if account is active if not user.is_active: return custom_response( success=False, data="Account is inactive. Please contact support.", status_code=403 ) # 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.datetime.utcnow() + datetime.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.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 # Get device information device_id = payload.get('device_id') or request.user_agent.string device_info = { 'ip_address': request.remote_addr, 'user_agent': request.user_agent.string, 'device_type': self._get_device_type(request.user_agent.string), 'device_id': device_id } # Check if this is a new device (by device_id) # trusted_device = ctx.session.query(TrustedDevice).filter( # TrustedDevice.user_id == user.id, # TrustedDevice.device_id == device_id, # TrustedDevice.expires_at > datetime.datetime.utcnow() # ).first() # is_new_device = not trusted_device # If it's a new device and 2FA is enabled, send security alert # if is_new_device and user.two_factor_enabled: # self._send_new_device_alert(user, device_info) # Force disable 2FA for testing if user.two_factor_enabled: user.two_factor_enabled = False ctx.session.commit() # Check if 2FA is enabled (disabled for direct authentication) # if user.two_factor_enabled: # # Generate OTP # otp = self.generate_otp() # user.otp = otp # user.otp_expiry = datetime.datetime.utcnow() + datetime.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 # } # ) # else: # return custom_response( # success=False, # data="Failed to send OTP", # status_code=500 # ) # else: # No 2FA required, proceed with direct login # Reset failed login attempts user.failed_login_attempts = 0 user.last_login = datetime.datetime.utcnow() ctx.session.commit() # Handle supervisor role preference preferred_role = payload.get('preferred_role') effective_role = self._determine_effective_role(user, preferred_role) response = self._complete_login(user, device_info, effective_role) # Always upsert trusted device after successful login self._add_trusted_device(user.id, device_info) return response def verify_otp(self, payload): """Verify OTP for 2FA login with enhanced security checks""" with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter( User.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.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.datetime.utcnow() + datetime.timedelta(minutes=30) ctx.session.commit() # Send security alert self._send_security_alert(user, "Too many failed OTP attempts") 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 ) # Get device information (ensure device_id is present) device_id = payload.get('device_id') or request.user_agent.string device_info = { 'ip_address': request.remote_addr, 'user_agent': request.user_agent.string, 'device_type': self._get_device_type(request.user_agent.string), 'device_id': device_id } # Check if user wants to trust this device if payload.get('trust_device', False): self._add_trusted_device(user.id, device_info) # 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.datetime.utcnow() ctx.session.commit() response = self.hile(user, device_info) 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.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.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.datetime.utcnow().isoformat(), "previous_lock_until": user.account_locked_until.isoformat() if user.account_locked_until else None }, status_code=200 ) return self._complete_login(user, device_info) def _determine_effective_role(self, user, preferred_role=None): """Determine the effective role for the user based on their type and preference""" from src.models.models import UserType # If user is not a supervisor, return their actual user type if user.user_type != UserType.supervisor: return user.user_type.value # If user is a supervisor and has a preferred role, use that if preferred_role and preferred_role in ['supervisor', 'tutor']: return preferred_role # Default to supervisor role for supervisors without preference return 'supervisor' def _complete_login(self, user, device_info=None, effective_role=None): """Complete the login process by generating auth token and logging device""" # Use effective role if provided, otherwise use user's actual type final_role = effective_role if effective_role else user.user_type.value # 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, "effective_role": final_role, "exp": datetime.datetime.utcnow() + datetime.timedelta(days=1), "iat": datetime.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, "effective_role": final_role }, "token": token # Add token to response }, status_code=200, success=True ) def _get_device_type(self, user_agent): """Determine device type from user agent string""" if 'mobile' in user_agent.lower(): return 'mobile' elif 'tablet' in user_agent.lower(): return 'tablet' else: return 'desktop' def _add_trusted_device(self, user_id, device_info): try: with DatabaseContextManager() as ctx: # First try to find existing device by device_id only existing_device = ctx.session.query(TrustedDevice).filter( TrustedDevice.device_id == device_info['device_id'] ).first() if existing_device: # Update existing device existing_device.user_id = user_id existing_device.device_name = f"{device_info['device_type'].capitalize()} Device" existing_device.ip_address = device_info['ip_address'] existing_device.device_type = device_info['device_type'] existing_device.browser = getattr(request.user_agent, 'browser', None) existing_device.os = getattr(request.user_agent, 'platform', None) existing_device.expires_at = datetime.datetime.utcnow() + datetime.timedelta(days=30) existing_device.last_used = datetime.datetime.utcnow() else: # Create new device device = TrustedDevice( id=str(uuid.uuid4()), user_id=user_id, device_name=f"{device_info['device_type'].capitalize()} Device", device_id=device_info['device_id'], ip_address=device_info['ip_address'], device_type=device_info['device_type'], browser=getattr(request.user_agent, 'browser', None), os=getattr(request.user_agent, 'platform', None), expires_at=datetime.datetime.utcnow() + datetime.timedelta(days=30), last_used=datetime.datetime.utcnow(), created_at=datetime.datetime.utcnow() ) ctx.session.add(device) ctx.session.commit() except Exception as e: current_app.logger.error(f"Failed to add trusted device: {str(e)}") # Don't fail the login if device tracking fails pass def _send_new_device_alert(self, user, device_info): """Send email alert about new device login attempt""" subject = "New Device Login Alert" login_time = datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC') message = f"""

New Login Detected

Hello {user.first_name},

A login attempt was made from a new device:

Device Details:

Type: {device_info['device_type'].capitalize()}

IP Address: {device_info['ip_address']}

Time: {login_time}

User Agent: {device_info['user_agent']}

If this was you, you can ignore this message. If you don't recognize this activity, please secure your account immediately.

Best regards,

Mutable Tech Enterprises Team

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 ) except Exception as e: current_app.logger.error(f"Failed to send new device alert: {str(e)}") def _send_security_alert(self, user, reason): """Send security alert email""" subject = "Security Alert for Your Account" alert_time = datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC') message = f"""

Security Alert

Hello {user.first_name},

We detected unusual activity on your account:

{reason}

Details:

Time: {alert_time}

IP Address: {request.remote_addr}

If this wasn't you, please secure your account immediately by changing your password.

Best regards,

Mutable Tech Enterprises Team

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 ) except Exception as e: current_app.logger.error(f"Failed to send security alert: {str(e)}") 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') # Update user last activity (if needed) with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter(User.id == user_id).first() if user: user.last_activity = datetime.datetime.utcnow() ctx.session.commit() except: # Token decoding failed, but we still want to clear cookies 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 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(User).filter(User.id == user_id).first() if not user or not user.is_active: return custom_response( success=False, data="Invalid token", status_code=401 ) # Update last activity user.last_activity = datetime.datetime.utcnow() ctx.session.commit() # Get role-specific data if needed user_data = { "id": user.id, "first_name": user.first_name, "last_name": user.last_name, "email": user.email, "user_type": user.user_type.value } return custom_response( success=True, data={"user": user_data}, 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 request_password_reset(self, payload): """Handle password reset request""" email = payload.get('email') if not email: return custom_response( success=False, data="Email is required", status_code=400 ) with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter( User.email == email ).first() if not user: return custom_response( success=False, data="If this email exists, a reset link has been sent", status_code=200 ) # Generate a unique reset token reset_token = uuid.uuid4().hex user.reset_token = reset_token user.reset_token_expiry = datetime.datetime.utcnow() + datetime.timedelta(minutes=30) ctx.session.commit() # Send reset email reset_link = f"{current_app.config['FRONTEND_URL']}/reset-password?token={reset_token}" subject = "Password Reset Request" message = f"""

Password Reset Request

Hello {user.first_name},

We received a request to reset your password for your LMS account.

Click the button below to securely set a new password. This link will expire in 30 minutes for your security.

Reset Password

If you did not request this, you can safely ignore this email. Your password will remain unchanged.

Best regards,

Mutable Tech Enterprises Team

LMS

""" if not send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=user.email, subject=subject, message=message ): return custom_response( success=False, data="Failed to send reset email", status_code=500 ) return custom_response( success=True, data="If this email exists, a reset link has been sent", status_code=200 ) def verify_reset_token(self, payload): """Verify the password reset token""" reset_token = payload.get('token') if not reset_token: return custom_response( success=False, data="Reset token is required", status_code=400 ) with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter( User.reset_token == reset_token, User.reset_token_expiry > datetime.datetime.utcnow() ).first() if not user: return custom_response( success=False, data="Invalid or expired reset token", status_code=400 ) return custom_response( success=True, data="Reset token is valid", status_code=200 ) def reset_password(self, payload): """Reset the user's password""" reset_token = payload.get('token') new_password = payload.get('new_password') if not reset_token or not new_password: return custom_response( success=False, data="Reset token and new password are required", status_code=400 ) with DatabaseContextManager() as ctx: user = ctx.session.query(User).filter( User.reset_token == reset_token, User.reset_token_expiry > datetime.datetime.utcnow() ).first() if not user: return custom_response( success=False, data="Invalid or expired reset token", status_code=400 ) # Update password user.password_hash = hash_password(new_password, salt=current_app.config['SECRET_KEY']) user.reset_token = None user.reset_token_expiry = None user.last_password_change = datetime.datetime.utcnow() ctx.session.commit() # Send confirmation email subject = "Password Successfully Updated - Action Required if Unauthorized" message = f"""

Password Successfully Updated

Dear {user.first_name},

We are writing to confirm that your account password has been successfully updated on {datetime.datetime.now().strftime('%B %d, %Y at %I:%M %p')}.

IMPORTANT SECURITY NOTICE:
If you did not initiate this password change, your account security may be compromised. Please take immediate action by:

  1. Contacting our support team immediately at info@mutabletech.co.ke
  2. Reviewing your recent account activity
  3. Ensuring your recovery information is up to date

If you did make this change, no further action is required. Your account remains secure.

Best regards,

Mutable Tech Enterprises Team

LMS

""" send_email( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config['MAIL_PASSWORD'], receiver_email=user.email, subject=subject, message=message ) return custom_response( success=True, data="Password reset successfully", status_code=200 ) def enable_two_factor(self, user_id, method): """Enable two-factor authentication for a user""" with DatabaseContextManager() as ctx: 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 ) if method not in [m.value for m in TwoFactorMethod]: return custom_response( success=False, data="Invalid 2FA method", status_code=400 ) user.two_factor_enabled = True user.two_factor_method = method # Generate a secret if using authenticator app if method == TwoFactorMethod.AUTHENTICATOR_APP.value: user.two_factor_secret = str(uuid.uuid4()) # Generate backup codes backup_codes = [generate_otp() for _ in range(5)] user.two_factor_backup_codes = ",".join(backup_codes) ctx.session.commit() return custom_response( success=True, data={ "message": "2FA enabled successfully", "backup_codes": backup_codes, "secret": user.two_factor_secret if method == TwoFactorMethod.AUTHENTICATOR_APP.value else None } ) def disable_two_factor(self, user_id): """Disable two-factor authentication for a user""" with DatabaseContextManager() as ctx: 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 ) user.two_factor_enabled = False user.two_factor_method = TwoFactorMethod.NONE.value user.two_factor_secret = None user.two_factor_backup_codes = None ctx.session.commit() return custom_response( success=True, data="2FA disabled successfully" ) def get_notification_preferences(self, user_id): """Get notification preferences for a user""" with DatabaseContextManager() as ctx: prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == user_id ).first() if not prefs: return custom_response( success=False, data="Preferences not found", status_code=404 ) return custom_response( success=True, data=prefs.to_json() ) def update_notification_preferences(self, user_id, payload): """Update notification preferences for a user""" with DatabaseContextManager() as ctx: prefs = ctx.session.query(NotificationPreference).filter( NotificationPreference.user_id == user_id ).first() if not prefs: return custom_response( success=False, data="Preferences not found", status_code=404 ) # Update fields that are present in payload for field in ['receive_email', 'receive_sms', 'receive_push', 'email_notification_time', 'sms_notification_time', 'push_notification_time', 'notify_attendance_updates', 'notify_grade_updates', 'notify_schedule_changes', 'notify_system_messages']: if field in payload: setattr(prefs, field, payload[field]) ctx.session.commit() return custom_response( success=True, data="Preferences updated successfully" ) def fetchAll(self, page=1, per_page=10): """Fetch all users with pagination""" with DatabaseContextManager() as ctx: users = ctx.session.query(User).filter( User.is_active == True ).offset((page - 1) * per_page).limit(per_page).all() total = ctx.session.query(User).filter( User.is_active == True ).count() # Serialize users properly based on their type serialized_users = [] for user in users: # Base user data user_data = { "id": user.id, "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "user_type": user.user_type.value if user.user_type else None, "is_active": user.is_active, "phone": user.phone, "profile_picture": user.profile_picture, "gender": user.gender, "date_of_birth": str(user.date_of_birth) if user.date_of_birth else None, "nationality": user.nationality, "created_at": user.created_at.isoformat() if user.created_at else None, "last_login": user.last_login.isoformat() if user.last_login else None, "two_factor_enabled": user.two_factor_enabled, "two_factor_method": user.two_factor_method.value if user.two_factor_method else None } # Add role-specific data if user.user_type == UserType.tutor: tutor = ctx.session.query(Tutor).filter(Tutor.id == user.id).first() if tutor: user_data.update(self._tutor_to_dict(tutor)) elif user.user_type == UserType.supervisor: supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == user.id).first() if supervisor: user_data.update(self._supervisor_to_dict(supervisor)) elif user.user_type == UserType.student: student = ctx.session.query(Student).filter(Student.id == user.id).first() if student: user_data.update(self._student_to_dict(student)) serialized_users.append(user_data) return custom_response( success=True, data={ "users": serialized_users, "total": total, "page": page, "per_page": per_page } ) def fetchAllForChat(self): """Fetch all users for chat system without pagination""" with DatabaseContextManager() as ctx: users = ctx.session.query(User).filter( User.is_active == True ).all() # Serialize users for chat (minimal data needed) serialized_users = [] for user in users: # Base user data for chat user_data = { "id": user.id, "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "user_type": user.user_type.value if user.user_type else None, "profile_picture": user.profile_picture } serialized_users.append(user_data) return custom_response( success=True, data=serialized_users ) def filter(self, filter_params, page=1, per_page=10): """Filter users based on criteria with pagination""" with DatabaseContextManager() as ctx: query = ctx.session.query(User).filter(User.is_active == True) # Apply filters if 'user_type' in filter_params: query = query.filter(User.user_type == filter_params['user_type']) if 'department' in filter_params: # Handle department filtering through the new many-to-many relationship from src.models.models import TutorDepartment query = query.join(Tutor).join(TutorDepartment).filter( TutorDepartment.department_name == filter_params['department'], TutorDepartment.is_active == True ) if 'program' in filter_params: query = query.join(Student).filter(Student.program == filter_params['program']) if 'is_active' in filter_params: query = query.filter(User.is_active == filter_params['is_active']) total = query.count() users = query.offset((page - 1) * per_page).limit(per_page).all() # Serialize users properly based on their type serialized_users = [] for user in users: # Base user data user_data = { "id": user.id, "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "user_type": user.user_type.value if user.user_type else None, "is_active": user.is_active, "phone": user.phone, "profile_picture": user.profile_picture, "gender": user.gender, "date_of_birth": str(user.date_of_birth) if user.date_of_birth else None, "nationality": user.nationality, "created_at": user.created_at.isoformat() if user.created_at else None, "last_login": user.last_login.isoformat() if user.last_login else None, "two_factor_enabled": user.two_factor_enabled, "two_factor_method": user.two_factor_method.value if user.two_factor_method else None } # Add role-specific data if user.user_type == UserType.tutor: tutor = ctx.session.query(Tutor).filter(Tutor.id == user.id).first() if tutor: user_data.update(self._tutor_to_dict(tutor)) elif user.user_type == UserType.supervisor: supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == user.id).first() if supervisor: user_data.update(self._supervisor_to_dict(supervisor)) elif user.user_type == UserType.student: student = ctx.session.query(Student).filter(Student.id == user.id).first() if student: user_data.update(self._student_to_dict(student)) serialized_users.append(user_data) return custom_response( success=True, data={ "users": serialized_users, "total": total, "page": page, "per_page": per_page } ) def get_user_dashboard_summary(self, user_id): """Get comprehensive summary data for user dashboard based on their role""" with DatabaseContextManager() as ctx: 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 ) summary = { "user_info": { "id": user.id, "name": f"{user.first_name} {user.last_name}", "email": user.email, "user_type": user.user_type.value, "profile_picture": user.profile_picture }, "notifications": self._get_recent_notifications(ctx, user_id), } if user.user_type == UserType.student: # Student dashboard summary summary.update(self._get_student_dashboard_summary(ctx, user_id)) elif user.user_type == UserType.tutor: # Tutor dashboard summary summary.update(self._get_tutor_dashboard_summary(ctx, user_id)) elif user.user_type == UserType.supervisor: # Supervisor dashboard summary summary.update(self._get_supervisor_dashboard_summary(ctx, user_id)) return custom_response( success=True, data=summary ) def debug_enrollment_data(self): """Debug function to check enrollment data in the database""" try: with DatabaseContextManager() as ctx: # Check total counts total_courses = ctx.session.query(func.count(Course.id)).scalar() total_students = ctx.session.query(func.count(Student.id)).scalar() # Note: course_student_association no longer exists - using enrollment model total_associations = ctx.session.query(func.count(Enrollment.id)).scalar() # Check if association table has any data if total_associations == 0: # Try to create a test association try: # Get first course and first student first_course = ctx.session.query(Course).first() first_student = ctx.session.query(Student).first() if first_course and first_student: # Create enrollment record enrollment = Enrollment( student_id=first_student.id, course_id=first_course.id, speciality_id=first_student.speciality_id, enrollment_date=datetime.utcnow().date(), status='active' ) ctx.session.add(enrollment) ctx.session.commit() except Exception as e: pass # Check sample data # Sample courses courses = ctx.session.query(Course).limit(3).all() # Sample students students = ctx.session.query(Student).limit(3).all() # Sample enrollments enrollments = ctx.session.query(Enrollment).limit(5).all() # Check specific course enrollment if courses: course = courses[0] # Direct query through enrollment direct_count = ctx.session.query(func.count(Enrollment.student_id)).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).filter( enrollment_courses.c.course_id == course.id, enrollment_courses.c.status == 'active', Enrollment.status == 'active' ).scalar() # Through Student model try: student_count = ctx.session.query(func.count(Student.id)).join( Enrollment, Student.id == Enrollment.student_id ).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).filter( enrollment_courses.c.course_id == course.id, enrollment_courses.c.status == 'active', Enrollment.status == 'active' ).scalar() except Exception as e: pass # Check database schema try: inspector = inspect(ctx.session.bind) tables = inspector.get_table_names() # Check for any table that might contain enrollment data for table_name in tables: if 'enroll' in table_name.lower() or 'student' in table_name.lower() or 'course' in table_name.lower(): try: # Get column info columns = inspector.get_columns(table_name) # Get row count using SQLAlchemy from sqlalchemy import text result = ctx.session.execute(text(f"SELECT COUNT(*) FROM {table_name}")) count = result.scalar() # If it's a small table, show sample data if count <= 10 and count > 0: result = ctx.session.execute(text(f"SELECT * FROM {table_name} LIMIT 3")) rows = result.fetchall() except Exception as e: pass except Exception as e: pass return custom_response( success=True, data={ "total_courses": total_courses, "total_students": total_students, "total_associations": total_associations, "sample_courses": [{"code": c.code, "id": c.id} for c in courses], "sample_students": [{"name": f"{s.first_name} {s.last_name}", "id": s.id} for s in students], "sample_associations": [{"course_id": a.course_id, "student_id": a.student_id} for a in enrollments] } ) except Exception as e: return custom_response( success=False, data=f"Debug function error: {str(e)}", status_code=500 ) def _get_recent_notifications(self, ctx, user_id): """Get the 5 most recent notifications for a user""" # Query real notifications from the database # Note: This would require a notifications table in a real implementation # For now, return empty array to avoid dummy data return [] def _get_student_dashboard_summary(self, ctx, user_id): """Generate comprehensive dashboard summary for a student""" from sqlalchemy import func, desc, and_ from datetime import datetime, timedelta, date # Get student information student = ctx.session.query(Student).filter(Student.id == user_id).first() # Get enrolled courses through enrollment courses = ctx.session.query(Course).join( enrollment_courses, Course.id == enrollment_courses.c.course_id ).join( Enrollment, enrollment_courses.c.enrollment_id == Enrollment.id ).filter( Enrollment.student_id == user_id, enrollment_courses.c.status == 'active', Enrollment.status == 'active', Course.is_active == True ).all() # Get upcoming classes (teaching sessions) - both recurring and daily sessions today = date.today() upcoming_classes = [] # Get recurring teaching sessions recurring_sessions = ctx.session.query(TeachingSession).join( enrollment_courses, TeachingSession.course_id == enrollment_courses.c.course_id ).join( Enrollment, enrollment_courses.c.enrollment_id == Enrollment.id ).join( Course, TeachingSession.course_id == Course.id ).filter( Enrollment.student_id == user_id, enrollment_courses.c.status == 'active', Enrollment.status == 'active', TeachingSession.status == 'scheduled' ).all() # Get daily teaching sessions for today and future dates daily_sessions = ctx.session.query(DailyTeachingSession).join( enrollment_courses, DailyTeachingSession.course_id == enrollment_courses.c.course_id ).join( Enrollment, enrollment_courses.c.enrollment_id == Enrollment.id ).join( Course, DailyTeachingSession.course_id == Course.id ).filter( Enrollment.student_id == user_id, enrollment_courses.c.status == 'active', Enrollment.status == 'active', DailyTeachingSession.session_date >= today, DailyTeachingSession.status == 'scheduled' ).order_by(DailyTeachingSession.session_date, DailyTeachingSession.start_time).limit(10).all() # Combine and format upcoming classes for session in daily_sessions: upcoming_classes.append({ "id": session.id, "title": f"{session.course.code} - {session.course.title}", "course_code": session.course.code, "course_title": session.course.title, "start_time": f"{session.session_date}T{session.start_time}", "end_time": f"{session.session_date}T{session.end_time}", "location": session.room or "TBD", "session_type": session.session_type, "tutor_name": f"{session.tutor.first_name} {session.tutor.last_name}" if session.tutor else "TBD" }) # Get upcoming assignments upcoming_assignments = ctx.session.query(Assignment).join( enrollment_courses, Assignment.course_id == enrollment_courses.c.course_id ).join( Enrollment, enrollment_courses.c.enrollment_id == Enrollment.id ).join( Course, Assignment.course_id == Course.id ).filter( Enrollment.student_id == user_id, enrollment_courses.c.status == 'active', Enrollment.status == 'active', Assignment.due_date > datetime.utcnow(), Assignment.is_published == True ).order_by(Assignment.due_date).limit(8).all() # Get recent submissions with grades recent_submissions = ctx.session.query(AssignmentSubmission).join( Assignment, AssignmentSubmission.assignment_id == Assignment.id ).join( Course, Assignment.course_id == Course.id ).filter( AssignmentSubmission.student_id == user_id ).order_by(desc(AssignmentSubmission.submitted_at)).limit(5).all() # Calculate comprehensive attendance statistics attendance_records = ctx.session.query(Attendance).filter( Attendance.student_id == user_id ).all() present_count = sum(1 for r in attendance_records if r.status == AttendanceStatus.present) absent_count = sum(1 for r in attendance_records if r.status == AttendanceStatus.absent) late_count = sum(1 for r in attendance_records if r.status == AttendanceStatus.late) attendance_percentage = round((present_count / len(attendance_records)) * 100, 1) if attendance_records else 0 # Get course-wise attendance course_attendance_stats = [] for course in courses: course_attendance = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( Attendance.student_id == user_id, TeachingSession.course_id == course.id ).all() course_present = sum(1 for r in course_attendance if r.status == AttendanceStatus.present) course_total = len(course_attendance) course_percentage = round((course_present / course_total) * 100, 1) if course_total > 0 else 0 course_attendance_stats.append({ "course_id": course.id, "course_code": course.code, "course_title": course.title, "attendance_rate": course_percentage, "present_sessions": course_present, "total_sessions": course_total, "missed_sessions": course_total - course_present }) # Get recent learning materials (real data from database) # Note: This would require a learning_materials table in a real implementation # For now, return empty array to avoid dummy data learning_materials = [] # Calculate course progress course_progress = [] for course in courses: # Get total assignments for this course total_assignments = ctx.session.query(func.count(Assignment.id)).filter( Assignment.course_id == course.id, Assignment.is_published == True ).scalar() # Get submitted assignments submitted_assignments = ctx.session.query(func.count(AssignmentSubmission.id)).join( Assignment, AssignmentSubmission.assignment_id == Assignment.id ).filter( AssignmentSubmission.student_id == user_id, Assignment.course_id == course.id ).scalar() # Calculate progress percentage progress_percentage = round((submitted_assignments / total_assignments) * 100, 1) if total_assignments > 0 else 0 # Get average grade for this course avg_grade = ctx.session.query(func.avg(AssignmentSubmission.grade)).join( Assignment, AssignmentSubmission.assignment_id == Assignment.id ).filter( AssignmentSubmission.student_id == user_id, Assignment.course_id == course.id, AssignmentSubmission.grade.isnot(None) ).scalar() course_progress.append({ "course_id": course.id, "course_code": course.code, "course_title": course.title, "progress_percentage": progress_percentage, "assignments_completed": submitted_assignments, "total_assignments": total_assignments, "average_grade": round(avg_grade, 1) if avg_grade else None, "next_deadline": None # Will be populated from upcoming assignments }) # Calculate overall academic metrics all_grades = ctx.session.query(AssignmentSubmission.grade).filter( AssignmentSubmission.student_id == user_id, AssignmentSubmission.grade.isnot(None) ).all() overall_gpa = 0 if all_grades: # Convert grades to GPA scale (assuming 0-100 scale) grade_values = [g[0] for g in all_grades] avg_grade = sum(grade_values) / len(grade_values) # Convert to 4.0 scale overall_gpa = min(4.0, (avg_grade / 100) * 4.0) # Assignment completion rate total_published_assignments = ctx.session.query(func.count(Assignment.id)).join( enrollment_courses, Assignment.course_id == enrollment_courses.c.course_id ).join( Enrollment, enrollment_courses.c.enrollment_id == Enrollment.id ).filter( Enrollment.student_id == user_id, enrollment_courses.c.status == 'active', Enrollment.status == 'active', Assignment.is_published == True ).scalar() total_submitted = ctx.session.query(func.count(AssignmentSubmission.id)).filter( AssignmentSubmission.student_id == user_id ).scalar() assignment_completion_rate = round((total_submitted / total_published_assignments) * 100, 1) if total_published_assignments > 0 else 0 return { "courses_count": len(courses), "courses": [{ "id": c.id, "code": c.code, "title": c.title, "credits": c.credits, "department": c.department, "semester": c.semester } for c in courses], "upcoming_classes": upcoming_classes, "upcoming_assignments": [{ "id": a.id, "title": a.title, "course_code": a.course.code, "course_title": a.course.title, "due_date": a.due_date.isoformat(), "assignment_type": a.assignment_type.value, "points": a.points, "description": a.description[:100] + "..." if a.description and len(a.description) > 100 else a.description } for a in upcoming_assignments], "recent_submissions": [{ "id": s.id, "assignment_title": s.assignment.title, "course_code": s.assignment.course.code, "course_title": s.assignment.course.title, "status": s.status.value, "grade": s.grade, "max_points": s.assignment.points, "submitted_at": s.submitted_at.isoformat() if s.submitted_at else None, "feedback": s.feedback[:100] + "..." if s.feedback and len(s.feedback) > 100 else s.feedback } for s in recent_submissions], "attendance": { "percentage": attendance_percentage, "present": present_count, "absent": absent_count, "late": late_count, "total": len(attendance_records), "trend": "improving" if attendance_percentage >= 80 else "stable" if attendance_percentage >= 70 else "needs_attention" }, "attendance_analytics": { "overall_attendance_rate": attendance_percentage, "perfect_attendance_days": present_count, "attendance_rank": "1st" if attendance_percentage >= 95 else "Top 10%" if attendance_percentage >= 85 else "Above Average", "weekly_trend": self._get_real_weekly_attendance_trend(ctx, user_id), "course_attendance_stats": course_attendance_stats }, "learning_materials": learning_materials, "course_progress": course_progress, "performance_metrics": { "overall_gpa": round(overall_gpa, 2), "assignment_completion_rate": assignment_completion_rate, "total_assignments": total_published_assignments, "completed_assignments": total_submitted, "average_grade": round(sum([g[0] for g in all_grades]) / len(all_grades), 1) if all_grades else 0 }, "academic_info": { "student_id": student.student_id if student else None, "program": student.program if student else None, "department": student.department if student else None, "year_of_study": student.year_of_study if student else None, "current_semester": student.current_semester if student else None, "enrollment_date": student.enrollment_date.isoformat() if student and student.enrollment_date else None } } def _get_real_weekly_attendance_trend(self, ctx, user_id): """Get real weekly attendance trends based on actual attendance data""" from datetime import datetime, timedelta # Get attendance records for the last 8 weeks eight_weeks_ago = datetime.utcnow() - timedelta(weeks=8) # Get all attendance records for the student in the last 8 weeks attendance_records = ctx.session.query(Attendance).filter( Attendance.student_id == user_id, Attendance.created_at >= eight_weeks_ago ).order_by(Attendance.created_at).all() # Group by week and calculate attendance rates weekly_data = {} for record in attendance_records: # Get the week number (1-8) week_number = ((record.created_at - eight_weeks_ago).days // 7) + 1 if week_number > 8: continue if week_number not in weekly_data: weekly_data[week_number] = {'total': 0, 'present': 0} weekly_data[week_number]['total'] += 1 if record.status == AttendanceStatus.present: weekly_data[week_number]['present'] += 1 # Format the data for the frontend trend_data = [] for week_num in range(1, 9): if week_num in weekly_data: total = weekly_data[week_num]['total'] present = weekly_data[week_num]['present'] attendance_rate = round((present / total) * 100, 1) if total > 0 else 0 else: attendance_rate = 0 trend_data.append({ "week": f"Week {week_num}", "attendance_rate": attendance_rate }) return trend_data def _get_tutor_dashboard_summary(self, ctx, user_id): """Generate comprehensive dashboard summary for a tutor""" # Get assigned courses courses = ctx.session.query(Course).join( tutor_course_association, Course.id == tutor_course_association.c.course_id ).filter( tutor_course_association.c.tutor_id == user_id, Course.is_active == True ).all() # Get upcoming daily teaching sessions (next 7 days) today = datetime.datetime.utcnow().date() next_week = today + datetime.timedelta(days=7) upcoming_daily_sessions = ctx.session.query(DailyTeachingSession).join( TeachingSession, DailyTeachingSession.teaching_session_id == TeachingSession.id ).filter( TeachingSession.tutor_id == user_id, DailyTeachingSession.session_date >= today, DailyTeachingSession.session_date <= next_week, DailyTeachingSession.status != 'cancelled' ).order_by(DailyTeachingSession.session_date, DailyTeachingSession.start_time).limit(10).all() # Get students count across all courses through enrollment students_count = ctx.session.query(func.count(func.distinct(Enrollment.student_id))).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).join( tutor_course_association, enrollment_courses.c.course_id == tutor_course_association.c.course_id ).filter( tutor_course_association.c.tutor_id == user_id, enrollment_courses.c.status == 'active', Enrollment.status == 'active' ).scalar() # Get assignments needing grading pending_grading = ctx.session.query(AssignmentSubmission).join( Assignment, AssignmentSubmission.assignment_id == Assignment.id ).join( tutor_course_association, Assignment.course_id == tutor_course_association.c.course_id ).filter( tutor_course_association.c.tutor_id == user_id, AssignmentSubmission.status == SubmissionStatus.submitted, AssignmentSubmission.grade == None ).order_by(AssignmentSubmission.submitted_at).limit(5).all() # Get recent attendance records recent_attendance = ctx.session.query(Attendance).filter( Attendance.tutor_id == user_id ).order_by(Attendance.timestamp.desc()).limit(5).all() # Get recent notifications for the tutor notifications = ctx.session.query(Notification).filter( Notification.user_id == user_id ).order_by(Notification.created_at.desc()).limit(10).all() # Calculate attendance percentage total_sessions = ctx.session.query(func.count(DailyTeachingSession.id)).join( TeachingSession, DailyTeachingSession.teaching_session_id == TeachingSession.id ).filter( TeachingSession.tutor_id == user_id, DailyTeachingSession.session_date <= today ).scalar() or 0 attended_sessions = ctx.session.query(func.count(func.distinct(Attendance.session_id))).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( TeachingSession.tutor_id == user_id ).scalar() or 0 attendance_percentage = round((attended_sessions / total_sessions * 100), 1) if total_sessions > 0 else 0 # Get teaching sessions by date for the next 7 days (for bar chart) teaching_sessions_by_date = [] for i in range(7): current_date = today + datetime.timedelta(days=i) session_count = ctx.session.query(func.count(DailyTeachingSession.id)).join( TeachingSession, DailyTeachingSession.teaching_session_id == TeachingSession.id ).filter( TeachingSession.tutor_id == user_id, DailyTeachingSession.session_date == current_date, DailyTeachingSession.status != 'cancelled' ).scalar() or 0 teaching_sessions_by_date.append({ "date": current_date.strftime('%a %d'), # e.g., "Mon 10" "sessions": session_count, "full_date": current_date.isoformat() }) return { "courses_count": len(courses), "courses": [{"id": c.id, "code": c.code, "title": c.title} for c in courses[:3]], # Top 3 courses "upcoming_classes": [{ "id": ds.id, "title": f"{ds.session.course.code} - {ds.session.course.title}" if ds.session and ds.session.course else "Teaching Session", "course_code": ds.session.course.code if ds.session and ds.session.course else "", "course_title": ds.session.course.title if ds.session and ds.session.course else "", "start_time": datetime.datetime.combine(ds.date, ds.start_time).isoformat() if ds.date and ds.start_time else None, "end_time": datetime.datetime.combine(ds.date, ds.end_time).isoformat() if ds.date and ds.end_time else None, "location": ds.session.room if ds.session else "TBA", "students_registered": ctx.session.query(func.count(Enrollment.id)).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).filter( enrollment_courses.c.course_id == ds.session.course_id, Enrollment.status == 'active' ).scalar() if ds.session else 0 } for ds in upcoming_daily_sessions], "students_count": students_count or 0, "pending_grading": [{ "id": s.id, "assignment_title": s.assignment.title if s.assignment else "Assignment", "student_name": f"{s.student.first_name} {s.student.last_name}" if s.student else "Student", "submitted_at": s.submitted_at.isoformat() if s.submitted_at else None, "days_late": s.late_days if hasattr(s, 'late_days') else 0 } for s in pending_grading], "recent_attendance": [{ "id": a.id, "student_name": f"{a.student.first_name} {a.student.last_name}" if a.student else "Student", "course_code": a.session.course.code if a.session and a.session.course else "", "status": a.status.value if hasattr(a.status, 'value') else str(a.status), "timestamp": a.timestamp.isoformat() if a.timestamp else None } for a in recent_attendance], "notifications": [{ "id": n.id, "title": n.title, "message": n.message, "timestamp": n.created_at.isoformat() if n.created_at else None, "read": n.is_read if hasattr(n, 'is_read') else False, "type": n.notification_type if hasattr(n, 'notification_type') else "general" } for n in notifications], "attendance": { "percentage": attendance_percentage, "present": attended_sessions, "total": total_sessions, "trend": "up" if attendance_percentage >= 80 else "down" if attendance_percentage < 60 else "stable" }, "teaching_sessions_by_date": teaching_sessions_by_date } def _get_supervisor_dashboard_summary(self, ctx, user_id): """Generate comprehensive dashboard summary for a supervisor""" # Get supervisor supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == user_id).first() if not supervisor: return {} # Get supervisor's departments supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] # Get managed courses (courses in supervisor's departments) courses = ctx.session.query(Course).filter( Course.department.in_(supervisor_departments), Course.is_active == True ).all() # Get managed tutors (tutors in supervisor's departments) tutors = ( ctx.session.query(Tutor) .join(TutorDepartment, Tutor.id == TutorDepartment.tutor_id) .filter( TutorDepartment.department_name.in_(supervisor_departments), TutorDepartment.is_active == True, Tutor.is_active == True ) .distinct() .all() ) # Get specialities in supervisor's departments total_specialities = ctx.session.query(Speciality).filter( Speciality.department.in_(supervisor_departments), Speciality.is_active == True ).count() # Get total courses count in supervisor's departments total_courses = ctx.session.query(Course).filter( Course.department.in_(supervisor_departments), Course.is_active == True ).count() # Get pending availability approvals pending_approvals = ctx.session.query(TutorAvailability).filter( TutorAvailability.is_approved == False, TutorAvailability.tutor_id.in_([t.id for t in tutors]) ).count() # Get teaching sessions needing verification sessions_needing_verification = ctx.session.query(TeachingSession).filter( TeachingSession.is_verified == False, TeachingSession.end_time < datetime.datetime.utcnow(), TeachingSession.course_id.in_([c.id for c in courses]) ).order_by(TeachingSession.end_time).limit(5).all() # Get recent timetable changes recent_timetable_changes = ctx.session.query(TimetableBlock).filter( TimetableBlock.created_by == user_id ).order_by(TimetableBlock.created_at.desc()).limit(5).all() # Get top 3 teaching lessons (most recent completed sessions) top_teaching_lessons = ctx.session.query(TeachingSession).filter( TeachingSession.course_id.in_([c.id for c in courses]), TeachingSession.status == 'completed', TeachingSession.end_time < datetime.datetime.utcnow() ).order_by(TeachingSession.end_time.desc()).limit(3).all() # Enhanced courses data with enrollment and attendance information enhanced_courses = [] for course in courses: # Get student count through enrollment student_count = ctx.session.query(func.count(func.distinct(Enrollment.student_id))).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).filter( enrollment_courses.c.course_id == course.id, enrollment_courses.c.status == 'active', Enrollment.status == 'active' ).scalar() try: # Alternative count through speciality student_count_alt = ctx.session.query(func.count(Student.id)).filter( Student.speciality_id == course.speciality_id ).scalar() # Use the alternative count if the first one is 0 if student_count == 0 and student_count_alt > 0: student_count = student_count_alt except Exception as e: print(f" Approach 2 failed: {e}") try: # Check if there are any enrollments in other tables # Sometimes enrollments might be stored in a different table try: # Check if there's an enrollments table inspector = inspect(ctx.session.bind) tables = inspector.get_table_names() # Check for any table that might contain enrollment data for table_name in tables: if 'enroll' in table_name.lower() or 'student' in table_name.lower(): try: # Try to query this table using SQLAlchemy text from sqlalchemy import text result = ctx.session.execute(text(f"SELECT COUNT(*) FROM {table_name}")) count = result.scalar() except Exception as e: pass except Exception as e: pass except Exception as e: pass # Calculate attendance rate for this course attendance_records = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( TeachingSession.course_id == course.id ).all() attendance_rate = 0 if attendance_records: present_count = sum(1 for r in attendance_records if r.status == AttendanceStatus.present) attendance_rate = round((present_count / len(attendance_records)) * 100, 1) # If no students found, skip creating test enrollments (enrollment model restructured) if student_count == 0: pass # Skip test enrollment creation as it's no longer valid with new model enhanced_courses.append({ "id": course.id, "code": course.code, "title": course.title, "department": course.department, "student_count": student_count, "attendance_rate": attendance_rate }) # Get exam evidence count for supervisor's departments exam_evidence_count = ctx.session.query(ExamEvidence).join( Student, ExamEvidence.student_id == Student.id ).filter( Student.department.in_(supervisor_departments), ExamEvidence.is_active == True ).count() # Get total daily teaching sessions for supervisor's courses total_daily_sessions = ctx.session.query(DailyTeachingSession).filter( DailyTeachingSession.course_id.in_([c.id for c in courses]) ).count() # Get sessions today today = date.today() sessions_today = ctx.session.query(DailyTeachingSession).filter( DailyTeachingSession.course_id.in_([c.id for c in courses]), DailyTeachingSession.session_date == today ).count() # Get total students in supervisor's departments total_students = ctx.session.query(Student).filter( Student.department.in_(supervisor_departments), Student.is_active == True ).count() # Get detailed tutor workload analysis (only for managed tutors) tutor_workload_analysis = self._get_supervisor_specific_tutor_workload_analysis(ctx, tutors, courses) # Get department-specific insights department_insights = self._get_department_insights(ctx, supervisor_departments, courses, tutors) return { "courses_count": len(courses), "total_courses": total_courses, "total_specialities": total_specialities, "total_students": total_students, "exam_evidence_count": exam_evidence_count, "total_daily_sessions": total_daily_sessions, "sessions_today": sessions_today, "courses": enhanced_courses, "tutors_count": len(tutors), "tutors": [{"id": t.id, "name": f"{t.first_name} {t.last_name}"} for t in tutors[:5]], "pending_approvals": pending_approvals, "sessions_needing_verification": [{ "id": s.id, "session_title": s.title or f"{s.course.code} - {s.course.title}", "tutor_name": f"{s.tutor.first_name} {s.tutor.last_name}", "course_code": s.course.code, "course_title": s.course.title, "start_time": s.start_time.isoformat() if s.start_time else None, "end_time": s.end_time.isoformat() if s.end_time else None, "session_date": s.created_at.isoformat() if s.created_at else None, "duration": int((s.end_time.hour * 60 + s.end_time.minute) - (s.start_time.hour * 60 + s.start_time.minute)) if s.start_time and s.end_time else 0 } for s in sessions_needing_verification], "recent_timetable_changes": [{ "id": b.id, "course_code": b.course.code, "tutor_name": f"{b.tutor.first_name} {b.tutor.last_name}" if b.tutor else "Unknown", "day": ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"][b.day_of_week], "time": f"{b.start_time.strftime('%H:%M')} - {b.end_time.strftime('%H:%M')}", "change_date": b.created_at.isoformat() if b.created_at else None } for b in recent_timetable_changes], "top_teaching_lessons": [{ "id": s.id, "title": s.title or f"{s.course.code} - {s.course.title}", "course_code": s.course.code, "course_title": s.course.title, "tutor_name": f"{s.tutor.first_name} {s.tutor.last_name}", "start_time": s.start_time.strftime('%H:%M:%S') if s.start_time else None, "end_time": s.end_time.strftime('%H:%M:%S') if s.end_time else None, "location": s.room, "session_date": s.start_time.isoformat() if s.start_time else None, "duration": int((s.end_time.hour * 60 + s.end_time.minute) - (s.start_time.hour * 60 + s.start_time.minute)) if s.start_time and s.end_time else 0, "attendance_count": len(s.attendance_records) if hasattr(s, 'attendance_records') else 0, "verification_status": s.is_verified } for s in top_teaching_lessons], "tutor_workload_analysis": tutor_workload_analysis, "department_insights": department_insights, "supervisor_departments": supervisor_departments } def _get_supervisor_specific_tutor_workload_analysis(self, ctx, managed_tutors, managed_courses): """Get tutor workload analysis for only the supervisor's managed tutors""" tutor_workloads = [] course_ids = [c.id for c in managed_courses] for tutor in managed_tutors: # Get courses assigned to this tutor (only from managed courses) 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.id.in_(course_ids), Course.is_active == True ).all() total_credits = sum(c.credits or 0 for c in tutor_courses) primary_courses = sum(1 for c in tutor_courses if self._is_tutor_primary_for_course(ctx, tutor.id, c.id)) estimated_students = sum(self._get_course_student_count(ctx, c.id) for c in tutor_courses) # Determine workload status based on credit hours if total_credits < 24: workload_status = 'underloaded' elif total_credits <= 28: workload_status = 'balanced' else: workload_status = 'overloaded' # Calculate efficiency score efficiency_score = min(100, (total_credits / 26) * 100) if total_credits > 0 else 0 tutor_workloads.append({ 'tutor_id': tutor.id, 'name': f"{tutor.first_name} {tutor.last_name}", 'email': tutor.email, 'department': tutor.get_primary_department() or 'Unknown', 'courses': [{ 'course_id': c.id, 'course_code': c.code, 'course_title': c.title, 'credits': c.credits or 0, 'is_primary': self._is_tutor_primary_for_course(ctx, tutor.id, c.id) } for c in tutor_courses], 'total_credits': total_credits, 'total_courses': len(tutor_courses), 'primary_courses': primary_courses, 'secondary_courses': len(tutor_courses) - primary_courses, 'estimated_students': estimated_students, 'workload_status': workload_status, 'efficiency_score': round(efficiency_score, 1) }) # Calculate summary statistics summary = { 'total_tutors': len(tutor_workloads), 'underloaded_count': sum(1 for t in tutor_workloads if t['workload_status'] == 'underloaded'), 'balanced_count': sum(1 for t in tutor_workloads if t['workload_status'] == 'balanced'), 'overloaded_count': sum(1 for t in tutor_workloads if t['workload_status'] == 'overloaded'), 'average_efficiency': round(sum(t['efficiency_score'] for t in tutor_workloads) / len(tutor_workloads), 1) if tutor_workloads else 0 } return { 'tutor_workloads': tutor_workloads, 'summary': summary } def _is_tutor_primary_for_course(self, ctx, tutor_id, course_id): """Check if tutor is primary for a course""" result = ctx.session.query(tutor_course_association).filter( tutor_course_association.c.tutor_id == tutor_id, tutor_course_association.c.course_id == course_id ).first() return result.is_primary if result else False def _get_course_student_count(self, ctx, course_id): """Get student count for a course""" count = ctx.session.query(func.count(func.distinct(Enrollment.student_id))).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).filter( enrollment_courses.c.course_id == course_id, enrollment_courses.c.status == 'active' ).scalar() return count or 0 def _get_department_insights(self, ctx, supervisor_departments, courses, tutors): """Get creative department insights for supervisor dashboard""" # Calculate various metrics total_credits = sum(c.credits or 0 for c in courses) avg_course_credits = round(total_credits / len(courses), 1) if courses else 0 # Get practical vs theory courses practical_courses = sum(1 for c in courses if c.has_practical) theory_courses = len(courses) - practical_courses # Get shared courses count shared_courses = sum(1 for c in courses if c.is_shared_course) # Get tutors on leave tutors_on_leave = sum(1 for t in tutors if t.is_on_leave) available_tutors = len(tutors) - tutors_on_leave # Get course distribution by semester semester_distribution = {} for course in courses: semester = course.semester or 'Unknown' semester_distribution[semester] = semester_distribution.get(semester, 0) + 1 # Get top performing tutor (by course count) tutor_course_counts = {} for tutor in tutors: tutor_courses_count = ctx.session.query(tutor_course_association).filter( tutor_course_association.c.tutor_id == tutor.id, tutor_course_association.c.course_id.in_([c.id for c in courses]) ).count() if tutor_courses_count > 0: tutor_course_counts[f"{tutor.first_name} {tutor.last_name}"] = tutor_courses_count top_tutor = max(tutor_course_counts.items(), key=lambda x: x[1]) if tutor_course_counts else ("No data", 0) return { 'total_credits': total_credits, 'avg_course_credits': avg_course_credits, 'practical_courses': practical_courses, 'theory_courses': theory_courses, 'shared_courses': shared_courses, 'tutors_on_leave': tutors_on_leave, 'available_tutors': available_tutors, 'semester_distribution': semester_distribution, 'top_tutor': { 'name': top_tutor[0], 'course_count': top_tutor[1] }, 'departments': supervisor_departments, 'tutor_efficiency': round((available_tutors / len(tutors)) * 100, 1) if tutors else 100 } def get_attendance_analytics(self, ctx): """Get comprehensive attendance analytics for supervisor dashboard""" try: # Get overall attendance statistics total_sessions = ctx.session.query(TeachingSession).count() total_attendance_records = ctx.session.query(Attendance).count() # Calculate overall attendance rate overall_attendance_rate = 0 if total_attendance_records > 0: present_records = ctx.session.query(Attendance).filter( Attendance.status == 'present' ).count() overall_attendance_rate = round((present_records / total_attendance_records) * 100, 1) # Get tutor attendance statistics tutor_attendance_stats = [] tutors = ctx.session.query(Tutor).all() for tutor in tutors: tutor_sessions = ctx.session.query(TeachingSession).filter( TeachingSession.tutor_id == tutor.id ).count() tutor_attendance_records = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter(TeachingSession.tutor_id == tutor.id).count() tutor_present_records = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( TeachingSession.tutor_id == tutor.id, Attendance.status == 'present' ).count() tutor_attendance_rate = 0 if tutor_attendance_records > 0: tutor_attendance_rate = round((tutor_present_records / tutor_attendance_records) * 100, 1) tutor_attendance_stats.append({ 'tutor_id': tutor.id, 'tutor_name': f"{tutor.first_name} {tutor.last_name}", 'department': tutor.get_primary_department() or 'Unknown', 'total_sessions': tutor_sessions, 'attendance_records': tutor_attendance_records, 'attendance_rate': tutor_attendance_rate, 'status': 'excellent' if tutor_attendance_rate >= 95 else 'good' if tutor_attendance_rate >= 85 else 'needs_improvement' if tutor_attendance_rate >= 70 else 'poor' }) # Sort by attendance rate (descending) tutor_attendance_stats.sort(key=lambda x: x['attendance_rate'], reverse=True) # Get course-wise attendance statistics course_attendance_stats = [] courses = ctx.session.query(Course).all() for course in courses: course_sessions = ctx.session.query(TeachingSession).filter( TeachingSession.course_id == course.id ).count() course_attendance_records = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter(TeachingSession.course_id == course.id).count() course_present_records = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( TeachingSession.course_id == course.id, Attendance.status == 'present' ).count() course_attendance_rate = 0 if course_attendance_records > 0: course_attendance_rate = round((course_present_records / course_attendance_records) * 100, 1) # Get student count for this course student_count = ctx.session.query(Student).join( Enrollment, Student.id == Enrollment.student_id ).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).filter( enrollment_courses.c.course_id == course.id, enrollment_courses.c.status == 'active', Enrollment.status == 'active' ).count() course_attendance_stats.append({ 'course_id': course.id, 'course_code': course.code, 'course_title': course.title, 'department': course.department or 'Unknown', 'student_count': student_count, 'total_sessions': course_sessions, 'attendance_records': course_attendance_records, 'attendance_rate': course_attendance_rate, 'status': 'excellent' if course_attendance_rate >= 95 else 'good' if course_attendance_rate >= 85 else 'needs_improvement' if course_attendance_rate >= 70 else 'poor' }) # Sort by attendance rate (descending) course_attendance_stats.sort(key=lambda x: x['attendance_rate'], reverse=True) # Get weekly attendance trends (last 6 weeks) weekly_trends = [] from datetime import datetime, timedelta for i in range(6): week_start = datetime.now() - timedelta(weeks=i+1) week_end = datetime.now() - timedelta(weeks=i) week_sessions = ctx.session.query(TeachingSession).filter( TeachingSession.start_time >= week_start, TeachingSession.start_time < week_end ).count() week_attendance_records = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( TeachingSession.start_time >= week_start, TeachingSession.start_time < week_end ).count() week_present_records = ctx.session.query(Attendance).join( TeachingSession, Attendance.session_id == TeachingSession.id ).filter( TeachingSession.start_time >= week_start, TeachingSession.start_time < week_end, Attendance.status == 'present' ).count() week_attendance_rate = 0 if week_attendance_records > 0: week_attendance_rate = round((week_present_records / week_attendance_records) * 100, 1) weekly_trends.append({ 'week': f"Week {6-i}", 'week_start': week_start.strftime('%Y-%m-%d'), 'week_end': week_end.strftime('%Y-%m-%d'), 'sessions': week_sessions, 'attendance_rate': week_attendance_rate, 'attendance_records': week_attendance_records }) # Reverse to get chronological order weekly_trends.reverse() # Get attendance alerts (courses/tutors with low attendance) attendance_alerts = [] # Course alerts for course_stat in course_attendance_stats: if course_stat['attendance_rate'] < 75 and course_stat['attendance_records'] > 0: attendance_alerts.append({ 'type': 'course_low_attendance', 'severity': 'high' if course_stat['attendance_rate'] < 60 else 'medium', 'message': f"{course_stat['course_code']} attendance dropped below 75%", 'course_code': course_stat['course_code'], 'course_title': course_stat['course_title'], 'attendance_rate': course_stat['attendance_rate'], 'student_count': course_stat['student_count'], 'department': course_stat['department'] }) # Tutor alerts for tutor_stat in tutor_attendance_stats: if tutor_stat['attendance_rate'] < 85 and tutor_stat['attendance_records'] > 0: attendance_alerts.append({ 'type': 'tutor_low_attendance', 'severity': 'high' if tutor_stat['attendance_rate'] < 70 else 'medium', 'message': f"{tutor_stat['tutor_name']} attendance below 85%", 'tutor_name': tutor_stat['tutor_name'], 'department': tutor_stat['department'], 'attendance_rate': tutor_stat['attendance_rate'], 'total_sessions': tutor_stat['total_sessions'] }) # Sort alerts by severity attendance_alerts.sort(key=lambda x: 0 if x['severity'] == 'high' else 1) return { 'overview': { 'total_sessions': total_sessions, 'total_attendance_records': total_attendance_records, 'overall_attendance_rate': overall_attendance_rate, 'total_tutors': len(tutors), 'total_courses': len(courses) }, 'tutor_attendance_stats': tutor_attendance_stats[:10], # Top 10 tutors 'course_attendance_stats': course_attendance_stats[:10], # Top 10 courses 'weekly_trends': weekly_trends, 'attendance_alerts': attendance_alerts[:5], # Top 5 alerts 'summary': { 'excellent_tutors': len([t for t in tutor_attendance_stats if t['attendance_rate'] >= 95]), 'good_tutors': len([t for t in tutor_attendance_stats if 85 <= t['attendance_rate'] < 95]), 'needs_improvement_tutors': len([t for t in tutor_attendance_stats if 70 <= t['attendance_rate'] < 85]), 'poor_tutors': len([t for t in tutor_attendance_stats if t['attendance_rate'] < 70]), 'excellent_courses': len([c for c in course_attendance_stats if c['attendance_rate'] >= 95]), 'good_courses': len([c for c in course_attendance_stats if 85 <= c['attendance_rate'] < 95]), 'needs_improvement_courses': len([c for c in course_attendance_stats if 70 <= c['attendance_rate'] < 85]), 'poor_courses': len([c for c in course_attendance_stats if c['attendance_rate'] < 70]) } } except Exception as e: current_app.logger.error(f"Error fetching attendance analytics: {str(e)}", exc_info=True) return { 'overview': { 'total_sessions': 0, 'total_attendance_records': 0, 'overall_attendance_rate': 0, 'total_tutors': 0, 'total_courses': 0 }, 'tutor_attendance_stats': [], 'course_attendance_stats': [], 'weekly_trends': [], 'attendance_alerts': [], 'summary': { 'excellent_tutors': 0, 'good_tutors': 0, 'needs_improvement_tutors': 0, 'poor_tutors': 0, 'excellent_courses': 0, 'good_courses': 0, 'needs_improvement_courses': 0, 'poor_courses': 0 } } def update_password_by_email(self, payload): """Update user password by email address""" with DatabaseContextManager() as ctx: try: email = payload.get('email') new_password = payload.get('new_password') if not email or not new_password: return custom_response( success=False, data="Email and new password are required", status_code=400 ) # Find user by email user = ctx.session.query(User).filter(User.email == email).first() if not user: return custom_response( success=False, data="User not found with the specified email", status_code=404 ) # Hash the new password hashed_password = hash_password(new_password, salt=current_app.config['SECRET_KEY']) # Update the password user.password_hash = hashed_password user.last_password_change = datetime.datetime.utcnow() user.must_change_password = False # Clear any existing reset tokens user.reset_token = None user.reset_token_expiry = None ctx.session.commit() return custom_response( success=True, data={ "message": "Password updated successfully", "user_id": user.id, "email": user.email, "updated_at": user.last_password_change.isoformat() }, status_code=200 ) except Exception as e: ctx.session.rollback() current_app.logger.error(f"Error updating password: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to update password due to server error", status_code=500 ) def get_user_department(self, user_id: str): """Get a user's department based on their role""" with DatabaseContextManager() as ctx: try: # Get 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 ) # Get department based on user type department = None if user.user_type.value == "tutor": tutor = ctx.session.query(Tutor).filter(Tutor.id == user_id).first() if tutor and tutor.departments: # Get primary department - handle both dict and object formats primary_dept = None for dept in tutor.departments: if hasattr(dept, 'is_primary'): # Object format if dept.is_primary: primary_dept = dept break elif isinstance(dept, dict) and dept.get('is_primary', False): # Dict format primary_dept = dept break if primary_dept: if hasattr(primary_dept, 'department_name'): department = primary_dept.department_name elif isinstance(primary_dept, dict): department = primary_dept.get('department_name') elif tutor.departments: # Fallback to first department first_dept = tutor.departments[0] if hasattr(first_dept, 'department_name'): department = first_dept.department_name elif isinstance(first_dept, dict): department = first_dept.get('department_name') elif user.user_type.value == "supervisor": supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == user_id).first() if supervisor and supervisor.departments: # Get primary department - handle both dict and object formats primary_dept = None for dept in supervisor.departments: if hasattr(dept, 'is_primary'): # Object format if dept.is_primary: primary_dept = dept break elif isinstance(dept, dict) and dept.get('is_primary', False): # Dict format primary_dept = dept break if primary_dept: if hasattr(primary_dept, 'department_name'): department = primary_dept.department_name elif isinstance(primary_dept, dict): department = primary_dept.get('department_name') elif supervisor.departments: # Fallback to first department first_dept = supervisor.departments[0] if hasattr(first_dept, 'department_name'): department = first_dept.department_name elif isinstance(first_dept, dict): department = first_dept.get('department_name') elif user.user_type.value == "student": student = ctx.session.query(Student).filter(Student.id == user_id).first() if student: department = student.program # Students might have program instead of department return custom_response( success=True, data={ "department": department, "user_type": user.user_type.value, "user_id": user.id }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error getting user department: {str(e)}", exc_info=True) return custom_response( success=False, data="Failed to get user department due to server error", status_code=500 ) def update_admin_password(self, admin_id, new_password, current_password=None, require_current_password=True): """Update admin password with optional current password verification""" with DatabaseContextManager() as ctx: try: # Find the admin user admin_user = ctx.session.query(User).filter( User.id == admin_id, User.user_type == 'admin' ).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, admin_user.salt): 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 password_hash, salt = hash_password(new_password) # Update the password admin_user.password_hash = password_hash admin_user.salt = salt admin_user.updated_at = datetime.datetime.utcnow() # 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, "updated_at": admin_user.updated_at.isoformat() }, status_code=200 ) except Exception as e: ctx.session.rollback() 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_id, new_password): """Reset admin password without requiring current password (for system use)""" with DatabaseContextManager() as ctx: try: # Find the admin user admin_user = ctx.session.query(User).filter( User.id == admin_id, User.user_type == 'admin' ).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 password_hash, salt = hash_password(new_password) # Update the password admin_user.password_hash = password_hash admin_user.salt = salt admin_user.updated_at = datetime.datetime.utcnow() # 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, "updated_at": admin_user.updated_at.isoformat() }, status_code=200 ) except Exception as e: ctx.session.rollback() 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 generate_admin_password_reset_token(self, admin_email): """Generate a password reset token for admin""" with DatabaseContextManager() as ctx: try: # Find the admin user admin_user = ctx.session.query(User).filter( User.email == admin_email, User.user_type == 'admin' ).first() if not admin_user: return custom_response( success=False, data="Admin user not found", status_code=404 ) # Generate reset token reset_token = secrets.token_urlsafe(32) expires_at = datetime.datetime.utcnow() + datetime.timedelta(hours=1) # Store token in user model (you might want to create a separate table for this) # For now, we'll use a simple approach admin_user.reset_token = reset_token admin_user.reset_token_expires = expires_at admin_user.updated_at = datetime.datetime.utcnow() ctx.session.commit() # Send reset email (implement this based on your email service) try: reset_link = f"{current_app.config.get('FRONTEND_URL', 'http://localhost:3000')}/reset-password?token={reset_token}" 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. Click the link below to reset your password: {reset_link} This link will expire in 1 hour. 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 token generated and email sent", "admin_id": admin_user.id, "email": admin_user.email, "expires_at": expires_at.isoformat() }, status_code=200 ) except Exception as e: ctx.session.rollback() current_app.logger.error(f"Error generating reset token: {str(e)}", exc_info=True) 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() # Calculate real workload data 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 real student count through enrollments estimated_students = 0 for course in tutor_courses: student_count = ctx.session.query(func.count(Enrollment.student_id)).join( enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id ).filter( enrollment_courses.c.course_id == course.id, enrollment_courses.c.status == 'active', Enrollment.status == 'active' ).scalar() estimated_students += student_count workload_analysis.append({ 'tutor_id': tutor.id, 'name': f"{tutor.first_name} {tutor.last_name}", 'email': tutor.email, 'department': tutor.get_primary_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 { 'tutor_workloads': workload_analysis, 'department_stats': {}, # Simplified for now 'summary': summary } except Exception as e: current_app.logger.error(f"Error fetching tutor workload analysis: {str(e)}", exc_info=True) return { 'tutor_workloads': [], 'department_stats': {}, 'summary': { 'total_tutors': 0, 'underloaded_count': 0, 'balanced_count': 0, 'overloaded_count': 0, 'average_efficiency': 0 } }