from datetime import datetime, timedelta, date
from sqlalchemy import func, and_, or_, case, text
from sqlalchemy.orm import joinedload
from src.models import DatabaseContextManager
from src.models.models import (
Supervisor, Tutor, Student, Course, TeachingSession, DailyTeachingSession, Attendance,
Assignment, AssignmentSubmission,NotificationPreference, TutorAvailability,
Timetable, TimetableBlock, tutor_course_association,
AttendanceStatus, SubmissionStatus, TeachingVerificationMethod,
SupervisorDepartment, TutorDepartment, SupervisorTeachingLog, SupervisorAvailability,
Speciality, User, course_department_association, Enrollment, enrollment_courses,
CourseProgress
)
from flask import current_app
from src.utils import (
ApiABC,
custom_response,
send_email
)
from typing import Any, Dict, List, Optional
import uuid
class SupervisorManager(ApiABC):
def __init__(self):
self.table = Supervisor
def create(self, payload: Dict[str, Any]) -> Dict[str, Any]:
return super().create(payload)
def update(self, supervisor_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Update supervisor information"""
with DatabaseContextManager() as ctx:
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Handle department update separately to maintain supervisor_departments table
department_updated = True
if 'department' in payload and payload['department']:
department_updated = self._update_supervisor_department(ctx, supervisor_id, payload['department'])
if not department_updated:
return custom_response(
success=False,
data="Failed to update supervisor department",
status_code=400
)
# Update allowed fields (excluding department as it's handled separately)
updatable_fields = [
'email', 'first_name', 'last_name', 'phone', 'profile_picture',
'staff_id', 'office_location', 'office_hours',
'is_head_of_department', 'years_of_experience'
]
for field in updatable_fields:
if field in payload:
setattr(supervisor, field, payload[field])
try:
ctx.session.commit()
return custom_response(
success=True,
data="Supervisor updated successfully",
status_code=200
)
except Exception as e:
ctx.session.rollback()
current_app.logger.error(f"Error updating supervisor: {str(e)}")
return custom_response(
success=False,
data=f"Error updating supervisor: {str(e)}",
status_code=400
)
def _update_supervisor_department(self, ctx, supervisor_id: str, new_department: str) -> bool:
"""Update supervisor's department in the supervisor_departments table"""
try:
# First, deactivate all existing department associations
existing_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
for dept in existing_departments:
dept.is_active = False
dept.is_primary = False
# Create new primary department association
new_dept_assoc = SupervisorDepartment(
supervisor_id=supervisor_id,
department_name=new_department,
is_primary=True,
is_active=True,
assigned_by=supervisor_id, # Self-assigned
notes=f"Department updated to {new_department}",
assigned_date=datetime.now().date()
)
ctx.session.add(new_dept_assoc)
# Note: The supervisor.department field has been removed in favor of the departments relationship
# The department information is now stored in the SupervisorDepartment table
return True
except Exception as e:
current_app.logger.error(f"Error updating supervisor department: {str(e)}")
return False
def delete_supervisor_department(self, supervisor_id: str, department_id: str) -> Dict[str, Any]:
"""Delete a specific department from a supervisor"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Find the department association to delete
department_assoc = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.id == department_id,
SupervisorDepartment.supervisor_id == supervisor_id
).first()
if not department_assoc:
return custom_response(
success=False,
data="Department association not found",
status_code=404
)
# Check if this is the primary department
if department_assoc.is_primary:
return custom_response(
success=False,
data="Cannot delete primary department. Please set another department as primary first.",
status_code=400
)
try:
# Delete the department association
ctx.session.delete(department_assoc)
ctx.session.commit()
return custom_response(
success=True,
data="Department removed successfully",
status_code=200
)
except Exception as e:
ctx.session.rollback()
current_app.logger.error(f"Error deleting supervisor department: {str(e)}")
return custom_response(
success=False,
data=f"Error deleting department: {str(e)}",
status_code=500
)
def remove_department_by_name(self, supervisor_id: str, department_name: str) -> Dict[str, Any]:
"""Remove a department from a supervisor by department name (soft delete)"""
try:
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
dept_assignment = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.department_name == department_name,
SupervisorDepartment.is_active == True
).first()
if not dept_assignment:
return custom_response(
success=False,
data=f"Department '{department_name}' not found for this supervisor",
status_code=404
)
# Soft delete
dept_assignment.is_active = False
dept_assignment.updated_at = datetime.utcnow()
dept_assignment.notes = f"Removed by admin at {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')}"
ctx.session.commit()
return custom_response(
success=True,
data={
'message': f"Department '{department_name}' removed successfully",
'department_id': dept_assignment.id,
'department_name': department_name,
'removed_at': datetime.utcnow().isoformat()
},
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error removing supervisor department by name: {str(e)}", exc_info=True)
return custom_response(
success=False,
data=f"Failed to remove supervisor department: {str(e)}",
status_code=500
)
def bulk_remove_departments(self, supervisor_id: str, department_names: list) -> Dict[str, Any]:
"""Remove multiple departments from a supervisor (soft delete)"""
try:
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
if not department_names:
return custom_response(
success=False,
data="No department names provided",
status_code=400
)
# Find all active department assignments for the specified departments
dept_assignments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.department_name.in_(department_names),
SupervisorDepartment.is_active == True
).all()
if not dept_assignments:
return custom_response(
success=False,
data="No active department assignments found for the specified departments",
status_code=404
)
removed_count = 0
removed_departments = []
for dept_assignment in dept_assignments:
# Soft delete
dept_assignment.is_active = False
dept_assignment.updated_at = datetime.utcnow()
dept_assignment.notes = f"Bulk removed by admin at {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')}"
removed_departments.append({
'id': dept_assignment.id,
'name': dept_assignment.department_name,
'was_primary': dept_assignment.is_primary
})
removed_count += 1
ctx.session.commit()
return custom_response(
success=True,
data={
'message': f"Successfully removed {removed_count} departments",
'removed_count': removed_count,
'removed_departments': removed_departments,
'removed_at': datetime.utcnow().isoformat()
},
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error in bulk supervisor department removal: {str(e)}", exc_info=True)
return custom_response(
success=False,
data=f"Failed to remove supervisor departments: {str(e)}",
status_code=500
)
def get(self, id) -> Dict[str, Any]:
return super().get(id)
def fetchAll(self) -> Dict[str, Any]:
return super().fetchAll()
def delete(self, supervisor_id: str) -> Dict[str, Any]:
"""Delete supervisor and all associated relations using raw SQL to avoid schema issues"""
with DatabaseContextManager() as ctx:
try:
# First, check if supervisor exists using raw SQL
result = ctx.session.execute(
text("SELECT s.id, u.first_name, u.last_name, u.email, s.staff_id FROM supervisors s JOIN users u ON s.id = u.id WHERE s.id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
supervisor_data = result.fetchone()
if not supervisor_data:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
supervisor_id_db, first_name, last_name, email, staff_id = supervisor_data
supervisor_name = f"{first_name} {last_name}"
# Delete related records using raw SQL to avoid ORM schema issues
# 1. Delete supervisor availability records
ctx.session.execute(
text("DELETE FROM supervisor_availability WHERE supervisor_id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
# 2. Delete supervisor teaching logs
ctx.session.execute(
text("DELETE FROM supervisor_teaching_logs WHERE supervisor_id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
# 3. Delete supervisor department associations
ctx.session.execute(
text("DELETE FROM supervisor_departments WHERE supervisor_id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
# 4. Update tutors to remove supervisor assignment
tutors_result = ctx.session.execute(
text("UPDATE tutors SET supervisor_id = NULL WHERE supervisor_id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
tutors_affected = tutors_result.rowcount
# 5. Update courses to remove supervisor assignment
courses_result = ctx.session.execute(
text("UPDATE courses SET supervisor_id = NULL WHERE supervisor_id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
courses_affected = courses_result.rowcount
# 6. Delete notification preferences
ctx.session.execute(
text("DELETE FROM notification_preferences WHERE user_id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
# 7. Delete reminders related to supervisor
ctx.session.execute(
text("DELETE FROM reminders WHERE supervisor_id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
# 8. Delete supervisor record
ctx.session.execute(
text("DELETE FROM supervisors WHERE id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
# 9. Delete associated user record
ctx.session.execute(
text("DELETE FROM users WHERE id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
# Commit all changes
ctx.session.commit()
# Log the deletion
current_app.logger.info(f"Supervisor deleted: {supervisor_name} ({email}) - ID: {supervisor_id}")
return custom_response(
success=True,
data={
"message": f"Supervisor {supervisor_name} and all associated records deleted successfully",
"deleted_supervisor": {
"id": supervisor_id,
"name": supervisor_name,
"email": email,
"staff_id": staff_id
},
"affected_records": {
"tutors_unassigned": tutors_affected,
"courses_unassigned": courses_affected
}
},
status_code=200
)
except Exception as e:
ctx.session.rollback()
current_app.logger.error(f"Error deleting supervisor {supervisor_id}: {str(e)}")
return custom_response(
success=False,
data=f"Error deleting supervisor: {str(e)}",
status_code=500
)
def delete_multiple_supervisors(self, supervisor_ids: List[str]) -> Dict[str, Any]:
"""Delete multiple supervisors and their associated relations"""
results = {
"successful_deletions": [],
"failed_deletions": [],
"summary": {
"total_requested": len(supervisor_ids),
"successful": 0,
"failed": 0
}
}
for supervisor_id in supervisor_ids:
try:
delete_result = self.delete(supervisor_id)
if delete_result.get('success'):
results["successful_deletions"].append({
"id": supervisor_id,
"data": delete_result.get('data')
})
results["summary"]["successful"] += 1
else:
results["failed_deletions"].append({
"id": supervisor_id,
"error": delete_result.get('data')
})
results["summary"]["failed"] += 1
except Exception as e:
results["failed_deletions"].append({
"id": supervisor_id,
"error": str(e)
})
results["summary"]["failed"] += 1
return custom_response(
success=True,
data=results,
status_code=200
)
def get_supervisor_deletion_impact(self, supervisor_id: str) -> Dict[str, Any]:
"""Get information about what will be affected when deleting a supervisor"""
with DatabaseContextManager() as ctx:
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Count related records
availability_count = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.supervisor_id == supervisor_id
).count()
teaching_logs_count = ctx.session.query(SupervisorTeachingLog).filter(
SupervisorTeachingLog.supervisor_id == supervisor_id
).count()
department_associations_count = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id
).count()
managed_tutors_count = ctx.session.query(Tutor).filter(
Tutor.supervisor_id == supervisor_id,
Tutor.is_active == True
).count()
managed_courses_count = ctx.session.query(Course).filter(
Course.supervisor_id == supervisor_id
).count()
notification_preferences_count = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == supervisor_id
).count()
reminders_count = 0
try:
# Use raw SQL to avoid ORM schema issues
result = ctx.session.execute(
text("SELECT COUNT(*) FROM reminders WHERE supervisor_id = :supervisor_id"),
{"supervisor_id": supervisor_id}
)
reminders_count = result.scalar() or 0
except Exception as e:
# Handle case where reminders table schema is outdated
current_app.logger.warning(f"Could not count reminders for supervisor {supervisor_id}: {str(e)}")
reminders_count = 0
impact_data = {
"supervisor": {
"id": supervisor_id,
"name": f"{supervisor.first_name} {supervisor.last_name}",
"email": supervisor.email,
"staff_id": supervisor.staff_id
},
"records_to_delete": {
"availability_records": availability_count,
"teaching_logs": teaching_logs_count,
"department_associations": department_associations_count,
"notification_preferences": notification_preferences_count,
"reminders": reminders_count
},
"records_to_modify": {
"tutors_to_deactivate": managed_tutors_count,
"courses_to_unassign": managed_courses_count
},
"total_records_affected": (
availability_count + teaching_logs_count + department_associations_count +
notification_preferences_count + reminders_count + managed_tutors_count + managed_courses_count
)
}
return custom_response(
success=True,
data=impact_data,
status_code=200
)
def _supervisor_to_dict(self, supervisor: Supervisor) -> Dict:
"""Convert Supervisor model to dictionary"""
return {
'id': supervisor.id,
'staff_id': supervisor.staff_id,
'first_name': supervisor.first_name,
'last_name': supervisor.last_name,
'email': supervisor.email,
'phone': supervisor.phone,
'department': supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Not assigned',
'departments': [dept.department_name for dept in supervisor.departments if dept.is_active] if hasattr(supervisor, 'departments') else [],
'office_location': supervisor.office_location,
'office_hours': supervisor.office_hours,
'is_head_of_department': supervisor.is_head_of_department,
'years_of_experience': supervisor.years_of_experience,
'profile_picture': supervisor.profile_picture,
'managed_tutors_count': len(supervisor.managed_tutors) if hasattr(supervisor, 'managed_tutors') else 0,
'managed_courses_count': len(supervisor.managed_courses) if hasattr(supervisor, 'managed_courses') else 0
}
def get_supervisor_tutors_lightweight(self, supervisor_id: str) -> Dict:
"""Get tutors under supervisor - comprehensive version with detailed information"""
try:
with DatabaseContextManager() as ctx:
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
department_names = [dept.department_name for dept in supervisor_departments]
if not department_names:
return custom_response(
success=True,
data={
'tutors': [],
'user_info': {
'id': supervisor_id,
'user_type': 'supervisor',
'department': None,
'departments': []
}
}
)
# Get tutors in the supervisor's departments with comprehensive data
tutors_query = (
ctx.session.query(
User.id,
User.first_name,
User.last_name,
User.email,
User.phone,
User.profile_picture,
User.created_at,
User.last_login,
Tutor.staff_id,
Tutor.qualification,
Tutor.specialization,
Tutor.years_of_teaching,
Tutor.is_full_time,
Tutor.is_on_leave,
Tutor.max_teaching_hours,
Tutor.hourly_rate,
Tutor.office_location,
Tutor.bio,
Tutor.verification_methods,
Tutor.verification_success_rate,
Tutor.supervisor_id
)
.join(Tutor, User.id == Tutor.id)
.join(TutorDepartment, User.id == TutorDepartment.tutor_id)
.join(
SupervisorDepartment,
TutorDepartment.department_name == SupervisorDepartment.department_name
)
.filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True,
TutorDepartment.is_active == True,
User.is_active == True
)
.distinct()
.limit(100) # Reasonable limit for detailed data
)
tutors = tutors_query.all()
# Get comprehensive tutor information
tutors_data = []
for tutor in tutors:
# Get tutor's departments
tutor_departments = ctx.session.query(TutorDepartment).filter(
TutorDepartment.tutor_id == tutor.id,
TutorDepartment.is_active == True
).all()
departments_data = []
for dept in tutor_departments:
departments_data.append({
'id': dept.id,
'department_name': dept.department_name,
'is_primary': dept.is_primary,
'assigned_date': dept.assigned_date.isoformat() if dept.assigned_date else None,
'assigned_by': dept.assigned_by,
'notes': dept.notes,
'is_active': dept.is_active
})
# Get supervisor name
supervisor_name = None
if tutor.supervisor_id:
supervisor = ctx.session.query(User).filter(User.id == tutor.supervisor_id).first()
if supervisor:
supervisor_name = f"{supervisor.first_name} {supervisor.last_name}"
# Get courses count
courses_count = 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
).count()
# Get students count across all courses
students_count = ctx.session.query(func.count(func.distinct(Enrollment.student_id))).join(
enrollment_courses, Enrollment.id == enrollment_courses.c.enrollment_id
).join(
Course, enrollment_courses.c.course_id == Course.id
).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
).scalar() or 0
# Get teaching sessions count from DailyTeachingSession
total_sessions = ctx.session.query(DailyTeachingSession).filter(
DailyTeachingSession.tutor_id == tutor.id
).count()
verified_sessions = ctx.session.query(DailyTeachingSession).filter(
DailyTeachingSession.tutor_id == tutor.id,
DailyTeachingSession.is_verified == True
).count()
verification_rate = (verified_sessions / total_sessions * 100) if total_sessions > 0 else 0
# Get current teaching hours for this semester from DailyTeachingSession
# Use SQLite-compatible date filtering and time calculation
current_month_start = datetime.now().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
# Get sessions for current month and calculate hours manually
sessions_this_month = ctx.session.query(DailyTeachingSession).filter(
DailyTeachingSession.tutor_id == tutor.id,
DailyTeachingSession.is_verified == True,
DailyTeachingSession.session_date >= current_month_start
).all()
current_hours = 0
for session in sessions_this_month:
if session.start_time and session.end_time:
# Calculate duration in hours
start_dt = datetime.combine(session.session_date, session.start_time)
end_dt = datetime.combine(session.session_date, session.end_time)
duration = (end_dt - start_dt).total_seconds() / 3600
current_hours += duration
# Calculate overall rating (placeholder - you can implement actual rating logic)
overall_rating = 0
if verification_rate > 90:
overall_rating = 4.5
elif verification_rate > 80:
overall_rating = 4.0
elif verification_rate > 70:
overall_rating = 3.5
elif verification_rate > 60:
overall_rating = 3.0
tutors_data.append({
'id': tutor.id,
'first_name': tutor.first_name,
'last_name': tutor.last_name,
'email': tutor.email,
'phone': tutor.phone,
'profile_picture': tutor.profile_picture,
'created_at': tutor.created_at.isoformat() if tutor.created_at else None,
'last_login': tutor.last_login.isoformat() if tutor.last_login else None,
'staff_id': tutor.staff_id,
'qualification': tutor.qualification,
'specialization': tutor.specialization,
'years_of_teaching': tutor.years_of_teaching or 0,
'is_full_time': tutor.is_full_time or False,
'is_on_leave': tutor.is_on_leave or False,
'max_teaching_hours': tutor.max_teaching_hours or 0,
'current_hours': float(current_hours),
'hourly_rate': float(tutor.hourly_rate) if tutor.hourly_rate else 0,
'office_location': tutor.office_location,
'bio': tutor.bio,
'verification_methods': tutor.verification_methods,
'verification_success_rate': float(tutor.verification_success_rate) if tutor.verification_success_rate else verification_rate,
'overall_rating': overall_rating,
'supervisor_id': tutor.supervisor_id,
'supervisor_name': supervisor_name,
'departments': departments_data,
'courses_count': courses_count,
'students_count': students_count,
'attendance_rate': verification_rate, # Using verification rate as proxy
'performance_score': verification_rate,
'teaching_stats': {
'total_sessions': total_sessions,
'verified_sessions': verified_sessions,
'verification_rate': verification_rate
}
})
# Get supervisor basic info
supervisor = ctx.session.query(User).filter(User.id == supervisor_id).first()
supervisor_info = None
if supervisor:
supervisor_info = {
'id': supervisor.id,
'name': f"{supervisor.first_name} {supervisor.last_name}",
'email': supervisor.email,
'profile_picture': supervisor.profile_picture,
'user_type': 'supervisor',
'department': department_names[0] if department_names else None,
'departments': department_names
}
return custom_response(
success=True,
data={
'tutors': tutors_data,
'user_info': supervisor_info or {
'id': supervisor_id,
'user_type': 'supervisor',
'department': department_names[0] if department_names else None,
'departments': department_names
}
}
)
except Exception as e:
current_app.logger.error(f"Error getting supervisor tutors: {str(e)}")
return custom_response(
success=False,
data=f"Failed to retrieve supervisor tutors: {str(e)}"
), 500
def get_supervisor_details(self, supervisor_id: str) -> Dict:
"""Get detailed information about a supervisor including comprehensive dashboard data"""
with DatabaseContextManager() as ctx:
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
department_names = [dept.department_name for dept in supervisor_departments]
# Get courses from supervisor's departments (not just direct supervision)
courses = ctx.session.query(Course).filter(
Course.department.in_(department_names) if department_names else False,
Course.is_active == True
).all()
# If no courses found by department, get all active courses as fallback
if not courses:
courses = ctx.session.query(Course).filter(Course.is_active == True).limit(100).all()
# Get tutors in the supervisor's departments
tutors = (
ctx.session.query(Tutor)
.join(TutorDepartment, Tutor.id == TutorDepartment.tutor_id)
.join(
SupervisorDepartment,
TutorDepartment.department_name == SupervisorDepartment.department_name
)
.filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True,
TutorDepartment.is_active == True,
Tutor.is_active == True
)
.distinct()
.all()
)
# Get teaching sessions for analytics
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id.in_([c.id for c in courses])
).all()
# Get attendance records
attendance_records = ctx.session.query(Attendance).join(
TeachingSession,
Attendance.session_id == TeachingSession.id
).filter(
TeachingSession.course_id.in_([c.id for c in courses])
).all()
# Calculate attendance analytics
attendance_analytics = self._calculate_attendance_analytics(attendance_records, sessions, courses, tutors)
# Calculate workload analysis
workload_analysis = self._calculate_workload_analysis(tutors, courses, sessions)
# Get sessions needing verification
sessions_needing_verification = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id.in_([c.id for c in courses]),
TeachingSession.status == 'completed',
TeachingSession.is_verified == False
).join(Tutor, TeachingSession.tutor_id == Tutor.id).join(Course, TeachingSession.course_id == Course.id).all()
# Prepare courses data with attendance rates
courses_data = []
for course in courses:
course_sessions = [s for s in sessions if s.course_id == course.id]
course_attendance = [a for a in attendance_records if any(s.id == a.session_id for s in course_sessions)]
# Calculate attendance rate for this course
if course_attendance:
present_count = sum(1 for a in course_attendance if a.status == AttendanceStatus.present)
attendance_rate = round((present_count / len(course_attendance)) * 100, 1)
else:
attendance_rate = 0
# Count students through speciality relationship
students_count = ctx.session.query(Student).join(
Speciality, Student.speciality_id == Speciality.id
).filter(
Speciality.id == course.speciality_id,
Student.is_active == True
).count()
courses_data.append({
'id': course.id,
'code': course.code,
'title': course.title,
'credits': course.credits,
'semester': course.semester,
'department': course.department,
'students_count': students_count,
'attendance_rate': attendance_rate,
'tutors_count': len(course.tutors) if hasattr(course, 'tutors') else 0,
'sessions_count': len(course_sessions)
})
# Get tutors data
tutors_data = []
for tutor in tutors:
tutor_sessions = [s for s in sessions if s.tutor_id == tutor.id]
tutor_attendance = [a for a in attendance_records if any(s.id == a.session_id for s in tutor_sessions)]
# Calculate tutor attendance rate
if tutor_attendance:
present_count = sum(1 for a in tutor_attendance if a.status == AttendanceStatus.present)
attendance_rate = round((present_count / len(tutor_attendance)) * 100, 1)
else:
attendance_rate = 0
tutors_data.append({
'id': tutor.id,
'name': f"{tutor.first_name} {tutor.last_name}",
'attendance_rate': attendance_rate,
'sessions_count': len(tutor_sessions),
'verified_sessions': len([s for s in tutor_sessions if s.is_verified])
})
# Get recent timetable changes (placeholder)
recent_timetable_changes = []
# Get top teaching lessons (placeholder)
top_teaching_lessons = []
# Get notifications (placeholder)
notifications = []
# Prepare comprehensive response
supervisor_data = {
'user_info': {
'id': supervisor.id,
'name': f"{supervisor.first_name} {supervisor.last_name}",
'email': supervisor.email,
'profile_picture': supervisor.profile_picture,
'user_type': 'supervisor',
'department': supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else None,
'departments': [dept.department_name for dept in supervisor_departments] if supervisor_departments else []
},
'courses': courses_data,
'courses_count': len(courses_data),
'tutors': tutors_data,
'tutors_count': len(tutors_data),
'total_courses': len(courses),
'total_specialities': ctx.session.query(Speciality).filter(Speciality.is_active == True).count(),
'pending_approvals': 0, # Placeholder
'notifications': notifications,
'recent_timetable_changes': recent_timetable_changes,
'top_teaching_lessons': top_teaching_lessons,
'sessions_needing_verification': [
{
'id': session.id,
'session_title': f"{session.course.code} - {session.course.title}",
'course_code': session.course.code,
'course_title': session.course.title,
'tutor_name': f"{session.tutor.first_name} {session.tutor.last_name}",
'session_date': session.start_time.isoformat(),
'start_time': session.start_time.strftime('%H:%M:%S'),
'end_time': session.end_time.strftime('%H:%M:%S'),
'duration': int(self._calculate_session_duration(session.start_time, session.end_time) / 60)
} for session in sessions_needing_verification[:5] # Limit to 5
],
'attendance_analytics': attendance_analytics,
'tutor_workload_analysis': workload_analysis
}
return custom_response(
success=True,
data=supervisor_data,
status_code=200
)
def _calculate_attendance_analytics(self, attendance_records, sessions, courses, tutors):
"""Calculate comprehensive attendance analytics"""
total_records = len(attendance_records)
present_count = sum(1 for a in attendance_records if a.status == AttendanceStatus.present)
late_count = sum(1 for a in attendance_records if a.status == AttendanceStatus.late)
absent_count = sum(1 for a in attendance_records if a.status == AttendanceStatus.absent)
overall_attendance_rate = round((present_count / total_records) * 100, 1) if total_records > 0 else 0
# Course attendance stats
course_attendance_stats = []
for course in courses[:10]: # Limit to 10 courses for performance
course_sessions = [s for s in sessions if s.course_id == course.id]
course_attendance = [a for a in attendance_records if any(s.id == a.session_id for s in course_sessions)]
if course_attendance:
course_present = sum(1 for a in course_attendance if a.status == AttendanceStatus.present)
course_rate = round((course_present / len(course_attendance)) * 100, 1)
else:
course_rate = 0
course_attendance_stats.append({
'course_id': course.id,
'course_code': course.code,
'course_title': course.title,
'attendance_rate': course_rate,
'total_records': len(course_attendance)
})
# Tutor attendance stats
tutor_attendance_stats = []
for tutor in tutors[:10]: # Limit to 10 tutors for performance
tutor_sessions = [s for s in sessions if s.tutor_id == tutor.id]
tutor_attendance = [a for a in attendance_records if any(s.id == a.session_id for s in tutor_sessions)]
if tutor_attendance:
tutor_present = sum(1 for a in tutor_attendance if a.status == AttendanceStatus.present)
tutor_rate = round((tutor_present / len(tutor_attendance)) * 100, 1)
else:
tutor_rate = 0
tutor_attendance_stats.append({
'tutor_id': tutor.id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}",
'attendance_rate': tutor_rate,
'total_records': len(tutor_attendance)
})
# Weekly trends (sample data for now)
weekly_trends = [
{'week': 'Week 1', 'attendance_rate': 85.5},
{'week': 'Week 2', 'attendance_rate': 87.2},
{'week': 'Week 3', 'attendance_rate': 89.1},
{'week': 'Week 4', 'attendance_rate': 86.8},
{'week': 'Week 5', 'attendance_rate': 88.3}
]
return {
'overview': {
'overall_attendance_rate': overall_attendance_rate,
'total_attendance_records': total_records,
'total_courses': len(courses),
'total_sessions': len(sessions),
'total_tutors': len(tutors)
},
'summary': {
'excellent_courses': len([c for c in course_attendance_stats if c['attendance_rate'] >= 90]),
'good_courses': len([c for c in course_attendance_stats if 80 <= c['attendance_rate'] < 90]),
'needs_improvement_courses': len([c for c in course_attendance_stats if 70 <= c['attendance_rate'] < 80]),
'poor_courses': len([c for c in course_attendance_stats if c['attendance_rate'] < 70]),
'excellent_tutors': len([t for t in tutor_attendance_stats if t['attendance_rate'] >= 90]),
'good_tutors': len([t for t in tutor_attendance_stats if 80 <= t['attendance_rate'] < 90]),
'needs_improvement_tutors': len([t for t in tutor_attendance_stats if 70 <= t['attendance_rate'] < 80]),
'poor_tutors': len([t for t in tutor_attendance_stats if t['attendance_rate'] < 70])
},
'course_attendance_stats': course_attendance_stats,
'tutor_attendance_stats': tutor_attendance_stats,
'weekly_trends': weekly_trends,
'attendance_alerts': [] # Can be populated with specific alerts
}
def _calculate_workload_analysis(self, tutors, courses, sessions):
"""Calculate tutor workload analysis"""
tutor_workloads = []
total_tutors = len(tutors)
for tutor in tutors:
tutor_sessions = [s for s in sessions if s.tutor_id == tutor.id]
tutor_courses = [c for c in courses if any(s.course_id == c.id for s in tutor_sessions)]
# Calculate workload metrics
sessions_count = len(tutor_sessions)
courses_count = len(tutor_courses)
hours_per_week = sessions_count * 2 # Assuming 2 hours per session
# Categorize workload
if hours_per_week <= 10:
workload_status = 'underloaded'
elif hours_per_week <= 20:
workload_status = 'balanced'
else:
workload_status = 'overloaded'
tutor_workloads.append({
'tutor_id': tutor.id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}",
'sessions_count': sessions_count,
'courses_count': courses_count,
'hours_per_week': hours_per_week,
'workload_status': workload_status,
'efficiency_score': min(100, max(0, 100 - (hours_per_week - 15) * 2)) # Simple efficiency calculation
})
# Calculate summary statistics
underloaded_count = len([t for t in tutor_workloads if t['workload_status'] == 'underloaded'])
balanced_count = len([t for t in tutor_workloads if t['workload_status'] == 'balanced'])
overloaded_count = len([t for t in tutor_workloads if t['workload_status'] == 'overloaded'])
average_efficiency = round(sum(t['efficiency_score'] for t in tutor_workloads) / total_tutors, 1) if total_tutors > 0 else 0
return {
'summary': {
'total_tutors': total_tutors,
'underloaded_count': underloaded_count,
'balanced_count': balanced_count,
'overloaded_count': overloaded_count,
'average_efficiency': average_efficiency
},
'tutor_workloads': tutor_workloads,
'department_stats': {} # Can be populated with department-specific stats
}
def get_supervisor_department(self, supervisor_id: str) -> Dict:
"""Get the supervisor's department information"""
with DatabaseContextManager() as ctx:
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get primary department from supervisor_departments table
primary_dept = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_primary == True,
SupervisorDepartment.is_active == True
).first()
# Get all active departments
all_depts = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
department_data = {
'primary_department': primary_dept.department_name if primary_dept else (supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Not assigned'),
'departments': [dept.department_name for dept in all_depts],
'total_departments': len(all_depts)
}
return custom_response(
success=True,
data=department_data,
status_code=200
)
def get_department_statistics(self, supervisor_id: str) -> Dict:
"""Get comprehensive statistics for the supervisor's department"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
department = supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Not assigned'
# Get basic counts
tutors_count = ctx.session.query(Tutor).filter(
Tutor.department == department,
Tutor.is_active == True
).count()
students_count = ctx.session.query(Student).filter(
Student.program.like(f"%{department}%"),
Student.is_active == True
).count()
courses_count = ctx.session.query(Course).filter(
Course.department == department,
Course.is_active == True
).count()
# Get attendance statistics
attendance_stats = ctx.session.query(
func.count(Attendance.id),
func.sum(case((Attendance.status == AttendanceStatus.present, 1), else_=0)),
func.sum(case((Attendance.status == AttendanceStatus.late, 1), else_=0)),
func.sum(case((Attendance.status == AttendanceStatus.absent, 1), else_=0))
).join(
TeachingSession,
Attendance.session_id == TeachingSession.id
).join(
Course,
TeachingSession.course_id == Course.id
).filter(
Course.department == department
).first()
# Get assignment statistics
assignment_stats = ctx.session.query(
func.count(AssignmentSubmission.id),
func.sum(case((AssignmentSubmission.status == SubmissionStatus.graded, 1), else_=0)),
func.avg(AssignmentSubmission.grade)
).join(
Assignment,
AssignmentSubmission.assignment_id == Assignment.id
).join(
Course,
Assignment.course_id == Course.id
).filter(
Course.department == department
).first()
# Get upcoming sessions (next 7 days)
upcoming_sessions = ctx.session.query(TeachingSession).join(
Course,
TeachingSession.course_id == Course.id
).filter(
Course.department == department,
TeachingSession.start_time >= datetime.utcnow(),
TeachingSession.start_time <= datetime.utcnow() + timedelta(days=7),
TeachingSession.status == 'scheduled'
).count()
# Get pending tutor availability approvals
pending_approvals = ctx.session.query(TutorAvailability).filter(
TutorAvailability.is_approved == False,
TutorAvailability.tutor_id.in_(
ctx.session.query(Tutor.id).filter(
Tutor.supervisor_id == supervisor_id
)
)
).count()
# Get pending attendance verifications
pending_verifications = ctx.session.query(TeachingSession).filter(
TeachingSession.is_verified == False,
TeachingSession.end_time < datetime.utcnow(),
TeachingSession.course_id.in_(
ctx.session.query(Course.id).filter(
Course.supervisor_id == supervisor_id
)
)
).count()
return custom_response(
success=True,
data={
'department': department,
'tutors_count': tutors_count,
'students_count': students_count,
'courses_count': courses_count,
'attendance': {
'total': attendance_stats[0] if attendance_stats else 0,
'present': attendance_stats[1] if attendance_stats else 0,
'late': attendance_stats[2] if attendance_stats else 0,
'absent': attendance_stats[3] if attendance_stats else 0,
'rate': round((attendance_stats[1] / attendance_stats[0]) * 100, 1) if attendance_stats and attendance_stats[0] else 0
},
'assignments': {
'total': assignment_stats[0] if assignment_stats else 0,
'graded': assignment_stats[1] if assignment_stats else 0,
'average_grade': round(assignment_stats[2], 1) if assignment_stats and assignment_stats[2] else None
},
'upcoming_sessions': upcoming_sessions,
'pending_approvals': pending_approvals,
'pending_verifications': pending_verifications
},
status_code=200
)
def get_student_statistics(self, supervisor_id: str, student_id: str) -> Dict:
"""Get detailed statistics for a specific student in the supervisor's department"""
with DatabaseContextManager() as ctx:
# Verify supervisor and student
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
student = ctx.session.query(Student).filter(Student.id == student_id).first()
if not supervisor or not student:
return custom_response(
success=False,
data="Supervisor or student not found",
status_code=404
)
# Check if student is in supervisor's department
# Get student and check if their speciality courses are in supervisor's department
student = ctx.session.query(Student).filter(Student.id == student_id).first()
if not student:
return {"error": "Student not found"}
student_courses = ctx.session.query(Course).filter(
Course.speciality_id == student.speciality_id,
Course.department == (supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Not assigned')
).all()
if not student_courses:
return custom_response(
success=False,
data="Student is not enrolled in any courses in your department",
status_code=403
)
# Get base student info
student_data = {
'id': student.id,
'student_id': student.student_id,
'name': f"{student.first_name} {student.last_name}",
'email': student.email,
'program': student.program,
'year_of_study': student.year_of_study,
'cumulative_gpa': student.cumulative_gpa,
'is_on_probation': student.is_on_probation,
'courses_count': len(student_courses)
}
# Get attendance statistics
attendance_stats = ctx.session.query(
func.count(Attendance.id),
func.sum(case((Attendance.status == AttendanceStatus.present, 1), else_=0)),
func.sum(case((Attendance.status == AttendanceStatus.late, 1), else_=0)),
func.sum(case((Attendance.status == AttendanceStatus.absent, 1), else_=0))
).join(
TeachingSession,
Attendance.session_id == TeachingSession.id
).join(
Course,
TeachingSession.course_id == Course.id
).filter(
Attendance.student_id == student_id,
Course.department == (supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Not assigned')
).first()
student_data['attendance'] = {
'total': attendance_stats[0] if attendance_stats else 0,
'present': attendance_stats[1] if attendance_stats else 0,
'late': attendance_stats[2] if attendance_stats else 0,
'absent': attendance_stats[3] if attendance_stats else 0,
'rate': round((attendance_stats[1] / attendance_stats[0]) * 100, 1) if attendance_stats and attendance_stats[0] else 0
}
# Get assignment statistics
assignment_stats = ctx.session.query(
func.count(AssignmentSubmission.id),
func.sum(case((AssignmentSubmission.status == SubmissionStatus.graded, 1), else_=0)),
func.avg(AssignmentSubmission.grade)
).join(
Assignment,
AssignmentSubmission.assignment_id == Assignment.id
).join(
Course,
Assignment.course_id == Course.id
).filter(
AssignmentSubmission.student_id == student_id,
Course.department == (supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Not assigned')
).first()
student_data['assignments'] = {
'total': assignment_stats[0] if assignment_stats else 0,
'graded': assignment_stats[1] if assignment_stats else 0,
'average_grade': round(assignment_stats[2], 1) if assignment_stats and assignment_stats[2] else None
}
# Get course-wise breakdown
courses_data = []
for course in student_courses:
# Attendance for this course
course_attendance = ctx.session.query(
func.count(Attendance.id),
func.sum(case((Attendance.status == AttendanceStatus.present, 1), else_=0))
).join(
TeachingSession,
Attendance.session_id == TeachingSession.id
).filter(
TeachingSession.course_id == course.id,
Attendance.student_id == student_id
).first()
# Assignments for this course
course_assignments = ctx.session.query(
func.count(AssignmentSubmission.id),
func.sum(case((AssignmentSubmission.status == SubmissionStatus.graded, 1), else_=0)),
func.avg(AssignmentSubmission.grade)
).join(
Assignment,
AssignmentSubmission.assignment_id == Assignment.id
).filter(
Assignment.course_id == course.id,
AssignmentSubmission.student_id == student_id
).first()
courses_data.append({
'course_id': course.id,
'course_code': course.code,
'course_title': course.title,
'attendance': {
'present': course_attendance[1] if course_attendance else 0,
'total': course_attendance[0] if course_attendance else 0,
'rate': round((course_attendance[1] / course_attendance[0]) * 100, 1) if course_attendance and course_attendance[0] else 0
},
'assignments': {
'graded': course_assignments[1] if course_assignments else 0,
'total': course_assignments[0] if course_assignments else 0,
'average_grade': round(course_assignments[2], 1) if course_assignments and course_assignments[2] else None
}
})
student_data['courses'] = courses_data
return custom_response(
success=True,
data=student_data,
status_code=200
)
def approve_tutor_availability(self, supervisor_id: str, availability_id: str, approve: bool = True, notes: str = None) -> Dict:
"""Approve or reject a tutor's availability request"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get the availability record
availability = ctx.session.query(TutorAvailability).filter(
TutorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Availability record not found",
status_code=404
)
# Check if supervisor manages this tutor
tutor = ctx.session.query(Tutor).filter(
Tutor.id == availability.tutor_id,
Tutor.supervisor_id == supervisor_id
).first()
if not tutor:
return custom_response(
success=False,
data="You are not authorized to approve this tutor's availability",
status_code=403
)
# Update the availability record
if approve:
availability.is_approved = True
availability.approval_date = datetime.utcnow()
availability.approved_by = supervisor_id
availability.approval_notes = notes
# Notify tutor
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = "Your Availability Has Been Approved"
message = f"""
Availability Approved
Hello {tutor.first_name},
Your availability request has been approved by your supervisor:
Day: {availability.day_of_week}
Time: {availability.start_time} - {availability.end_time}
Type: {availability.availability_type}
Notes: {notes or 'None'}
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send approval notification to tutor: {str(e)}")
message = "Availability approved successfully"
else:
availability.is_approved = False
availability.is_cancelled = True
availability.cancellation_reason = notes
availability.cancellation_date = datetime.utcnow()
# Notify tutor
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = "Your Availability Has Been Rejected"
message = f"""
Availability Rejected
Hello {tutor.first_name},
Your availability request has been rejected by your supervisor:
Day: {availability.day_of_week}
Time: {availability.start_time} - {availability.end_time}
Type: {availability.availability_type}
Reason: {notes or 'Not specified'}
Please adjust your availability and resubmit for approval.
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send rejection notification to tutor: {str(e)}")
message = "Availability rejected successfully"
ctx.session.commit()
return custom_response(
success=True,
data=message,
status_code=200
)
def generate_timetable(self, supervisor_id: str, payload: Dict) -> Dict:
"""Generate a master timetable for the department"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
department = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor.id
).first()
# Create timetable record
timetable = Timetable(
id=str(uuid.uuid4()),
created_by=supervisor_id,
name=payload['name'],
description=payload.get('description'),
semester=payload['semester'],
academic_year=payload['academic_year'],
approval_status='draft',
department=department.department_name,
term=payload['semester']
)
ctx.session.add(timetable)
# Add timetable blocks
for block_data in payload['blocks']:
block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable.id,
tutor_id=block_data['tutor_id'],
course_id=block_data['course_id'],
day_of_week=block_data['day_of_week'],
start_time=block_data['start_time'],
end_time=block_data['end_time'],
room=block_data.get('room'),
block_type=block_data['block_type'],
recurring=block_data.get('recurring', True),
start_date=block_data.get('start_date'),
end_date=block_data.get('end_date'),
created_by=supervisor_id
)
ctx.session.add(block)
# Create initial teaching sessions for the next 2 weeks
if block_data.get('create_sessions', True):
self._create_initial_sessions(ctx, block, supervisor_id)
ctx.session.commit()
return custom_response(
success=True,
data={
'timetable_id': timetable.id,
'name': timetable.name,
'blocks_count': len(payload['blocks'])
},
status_code=201
)
def _create_initial_sessions(self, ctx, block: TimetableBlock, supervisor_id: str) -> None:
"""Create initial teaching sessions for a timetable block"""
start_date = block.start_date if block.start_date else datetime.utcnow().date()
end_date = block.end_date if block.end_date else start_date + timedelta(days=14)
current_date = start_date
while current_date <= end_date:
if current_date.weekday() == block.day_of_week:
session = TeachingSession(
id=str(uuid.uuid4()),
course_id=block.course_id,
tutor_id=block.tutor_id,
timetable_id=block.timetable_id,
title=f"{block.course.code} - {block.block_type}",
description=f"Regular {block.block_type} session",
start_datetime=datetime.combine(current_date, block.start_time),
end_datetime=datetime.combine(current_date, block.end_time),
location=block.room,
status='scheduled',
created_by=supervisor_id,
is_mandatory=True,
session_type=block.block_type
)
ctx.session.add(session)
current_date += timedelta(days=1)
def approve_timetable(self, supervisor_id: str, timetable_id: str, approve: bool = True, notes: str = None) -> Dict:
"""Approve or reject a timetable"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get the timetable
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Check if supervisor is authorized to approve
if timetable.created_by != supervisor_id and not supervisor.is_head_of_department:
return custom_response(
success=False,
data="You are not authorized to approve this timetable",
status_code=403
)
# Update timetable status
if approve:
timetable.approval_status = 'approved'
timetable.approved_by = supervisor_id
timetable.approved_at = datetime.utcnow()
# Notify tutors
tutors = ctx.session.query(Tutor).join(
TimetableBlock,
Tutor.id == TimetableBlock.tutor_id
).filter(
TimetableBlock.timetable_id == timetable_id
).distinct().all()
for tutor in tutors:
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = "New Timetable Approved"
message = f"""
Timetable Approved
Hello {tutor.first_name},
A new timetable has been approved by your supervisor:
Timetable: {timetable.name}
Semester: {timetable.semester}
Academic Year: {timetable.academic_year}
Notes: {notes or 'None'}
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable notification to tutor {tutor.id}: {str(e)}")
message = "Timetable approved successfully"
else:
timetable.approval_status = 'rejected'
timetable.approved_by = supervisor_id
timetable.approved_at = datetime.utcnow()
# Notify creator
creator = ctx.session.query(Supervisor).filter(
Supervisor.id == timetable.created_by
).first()
if creator:
creator_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == creator.id
).first()
if creator_prefs and creator_prefs.receive_email:
subject = "Timetable Rejected"
message = f"""
Timetable Rejected
Hello {creator.first_name},
Your timetable has been rejected by {supervisor.first_name} {supervisor.last_name}:
Timetable: {timetable.name}
Reason: {notes or 'Not specified'}
Please review and resubmit the timetable for approval.
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=creator.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send rejection notification to creator {creator.id}: {str(e)}")
message = "Timetable rejected successfully"
ctx.session.commit()
return custom_response(
success=True,
data=message,
status_code=200
)
def verify_teaching_session(self, supervisor_id: str, session_id: str, verify: bool = True, notes: str = None) -> Dict:
"""Verify a completed teaching session and its attendance"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get the session
session = ctx.session.query(TeachingSession).filter(
TeachingSession.id == session_id
).first()
if not session:
return custom_response(
success=False,
data="Teaching session not found",
status_code=404
)
# Check if supervisor manages this course
course = ctx.session.query(Course).filter(
Course.id == session.course_id,
Course.supervisor_id == supervisor_id
).first()
if not course:
return custom_response(
success=False,
data="You are not authorized to verify this session",
status_code=403
)
# Verify the session
if verify:
session.is_verified = True
session.verified_by = supervisor_id
session.verified_at = datetime.utcnow()
session.verification_method = TeachingVerificationMethod.manual
session.supervisor_notes = notes
# Approve all attendance records for this session
attendance_records = ctx.session.query(Attendance).filter(
Attendance.session_id == session_id
).all()
for record in attendance_records:
if record.status == AttendanceStatus.pending_approval:
record.status = AttendanceStatus.present
record.approved_by = supervisor_id
record.approval_date = datetime.utcnow()
record.approval_notes = "Approved with session verification"
# Notify tutor
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == session.tutor_id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = "Session Verified"
message = f"""
Session Verified
Hello {session.tutor.first_name},
Your teaching session has been verified by your supervisor:
Course: {session.course.code} - {session.course.title}
Session: {session.course.code} - {session.course.title}
Date: {session.start_time.strftime('%A, %B %d, %Y') if session.start_time else 'TBD'}
Time: {session.start_time.strftime('%H:%M') if session.start_time else 'TBD'} - {session.end_time.strftime('%H:%M') if session.end_time else 'TBD'}
Notes: {notes or 'None'}
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=session.tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send verification notification to tutor: {str(e)}")
message = "Session verified successfully"
else:
session.is_verified = False
session.supervisor_notes = notes
# Notify tutor
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == session.tutor_id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = "Session Verification Issues"
message = f"""
Session Verification Issues
Hello {session.tutor.first_name},
Your supervisor has identified issues with your teaching session:
Course: {session.course.code} - {session.course.title}
Session: {session.course.code} - {session.course.title}
Date: {session.start_time.strftime('%A, %B %d, %Y') if session.start_time else 'TBD'}
Time: {session.start_time.strftime('%H:%M') if session.start_time else 'TBD'} - {session.end_time.strftime('%H:%M') if session.end_time else 'TBD'}
Issues: {notes or 'Not specified'}
Please address these issues and contact your supervisor.
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=session.tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send verification issue notification to tutor: {str(e)}")
message = "Session verification issues noted"
ctx.session.commit()
return custom_response(
success=True,
data=message,
status_code=200
)
def resolve_attendance_dispute(self, supervisor_id: str, attendance_id: str, resolution: Dict) -> Dict:
"""Resolve an attendance dispute"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get the attendance record
attendance = ctx.session.query(Attendance).filter(
Attendance.id == attendance_id,
Attendance.is_disputed == True
).first()
if not attendance:
return custom_response(
success=False,
data="Attendance dispute not found",
status_code=404
)
# Check if supervisor manages this course
course = ctx.session.query(Course).filter(
Course.id == attendance.session.course_id,
Course.supervisor_id == supervisor_id
).first()
if not course:
return custom_response(
success=False,
data="You are not authorized to resolve this dispute",
status_code=403
)
# Resolve the dispute
attendance.is_disputed = False
attendance.dispute_resolved_by = supervisor_id
attendance.dispute_resolution_date = datetime.utcnow()
attendance.resolution_notes = resolution.get('notes')
if 'new_status' in resolution:
attendance.status = resolution['new_status']
# Notify student
student_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == attendance.student_id
).first()
if student_prefs and student_prefs.receive_email:
subject = "Attendance Dispute Resolved"
message = f"""
Attendance Dispute Resolution
Hello {attendance.student.first_name},
Your attendance dispute has been resolved by your supervisor:
Course: {attendance.session.course.code} - {attendance.session.course.title}
Session: {attendance.session.course.code} - {attendance.session.course.title}
Date: {attendance.session.start_time.strftime('%A, %B %d, %Y') if attendance.session.start_time else 'TBD'}
Original Status: {attendance.status.value}
Resolution: {resolution.get('notes', 'No details provided')}
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=attendance.student.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send dispute resolution notification to student: {str(e)}")
# Notify tutor
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == attendance.tutor_id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = "Attendance Dispute Resolved"
message = f"""
Attendance Dispute Resolution
Hello {attendance.tutor.first_name},
An attendance dispute you were involved in has been resolved by your supervisor:
Student: {attendance.student.first_name} {attendance.student.last_name}
Course: {attendance.session.course.code} - {attendance.session.course.title}
Session: {attendance.session.course.code} - {attendance.session.course.title}
Date: {attendance.session.start_time.strftime('%A, %B %d, %Y') if attendance.session.start_time else 'TBD'}
Original Status: {attendance.status.value}
Resolution: {resolution.get('notes', 'No details provided')}
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=attendance.tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send dispute resolution notification to tutor: {str(e)}")
ctx.session.commit()
return custom_response(
success=True,
data="Attendance dispute resolved successfully",
status_code=200
)
def get_pending_verifications(self, supervisor_id: str, page: int = 1, per_page: int = 10) -> Dict:
"""Get daily teaching sessions that need verification"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get daily teaching sessions that need verification
query = ctx.session.query(DailyTeachingSession).join(
Course,
DailyTeachingSession.course_id == Course.id
).options(
joinedload(DailyTeachingSession.attendance_records),
joinedload(DailyTeachingSession.tutor),
joinedload(DailyTeachingSession.course)
).filter(
Course.supervisor_id == supervisor_id,
DailyTeachingSession.end_time < datetime.utcnow(),
DailyTeachingSession.is_verified == False
).order_by(DailyTeachingSession.end_time.desc())
total = query.count()
sessions = query.offset((page - 1) * per_page).limit(per_page).all()
return custom_response(
success=True,
data={
'sessions': [{
'id': session.id,
'course_id': session.course_id,
'course_code': session.course.code,
'course_title': session.course.title,
'session_title': f"{session.course.code} - {session.course.title}",
'tutor_id': session.tutor_id,
'tutor_name': f"{session.tutor.first_name} {session.tutor.last_name}" if session.tutor else None,
'tutor_email': session.tutor.email if session.tutor else None,
'start_time': session.start_time.isoformat() if session.start_time else None,
'end_time': session.end_time.isoformat() if session.end_time else None,
'room': session.room,
'session_date': session.session_date.isoformat() if session.session_date else None,
'day_name': session.session_date.strftime('%A') if session.session_date else None,
'session_type': session.session_type,
'status': session.status,
'notes': session.notes,
'created_at': session.created_at.isoformat() if session.created_at else None,
'updated_at': session.updated_at.isoformat() if session.updated_at else None,
'attendance_count': len(session.attendance_records) if session.attendance_records else 0,
'pending_attendance': len([a for a in session.attendance_records if a.status == AttendanceStatus.pending_approval]) if session.attendance_records else 0,
'disputed_attendance': len([a for a in session.attendance_records if getattr(a, 'is_disputed', False)]) if session.attendance_records else 0,
'present_attendance': len([a for a in session.attendance_records if a.status == AttendanceStatus.present]) if session.attendance_records else 0,
'absent_attendance': len([a for a in session.attendance_records if a.status == AttendanceStatus.absent]) if session.attendance_records else 0,
'late_attendance': len([a for a in session.attendance_records if a.status == AttendanceStatus.late]) if session.attendance_records else 0,
'attendance_rate': round((len([a for a in session.attendance_records if a.status == AttendanceStatus.present]) / len(session.attendance_records)) * 100, 1) if session.attendance_records and len(session.attendance_records) > 0 else 0,
'duration_minutes': int((session.end_time.hour * 60 + session.end_time.minute) - (session.start_time.hour * 60 + session.start_time.minute)) if session.start_time and session.end_time else 0,
'is_verified': session.is_verified,
'verified_by': session.verified_by,
'verified_at': session.verified_at.isoformat() if session.verified_at else None,
'verification_method': session.verification_method,
'timetable_block_id': session.timetable_block_id,
'department': session.course.department if session.course else None,
'semester': getattr(session.course, 'semester', None) if session.course else None,
'academic_year': getattr(session.course, 'academic_year', None) if session.course else None,
'attendance_taken': session.attendance_taken
} for session in sessions],
'total': total,
'page': page,
'per_page': per_page
}
)
def get_pending_approvals(self, supervisor_id: str, page: int = 1, per_page: int = 10) -> Dict:
"""Get tutor availability requests that need approval"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get availability requests that need approval
query = ctx.session.query(TutorAvailability).filter(
TutorAvailability.is_approved == False,
TutorAvailability.is_cancelled == False,
TutorAvailability.tutor_id.in_(
ctx.session.query(Tutor.id).filter(
Tutor.supervisor_id == supervisor_id
)
)
).order_by(TutorAvailability.created_at.desc())
total = query.count()
availabilities = query.offset((page - 1) * per_page).limit(per_page).all()
return custom_response(
success=True,
data={
'availabilities': [{
'id': avail.id,
'tutor_id': avail.tutor_id,
'tutor_name': f"{avail.tutor.first_name} {avail.tutor.last_name}",
'day_of_week': avail.day_of_week,
'start_time': avail.start_time.isoformat(),
'end_time': avail.end_time.isoformat(),
'availability_type': avail.availability_type,
'valid_from': str(avail.valid_from) if avail.valid_from else None,
'valid_to': str(avail.valid_to) if avail.valid_to else None,
'notes': avail.notes,
'created_at': avail.created_at.isoformat()
} for avail in availabilities],
'total': total,
'page': page,
'per_page': per_page
}
)
def get_pending_timetables(self, supervisor_id: str, page: int = 1, per_page: int = 10) -> Dict:
"""Get timetables that need approval"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get timetables that need approval
query = ctx.session.query(Timetable).filter(
Timetable.approval_status == 'pending',
or_(
Timetable.created_by == supervisor_id,
supervisor.is_head_of_department == True
)
).order_by(Timetable.created_at.desc())
total = query.count()
timetables = query.offset((page - 1) * per_page).limit(per_page).all()
return custom_response(
success=True,
data={
'timetables': [{
'id': timetable.id,
'name': timetable.name,
'semester': timetable.semester,
'academic_year': timetable.academic_year,
'created_at': timetable.created_at.isoformat(),
'created_by': timetable.creator.first_name + " " + timetable.creator.last_name,
'blocks_count': len(timetable.timetable_blocks)
} for timetable in timetables],
'total': total,
'page': page,
'per_page': per_page
}
)
def send_session_completion_alerts(self) -> Dict:
"""Send alerts to supervisors when sessions are completed (to be run periodically)"""
with DatabaseContextManager() as ctx:
# Get sessions that ended in the last hour and haven't had alerts sent
alert_window_start = datetime.utcnow() - timedelta(hours=1)
alert_window_end = datetime.utcnow()
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.end_time >= alert_window_start,
TeachingSession.end_time <= alert_window_end,
TeachingSession.completion_alert_sent == False,
TeachingSession.status == 'completed'
).all()
results = []
for session in sessions:
# Get supervisor
supervisor = session.course.supervisor
if not supervisor:
continue
# Get supervisor notification preferences
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == supervisor.id
).first()
if not prefs or not prefs.receive_email:
continue
# Send email alert
subject = f"Session Completed: {session.course.code} - {session.course.title}"
message = f"""
Session Completion Alert
Hello {supervisor.first_name},
A teaching session has just been completed:
Course: {session.course.code} - {session.course.title}
Session: {session.course.code} - {session.course.title}
Tutor: {session.tutor.first_name} {session.tutor.last_name}
Time: {session.start_time.strftime('%H:%M') if session.start_time else 'TBD'} - {session.end_time.strftime('%H:%M') if session.end_time else 'TBD'}
Room: {session.room}
Attendance: {len([a for a in session.attendance_records if a.status == AttendanceStatus.present])} present
Please review and verify this session when you have time.
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email="kisiwa@mutabletech.co.ke",
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=supervisor.email,
subject=subject,
message=message
)
# Mark alert as sent
session.completion_alert_sent = True
results.append({
'session_id': session.id,
'supervisor_id': supervisor.id,
'status': 'success'
})
except Exception as e:
current_app.logger.error(f"Failed to send completion alert to supervisor {supervisor.id}: {str(e)}")
results.append({
'session_id': session.id,
'supervisor_id': supervisor.id,
'status': 'failed',
'error': str(e)
})
ctx.session.commit()
return custom_response(
success=True,
data={
'alerts_sent': len([r for r in results if r['status'] == 'success']),
'alerts_failed': len([r for r in results if r['status'] == 'failed']),
'details': results
}
)
def update_tutor_availability(self, supervisor_id: str, availability_id: str, payload: Dict) -> Dict:
"""Update a specific availability slot (supervisor route)"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get the availability record
availability = ctx.session.query(TutorAvailability).filter(
TutorAvailability.id == availability_id
).first()
if not availability:
return custom_response(
success=False,
data="Availability record not found",
status_code=404
)
# Check if supervisor manages this tutor
tutor = ctx.session.query(Tutor).filter(
Tutor.id == availability.tutor_id,
Tutor.supervisor_id == supervisor_id
).first()
if not tutor:
return custom_response(
success=False,
data="You are not authorized to update this tutor's availability",
status_code=403
)
# Update fields
if 'day_of_week' in payload:
availability.day_of_week = payload['day_of_week']
if 'start_time' in payload:
availability.start_time = payload['start_time']
if 'end_time' in payload:
availability.end_time = payload['end_time']
if 'is_recurring' in payload:
availability.is_recurring = payload['is_recurring']
if 'valid_from' in payload:
availability.valid_from = payload['valid_from']
if 'valid_to' in payload:
availability.valid_to = payload['valid_to']
if 'availability_type' in payload:
availability.availability_type = payload['availability_type']
if 'location' in payload:
availability.location = payload['location']
if 'notes' in payload:
availability.notes = payload['notes']
# If any changes were made, set approval to pending
availability.is_approved = False
availability.approval_date = None
ctx.session.commit()
return custom_response(
success=True,
data="Availability updated successfully. Requires re-approval.",
status_code=200
)
def bulk_edit_tutor_availability(self, supervisor_id: str, payload: Dict) -> Dict:
"""Bulk edit availability slots for a specific tutor"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Extract payload data
tutor_id = payload.get('tutor_id')
changes = payload.get('changes', {})
if not tutor_id:
return custom_response(
success=False,
data="Tutor ID is required",
status_code=400
)
# Check if supervisor manages this tutor
tutor = ctx.session.query(Tutor).filter(
Tutor.id == tutor_id,
Tutor.supervisor_id == supervisor_id
).first()
if not tutor:
return custom_response(
success=False,
data="You are not authorized to edit this tutor's availability",
status_code=403
)
# Get all availability records for this tutor
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id
).all()
if not availabilities:
return custom_response(
success=False,
data="No availability records found for this tutor",
status_code=404
)
# Track changes made
updated_count = 0
errors = []
# Validate time changes if both start and end time are provided
if 'start_time' in changes and changes['start_time'] and 'end_time' in changes and changes['end_time']:
try:
start_time = datetime.strptime(changes['start_time'], '%H:%M').time()
end_time = datetime.strptime(changes['end_time'], '%H:%M').time()
if start_time >= end_time:
return custom_response(
success=False,
data="Start time must be before end time",
status_code=400
)
except ValueError:
return custom_response(
success=False,
data="Invalid time format. Use HH:MM format",
status_code=400
)
# Apply bulk changes to each availability record
for availability in availabilities:
try:
# Update fields if provided in changes
if 'availability_type' in changes and changes['availability_type']:
availability.availability_type = changes['availability_type']
if 'start_time' in changes and changes['start_time']:
availability.start_time = changes['start_time']
if 'end_time' in changes and changes['end_time']:
availability.end_time = changes['end_time']
if 'location' in changes and changes['location']:
availability.location = changes['location']
if 'notes' in changes and changes['notes']:
availability.notes = changes['notes']
# If any changes were made, set approval to pending
if any([
'availability_type' in changes and changes['availability_type'],
'start_time' in changes and changes['start_time'],
'end_time' in changes and changes['end_time'],
'location' in changes and changes['location'],
'notes' in changes and changes['notes']
]):
availability.is_approved = False
availability.approval_date = None
updated_count += 1
except Exception as e:
errors.append({
'availability_id': availability.id,
'error': str(e)
})
if errors:
ctx.session.rollback()
return custom_response(
success=False,
data=f"Failed to update {len(errors)} availability records. Errors: {errors}",
status_code=500
)
# Commit all changes
ctx.session.commit()
return custom_response(
success=True,
data={
'message': f'Successfully updated {updated_count} availability records',
'updated_count': updated_count,
'total_records': len(availabilities),
'tutor_name': f"{tutor.first_name} {tutor.last_name}",
'changes_applied': {k: v for k, v in changes.items() if v}
},
status_code=200
)
def send_availability_reminders(self, supervisor_id: str, session_id: str) -> Dict:
"""Send email reminders to tutors for availability slots"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get the opening session
from src.models.models import AvailabilityOpeningSession
opening_session = ctx.session.query(AvailabilityOpeningSession).filter(
AvailabilityOpeningSession.id == session_id
).first()
if not opening_session:
return custom_response(
success=False,
data="Opening session not found",
status_code=404
)
# Check if session is active
if opening_session.status != 'active' or not opening_session.is_active:
return custom_response(
success=False,
data="Can only send reminders for active opening sessions",
status_code=400
)
# Get all tutors managed by this supervisor
managed_tutors = ctx.session.query(Tutor).filter(
Tutor.supervisor_id == supervisor_id,
Tutor.is_active == True
).all()
if not managed_tutors:
return custom_response(
success=False,
data="No tutors found under your supervision",
status_code=404
)
# Track reminder sending
reminders_sent = 0
failed_reminders = []
# Send reminders to each tutor
for tutor in managed_tutors:
try:
# Check if tutor has already submitted availability
existing_availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor.id,
TutorAvailability.is_approved == True
).count()
# Only send reminder if tutor hasn't submitted enough availability
if existing_availabilities < opening_session.max_slots_per_tutor:
# Prepare email content
email_subject = f"Reminder: Submit Your Availability for {opening_session.name}"
email_body = f"""
Dear {tutor.first_name} {tutor.last_name},
This is a friendly reminder that the availability submission period for "{opening_session.name}" is currently open.
Session Details:
- Name: {opening_session.name}
- Description: {opening_session.description or 'No description provided'}
- Valid Period: {opening_session.start_time.split('T')[0]} to {opening_session.end_time.split('T')[0]}
- Time Range: {opening_session.time_range_start} - {opening_session.time_range_end}
- Maximum Slots: {opening_session.max_slots_per_tutor}
- Your Current Slots: {existing_availabilities}
Please log into the system and submit your availability slots before the session expires.
If you have any questions, please contact your supervisor.
Best regards,
{supervisor.first_name} {supervisor.last_name}
Supervisor
"""
# Send email using the utility function
email_sent = send_email(
to_email=tutor.email,
subject=email_subject,
body=email_body
)
if email_sent:
reminders_sent += 1
current_app.logger.info(f"Reminder sent to tutor {tutor.id} ({tutor.email})")
else:
failed_reminders.append({
'tutor_id': tutor.id,
'email': tutor.email,
'reason': 'Email sending failed'
})
else:
current_app.logger.info(f"Tutor {tutor.id} already has sufficient availability slots")
except Exception as e:
current_app.logger.error(f"Failed to send reminder to tutor {tutor.id}: {str(e)}")
failed_reminders.append({
'tutor_id': tutor.id,
'email': tutor.email,
'reason': str(e)
})
# Log summary
current_app.logger.info(f"Reminder sending completed. Sent: {reminders_sent}, Failed: {len(failed_reminders)}")
if failed_reminders:
return custom_response(
success=True,
data={
'message': f'Reminders sent to {reminders_sent} tutors. {len(failed_reminders)} failed.',
'reminders_sent': reminders_sent,
'failed_count': len(failed_reminders),
'failed_details': failed_reminders,
'session_name': opening_session.name,
'supervisor_name': f"{supervisor.first_name} {supervisor.last_name}"
},
status_code=200
)
else:
return custom_response(
success=True,
data={
'message': f'Successfully sent reminders to {reminders_sent} tutors',
'reminders_sent': reminders_sent,
'failed_count': 0,
'session_name': opening_session.name,
'supervisor_name': f"{supervisor.first_name} {supervisor.last_name}"
},
status_code=200
)
def get_approved_timetables_for_supervisor(self, supervisor_id: str) -> Dict:
"""
Get approved timetables for a supervisor with comprehensive statistics and blocks
Args:
supervisor_id: Supervisor ID
Returns:
Response with approved timetables including blocks and statistics
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Get approved timetables created by this supervisor
approved_timetables = ctx.session.query(Timetable).filter(
and_(
Timetable.approval_status == 'approved',
Timetable.is_active == True,
Timetable.created_by == supervisor_id
)
).order_by(Timetable.created_at.desc()).all()
# If no timetables created by supervisor, get all approved timetables as fallback
if not approved_timetables:
all_approved_timetables = ctx.session.query(Timetable).filter(
and_(
Timetable.approval_status == 'approved',
Timetable.is_active == True
)
).order_by(Timetable.created_at.desc()).all()
if all_approved_timetables:
approved_timetables = all_approved_timetables
else:
return custom_response(
success=False,
data="No approved timetables found",
status_code=404
)
# Format timetables with comprehensive data
formatted_timetables = []
for timetable in approved_timetables: # Get all blocks for this timetable
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).all()
# Calculate comprehensive statistics
total_courses = len(set(block.course_id for block in blocks if block.course_id))
total_tutors = len(set(block.tutor_id for block in blocks if block.tutor_id))
total_students = 0 # This would need to be calculated from enrollments
total_hours = sum(
(datetime.combine(datetime.min.date(), block.end_time) -
datetime.combine(datetime.min.date(), block.start_time)).total_seconds() / 3600
for block in blocks if block.start_time and block.end_time
)
room_assignments = len(set(block.room for block in blocks if block.room))
conflict_count = 0 # This would need conflict detection logic
completion_rate = min(100.0, (len(blocks) / max(1, total_courses)) * 100) if total_courses > 0 else 0.0
# Get department and speciality from the first course if available
department = "General"
speciality = "General"
if blocks:
first_course = ctx.session.query(Course).filter(Course.id == blocks[0].course_id).first()
if first_course:
department = first_course.department or "General"
speciality = first_course.speciality.name if first_course.speciality else "General"
# Convert blocks to dictionary format using the same logic as timetable utils
blocks_data = []
for i, block in enumerate(blocks):
# Get course and tutor information
course = ctx.session.query(Course).filter(Course.id == block.course_id).first()
tutor = ctx.session.query(Tutor).filter(Tutor.id == block.tutor_id).first()
# Fallback: If tutor not found in Tutor table, look in User table
if not tutor and block.tutor_id:
user_tutor = ctx.session.query(User).filter(User.id == block.tutor_id).first()
if user_tutor:
class MockTutor:
def __init__(self, user):
self.id = user.id
self.first_name = user.first_name
self.last_name = user.last_name
self.email = user.email
self.staff_id = getattr(user, 'staff_id', None)
self.qualification = getattr(user, 'qualification', None)
tutor = MockTutor(user_tutor)
block_data = {
'id': block.id,
'timetable_id': block.timetable_id,
'course_id': block.course_id,
'course_code': course.code if course else 'Unknown',
'course_title': course.title if course else 'Unknown',
'tutor_id': block.tutor_id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}" if tutor else 'Unknown',
'tutor_email': tutor.email if tutor else None,
'room': block.room,
'day_of_week': block.day_of_week,
'start_time': block.start_time.isoformat() if block.start_time else None,
'end_time': block.end_time.isoformat() if block.end_time else None,
'block_type': block.block_type,
'recurring': block.recurring,
'created_at': block.created_at.isoformat() if block.created_at else None,
'created_by': block.created_by,
'supervisor_tutor_id': block.supervisor_tutor_id,
}
blocks_data.append(block_data)
# Create comprehensive timetable data structure
timetable_data = {
'id': timetable.id,
'name': timetable.name,
'description': timetable.description,
'academic_year': timetable.academic_year,
'term': getattr(timetable, 'term', 'Term 1'),
'semester': timetable.semester,
'is_active': timetable.is_active,
'created_at': timetable.created_at.isoformat() if timetable.created_at else None,
'updated_at': getattr(timetable, 'updated_at', timetable.created_at).isoformat() if getattr(timetable, 'updated_at', timetable.created_at) else None,
'supervisor_id': timetable.created_by,
'supervisor_name': f"{supervisor.first_name} {supervisor.last_name}",
'total_courses': total_courses,
'total_students': total_students,
'total_tutors': total_tutors,
'total_hours': total_hours,
'status': getattr(timetable, 'status', 'draft'),
'approval_status': timetable.approval_status,
'last_modified': getattr(timetable, 'updated_at', timetable.created_at).isoformat() if getattr(timetable, 'updated_at', timetable.created_at) else None,
'version': getattr(timetable, 'version', '1.0'),
'department': department,
'speciality': speciality,
'room_assignments': room_assignments,
'conflict_count': conflict_count,
'completion_rate': completion_rate,
'blocks': blocks_data,
'blocks_count': len(blocks)
}
formatted_timetables.append(timetable_data)
# Return the first timetable (most relevant) instead of an array
if formatted_timetables:
return custom_response(
success=True,
data=formatted_timetables[0], # Return single timetable object
status_code=200
)
else:
return custom_response(
success=True,
data=None,
status_code=200
)
def get_course_analytics_for_supervisor(self, supervisor_id: str) -> Dict:
"""
Get comprehensive course analytics for a supervisor
Args:
supervisor_id: Supervisor ID
Returns:
Response with course analytics including specialities, courses, and timetable blocks
"""
with DatabaseContextManager() as ctx:
current_app.logger.info(f"[SUPERVISOR ANALYTICS DEBUG] Starting get_course_analytics_for_supervisor for supervisor_id: {supervisor_id}")
# Verify user exists and has supervisor privileges
user = ctx.session.query(User).filter(
User.id == supervisor_id,
User.is_active == True
).first()
if not user:
current_app.logger.error(f"[SUPERVISOR ANALYTICS DEBUG] User not found: {supervisor_id}")
return custom_response(
success=False,
data="User not found or inactive",
status_code=404
)
# Check if user has supervisor privileges (either Supervisor record or supervisor role)
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor and user.role != 'supervisor':
current_app.logger.error(f"[SUPERVISOR ANALYTICS DEBUG] User {supervisor_id} does not have supervisor privileges")
return custom_response(
success=False,
data="User does not have supervisor privileges",
status_code=403
)
current_app.logger.info(f"[SUPERVISOR ANALYTICS DEBUG] Found user: {user.first_name} {user.last_name}")
# Get supervisor's department(s)
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
if not supervisor_departments:
current_app.logger.warning(f"[SUPERVISOR ANALYTICS DEBUG] No departments found for supervisor {supervisor_id}")
return custom_response(
success=False,
data="Supervisor has no assigned departments",
status_code=404
)
# Get department names
department_names = [dept.department_name for dept in supervisor_departments]
current_app.logger.info(f"[SUPERVISOR ANALYTICS DEBUG] Supervisor departments: {department_names}")
# Get courses filtered by supervisor's departments
courses = ctx.session.query(Course).filter(
and_(
Course.is_active == True,
Course.department.in_(department_names)
)
).all()
current_app.logger.info(f"[SUPERVISOR ANALYTICS DEBUG] Found {len(courses)} active courses in supervisor's departments")
# Get specialities filtered by supervisor's departments
specialities = ctx.session.query(Speciality).filter(
and_(
Speciality.is_active == True,
Speciality.department.in_(department_names)
)
).all()
current_app.logger.info(f"[SUPERVISOR ANALYTICS DEBUG] Found {len(specialities)} active specialities in supervisor's departments")
# Get timetable blocks from approved timetables filtered by supervisor's departments
approved_timetables = ctx.session.query(Timetable).filter(
and_(
Timetable.approval_status == 'approved',
Timetable.is_active == True,
Timetable.department.in_(department_names)
)
).all()
timetable_ids = [t.id for t in approved_timetables]
timetable_blocks = []
if timetable_ids:
timetable_blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id.in_(timetable_ids)
).all()
current_app.logger.info(f"[SUPERVISOR ANALYTICS DEBUG] Found {len(timetable_blocks)} timetable blocks from {len(approved_timetables)} approved timetables in supervisor's departments")
# Process courses data
courses_data = []
for course in courses:
# Get speciality information
speciality_info = None
if course.speciality_id:
speciality = ctx.session.query(Speciality).filter(Speciality.id == course.speciality_id).first()
if speciality:
speciality_info = {
'id': speciality.id,
'name': speciality.name,
'description': speciality.description,
'department': speciality.department
}
course_data = {
'id': course.id,
'code': course.code,
'title': course.title,
'department': course.department,
'speciality': speciality_info,
'credits': course.credits,
'semester': course.semester,
'academic_year': course.academic_session.year if course.academic_session else None,
'is_active': course.is_active,
'course_level': course.course_level,
'prerequisites': [], # Course model doesn't have prerequisites field
'learning_outcomes': course.learning_outcomes.split(',') if course.learning_outcomes else []
}
courses_data.append(course_data)
# Process specialities data
specialities_data = []
for speciality in specialities:
speciality_data = {
'id': speciality.id,
'name': speciality.name,
'description': speciality.description,
'department': speciality.department,
'is_active': speciality.is_active
}
specialities_data.append(speciality_data)
# Process timetable blocks data
blocks_data = []
total_hours = 0
for block in timetable_blocks:
# Get course information
course = ctx.session.query(Course).filter(Course.id == block.course_id).first()
if not course:
continue
# Get tutor information with User table fallback
tutor = ctx.session.query(Tutor).filter(Tutor.id == block.tutor_id).first()
if not tutor and block.tutor_id:
user_tutor = ctx.session.query(User).filter(User.id == block.tutor_id).first()
if user_tutor:
class MockTutor:
def __init__(self, user):
self.id = user.id
self.first_name = user.first_name
self.last_name = user.last_name
self.email = user.email
tutor = MockTutor(user_tutor)
# Get speciality information
speciality_name = "General"
if course.speciality_id:
speciality = ctx.session.query(Speciality).filter(Speciality.id == course.speciality_id).first()
if speciality:
speciality_name = speciality.name
# Calculate hours
if block.start_time and block.end_time:
start_dt = datetime.combine(datetime.min.date(), block.start_time)
end_dt = datetime.combine(datetime.min.date(), block.end_time)
hours = (end_dt - start_dt).total_seconds() / 3600
total_hours += hours
block_data = {
'id': block.id,
'course_id': block.course_id,
'course_code': course.code,
'course_title': course.title,
'tutor_id': block.tutor_id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}" if tutor else 'Unknown',
'day_of_week': block.day_of_week,
'start_time': block.start_time.isoformat() if block.start_time else None,
'end_time': block.end_time.isoformat() if block.end_time else None,
'room': block.room,
'block_type': block.block_type,
'recurring': block.recurring,
'speciality': speciality_name
}
blocks_data.append(block_data)
# Calculate statistics
total_courses = len(courses_data)
total_specialities = len(specialities_data)
total_blocks = len(blocks_data)
average_blocks_per_course = total_blocks / total_courses if total_courses > 0 else 0
average_hours_per_speciality = total_hours / total_specialities if total_specialities > 0 else 0
# Calculate credit hours by course level
credit_hours_by_level = {}
for course in courses_data:
level = course.get('course_level', 'Unknown')
credits = course.get('credits', 0) or 0
if level in credit_hours_by_level:
credit_hours_by_level[level] += credits
else:
credit_hours_by_level[level] = credits
# Convert to list format for frontend
credit_hours_by_level_list = [
{'level': level, 'credits': credits}
for level, credits in credit_hours_by_level.items()
]
statistics = {
'total_courses': total_courses,
'total_specialities': total_specialities,
'total_blocks': total_blocks,
'total_hours': round(total_hours, 2),
'average_blocks_per_course': round(average_blocks_per_course, 2),
'average_hours_per_speciality': round(average_hours_per_speciality, 2),
'credit_hours_by_level': credit_hours_by_level_list
}
current_app.logger.info(f"[SUPERVISOR ANALYTICS DEBUG] Calculated statistics: {statistics}")
analytics_data = {
'courses': courses_data,
'specialities': specialities_data,
'timetable_blocks': blocks_data,
'statistics': statistics
}
current_app.logger.info(f"[SUPERVISOR ANALYTICS DEBUG] Returning analytics data with {len(courses_data)} courses, {len(specialities_data)} specialities, {len(blocks_data)} blocks")
return custom_response(
success=True,
data=analytics_data,
status_code=200
)
def get_approved_timetables_for_supervisor_by_speciality(self, supervisor_id: str, speciality_id: str = None) -> Dict:
"""
Get approved timetables for a supervisor's department, filtered by speciality
Args:
supervisor_id: Supervisor ID
speciality_id: Optional speciality ID to filter by
Returns:
Response with approved timetables filtered by speciality
"""
with DatabaseContextManager() as ctx:
# First, get the supervisor and their department
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get the supervisor's primary department
primary_department = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_primary == True,
SupervisorDepartment.is_active == True
).first()
if not primary_department:
# Fallback: Check if supervisor has any department assignment (even if not primary)
any_department = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).first()
if any_department:
department_name = any_department.department_name
else:
# Auto-assign to Business department if no assignment exists
business_timetables = ctx.session.query(Timetable).filter(
Timetable.department == 'Business'
).count()
if business_timetables > 0:
# Create department assignment
from datetime import date
new_dept_assignment = SupervisorDepartment(
supervisor_id=supervisor_id,
department_name='Business',
is_primary=True,
is_active=True,
assigned_by=supervisor_id,
notes='Auto-assigned to Business department',
assigned_date=date.today()
)
ctx.session.add(new_dept_assignment)
ctx.session.commit()
department_name = 'Business'
else:
return custom_response(
success=False,
data="Supervisor has no assigned department and no Business timetables found",
status_code=404
)
else:
department_name = primary_department.department_name
# Get approved timetables for the supervisor's department
timetables = ctx.session.query(Timetable).filter(
and_(
Timetable.department == department_name,
Timetable.approval_status == 'approved',
Timetable.is_active == True
)
).order_by(Timetable.created_at.desc()).all()
if not timetables:
# Fallback: If no department-specific timetables, return any approved timetables
all_approved_timetables = ctx.session.query(Timetable).filter(
and_(
Timetable.approval_status == 'approved',
Timetable.is_active == True
)
).all()
if all_approved_timetables:
timetables = all_approved_timetables
else:
return custom_response(
success=False,
data=f"No approved timetables found for {department_name} department",
status_code=404
)
# Format timetables with blocks filtered by speciality
formatted_timetables = []
for timetable in timetables:
# Get all blocks for this approved timetable (no department filtering)
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).all()
current_app.logger.info(f"[SUPERVISOR SPECIALITY DEBUG] Found {len(blocks)} blocks for approved timetable {timetable.id} (no department filtering)")
# Add speciality filter if provided
if speciality_id:
current_app.logger.info(f"[SUPERVISOR SPECIALITY DEBUG] Adding speciality filter: {speciality_id}")
# Filter blocks by speciality
filtered_blocks = []
for block in blocks:
course = ctx.session.query(Course).filter(Course.id == block.course_id).first()
if course and course.speciality_id == speciality_id:
filtered_blocks.append(block)
blocks = filtered_blocks
current_app.logger.info(f"[SUPERVISOR SPECIALITY DEBUG] After speciality filter: Found {len(blocks)} blocks")
if len(blocks) == 0:
current_app.logger.warning(f"[SUPERVISOR SPECIALITY DEBUG] No blocks found for timetable {timetable.id}!")
else:
current_app.logger.info(f"[SUPERVISOR SPECIALITY DEBUG] Successfully found {len(blocks)} blocks for approved timetable {timetable.id}")
# Convert timetable to dict
timetable_data = {
'id': timetable.id,
'name': timetable.name,
'department': timetable.department,
'semester': timetable.semester,
'academic_year': timetable.academic_year,
'status': timetable.status,
'approval_status': timetable.approval_status,
'created_at': timetable.created_at.isoformat() if timetable.created_at else None,
'updated_at': timetable.updated_at.isoformat() if timetable.updated_at else None,
'created_by': timetable.created_by,
'approved_by': timetable.approved_by,
'is_active': timetable.is_active,
'blocks': []
}
# Convert blocks to dict with speciality information
for block in blocks:
# Get course, tutor, and speciality information
course = ctx.session.query(Course).filter(Course.id == block.course_id).first()
tutor = ctx.session.query(Tutor).filter(Tutor.id == block.tutor_id).first()
# Fallback: If tutor not found in Tutor table, look in User table
if not tutor and block.tutor_id:
current_app.logger.info(f"[SUPERVISOR SPECIALITY DEBUG] Tutor {block.tutor_id} not found in Tutor table, checking User table")
user_tutor = ctx.session.query(User).filter(User.id == block.tutor_id).first()
if user_tutor:
current_app.logger.info(f"[SUPERVISOR SPECIALITY DEBUG] Found tutor {block.tutor_id} in User table: {user_tutor.first_name} {user_tutor.last_name}")
# Create a mock tutor object with User data
class MockTutor:
def __init__(self, user):
self.id = user.id
self.first_name = user.first_name
self.last_name = user.last_name
self.email = user.email
self.staff_id = getattr(user, 'staff_id', None)
self.qualification = getattr(user, 'qualification', None)
tutor = MockTutor(user_tutor)
else:
current_app.logger.warning(f"[SUPERVISOR SPECIALITY DEBUG] Tutor {block.tutor_id} not found in User table either!")
# Get speciality information if course has one
speciality_info = None
if course and course.speciality_id:
speciality = ctx.session.query(Speciality).filter(Speciality.id == course.speciality_id).first()
if speciality:
speciality_info = {
'id': speciality.id,
'name': speciality.name,
'department': speciality.department,
'code': speciality.code,
'speciality_level': speciality.speciality_level
}
block_data = {
'id': block.id,
'timetable_id': block.timetable_id,
'course_id': block.course_id,
'course_code': course.code if course else 'Unknown',
'course_title': course.title if course else 'Unknown',
'course_speciality': speciality_info,
'tutor_id': block.tutor_id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}" if tutor else 'Unknown',
'tutor_email': tutor.email if tutor else None,
'room': block.room,
'day_of_week': block.day_of_week,
'start_time': block.start_time.isoformat() if block.start_time else None,
'end_time': block.end_time.isoformat() if block.end_time else None,
'block_type': block.block_type,
'recurring': block.recurring,
'created_at': block.created_at.isoformat() if block.created_at else None,
'created_by': block.created_by,
'supervisor_tutor_id': block.supervisor_tutor_id,
}
timetable_data['blocks'].append(block_data)
# Only include timetables that have blocks after speciality filtering
if timetable_data['blocks']:
formatted_timetables.append(timetable_data)
# Add speciality information to response if filtering by speciality
response_data = {
'timetables': formatted_timetables,
'total_blocks': sum(len(t['blocks']) for t in formatted_timetables),
'filtered_by_speciality': speciality_id is not None
}
if speciality_id:
speciality = ctx.session.query(Speciality).filter(Speciality.id == speciality_id).first()
if speciality:
response_data['speciality_info'] = {
'id': speciality.id,
'name': speciality.name,
'department': speciality.department,
'code': speciality.code,
'speciality_level': speciality.speciality_level
}
return custom_response(
success=True,
data=response_data,
status_code=200
)
def get_supervisor_dashboard_stats(self, supervisor_id: str) -> Dict:
"""
Get comprehensive dashboard statistics for a supervisor including
tutor performance, session statistics, and department metrics.
Args:
supervisor_id: ID of the supervisor
Returns:
Response with comprehensive dashboard statistics
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get all courses supervised by this supervisor
courses = ctx.session.query(Course).filter(
Course.supervisor_id == supervisor_id,
Course.is_active == True
).all()
if not courses:
return custom_response(
success=True,
data={
'summary_stats': {
'total_tutors': 0,
'active_tutors': 0,
'average_attendance': 0,
'total_courses': 0,
'total_sessions': 0,
'completed_sessions': 0,
'verified_sessions': 0,
'upcoming_sessions': 0,
'average_verification_rate': 0,
'high_performers': 0,
'needs_attention': 0
},
'department_stats': [],
'recent_activity': [],
'performance_trends': []
}
)
# Get tutors in the supervisor's departments
tutors = (
ctx.session.query(Tutor)
.join(TutorDepartment, Tutor.id == TutorDepartment.tutor_id)
.join(
SupervisorDepartment,
TutorDepartment.department_name == SupervisorDepartment.department_name
)
.filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True,
TutorDepartment.is_active == True,
Tutor.is_active == True
)
.distinct()
.all()
)
# Calculate comprehensive statistics
total_tutors = len(tutors)
active_tutors = 0
high_performers = 0
needs_attention = 0
total_attendance_rate = 0
total_verification_rate = 0
total_sessions = 0
completed_sessions = 0
verified_sessions = 0
upcoming_sessions = 0
tutor_performance_data = []
for tutor in tutors:
# Get assigned courses (only those supervised by this supervisor)
assigned_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.supervisor_id == supervisor_id,
Course.is_active == True
).all()
# Calculate teaching statistics
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.tutor_id == tutor.id,
TeachingSession.course_id.in_([c.id for c in courses])
).all()
# Get attendance records for these sessions
attendance_records = ctx.session.query(Attendance).join(
TeachingSession,
Attendance.session_id == TeachingSession.id
).filter(
TeachingSession.tutor_id == tutor.id,
TeachingSession.course_id.in_([c.id for c in courses])
).all()
# Calculate individual tutor statistics
tutor_total_sessions = len(sessions)
tutor_completed_sessions = len([s for s in sessions if s.status == 'completed'])
tutor_verified_sessions = len([s for s in sessions if s.is_verified])
present_count = sum(1 for r in attendance_records if r.status == AttendanceStatus.present)
tutor_attendance_rate = round((present_count / len(attendance_records)) * 100, 1) if attendance_records else 0
tutor_verification_rate = round((tutor_verified_sessions / tutor_completed_sessions) * 100, 1) if tutor_completed_sessions else 0
# Get upcoming sessions count
tutor_upcoming_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.tutor_id == tutor.id,
TeachingSession.course_id.in_([c.id for c in courses]),
TeachingSession.start_time >= datetime.utcnow(),
TeachingSession.status == 'scheduled'
).count()
# Accumulate totals
total_sessions += tutor_total_sessions
completed_sessions += tutor_completed_sessions
verified_sessions += tutor_verified_sessions
upcoming_sessions += tutor_upcoming_sessions
total_attendance_rate += tutor_attendance_rate
total_verification_rate += tutor_verification_rate
# Categorize tutors
if tutor_attendance_rate >= 75:
active_tutors += 1
if tutor_attendance_rate >= 90 and tutor_verification_rate >= 80:
high_performers += 1
elif tutor_attendance_rate < 70 or tutor_verification_rate < 60:
needs_attention += 1
# Store tutor performance data
tutor_performance_data.append({
'tutor_id': tutor.id,
'name': f"{tutor.first_name} {tutor.last_name}",
'attendance_rate': tutor_attendance_rate,
'verification_rate': tutor_verification_rate,
'total_sessions': tutor_total_sessions,
'completed_sessions': tutor_completed_sessions,
'verified_sessions': tutor_verified_sessions,
'upcoming_sessions': tutor_upcoming_sessions,
'courses_count': len(assigned_courses)
})
# Calculate averages
average_attendance = round(total_attendance_rate / total_tutors, 1) if total_tutors > 0 else 0
average_verification_rate = round(total_verification_rate / total_tutors, 1) if total_tutors > 0 else 0
# Get department-wise statistics
department_stats = []
for department in supervisor.departments:
dept_courses = [c for c in courses if c.department == department.department_name]
dept_tutors = (
ctx.session.query(Tutor)
.join(TutorDepartment, Tutor.id == TutorDepartment.tutor_id)
.filter(
TutorDepartment.department_name == department.department_name,
TutorDepartment.is_active == True,
Tutor.is_active == True
)
.distinct()
.all()
)
dept_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id.in_([c.id for c in dept_courses])
).all()
department_stats.append({
'department_name': department.department_name,
'courses_count': len(dept_courses),
'tutors_count': len(dept_tutors),
'sessions_count': len(dept_sessions),
'completed_sessions': len([s for s in dept_sessions if s.status == 'completed'])
})
# Get recent activity (last 10 sessions)
recent_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id.in_([c.id for c in courses])
).order_by(TeachingSession.start_time.desc()).limit(10).all()
recent_activity = []
for session in recent_sessions:
tutor = ctx.session.query(Tutor).filter(Tutor.id == session.tutor_id).first()
recent_activity.append({
'session_id': session.id,
'title': session.title,
'course_code': session.course.code,
'tutor_name': f"{tutor.first_name} {tutor.last_name}" if tutor else "Unknown",
'start_time': session.start_time.isoformat(),
'status': session.status,
'is_verified': session.is_verified
})
# Prepare comprehensive response
response_data = {
'summary_stats': {
'total_tutors': total_tutors,
'active_tutors': active_tutors,
'average_attendance': average_attendance,
'total_courses': len(courses),
'total_sessions': total_sessions,
'completed_sessions': completed_sessions,
'verified_sessions': verified_sessions,
'upcoming_sessions': upcoming_sessions,
'average_verification_rate': average_verification_rate,
'high_performers': high_performers,
'needs_attention': needs_attention
},
'tutor_performance': tutor_performance_data,
'department_stats': department_stats,
'recent_activity': recent_activity,
'supervisor_info': {
'id': supervisor.id,
'name': f"{supervisor.first_name} {supervisor.last_name}",
'email': supervisor.email,
'departments': [dept.department_name for dept in supervisor.departments]
}
}
return custom_response(
success=True,
data=response_data,
status_code=200
)
def get_supervisor_timetable_blocks(self, supervisor_id: str) -> Dict:
"""
Get timetable blocks for authenticated supervisor from approved timetables
Args:
supervisor_id: ID of the supervisor
Returns:
Response with timetable blocks data
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get approved timetables that the supervisor can access
# First, try to get timetables created by this supervisor
supervisor_timetables = ctx.session.query(Timetable).filter(
and_(
Timetable.created_by == supervisor_id,
Timetable.approval_status == 'approved',
Timetable.is_active == True
)
).order_by(Timetable.created_at.desc()).all()
# If no timetables created by supervisor, get timetables for their departments
if not supervisor_timetables and supervisor.departments:
department_names = [dept.department_name for dept in supervisor.departments]
# Get timetables for supervisor's departments
department_timetables = ctx.session.query(Timetable).join(
TimetableBlock, Timetable.id == TimetableBlock.timetable_id
).join(
Course, TimetableBlock.course_id == Course.id
).join(
course_department_association, Course.id == course_department_association.course_id
).filter(
and_(
course_department_association.department_name.in_(department_names),
Timetable.approval_status == 'approved',
Timetable.is_active == True
)
).distinct().order_by(Timetable.created_at.desc()).all()
supervisor_timetables = department_timetables
# If still no timetables, get all approved timetables as fallback
if not supervisor_timetables:
supervisor_timetables = ctx.session.query(Timetable).filter(
and_(
Timetable.approval_status == 'approved',
Timetable.is_active == True
)
).order_by(Timetable.created_at.desc()).all()
if not supervisor_timetables:
return custom_response(
success=False,
data="No approved timetables found",
status_code=404
)
# Get timetable blocks for these timetables where supervisor is the tutor
timetable_ids = [t.id for t in supervisor_timetables]
blocks = ctx.session.query(TimetableBlock).filter(
and_(
TimetableBlock.timetable_id.in_(timetable_ids),
TimetableBlock.tutor_id == supervisor_id
)
).all()
# Also include blocks where supervisor is assigned as supervisor tutor
supervisor_blocks = ctx.session.query(TimetableBlock).filter(
and_(
TimetableBlock.timetable_id.in_(timetable_ids),
TimetableBlock.supervisor_tutor_id == supervisor_id
)
).all()
# Combine both sets of blocks, removing duplicates
all_blocks = blocks + supervisor_blocks
unique_blocks = []
seen_ids = set()
for block in all_blocks:
if block.id not in seen_ids:
unique_blocks.append(block)
seen_ids.add(block.id)
# Format blocks data similar to the existing timetable utils format
blocks_data = []
for block in unique_blocks:
# Get course information
course = ctx.session.query(Course).filter(Course.id == block.course_id).first()
# Get tutor information
tutor = ctx.session.query(Tutor).filter(Tutor.id == block.tutor_id).first()
# Get supervisor tutor information if exists
supervisor_tutor = None
if hasattr(block, 'supervisor_tutor_id') and block.supervisor_tutor_id:
supervisor_tutor = ctx.session.query(Supervisor).filter(
Supervisor.id == block.supervisor_tutor_id
).first()
# Format block data
block_dict = {
'id': block.id,
'tutor_id': block.tutor_id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}" if tutor else "Unknown",
'course_id': block.course_id,
'course_code': course.code if course else "Unknown",
'course_title': course.title if course else "Unknown",
'day_of_week': block.day_of_week,
'start_time': block.start_time.strftime('%H:%M') if block.start_time else None,
'end_time': block.end_time.strftime('%H:%M') if block.end_time else None,
'room': block.room,
'block_type': block.block_type,
'recurring': block.recurring,
'timetable_id': block.timetable_id,
'department': course.department if course else "General",
'speciality_id': course.speciality_id if course else None,
'speciality': course.speciality.name if course and course.speciality else "General",
'term': course.semester if course else None,
'academic_year': course.academic_session.name if course.academic_session else None,
'created_at': block.created_at.isoformat() if block.created_at else None,
'created_by': block.created_by
}
# Add supervisor tutor information if exists
if supervisor_tutor:
block_dict['supervisor_tutor_id'] = supervisor_tutor.id
block_dict['supervisor_tutor_name'] = f"{supervisor_tutor.first_name} {supervisor_tutor.last_name}"
block_dict['supervisor_tutor_staff_id'] = supervisor_tutor.staff_id
# If no regular tutor, use supervisor tutor as the main tutor name
if not tutor:
block_dict['tutor_name'] = f"{supervisor_tutor.first_name} {supervisor_tutor.last_name}"
blocks_data.append(block_dict)
# Prepare response data
response_data = {
'supervisor_id': supervisor_id,
'supervisor_name': f"{supervisor.first_name} {supervisor.last_name}",
'total_blocks': len(blocks_data),
'timetables_count': len(supervisor_timetables),
'blocks': blocks_data,
'timetables': [
{
'id': t.id,
'name': t.name,
'semester': t.semester,
'academic_year': t.academic_year,
'status': t.approval_status,
'created_at': t.created_at.isoformat() if t.created_at else None,
'created_by': t.created_by
}
for t in supervisor_timetables
]
}
return custom_response(
success=True,
data=response_data,
status_code=200
)
def get_department_students(self, supervisor_id: str, page: int = 1, per_page: int = 10) -> Dict:
"""Get all students in the supervisor's department with comprehensive information"""
with DatabaseContextManager() as ctx:
# Get supervisor and their departments
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
if not supervisor_departments:
return custom_response(
success=False,
data="Supervisor has no assigned departments",
status_code=404
)
department_names = [dept.department_name for dept in supervisor_departments]
# Get students directly by department - much simpler and more reliable
students_query = ctx.session.query(Student).filter(
Student.department.in_(department_names),
Student.is_active == True
)
# Get total count
total_students = students_query.count()
# Apply pagination
offset = (page - 1) * per_page
students = students_query.offset(offset).limit(per_page).all()
# Build comprehensive student data
students_data = []
for student in students:
# Get student's courses through their speciality
student_courses = ctx.session.query(Course).filter(
Course.speciality_id == student.speciality_id,
Course.is_active == True
).all()
# Calculate attendance rate
attendance_records = ctx.session.query(Attendance).join(
TeachingSession, Attendance.session_id == TeachingSession.id
).join(
Course, TeachingSession.course_id == Course.id
).filter(
Course.speciality_id == student.speciality_id,
Attendance.student_id == student.id
).all()
total_sessions = len(attendance_records)
present_sessions = len([a for a in attendance_records if a.status.value == 'present'])
attendance_rate = round((present_sessions / total_sessions) * 100, 1) if total_sessions > 0 else 0
# Calculate assignment completion
assignments = ctx.session.query(Assignment).join(
Course, Assignment.course_id == Course.id
).filter(
Course.speciality_id == student.speciality_id
).all()
total_assignments = len(assignments)
completed_assignments = ctx.session.query(AssignmentSubmission).join(
Assignment, AssignmentSubmission.assignment_id == Assignment.id
).join(
Course, Assignment.course_id == Course.id
).filter(
Course.speciality_id == student.speciality_id,
AssignmentSubmission.student_id == student.id,
AssignmentSubmission.status.in_(['submitted', 'graded'])
).count()
assignment_completion = round((completed_assignments / total_assignments) * 100, 1) if total_assignments > 0 else 0
# Calculate average grade
graded_submissions = ctx.session.query(AssignmentSubmission).join(
Assignment, AssignmentSubmission.assignment_id == Assignment.id
).join(
Course, Assignment.course_id == Course.id
).filter(
Course.speciality_id == student.speciality_id,
AssignmentSubmission.student_id == student.id,
AssignmentSubmission.status == 'graded',
AssignmentSubmission.grade.isnot(None)
).all()
average_grade = round(sum(sub.grade for sub in graded_submissions) / len(graded_submissions), 2) if graded_submissions else None
# Build course data
courses_data = []
for course in student_courses:
course_attendance = ctx.session.query(Attendance).join(
TeachingSession, Attendance.session_id == TeachingSession.id
).filter(
TeachingSession.course_id == course.id,
Attendance.student_id == student.id
).all()
course_total = len(course_attendance)
course_present = len([a for a in course_attendance if a.status.value == 'present'])
course_attendance_rate = round((course_present / course_total) * 100, 1) if course_total > 0 else 0
# Get course average grade
course_assignments = ctx.session.query(AssignmentSubmission).join(
Assignment, AssignmentSubmission.assignment_id == Assignment.id
).filter(
Assignment.course_id == course.id,
AssignmentSubmission.student_id == student.id,
AssignmentSubmission.status == 'graded',
AssignmentSubmission.grade.isnot(None)
).all()
course_avg_grade = round(sum(sub.grade for sub in course_assignments) / len(course_assignments), 2) if course_assignments else None
courses_data.append({
'id': course.id,
'code': course.code,
'title': course.title,
'attendance_rate': course_attendance_rate,
'average_grade': course_avg_grade
})
student_data = {
'id': student.id,
'student_id': student.student_id,
'first_name': student.first_name,
'last_name': student.last_name,
'email': student.email,
'phone': student.phone,
'program': student.program,
'year_of_study': student.year_of_study,
'enrollment_date': student.enrollment_date.isoformat() if student.enrollment_date else None,
'cumulative_gpa': student.cumulative_gpa,
'is_on_probation': student.is_on_probation,
'probation_start_date': student.probation_start_date.isoformat() if student.probation_start_date else None,
'probation_end_date': student.probation_end_date.isoformat() if student.probation_end_date else None,
'probation_reason': student.probation_reason,
'attendance_rate': attendance_rate,
'assignment_completion': assignment_completion,
'average_grade': average_grade,
'courses': courses_data,
'speciality': {
'id': student.speciality.id if student.speciality else None,
'name': student.speciality.name if student.speciality else None,
'description': student.speciality.description if student.speciality else None
},
'guardian_name': student.guardian_name,
'guardian_contact': student.guardian_contact,
'emergency_contact': student.emergency_contact,
'address': student.address,
'city': student.city,
'country': student.country,
'postal_code': student.postal_code,
'profile_picture': student.profile_picture,
'created_at': student.created_at.isoformat() if student.created_at else None
}
students_data.append(student_data)
return custom_response(
success=True,
data={
'students': students_data,
'total': total_students,
'page': page,
'per_page': per_page,
'total_pages': (total_students + per_page - 1) // per_page
},
status_code=200
)
def assign_student_to_speciality(self, supervisor_id: str, student_id: str, speciality_id: str, term: str = None) -> Dict:
"""Assign a student to a speciality"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Verify student exists
student = ctx.session.query(Student).filter(Student.id == student_id).first()
if not student:
return custom_response(
success=False,
data="Student not found",
status_code=404
)
# Verify speciality exists
speciality = ctx.session.query(Speciality).filter(Speciality.id == speciality_id).first()
if not speciality:
return custom_response(
success=False,
data="Speciality not found",
status_code=404
)
# Check if student is in supervisor's department
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
department_names = [dept.department_name for dept in supervisor_departments]
if student.department not in department_names:
return custom_response(
success=False,
data="Student is not in your department",
status_code=403
)
# Check if speciality has courses in supervisor's departments
speciality_courses_query = ctx.session.query(Course).filter(
Course.speciality_id == speciality_id,
Course.department.in_(department_names),
Course.is_active == True
)
speciality_courses_count = speciality_courses_query.count()
if speciality_courses_count == 0:
return custom_response(
success=False,
data="This speciality has no courses in your department",
status_code=403
)
# Get the actual course objects
speciality_courses = speciality_courses_query.all()
# Update student's speciality
student.speciality_id = speciality_id
# Create or update enrollment for the student in this speciality
existing_enrollment = ctx.session.query(Enrollment).filter(
Enrollment.student_id == student_id,
Enrollment.speciality_id == speciality_id,
Enrollment.status == 'active'
).first()
if not existing_enrollment:
# Create new enrollment
enrollment = Enrollment(
student_id=student_id,
speciality_id=speciality_id,
enrollment_date=date.today(),
status='active',
enrollment_type='regular',
term=term
)
ctx.session.add(enrollment)
ctx.session.flush() # Flush to get the enrollment ID
# Add courses to the enrollment (filter by term if specified)
courses_to_add = speciality_courses
if term:
courses_to_add = [course for course in speciality_courses if course.semester == term]
for course in courses_to_add:
# Insert into enrollment_courses association table
ctx.session.execute(
enrollment_courses.insert().values(
enrollment_id=enrollment.id,
course_id=course.id,
enrollment_date=date.today(),
status='active'
)
)
else:
# Update existing enrollment with new term if provided
if term:
existing_enrollment.term = term
# Update courses in the enrollment if term is specified
if term:
# Remove existing course associations
ctx.session.execute(
enrollment_courses.delete().where(
enrollment_courses.c.enrollment_id == existing_enrollment.id
)
)
# Add filtered courses for the specific term
courses_to_add = [course for course in speciality_courses if course.semester == term]
for course in courses_to_add:
ctx.session.execute(
enrollment_courses.insert().values(
enrollment_id=existing_enrollment.id,
course_id=course.id,
enrollment_date=date.today(),
status='active'
)
)
ctx.session.commit()
# Send email notification to student
try:
from src.utils import send_email
from flask import current_app
# Use the already filtered courses for email
email_courses = speciality_courses
# If term is specified, filter the courses for email
if term:
email_courses = [course for course in speciality_courses if course.semester == term]
# Create email content
email_subject = f"Speciality Assignment - {speciality.name}"
# Create professional HTML email template
courses_html = ""
if email_courses:
term_info = f" for {term}" if term else ""
courses_html = f"""
📚 Your Course Schedule{term_info}
"""
for course in email_courses:
courses_html += f"""
{course.title}
Course Code: {course.code}
Credits: {course.credits} | Department: {course.department}
{f'
Semester: {course.semester}
' if course.semester else ''}
Level {course.course_level or 'N/A'}
"""
courses_html += """
"""
else:
courses_html = """
📋 Course information will be updated soon. Please check back later.
"""
email_message = f"""
Speciality Assignment Notification
🎓 KISNAP
Speciality Assignment Notification
✓
Congratulations!
You have been assigned to a new speciality
👤 Student Information
Student Name
{student.first_name} {student.last_name}
Student ID
{student.student_id}
Program
{student.program}
Year of Study
Year {student.year_of_study}
🎯 Your New Speciality
{speciality.name}
{speciality.description or 'A comprehensive speciality program designed to enhance your skills and knowledge in your chosen field.'}
Speciality ID: {speciality.id}
{courses_html}
🚀 Next Steps
- Log in to your KISNAP student portal to view your updated academic profile
- Review your course schedule and speciality requirements
- Check for any additional documentation or prerequisites
- Contact your academic advisor if you have any questions
Need Help?
If you have any questions about your speciality assignment, please contact our support team.
📧 support@kisiwa-tech.ac.ke
© 2024 KISNAP - Kisiwa Technical Training Institute
This is an automated notification. Please do not reply to this email.
"""
# Send email
success, message = send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['NOREPLY_PASSWORD'],
receiver_email=student.email,
subject=email_subject,
message=email_message
)
if success:
print(f"Email sent successfully to {student.email}")
else:
print(f"Failed to send email to {student.email}: {message}")
except Exception as e:
print(f"Error sending email notification: {str(e)}")
# Don't fail the main operation if email fails
return custom_response(
success=True,
data=f"Student {student.first_name} {student.last_name} has been assigned to {speciality.name}. Email notification sent.",
status_code=200
)
def update_student_info(self, supervisor_id: str, student_id: str, payload: Dict) -> Dict:
"""Update student information"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found",
status_code=404
)
# Verify student exists
student = ctx.session.query(Student).filter(Student.id == student_id).first()
if not student:
return custom_response(
success=False,
data="Student not found",
status_code=404
)
# Check if student is in supervisor's department
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
department_names = [dept.department_name for dept in supervisor_departments]
if student.department not in department_names:
return custom_response(
success=False,
data="Student is not in your department",
status_code=403
)
# Update allowed fields
updatable_fields = [
'first_name', 'last_name', 'phone', 'program', 'year_of_study',
'guardian_name', 'guardian_contact', 'emergency_contact',
'address', 'city', 'country', 'postal_code', 'profile_picture'
]
for field in updatable_fields:
if field in payload:
setattr(student, field, payload[field])
# Handle probation status
if 'is_on_probation' in payload:
student.is_on_probation = payload['is_on_probation']
if payload['is_on_probation']:
student.probation_start_date = payload.get('probation_start_date')
student.probation_end_date = payload.get('probation_end_date')
student.probation_reason = payload.get('probation_reason')
else:
student.probation_start_date = None
student.probation_end_date = None
student.probation_reason = None
ctx.session.commit()
return custom_response(
success=True,
data=f"Student {student.first_name} {student.last_name} information updated successfully",
status_code=200
)
def get_tutor_analytics(self, supervisor_id: str) -> Dict:
"""
Get comprehensive tutor analytics for the supervisor's department
Args:
supervisor_id: Supervisor ID
Returns:
Response with comprehensive tutor analytics
"""
with DatabaseContextManager() as ctx:
current_app.logger.info(f"[TUTOR ANALYTICS DEBUG] Starting get_tutor_analytics for supervisor_id: {supervisor_id}")
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
if not supervisor_departments:
return custom_response(
success=False,
data="Supervisor has no assigned departments",
status_code=404
)
department_names = [dept.department_name for dept in supervisor_departments]
current_app.logger.info(f"[TUTOR ANALYTICS DEBUG] Supervisor departments: {department_names}")
# Get tutors in supervisor's departments
tutors = ctx.session.query(Tutor).join(TutorDepartment).filter(
and_(
Tutor.is_active == True,
TutorDepartment.department_name.in_(department_names),
TutorDepartment.is_active == True
)
).all()
current_app.logger.info(f"[TUTOR ANALYTICS DEBUG] Found {len(tutors)} tutors in supervisor's departments")
# Get specialities in supervisor's departments
specialities = ctx.session.query(Speciality).filter(
and_(
Speciality.is_active == True,
Speciality.department.in_(department_names)
)
).all()
# Get courses in supervisor's departments
courses = ctx.session.query(Course).filter(
and_(
Course.is_active == True,
Course.department.in_(department_names)
)
).all()
# Get teaching sessions for tutors
tutor_ids = [tutor.id for tutor in tutors]
teaching_sessions = []
if tutor_ids:
teaching_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.tutor_id.in_(tutor_ids)
).all()
# Get attendance records for tutors
attendance_records = []
if tutor_ids:
attendance_records = ctx.session.query(Attendance).filter(
Attendance.tutor_id.in_(tutor_ids)
).all()
# Process analytics data
analytics_data = self._process_tutor_analytics_data(
tutors, specialities, courses, teaching_sessions, attendance_records, department_names
)
return custom_response(
success=True,
data=analytics_data,
status_code=200
)
def get_tutor_analytics_summary(self, supervisor_id: str) -> Dict:
"""
Get tutor analytics summary with key metrics
Args:
supervisor_id: Supervisor ID
Returns:
Response with tutor analytics summary
"""
with DatabaseContextManager() as ctx:
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
if not supervisor_departments:
return custom_response(
success=False,
data="Supervisor has no assigned departments",
status_code=404
)
department_names = [dept.department_name for dept in supervisor_departments]
# Get basic counts
total_tutors = ctx.session.query(Tutor).join(TutorDepartment).filter(
and_(
Tutor.is_active == True,
TutorDepartment.department_name.in_(department_names),
TutorDepartment.is_active == True
)
).count()
active_tutors = ctx.session.query(Tutor).join(TutorDepartment).filter(
and_(
Tutor.is_active == True,
Tutor.is_on_leave == False,
TutorDepartment.department_name.in_(department_names),
TutorDepartment.is_active == True
)
).count()
total_courses = ctx.session.query(Course).filter(
and_(
Course.is_active == True,
Course.department.in_(department_names)
)
).count()
total_specialities = ctx.session.query(Speciality).filter(
and_(
Speciality.is_active == True,
Speciality.department.in_(department_names)
)
).count()
# Calculate average performance (mock data for now)
avg_performance = 85.5 # This would be calculated from actual performance metrics
summary = {
'total_tutors': total_tutors,
'active_tutors': active_tutors,
'total_courses': total_courses,
'total_specialities': total_specialities,
'avg_performance': avg_performance,
'departments': department_names
}
return custom_response(
success=True,
data=summary,
status_code=200
)
def get_tutor_performance_analytics(self, supervisor_id: str) -> Dict:
"""
Get tutor performance analytics and trends
Args:
supervisor_id: Supervisor ID
Returns:
Response with tutor performance analytics
"""
with DatabaseContextManager() as ctx:
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
if not supervisor_departments:
return custom_response(
success=False,
data="Supervisor has no assigned departments",
status_code=404
)
department_names = [dept.department_name for dept in supervisor_departments]
# Get tutors in supervisor's departments
tutors = ctx.session.query(Tutor).join(TutorDepartment).filter(
and_(
Tutor.is_active == True,
TutorDepartment.department_name.in_(department_names),
TutorDepartment.is_active == True
)
).all()
# Mock performance data (in real implementation, this would be calculated from actual metrics)
performance_data = []
for i, tutor in enumerate(tutors):
performance_data.append({
'tutor_id': tutor.id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}",
'department': tutor.get_primary_department() or 'Unknown',
'courses': len(tutor.courses),
'teaching_hours': 25 + (i * 5), # Mock data
'students': 15 + (i * 3), # Mock data
'performance': 80 + (i * 2), # Mock data
'attendance': 95 - (i * 1), # Mock data
'status': 'active' if not tutor.is_on_leave else 'on_leave'
})
# Mock chart data for performance distribution
chart_data = [
{'name': 'Week 1', 'performance': 85, 'attendance': 92},
{'name': 'Week 2', 'performance': 87, 'attendance': 94},
{'name': 'Week 3', 'performance': 89, 'attendance': 96},
{'name': 'Week 4', 'performance': 91, 'attendance': 98},
{'name': 'Week 5', 'performance': 88, 'attendance': 95},
{'name': 'Week 6', 'performance': 90, 'attendance': 97},
{'name': 'Week 7', 'performance': 92, 'attendance': 99},
{'name': 'Week 8', 'performance': 89, 'attendance': 96}
]
analytics = {
'tutor_performance': performance_data,
'chart_data': chart_data,
'total_tutors': len(tutors),
'avg_performance': sum(t['performance'] for t in performance_data) / len(performance_data) if performance_data else 0
}
return custom_response(
success=True,
data=analytics,
status_code=200
)
def get_tutor_speciality_distribution(self, supervisor_id: str) -> Dict:
"""
Get tutor distribution by speciality
Args:
supervisor_id: Supervisor ID
Returns:
Response with tutor speciality distribution
"""
with DatabaseContextManager() as ctx:
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
if not supervisor_departments:
return custom_response(
success=False,
data="Supervisor has no assigned departments",
status_code=404
)
department_names = [dept.department_name for dept in supervisor_departments]
# Get specialities in supervisor's departments
specialities = ctx.session.query(Speciality).filter(
and_(
Speciality.is_active == True,
Speciality.department.in_(department_names)
)
).all()
# Get tutors in supervisor's departments
tutors = ctx.session.query(Tutor).join(TutorDepartment).filter(
and_(
Tutor.is_active == True,
TutorDepartment.department_name.in_(department_names),
TutorDepartment.is_active == True
)
).all()
# Count tutors by speciality (mock data for now)
speciality_distribution = []
for speciality in specialities:
# In real implementation, this would count actual tutors per speciality
tutor_count = len([t for t in tutors if t.specialization == speciality.name]) or (len(tutors) // len(specialities) + 1)
speciality_distribution.append({
'name': speciality.name,
'tutors': tutor_count,
'department': speciality.department
})
# Mock bar chart data
bar_chart_data = [
{'name': 'Computer Science', 'tutors': 12},
{'name': 'Mathematics', 'tutors': 8},
{'name': 'Physics', 'tutors': 6},
{'name': 'Chemistry', 'tutors': 5},
{'name': 'Biology', 'tutors': 4}
]
distribution = {
'speciality_distribution': speciality_distribution,
'bar_chart_data': bar_chart_data,
'total_tutors': len(tutors),
'total_specialities': len(specialities)
}
return custom_response(
success=True,
data=distribution,
status_code=200
)
def get_tutor_teaching_hours_analytics(self, supervisor_id: str) -> Dict:
"""
Get tutor teaching hours analytics and trends
Args:
supervisor_id: Supervisor ID
Returns:
Response with tutor teaching hours analytics
"""
with DatabaseContextManager() as ctx:
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
if not supervisor_departments:
return custom_response(
success=False,
data="Supervisor has no assigned departments",
status_code=404
)
department_names = [dept.department_name for dept in supervisor_departments]
# Get tutors in supervisor's departments
tutors = ctx.session.query(Tutor).join(TutorDepartment).filter(
and_(
Tutor.is_active == True,
TutorDepartment.department_name.in_(department_names),
TutorDepartment.is_active == True
)
).all()
# Mock teaching hours data
teaching_hours_data = []
for i, tutor in enumerate(tutors):
teaching_hours_data.append({
'tutor_id': tutor.id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}",
'teaching_hours': 20 + (i * 3), # Mock data
'max_hours': tutor.max_teaching_hours or 25,
'utilization': round((20 + (i * 3)) / (tutor.max_teaching_hours or 25) * 100, 1)
})
# Mock trend data
trend_data = [
{'name': 'Jan', 'teachingHours': 450},
{'name': 'Feb', 'teachingHours': 520},
{'name': 'Mar', 'teachingHours': 480},
{'name': 'Apr', 'teachingHours': 600},
{'name': 'May', 'teachingHours': 550},
{'name': 'Jun', 'teachingHours': 580}
]
analytics = {
'teaching_hours_data': teaching_hours_data,
'trend_data': trend_data,
'total_hours': sum(t['teaching_hours'] for t in teaching_hours_data),
'avg_hours': sum(t['teaching_hours'] for t in teaching_hours_data) / len(teaching_hours_data) if teaching_hours_data else 0
}
return custom_response(
success=True,
data=analytics,
status_code=200
)
def get_top_performing_tutors(self, supervisor_id: str) -> Dict:
"""
Get top performing tutors in the department
Args:
supervisor_id: Supervisor ID
Returns:
Response with top performing tutors
"""
with DatabaseContextManager() as ctx:
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
if not supervisor_departments:
return custom_response(
success=False,
data="Supervisor has no assigned departments",
status_code=404
)
department_names = [dept.department_name for dept in supervisor_departments]
# Get tutors in supervisor's departments
tutors = ctx.session.query(Tutor).join(TutorDepartment).filter(
and_(
Tutor.is_active == True,
TutorDepartment.department_name.in_(department_names),
TutorDepartment.is_active == True
)
).all()
# Mock top performers data
top_performers = []
for i, tutor in enumerate(tutors[:5]): # Top 5
top_performers.append({
'rank': i + 1,
'tutor_id': tutor.id,
'tutor_name': f"{tutor.first_name} {tutor.last_name}",
'department': tutor.get_primary_department() or 'Unknown',
'courses': len(tutor.courses),
'students': 20 + (i * 5), # Mock data
'performance': 95 - (i * 2), # Mock data
'teaching_hours': 25 + (i * 2) # Mock data
})
# Get real top performing courses/units from database
top_units = []
# Query courses in supervisor's departments with enrollment and performance data
courses_query = ctx.session.query(
Course.id,
Course.title,
Course.code,
func.count(enrollment_courses.c.course_id).label('enrollment_count'),
func.avg(CourseProgress.completion_percentage).label('avg_completion'),
func.avg(CourseProgress.attendance_rate).label('avg_attendance')
).outerjoin(
enrollment_courses, Course.id == enrollment_courses.c.course_id
).outerjoin(
Enrollment, enrollment_courses.c.enrollment_id == Enrollment.id
).outerjoin(
CourseProgress, Enrollment.id == CourseProgress.enrollment_id
).filter(
and_(
Course.is_active == True,
Course.department.in_(department_names)
)
).group_by(
Course.id, Course.title, Course.code
).order_by(
func.avg(CourseProgress.completion_percentage).desc().nullslast(),
func.count(enrollment_courses.c.course_id).desc()
).limit(5).all()
for i, course in enumerate(courses_query):
# Calculate performance score based on completion and attendance
completion_rate = float(course.avg_completion) if course.avg_completion else 0
attendance_rate = float(course.avg_attendance) if course.avg_attendance else 0
performance_score = (completion_rate + attendance_rate) / 2
# Get course count (how many instances of this course)
course_count = ctx.session.query(Course).filter(
and_(
Course.title == course.title,
Course.department.in_(department_names),
Course.is_active == True
)
).count()
top_units.append({
'name': course.title,
'courses': course_count,
'students': int(course.enrollment_count) if course.enrollment_count else 0,
'performance': round(performance_score, 1) if performance_score > 0 else 75.0 # Default performance
})
# If no courses found, get basic course information
if not top_units:
basic_courses = ctx.session.query(Course).filter(
and_(
Course.is_active == True,
Course.department.in_(department_names)
)
).limit(5).all()
for i, course in enumerate(basic_courses):
# Get enrollment count for this course
enrollment_count = ctx.session.query(enrollment_courses).filter(
enrollment_courses.c.course_id == course.id
).count()
top_units.append({
'name': course.title,
'courses': 1,
'students': enrollment_count,
'performance': 85.0 - (i * 2) # Default performance decreasing
})
analytics = {
'top_performers': top_performers,
'top_units': top_units,
'total_tutors': len(tutors)
}
return custom_response(
success=True,
data=analytics,
status_code=200
)
def _process_tutor_analytics_data(self, tutors, specialities, courses, teaching_sessions, attendance_records, department_names):
"""
Process and format tutor analytics data
Args:
tutors: List of tutors
specialities: List of specialities
courses: List of courses
teaching_sessions: List of teaching sessions
attendance_records: List of attendance records
department_names: List of department names
Returns:
Formatted analytics data
"""
# Process tutors data
tutors_data = []
for tutor in tutors:
tutor_data = {
'id': tutor.id,
'name': f"{tutor.first_name} {tutor.last_name}",
'email': tutor.email,
'department': tutor.get_primary_department() or 'Unknown',
'specialization': tutor.specialization or 'General',
'qualification': tutor.qualification or 'Not specified',
'years_of_teaching': tutor.years_of_teaching or 0,
'hourly_rate': tutor.hourly_rate or 0,
'max_teaching_hours': tutor.max_teaching_hours or 20,
'is_full_time': tutor.is_full_time,
'is_on_leave': tutor.is_on_leave,
'courses_count': len(tutor.courses),
'courses': [{'id': c.id, 'title': c.title, 'code': c.code} for c in tutor.courses]
}
tutors_data.append(tutor_data)
# Process specialities data
specialities_data = []
for speciality in specialities:
speciality_data = {
'id': speciality.id,
'name': speciality.name,
'description': speciality.description or '',
'department': speciality.department,
'code': speciality.code or '',
'is_active': speciality.is_active
}
specialities_data.append(speciality_data)
# Process courses data
courses_data = []
for course in courses:
course_data = {
'id': course.id,
'title': course.title,
'code': course.code,
'description': course.description or '',
'department': course.department,
'credits': course.credits or 0,
'is_active': course.is_active,
'supervisor_id': course.supervisor_id
}
courses_data.append(course_data)
# Calculate statistics
total_tutors = len(tutors_data)
active_tutors = len([t for t in tutors_data if not t['is_on_leave']])
total_courses = len(courses_data)
total_specialities = len(specialities_data)
total_teaching_sessions = len(teaching_sessions)
total_attendance_records = len(attendance_records)
# Calculate average metrics
avg_courses_per_tutor = total_courses / total_tutors if total_tutors > 0 else 0
avg_teaching_hours = sum(t['max_teaching_hours'] for t in tutors_data) / total_tutors if total_tutors > 0 else 0
statistics = {
'total_tutors': total_tutors,
'active_tutors': active_tutors,
'total_courses': total_courses,
'total_specialities': total_specialities,
'total_teaching_sessions': total_teaching_sessions,
'total_attendance_records': total_attendance_records,
'avg_courses_per_tutor': round(avg_courses_per_tutor, 2),
'avg_teaching_hours': round(avg_teaching_hours, 2),
'departments': department_names
}
return {
'tutors': tutors_data,
'specialities': specialities_data,
'courses': courses_data,
'statistics': statistics,
'teaching_sessions_count': total_teaching_sessions,
'attendance_records_count': total_attendance_records
}
def get_supervisor_workload_data(self) -> Dict[str, Any]:
"""Get comprehensive workload data for all supervisors"""
with DatabaseContextManager() as ctx:
try:
# Get all supervisors with their related data
supervisors = ctx.session.query(Supervisor).options(
joinedload(Supervisor.departments),
joinedload(Supervisor.managed_courses),
joinedload(Supervisor.managed_tutors)
).all()
supervisor_workload_data = []
for supervisor in supervisors:
# Calculate workload statistics
workload_stats = self._calculate_supervisor_workload(ctx, supervisor)
# Get courses data
courses_data = self._get_supervisor_courses_data(ctx, supervisor)
# Get recent activities
recent_activities = self._get_supervisor_recent_activities(ctx, supervisor)
# Get departments data
departments_data = self._get_supervisor_departments_data(ctx, supervisor)
supervisor_data = {
'id': supervisor.id,
'first_name': supervisor.first_name,
'last_name': supervisor.last_name,
'email': supervisor.email,
'phone': supervisor.phone,
'office_location': supervisor.office_location,
'departments': departments_data,
'specialization': supervisor.specialization,
'qualification': supervisor.qualification,
'years_of_experience': supervisor.years_of_experience,
'is_head_of_department': supervisor.is_head_of_department,
'profile_picture': supervisor.profile_picture,
'employee_id': supervisor.staff_id,
'workload_stats': workload_stats,
'courses': courses_data,
'recent_activities': recent_activities
}
supervisor_workload_data.append(supervisor_data)
return custom_response(
success=True,
data={'supervisors': supervisor_workload_data},
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error fetching supervisor workload data: {str(e)}")
return custom_response(
success=False,
data=f"Error fetching supervisor workload data: {str(e)}",
status_code=500
)
def _calculate_supervisor_workload(self, ctx, supervisor: Supervisor) -> Dict[str, Any]:
"""Calculate workload statistics for a supervisor"""
try:
# Get teaching sessions for the supervisor (using supervisor_tutor_id)
teaching_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.supervisor_tutor_id == supervisor.id
).all()
# Calculate total teaching hours (using duration calculation from start/end time)
total_teaching_hours = 0
for session in teaching_sessions:
if session.start_time and session.end_time:
# Calculate duration in hours
start_dt = datetime.combine(date.today(), session.start_time)
end_dt = datetime.combine(date.today(), session.end_time)
if end_dt < start_dt: # Handle overnight sessions
end_dt += timedelta(days=1)
duration = (end_dt - start_dt).total_seconds() / 3600
total_teaching_hours += duration
# Get courses count
courses_count = len(supervisor.managed_courses)
# Get students count from managed courses using enrollment_courses table
students_count = 0
for course in supervisor.managed_courses:
# Get enrollment count for this course through enrollment_courses table
enrollment_count = ctx.session.query(func.count(enrollment_courses.c.enrollment_id)).filter(
enrollment_courses.c.course_id == course.id,
enrollment_courses.c.status == 'active'
).scalar() or 0
students_count += enrollment_count
# Calculate utilization percentage (ideal hours = 20)
ideal_hours = 20
actual_hours = total_teaching_hours
utilization_percentage = (actual_hours / ideal_hours * 100) if ideal_hours > 0 else 0
# Determine workload status
if actual_hours > ideal_hours * 1.25: # > 25 hours
status = 'overloaded'
elif actual_hours < ideal_hours * 0.75: # < 15 hours
status = 'underutilized'
else:
status = 'optimal'
# Get sessions count
sessions_count = len(teaching_sessions)
# Calculate peak hours (simplified - could be enhanced)
peak_hours = ['9:00 AM', '2:00 PM'] # Default peak hours
# Get preferred schedule (simplified)
preferred_schedule = ['Monday', 'Wednesday', 'Friday'] # Default schedule
return {
'current_load': actual_hours,
'max_load': supervisor.max_teaching_hours or 25,
'utilization_percentage': round(utilization_percentage, 2),
'ideal_hours': ideal_hours,
'actual_hours': actual_hours,
'courses_count': courses_count,
'sessions_count': sessions_count,
'students_count': students_count,
'peak_hours': peak_hours,
'preferred_schedule': preferred_schedule,
'status': status
}
except Exception as e:
current_app.logger.error(f"Error calculating workload for supervisor {supervisor.id}: {str(e)}")
return {
'current_load': 0,
'max_load': 25,
'utilization_percentage': 0,
'ideal_hours': 20,
'actual_hours': 0,
'courses_count': 0,
'sessions_count': 0,
'students_count': 0,
'peak_hours': [],
'preferred_schedule': [],
'status': 'unknown'
}
def _get_supervisor_courses_data(self, ctx, supervisor: Supervisor) -> List[Dict[str, Any]]:
"""Get courses data for a supervisor"""
courses_data = []
for course in supervisor.managed_courses:
# Get student count for this course using enrollment_courses table
student_count = ctx.session.query(func.count(enrollment_courses.c.enrollment_id)).filter(
enrollment_courses.c.course_id == course.id,
enrollment_courses.c.status == 'active'
).scalar() or 0
course_data = {
'id': course.id,
'title': course.title,
'code': course.code,
'department': course.department,
'semester': course.semester,
'academic_year': course.academic_session.year if course.academic_session else None,
'student_count': student_count,
'credits': course.credits,
'level': course.course_level,
'hours_per_week': course.total_hours or 0,
'status': 'Active' # Default status
}
courses_data.append(course_data)
return courses_data
def _get_supervisor_recent_activities(self, ctx, supervisor: Supervisor) -> List[Dict[str, Any]]:
"""Get recent activities for a supervisor"""
recent_activities = []
try:
# Get recent teaching sessions (using supervisor_tutor_id)
recent_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.supervisor_tutor_id == supervisor.id
).order_by(TeachingSession.created_at.desc()).limit(5).all()
for session in recent_sessions:
# Calculate session duration
duration_hours = 0
if session.start_time and session.end_time:
start_dt = datetime.combine(date.today(), session.start_time)
end_dt = datetime.combine(date.today(), session.end_time)
if end_dt < start_dt:
end_dt += timedelta(days=1)
duration_hours = (end_dt - start_dt).total_seconds() / 3600
activity = {
'id': str(session.id),
'type': 'Teaching',
'description': f"Conducted {session.title or 'teaching session'}",
'timestamp': session.created_at.isoformat() if session.created_at else datetime.utcnow().isoformat(),
'hours_logged': duration_hours
}
recent_activities.append(activity)
# Get recent timetable approvals
recent_approvals = ctx.session.query(Timetable).filter(
Timetable.approved_by == supervisor.id
).order_by(Timetable.updated_at.desc()).limit(3).all()
for timetable in recent_approvals:
activity = {
'id': str(timetable.id),
'type': 'Approval',
'description': f"Approved timetable: {timetable.name or 'Untitled'}",
'timestamp': timetable.updated_at.isoformat() if timetable.updated_at else datetime.utcnow().isoformat(),
'hours_logged': 0
}
recent_activities.append(activity)
# Sort by timestamp and limit to 10
recent_activities.sort(key=lambda x: x['timestamp'], reverse=True)
return recent_activities[:10]
except Exception as e:
current_app.logger.error(f"Error getting recent activities for supervisor {supervisor.id}: {str(e)}")
return []
def _get_supervisor_departments_data(self, ctx, supervisor: Supervisor) -> List[Dict[str, Any]]:
"""Get departments data for a supervisor"""
departments_data = []
try:
# Get supervisor departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor.id,
SupervisorDepartment.is_active == True
).all()
for dept in supervisor_departments:
dept_data = {
'department_name': dept.department_name,
'is_primary': dept.is_primary,
'role': 'Head of Department' if supervisor.is_head_of_department else 'Supervisor'
}
departments_data.append(dept_data)
# If no departments found, create a default one
if not departments_data:
departments_data.append({
'department_name': 'General',
'is_primary': True,
'role': 'Supervisor'
})
return departments_data
except Exception as e:
current_app.logger.error(f"Error getting departments for supervisor {supervisor.id}: {str(e)}")
return [{
'department_name': 'General',
'is_primary': True,
'role': 'Supervisor'
}]
def _calculate_session_duration(self, start_time, end_time):
"""Calculate session duration in seconds, handling both datetime and time objects"""
if not start_time or not end_time:
return 0
# Handle both datetime and time objects
if hasattr(start_time, 'time'):
start_time = start_time.time()
if hasattr(end_time, 'time'):
end_time = end_time.time()
# Calculate duration in seconds
start_minutes = start_time.hour * 60 + start_time.minute
end_minutes = end_time.hour * 60 + end_time.minute
duration_minutes = end_minutes - start_minutes
return duration_minutes * 60 # Convert to seconds
def get_supervisor_analytics_data(self):
"""Get comprehensive analytics data for all supervisors"""
try:
with DatabaseContextManager() as ctx:
# Get all supervisors with their relationships
supervisors = ctx.session.query(Supervisor).options(
joinedload(Supervisor.managed_courses),
joinedload(Supervisor.departments),
joinedload(Supervisor.managed_tutors)
).all()
supervisors_data = []
for supervisor in supervisors:
# Calculate analytics metrics
analytics = self._calculate_supervisor_analytics(ctx, supervisor)
# Get trends data
trends = self._calculate_supervisor_trends(ctx, supervisor)
# Get recent activities
recent_activities = self._get_supervisor_recent_activities(ctx, supervisor)
# Get courses data
courses_data = self._get_supervisor_courses_data(ctx, supervisor)
# Get departments data
departments_data = self._get_supervisor_departments_data(ctx, supervisor)
supervisor_data = {
'id': supervisor.id,
'first_name': supervisor.first_name,
'last_name': supervisor.last_name,
'email': supervisor.email,
'phone': supervisor.phone,
'office_location': supervisor.office_location,
'departments': departments_data,
'specialization': supervisor.specialization,
'qualification': supervisor.qualification,
'years_of_experience': supervisor.years_of_experience,
'is_head_of_department': supervisor.is_head_of_department,
'profile_picture': supervisor.profile_picture,
'employee_id': supervisor.staff_id,
'analytics': analytics,
'trends': trends,
'recent_activities': recent_activities,
'courses': courses_data
}
supervisors_data.append(supervisor_data)
return {
'success': True,
'data': {
'supervisors': supervisors_data
}
}
except Exception as e:
current_app.logger.error(f"Error fetching supervisor analytics data: {str(e)}")
return {
'success': False,
'message': f'Error fetching supervisor analytics data: {str(e)}',
'data': None
}
def get_supervisor_detail_analytics(self, supervisor_id):
"""Get detailed analytics data for a specific supervisor"""
try:
with DatabaseContextManager() as ctx:
# Get the supervisor with relationships
supervisor = ctx.session.query(Supervisor).options(
joinedload(Supervisor.managed_courses),
joinedload(Supervisor.departments),
joinedload(Supervisor.managed_tutors)
).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
return {
'success': False,
'message': 'Supervisor not found',
'data': None
}
# Calculate department analytics
department_analytics = self._calculate_department_analytics(ctx, supervisor)
# Get teaching sessions data
teaching_sessions = self._get_teaching_sessions_data(ctx, supervisor)
# Get attendance analytics
attendance_analytics = self._get_attendance_analytics(ctx, supervisor)
# Get student analytics
student_analytics = self._get_student_analytics(ctx, supervisor)
# Get performance trends
performance_trends = self._get_performance_trends(ctx, supervisor)
supervisor_data = {
'id': supervisor.id,
'first_name': supervisor.first_name,
'last_name': supervisor.last_name,
'email': supervisor.email,
'phone': supervisor.phone,
'office_location': supervisor.office_location,
'departments': self._get_supervisor_departments_data(ctx, supervisor),
'specialization': supervisor.specialization,
'qualification': supervisor.qualification,
'years_of_experience': supervisor.years_of_experience,
'is_head_of_department': supervisor.is_head_of_department,
'profile_picture': supervisor.profile_picture,
'staff_id': supervisor.staff_id,
'department_analytics': department_analytics,
'teaching_sessions': teaching_sessions,
'attendance_analytics': attendance_analytics,
'student_analytics': student_analytics,
'performance_trends': performance_trends
}
return {
'success': True,
'data': supervisor_data
}
except Exception as e:
current_app.logger.error(f"Error fetching supervisor detail analytics: {str(e)}")
return {
'success': False,
'message': f'Error fetching supervisor detail analytics: {str(e)}',
'data': None
}
def _calculate_department_analytics(self, ctx, supervisor: Supervisor) -> Dict[str, Any]:
"""Calculate department analytics for a supervisor"""
try:
# Get managed tutors
managed_tutors = ctx.session.query(Tutor).filter(Tutor.supervisor_id == supervisor.id).all()
# Get total students in managed courses
total_students = 0
total_courses = len(supervisor.managed_courses)
total_sessions = 0
total_hours = 0
for course in supervisor.managed_courses:
# Get student count for this course
student_count = ctx.session.query(enrollment_courses).filter(
enrollment_courses.c.course_id == course.id
).count()
total_students += student_count
# Get sessions for this course
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id == course.id
).all()
total_sessions += len(sessions)
for session in sessions:
if session.start_time and session.end_time:
# Handle both datetime and time objects
if hasattr(session.start_time, 'time'):
start_time = session.start_time.time()
else:
start_time = session.start_time
if hasattr(session.end_time, 'time'):
end_time = session.end_time.time()
else:
end_time = session.end_time
# Calculate duration in hours
start_minutes = start_time.hour * 60 + start_time.minute
end_minutes = end_time.hour * 60 + end_time.minute
duration_hours = (end_minutes - start_minutes) / 60
total_hours += duration_hours
# Calculate average attendance
attendance_records = ctx.session.query(Attendance).join(TeachingSession).filter(
TeachingSession.course_id.in_([c.id for c in supervisor.managed_courses])
).all()
total_attendance = len(attendance_records)
total_possible_attendance = total_students * total_sessions if total_sessions > 0 else 0
average_attendance = (total_attendance / total_possible_attendance * 100) if total_possible_attendance > 0 else 0
# Calculate performance score
performance_score = min(100, (average_attendance * 0.4 + (total_hours / max(1, total_sessions)) * 0.3 + (total_courses * 5) * 0.3))
# Count active vs inactive tutors
active_trainers = len([t for t in managed_tutors if not t.is_on_leave])
inactive_trainers = len([t for t in managed_tutors if t.is_on_leave])
# Calculate completion rate (simplified)
completion_rate = min(100, average_attendance * 1.1)
return {
'total_trainers': len(managed_tutors),
'total_students': total_students,
'total_courses': total_courses,
'total_sessions': total_sessions,
'average_attendance': average_attendance,
'department_performance_score': performance_score,
'active_trainers': active_trainers,
'inactive_trainers': inactive_trainers,
'total_hours_taught': total_hours,
'completion_rate': completion_rate
}
except Exception as e:
current_app.logger.error(f"Error calculating department analytics: {str(e)}")
return {
'total_trainers': 0,
'total_students': 0,
'total_courses': 0,
'total_sessions': 0,
'average_attendance': 0,
'department_performance_score': 0,
'active_trainers': 0,
'inactive_trainers': 0,
'total_hours_taught': 0,
'completion_rate': 0
}
def _get_teaching_sessions_data(self, ctx, supervisor: Supervisor) -> List[Dict[str, Any]]:
"""Get teaching sessions data for a supervisor's department"""
try:
sessions_data = []
# Get sessions for managed courses
sessions = ctx.session.query(TeachingSession).join(Course).filter(
Course.supervisor_id == supervisor.id
).order_by(TeachingSession.start_time.desc()).limit(50).all()
for session in sessions:
# Get attendance count
attendance_count = ctx.session.query(Attendance).filter(
Attendance.session_id == session.id
).count()
# Get total enrolled students
total_enrolled = ctx.session.query(enrollment_courses).filter(
enrollment_courses.c.course_id == session.course_id
).count()
attendance_percentage = (attendance_count / total_enrolled * 100) if total_enrolled > 0 else 0
# Get trainer name
trainer_name = "Unknown"
if session.tutor_id:
tutor = ctx.session.query(Tutor).filter(Tutor.id == session.tutor_id).first()
if tutor:
trainer_name = f"{tutor.first_name} {tutor.last_name}"
elif session.supervisor_tutor_id:
supervisor_tutor = ctx.session.query(Supervisor).filter(Supervisor.id == session.supervisor_tutor_id).first()
if supervisor_tutor:
trainer_name = f"{supervisor_tutor.first_name} {supervisor_tutor.last_name}"
# Calculate duration
duration_hours = 0
if session.start_time and session.end_time:
# Handle both datetime and time objects
if hasattr(session.start_time, 'time'):
start_time = session.start_time.time()
else:
start_time = session.start_time
if hasattr(session.end_time, 'time'):
end_time = session.end_time.time()
else:
end_time = session.end_time
# Calculate duration in hours
start_minutes = start_time.hour * 60 + start_time.minute
end_minutes = end_time.hour * 60 + end_time.minute
duration_hours = (end_minutes - start_minutes) / 60
session_data = {
'id': str(session.id),
'date': session.start_time.date().isoformat() if session.start_time else '',
'course_name': session.course.title if session.course else 'Unknown Course',
'trainer_name': trainer_name,
'start_time': session.start_time.time().isoformat() if session.start_time else '',
'end_time': session.end_time.time().isoformat() if session.end_time else '',
'duration_hours': duration_hours,
'attendance_count': attendance_count,
'total_enrolled': total_enrolled,
'attendance_percentage': attendance_percentage,
'room': session.room or 'TBD',
'status': 'completed' if session.end_time else 'ongoing'
}
sessions_data.append(session_data)
return sessions_data
except Exception as e:
current_app.logger.error(f"Error getting teaching sessions data: {str(e)}")
return []
def _get_attendance_analytics(self, ctx, supervisor: Supervisor) -> Dict[str, Any]:
"""Get attendance analytics for a supervisor's department"""
try:
# Get daily attendance data (last 30 days)
daily_attendance = []
for i in range(30):
date = datetime.now().date() - timedelta(days=i)
# Get sessions for this date
sessions = ctx.session.query(TeachingSession).join(Course).filter(
Course.supervisor_id == supervisor.id,
func.date(TeachingSession.start_time) == date
).all()
total_sessions = len(sessions)
total_attendance = 0
total_possible = 0
for session in sessions:
attendance_count = ctx.session.query(Attendance).filter(
Attendance.session_id == session.id
).count()
total_attendance += attendance_count
enrolled_count = ctx.session.query(enrollment_courses).filter(
enrollment_courses.c.course_id == session.course_id
).count()
total_possible += enrolled_count
average_attendance = total_attendance / total_sessions if total_sessions > 0 else 0
attendance_rate = (total_attendance / total_possible * 100) if total_possible > 0 else 0
daily_attendance.append({
'date': date.isoformat(),
'total_sessions': total_sessions,
'total_attendance': total_attendance,
'average_attendance': average_attendance,
'attendance_rate': attendance_rate
})
# Get course attendance data
course_attendance = []
for course in supervisor.managed_courses:
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id == course.id
).all()
total_sessions = len(sessions)
total_attendance = 0
total_possible = 0
for session in sessions:
attendance_count = ctx.session.query(Attendance).filter(
Attendance.session_id == session.id
).count()
total_attendance += attendance_count
enrolled_count = ctx.session.query(enrollment_courses).filter(
enrollment_courses.c.course_id == session.course_id
).count()
total_possible += enrolled_count
average_attendance = total_attendance / total_sessions if total_sessions > 0 else 0
attendance_rate = (total_attendance / total_possible * 100) if total_possible > 0 else 0
course_attendance.append({
'course_name': course.title,
'total_sessions': total_sessions,
'average_attendance': average_attendance,
'attendance_rate': attendance_rate
})
# Get trainer performance data
trainer_performance = []
managed_tutors = ctx.session.query(Tutor).filter(Tutor.supervisor_id == supervisor.id).all()
for tutor in managed_tutors:
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.tutor_id == tutor.id
).all()
total_sessions = len(sessions)
total_attendance = 0
total_possible = 0
for session in sessions:
attendance_count = ctx.session.query(Attendance).filter(
Attendance.session_id == session.id
).count()
total_attendance += attendance_count
enrolled_count = ctx.session.query(enrollment_courses).filter(
enrollment_courses.c.course_id == session.course_id
).count()
total_possible += enrolled_count
average_attendance = total_attendance / total_sessions if total_sessions > 0 else 0
attendance_rate = (total_attendance / total_possible * 100) if total_possible > 0 else 0
trainer_performance.append({
'trainer_name': f"{tutor.first_name} {tutor.last_name}",
'total_sessions': total_sessions,
'average_attendance': average_attendance,
'attendance_rate': attendance_rate
})
return {
'daily_attendance': daily_attendance,
'course_attendance': course_attendance,
'trainer_performance': trainer_performance
}
except Exception as e:
current_app.logger.error(f"Error getting attendance analytics: {str(e)}")
return {
'daily_attendance': [],
'course_attendance': [],
'trainer_performance': []
}
def _get_student_analytics(self, ctx, supervisor: Supervisor) -> Dict[str, Any]:
"""Get student analytics for a supervisor's department"""
try:
# Get all students in managed courses
student_ids = set()
for course in supervisor.managed_courses:
enrollments = ctx.session.query(enrollment_courses).filter(
enrollment_courses.c.course_id == course.id
).all()
for enrollment in enrollments:
student_ids.add(enrollment.enrollment_id)
# Get student details
students = ctx.session.query(Student).filter(Student.id.in_(student_ids)).all()
total_students = len(students)
active_students = len([s for s in students if not s.is_graduated])
graduated_students = len([s for s in students if s.is_graduated])
# Calculate dropout rate (simplified)
dropout_rate = max(0, (total_students - active_students - graduated_students) / total_students * 100) if total_students > 0 else 0
# Calculate average performance (simplified)
average_performance = 85.0 # Placeholder
# Calculate student satisfaction (simplified)
student_satisfaction = 90.0 # Placeholder
# Calculate course completion rate
course_completion_rate = min(100, (active_students / total_students * 100)) if total_students > 0 else 0
return {
'total_students': total_students,
'active_students': active_students,
'graduated_students': graduated_students,
'dropout_rate': dropout_rate,
'average_performance': average_performance,
'student_satisfaction': student_satisfaction,
'course_completion_rate': course_completion_rate
}
except Exception as e:
current_app.logger.error(f"Error getting student analytics: {str(e)}")
return {
'total_students': 0,
'active_students': 0,
'graduated_students': 0,
'dropout_rate': 0,
'average_performance': 0,
'student_satisfaction': 0,
'course_completion_rate': 0
}
def _get_performance_trends(self, ctx, supervisor: Supervisor) -> Dict[str, Any]:
"""Get performance trends for a supervisor's department"""
try:
# Weekly performance (last 12 weeks)
weekly_performance = []
for i in range(12):
week_start = datetime.now().date() - timedelta(weeks=i+1)
week_end = week_start + timedelta(days=6)
sessions = ctx.session.query(TeachingSession).join(Course).filter(
Course.supervisor_id == supervisor.id,
func.date(TeachingSession.start_time) >= week_start,
func.date(TeachingSession.start_time) <= week_end
).all()
total_sessions = len(sessions)
total_attendance = 0
for session in sessions:
attendance_count = ctx.session.query(Attendance).filter(
Attendance.session_id == session.id
).count()
total_attendance += attendance_count
attendance_rate = (total_attendance / (total_sessions * 25)) * 100 if total_sessions > 0 else 0 # Assuming 25 students per session
performance_score = min(100, attendance_rate * 1.1)
weekly_performance.append({
'week': f"Week {12-i}",
'sessions': total_sessions,
'attendance': attendance_rate,
'performance_score': performance_score
})
# Monthly trends (last 6 months)
monthly_trends = []
for i in range(6):
month_start = datetime.now().date().replace(day=1) - timedelta(days=30*i)
month_end = month_start + timedelta(days=30)
sessions = ctx.session.query(TeachingSession).join(Course).filter(
Course.supervisor_id == supervisor.id,
func.date(TeachingSession.start_time) >= month_start,
func.date(TeachingSession.start_time) <= month_end
).all()
total_hours = 0
for session in sessions:
if session.start_time and session.end_time:
# Handle both datetime and time objects
if hasattr(session.start_time, 'time'):
start_time = session.start_time.time()
else:
start_time = session.start_time
if hasattr(session.end_time, 'time'):
end_time = session.end_time.time()
else:
end_time = session.end_time
# Calculate duration in hours
start_minutes = start_time.hour * 60 + start_time.minute
end_minutes = end_time.hour * 60 + end_time.minute
duration_hours = (end_minutes - start_minutes) / 60
total_hours += duration_hours
completion_rate = 90.0 # Placeholder
satisfaction_score = 88.0 # Placeholder
monthly_trends.append({
'month': month_start.strftime('%b'),
'total_hours': total_hours,
'completion_rate': completion_rate,
'satisfaction_score': satisfaction_score
})
return {
'weekly_performance': weekly_performance,
'monthly_trends': monthly_trends
}
except Exception as e:
current_app.logger.error(f"Error getting performance trends: {str(e)}")
return {
'weekly_performance': [],
'monthly_trends': []
}
def _calculate_supervisor_analytics(self, ctx, supervisor: Supervisor) -> Dict[str, Any]:
"""Calculate comprehensive analytics for a supervisor"""
try:
# Get teaching sessions for the supervisor
teaching_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.supervisor_tutor_id == supervisor.id
).all()
# Calculate total hours
total_hours = 0
completed_sessions = 0
cancelled_sessions = 0
for session in teaching_sessions:
if session.start_time and session.end_time:
# Calculate duration in hours
start_dt = datetime.combine(date.today(), session.start_time)
end_dt = datetime.combine(date.today(), session.end_time)
if end_dt < start_dt: # Handle overnight sessions
end_dt += timedelta(days=1)
duration = (end_dt - start_dt).total_seconds() / 3600
total_hours += duration
if session.status == 'completed':
completed_sessions += 1
elif session.status == 'cancelled':
cancelled_sessions += 1
# Get courses count
total_courses = len(supervisor.managed_courses)
# Get students count from managed courses
total_students = 0
for course in supervisor.managed_courses:
enrollment_count = ctx.session.query(func.count(enrollment_courses.c.enrollment_id)).filter(
enrollment_courses.c.course_id == course.id,
enrollment_courses.c.status == 'active'
).scalar() or 0
total_students += enrollment_count
# Calculate performance metrics (simplified calculations)
performance_score = min(100, max(0,
(completed_sessions * 10) +
(total_courses * 5) +
(total_students * 2) +
(supervisor.years_of_experience * 2)
))
# Calculate other metrics
average_rating = min(5.0, max(1.0, performance_score / 20)) # Convert to 1-5 scale
completion_rate = min(100, (completed_sessions / max(1, len(teaching_sessions))) * 100)
approval_rate = min(100, performance_score * 0.9) # Simplified
workload_utilization = min(100, (total_hours / 20) * 100) # Assuming 20h ideal
engagement_score = min(100, performance_score * 0.85)
student_satisfaction = min(100, average_rating * 20)
course_quality_score = min(100, performance_score * 0.95)
timeliness_score = min(100, performance_score * 0.88)
communication_score = min(100, performance_score * 0.92)
# Calculate response time (simplified)
response_time_hours = max(0.5, 24 - (performance_score / 10))
return {
'total_courses': total_courses,
'total_students': total_students,
'total_hours': round(total_hours, 2),
'average_rating': round(average_rating, 1),
'completion_rate': round(completion_rate, 1),
'approval_rate': round(approval_rate, 1),
'response_time_hours': round(response_time_hours, 1),
'workload_utilization': round(workload_utilization, 1),
'performance_score': round(performance_score, 1),
'engagement_score': round(engagement_score, 1),
'recent_activities_count': len(teaching_sessions),
'upcoming_sessions': len([s for s in teaching_sessions if s.status == 'scheduled']),
'completed_sessions': completed_sessions,
'cancelled_sessions': cancelled_sessions,
'student_satisfaction': round(student_satisfaction, 1),
'course_quality_score': round(course_quality_score, 1),
'timeliness_score': round(timeliness_score, 1),
'communication_score': round(communication_score, 1)
}
except Exception as e:
current_app.logger.error(f"Error calculating analytics for supervisor {supervisor.id}: {str(e)}")
return {
'total_courses': 0,
'total_students': 0,
'total_hours': 0,
'average_rating': 0,
'completion_rate': 0,
'approval_rate': 0,
'response_time_hours': 24,
'workload_utilization': 0,
'performance_score': 0,
'engagement_score': 0,
'recent_activities_count': 0,
'upcoming_sessions': 0,
'completed_sessions': 0,
'cancelled_sessions': 0,
'student_satisfaction': 0,
'course_quality_score': 0,
'timeliness_score': 0,
'communication_score': 0
}
def _calculate_supervisor_trends(self, ctx, supervisor: Supervisor) -> Dict[str, str]:
"""Calculate trends for a supervisor"""
try:
# Get recent teaching sessions (last 30 days)
thirty_days_ago = datetime.now() - timedelta(days=30)
recent_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.supervisor_tutor_id == supervisor.id,
TeachingSession.created_at >= thirty_days_ago
).all()
# Get older sessions (30-60 days ago)
sixty_days_ago = datetime.now() - timedelta(days=60)
older_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.supervisor_tutor_id == supervisor.id,
TeachingSession.created_at >= sixty_days_ago,
TeachingSession.created_at < thirty_days_ago
).all()
# Calculate trends (simplified logic)
recent_count = len(recent_sessions)
older_count = len(older_sessions)
courses_trend = 'stable'
students_trend = 'stable'
performance_trend = 'stable'
workload_trend = 'stable'
if recent_count > older_count:
courses_trend = 'up'
students_trend = 'up'
performance_trend = 'up'
elif recent_count < older_count:
courses_trend = 'down'
students_trend = 'down'
performance_trend = 'down'
# Workload trend based on session duration
recent_hours = sum([
(datetime.combine(date.today(), s.end_time) - datetime.combine(date.today(), s.start_time)).total_seconds() / 3600
for s in recent_sessions if s.start_time and s.end_time
])
older_hours = sum([
(datetime.combine(date.today(), s.end_time) - datetime.combine(date.today(), s.start_time)).total_seconds() / 3600
for s in older_sessions if s.start_time and s.end_time
])
if recent_hours > older_hours * 1.1:
workload_trend = 'up'
elif recent_hours < older_hours * 0.9:
workload_trend = 'down'
return {
'courses_trend': courses_trend,
'students_trend': students_trend,
'performance_trend': performance_trend,
'workload_trend': workload_trend
}
except Exception as e:
current_app.logger.error(f"Error calculating trends for supervisor {supervisor.id}: {str(e)}")
return {
'courses_trend': 'stable',
'students_trend': 'stable',
'performance_trend': 'stable',
'workload_trend': 'stable'
}
def get_supervisor_attendance_analytics(self):
"""Get comprehensive attendance and verification analytics for all supervisors"""
try:
with DatabaseContextManager() as ctx:
# Get all supervisors with their relationships
supervisors = ctx.session.query(Supervisor).options(
joinedload(Supervisor.managed_courses),
joinedload(Supervisor.departments),
joinedload(Supervisor.managed_tutors)
).all()
supervisor_analytics = []
for supervisor in supervisors:
try:
# Calculate verification analytics
verification_analytics = self._calculate_verification_analytics(ctx, supervisor)
# Calculate attendance analytics
attendance_analytics = self._calculate_supervisor_attendance_analytics(ctx, supervisor)
# Calculate performance metrics
performance_metrics = self._calculate_performance_metrics(ctx, supervisor)
supervisor_data = {
'id': supervisor.id,
'first_name': supervisor.first_name,
'last_name': supervisor.last_name,
'email': supervisor.email,
'department': supervisor.departments[0].department_name if supervisor.departments else 'Unknown',
'specialization': supervisor.specialization,
'staff_id': supervisor.staff_id,
'profile_picture': supervisor.profile_picture,
'verification_analytics': verification_analytics,
'attendance_analytics': attendance_analytics,
'performance_metrics': performance_metrics
}
supervisor_analytics.append(supervisor_data)
except Exception as e:
current_app.logger.error(f"Error processing supervisor {supervisor.id}: {str(e)}")
continue
return {
'success': True,
'data': supervisor_analytics
}
except Exception as e:
current_app.logger.error(f"Error fetching supervisor attendance analytics: {str(e)}")
return {
'success': False,
'message': f'Error fetching supervisor attendance analytics: {str(e)}',
'data': []
}
def _calculate_verification_analytics(self, ctx, supervisor):
"""Calculate verification analytics for a supervisor"""
try:
# Get all teaching sessions for courses managed by this supervisor
managed_course_ids = [course.id for course in supervisor.managed_courses]
if not managed_course_ids:
return {
'total_sessions': 0,
'verified_sessions': 0,
'pending_verification': 0,
'verification_rate': 0,
'avg_verification_time_hours': 0,
'verification_trend': []
}
# Get daily teaching sessions for managed courses (these have session_date)
from src.models.models import DailyTeachingSession
sessions = ctx.session.query(DailyTeachingSession).filter(
DailyTeachingSession.course_id.in_(managed_course_ids)
).all()
total_sessions = len(sessions)
verified_sessions = len([s for s in sessions if s.is_verified])
pending_verification = total_sessions - verified_sessions
verification_rate = (verified_sessions / total_sessions * 100) if total_sessions > 0 else 0
# Calculate average verification time
verified_sessions_with_time = [
s for s in sessions
if s.is_verified and s.verified_at and s.created_at
]
avg_verification_time_hours = 0
if verified_sessions_with_time:
total_verification_time = sum([
(s.verified_at - s.created_at).total_seconds() / 3600
for s in verified_sessions_with_time
])
avg_verification_time_hours = total_verification_time / len(verified_sessions_with_time)
# Calculate verification trend (last 30 days)
thirty_days_ago = datetime.now() - timedelta(days=30)
recent_sessions = [
s for s in sessions
if s.session_date and s.session_date >= thirty_days_ago.date()
]
verification_trend = []
for i in range(30):
date = (datetime.now() - timedelta(days=i)).date()
day_sessions = [s for s in recent_sessions if s.session_date == date]
verified_count = len([s for s in day_sessions if s.is_verified])
pending_count = len(day_sessions) - verified_count
verification_trend.append({
'date': date.isoformat(),
'verified': verified_count,
'pending': pending_count,
'total': len(day_sessions)
})
verification_trend.reverse() # Oldest to newest
return {
'total_sessions': total_sessions,
'verified_sessions': verified_sessions,
'pending_verification': pending_verification,
'verification_rate': verification_rate,
'avg_verification_time_hours': avg_verification_time_hours,
'verification_trend': verification_trend
}
except Exception as e:
current_app.logger.error(f"Error calculating verification analytics for supervisor {supervisor.id}: {str(e)}")
return {
'total_sessions': 0,
'verified_sessions': 0,
'pending_verification': 0,
'verification_rate': 0,
'avg_verification_time_hours': 0,
'verification_trend': []
}
def _calculate_supervisor_attendance_analytics(self, ctx, supervisor):
"""Calculate attendance analytics for a supervisor"""
try:
# Get all teaching sessions for courses managed by this supervisor
managed_course_ids = [course.id for course in supervisor.managed_courses]
if not managed_course_ids:
return {
'total_attendance_records': 0,
'average_attendance_rate': 0,
'attendance_trend': [],
'course_attendance': [],
'monthly_performance': []
}
# Get daily teaching sessions for managed courses (these have session_date)
from src.models.models import DailyTeachingSession
sessions = ctx.session.query(DailyTeachingSession).filter(
DailyTeachingSession.course_id.in_(managed_course_ids)
).all()
session_ids = [s.id for s in sessions]
# Get attendance records for these sessions
attendance_records = ctx.session.query(Attendance).filter(
Attendance.session_id.in_(session_ids)
).all()
total_attendance_records = len(attendance_records)
# Calculate average attendance rate
attendance_rates = []
for session in sessions:
session_attendance = [a for a in attendance_records if a.session_id == session.id]
if session_attendance:
present_count = len([a for a in session_attendance if a.status == 'present'])
total_count = len(session_attendance)
if total_count > 0:
attendance_rates.append((present_count / total_count) * 100)
average_attendance_rate = sum(attendance_rates) / len(attendance_rates) if attendance_rates else 0
# Calculate attendance trend (last 30 days)
thirty_days_ago = datetime.now() - timedelta(days=30)
recent_sessions = [
s for s in sessions
if s.session_date and s.session_date >= thirty_days_ago.date()
]
attendance_trend = []
for i in range(30):
date = (datetime.now() - timedelta(days=i)).date()
day_sessions = [s for s in recent_sessions if s.session_date == date]
total_students = 0
present_students = 0
for session in day_sessions:
session_attendance = [a for a in attendance_records if a.session_id == session.id]
total_students += len(session_attendance)
present_students += len([a for a in session_attendance if a.status == 'present'])
attendance_rate = (present_students / total_students * 100) if total_students > 0 else 0
attendance_trend.append({
'date': date.isoformat(),
'attendance_rate': attendance_rate,
'total_students': total_students,
'present_students': present_students
})
attendance_trend.reverse() # Oldest to newest
# Calculate course-specific attendance
course_attendance = []
for course in supervisor.managed_courses:
course_sessions = [s for s in sessions if s.course_id == course.id]
course_session_ids = [s.id for s in course_sessions]
course_attendance_records = [a for a in attendance_records if a.session_id in course_session_ids]
if course_attendance_records:
present_count = len([a for a in course_attendance_records if a.status == 'present'])
total_count = len(course_attendance_records)
attendance_rate = (present_count / total_count * 100) if total_count > 0 else 0
course_attendance.append({
'course_name': course.name,
'course_code': course.code,
'total_sessions': len(course_sessions),
'average_attendance': attendance_rate,
'attendance_rate': attendance_rate
})
# Calculate monthly performance (last 3 months)
monthly_performance = []
for i in range(3):
month_start = datetime.now().replace(day=1) - timedelta(days=i*30)
month_end = month_start.replace(day=1) + timedelta(days=32)
month_end = month_end.replace(day=1) - timedelta(days=1)
month_sessions = [
s for s in sessions
if s.session_date and month_start.date() <= s.session_date <= month_end.date()
]
sessions_conducted = len(month_sessions)
verified_count = len([s for s in month_sessions if s.is_verified])
verification_rate = (verified_count / sessions_conducted * 100) if sessions_conducted > 0 else 0
# Calculate monthly attendance rate
month_session_ids = [s.id for s in month_sessions]
month_attendance = [a for a in attendance_records if a.session_id in month_session_ids]
if month_attendance:
present_count = len([a for a in month_attendance if a.status == 'present'])
total_count = len(month_attendance)
attendance_rate = (present_count / total_count * 100) if total_count > 0 else 0
else:
attendance_rate = 0
performance_score = (verification_rate + attendance_rate) / 2
monthly_performance.append({
'month': month_start.strftime('%Y-%m'),
'sessions_conducted': sessions_conducted,
'verification_rate': verification_rate,
'attendance_rate': attendance_rate,
'performance_score': performance_score
})
monthly_performance.reverse() # Oldest to newest
return {
'total_attendance_records': total_attendance_records,
'average_attendance_rate': average_attendance_rate,
'attendance_trend': attendance_trend,
'course_attendance': course_attendance,
'monthly_performance': monthly_performance
}
except Exception as e:
current_app.logger.error(f"Error calculating attendance analytics for supervisor {supervisor.id}: {str(e)}")
return {
'total_attendance_records': 0,
'average_attendance_rate': 0,
'attendance_trend': [],
'course_attendance': [],
'monthly_performance': []
}
def _calculate_performance_metrics(self, ctx, supervisor):
"""Calculate comprehensive performance metrics for a supervisor"""
try:
# Get verification analytics
verification_analytics = self._calculate_verification_analytics(ctx, supervisor)
# Get attendance analytics
attendance_analytics = self._calculate_supervisor_attendance_analytics(ctx, supervisor)
# Calculate overall performance score
verification_performance = verification_analytics['verification_rate']
attendance_performance = attendance_analytics['average_attendance_rate']
# Calculate punctuality score based on verification time
avg_verification_time = verification_analytics['avg_verification_time_hours']
if avg_verification_time <= 2:
punctuality_score = 95
elif avg_verification_time <= 4:
punctuality_score = 85
elif avg_verification_time <= 8:
punctuality_score = 75
elif avg_verification_time <= 24:
punctuality_score = 65
else:
punctuality_score = 50
# Calculate student satisfaction (mock calculation based on attendance and verification)
student_satisfaction = (attendance_performance + verification_performance) / 2
# Calculate overall score
overall_score = (
verification_performance * 0.3 +
attendance_performance * 0.3 +
punctuality_score * 0.2 +
student_satisfaction * 0.2
)
return {
'overall_score': overall_score,
'verification_performance': verification_performance,
'attendance_performance': attendance_performance,
'punctuality_score': punctuality_score,
'student_satisfaction': student_satisfaction,
'last_updated': datetime.now().isoformat()
}
except Exception as e:
current_app.logger.error(f"Error calculating performance metrics for supervisor {supervisor.id}: {str(e)}")
return {
'overall_score': 0,
'verification_performance': 0,
'attendance_performance': 0,
'punctuality_score': 0,
'student_satisfaction': 0,
'last_updated': datetime.now().isoformat()
}