from datetime import datetime, date, time, timedelta from typing import Dict, List, Any from sqlalchemy.orm import joinedload from sqlalchemy import func, and_, or_ from src.models import DatabaseContextManager from src.models.models import ( ExamTimetable, ExamSession, ExamStudentRegistration, Supervisor, SupervisorDepartment, Course, Speciality ) from src.utils import custom_response from flask import current_app class ExamTimetableManager: """Manages exam timetable operations for supervisors""" def get_timetables(self, requester_id: str, requester_type: str, department: str = None, accreditation_body: str = None, status: str = None) -> Dict: """Get exam timetables for supervisor's departments""" try: with DatabaseContextManager() as ctx: # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can access exam timetables", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get supervisor's departments supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if not supervisor_departments: return custom_response( success=True, data={'timetables': []}, status_code=200 ) # Build query query = ctx.session.query(ExamTimetable).filter( ExamTimetable.department.in_(supervisor_departments), ExamTimetable.is_active == True ) # Apply filters if department: query = query.filter(ExamTimetable.department == department) if accreditation_body: query = query.filter(ExamTimetable.accreditation_body == accreditation_body) if status: query = query.filter(ExamTimetable.status == status) timetables = query.order_by(ExamTimetable.created_at.desc()).all() return custom_response( success=True, data={ 'timetables': [self._format_timetable(t) for t in timetables], 'departments': supervisor_departments }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching exam timetables: {str(e)}") return custom_response( success=False, data=f"Error fetching timetables: {str(e)}", status_code=500 ) def create_timetable(self, payload: Dict) -> Dict: """Create a new exam timetable""" try: with DatabaseContextManager() as ctx: requester_id = payload.get('requester_id') requester_type = payload.get('requester_type') # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can create exam timetables", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Verify supervisor has access to the department department = payload.get('department') supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to create timetables for this department", status_code=403 ) # Create timetable timetable = ExamTimetable( id=str(uuid.uuid4()), title=payload.get('title'), description=payload.get('description'), department=department, accreditation_body=payload.get('accreditation_body'), exam_period=payload.get('exam_period'), academic_year=payload.get('academic_year'), start_date=datetime.fromisoformat(payload['start_date'].replace('Z', '+00:00')).date() if payload.get('start_date') else None, end_date=datetime.fromisoformat(payload['end_date'].replace('Z', '+00:00')).date() if payload.get('end_date') else None, exam_duration_hours=payload.get('exam_duration_hours', 3), status='draft', is_published=False, created_by=requester_id, notes=payload.get('notes'), is_active=True ) ctx.session.add(timetable) ctx.session.commit() return custom_response( success=True, data={ 'message': 'Exam timetable created successfully', 'timetable': self._format_timetable(timetable) }, status_code=201 ) except Exception as e: current_app.logger.error(f"Error creating exam timetable: {str(e)}") import traceback current_app.logger.error(traceback.format_exc()) return custom_response( success=False, data=f"Error creating timetable: {str(e)}", status_code=500 ) def get_timetable(self, timetable_id: str, requester_id: str, requester_type: str) -> Dict: """Get exam timetable details with sessions""" try: with DatabaseContextManager() as ctx: # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can access exam timetables", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get timetable with sessions timetable = ctx.session.query(ExamTimetable).options( joinedload(ExamTimetable.exam_sessions).joinedload(ExamSession.course) ).filter( ExamTimetable.id == timetable_id, ExamTimetable.is_active == True ).first() if not timetable: return custom_response( success=False, data="Exam timetable not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to access this timetable", status_code=403 ) return custom_response( success=True, data={ 'timetable': self._format_timetable(timetable, include_sessions=True) }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching exam timetable: {str(e)}") return custom_response( success=False, data=f"Error fetching timetable: {str(e)}", status_code=500 ) def update_timetable(self, timetable_id: str, payload: Dict) -> Dict: """Update exam timetable""" try: with DatabaseContextManager() as ctx: requester_id = payload.get('requester_id') requester_type = payload.get('requester_type') # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can update exam timetables", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get timetable timetable = ctx.session.query(ExamTimetable).filter( ExamTimetable.id == timetable_id, ExamTimetable.is_active == True ).first() if not timetable: return custom_response( success=False, data="Exam timetable not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to update this timetable", status_code=403 ) # Update fields updatable_fields = ['title', 'description', 'exam_period', 'academic_year', 'notes'] for field in updatable_fields: if field in payload: setattr(timetable, field, payload[field]) if 'start_date' in payload: timetable.start_date = datetime.fromisoformat(payload['start_date'].replace('Z', '+00:00')).date() if 'end_date' in payload: timetable.end_date = datetime.fromisoformat(payload['end_date'].replace('Z', '+00:00')).date() if 'exam_duration_hours' in payload: timetable.exam_duration_hours = payload['exam_duration_hours'] timetable.updated_at = datetime.utcnow() timetable.updated_by = requester_id ctx.session.commit() return custom_response( success=True, data={ 'message': 'Exam timetable updated successfully', 'timetable': self._format_timetable(timetable) }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error updating exam timetable: {str(e)}") return custom_response( success=False, data=f"Error updating timetable: {str(e)}", status_code=500 ) def delete_timetable(self, timetable_id: str, requester_id: str, requester_type: str) -> Dict: """Delete exam timetable""" try: with DatabaseContextManager() as ctx: # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can delete exam timetables", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get timetable timetable = ctx.session.query(ExamTimetable).filter( ExamTimetable.id == timetable_id, ExamTimetable.is_active == True ).first() if not timetable: return custom_response( success=False, data="Exam timetable not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to delete this timetable", status_code=403 ) # Soft delete timetable.is_active = False timetable.updated_at = datetime.utcnow() timetable.updated_by = requester_id ctx.session.commit() return custom_response( success=True, data={'message': 'Exam timetable deleted successfully'}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error deleting exam timetable: {str(e)}") return custom_response( success=False, data=f"Error deleting timetable: {str(e)}", status_code=500 ) def publish_timetable(self, timetable_id: str, requester_id: str, requester_type: str) -> Dict: """Publish exam timetable""" try: with DatabaseContextManager() as ctx: # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can publish exam timetables", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get timetable timetable = ctx.session.query(ExamTimetable).filter( ExamTimetable.id == timetable_id, ExamTimetable.is_active == True ).first() if not timetable: return custom_response( success=False, data="Exam timetable not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to publish this timetable", status_code=403 ) # Publish timetable timetable.status = 'published' timetable.is_published = True timetable.published_at = datetime.utcnow() timetable.published_by = requester_id timetable.updated_at = datetime.utcnow() timetable.updated_by = requester_id ctx.session.commit() return custom_response( success=True, data={ 'message': 'Exam timetable published successfully', 'timetable': self._format_timetable(timetable) }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error publishing exam timetable: {str(e)}") return custom_response( success=False, data=f"Error publishing timetable: {str(e)}", status_code=500 ) def add_exam_session(self, payload: Dict) -> Dict: """Add exam session to timetable""" try: with DatabaseContextManager() as ctx: requester_id = payload.get('requester_id') requester_type = payload.get('requester_type') timetable_id = payload.get('timetable_id') # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can add exam sessions", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get timetable timetable = ctx.session.query(ExamTimetable).filter( ExamTimetable.id == timetable_id, ExamTimetable.is_active == True ).first() if not timetable: return custom_response( success=False, data="Exam timetable not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to add sessions to this timetable", status_code=403 ) # Create exam session session = ExamSession( id=str(uuid.uuid4()), timetable_id=timetable_id, course_id=payload.get('course_id'), exam_title=payload.get('exam_title'), exam_code=payload.get('exam_code'), exam_type=payload.get('exam_type', 'written'), exam_level=payload.get('exam_level'), exam_module=payload.get('exam_module'), exam_date=datetime.fromisoformat(payload['exam_date'].replace('Z', '+00:00')).date() if payload.get('exam_date') else None, start_time=datetime.fromisoformat(payload['start_time'].replace('Z', '+00:00')).time() if payload.get('start_time') else None, end_time=datetime.fromisoformat(payload['end_time'].replace('Z', '+00:00')).time() if payload.get('end_time') else None, duration_minutes=payload.get('duration_minutes'), venue=payload.get('venue'), room_capacity=payload.get('room_capacity'), invigilator_required=payload.get('invigilator_required', True), invigilator_count=payload.get('invigilator_count', 1), expected_students=payload.get('expected_students', 0), max_students=payload.get('max_students'), instructions=payload.get('instructions'), materials_required=payload.get('materials_required'), special_requirements=payload.get('special_requirements'), notes=payload.get('notes'), created_by=requester_id ) ctx.session.add(session) # Update timetable stats timetable.total_exams += 1 timetable.updated_at = datetime.utcnow() timetable.updated_by = requester_id ctx.session.commit() return custom_response( success=True, data={ 'message': 'Exam session added successfully', 'session': self._format_exam_session(session) }, status_code=201 ) except Exception as e: current_app.logger.error(f"Error adding exam session: {str(e)}") import traceback current_app.logger.error(traceback.format_exc()) return custom_response( success=False, data=f"Error adding session: {str(e)}", status_code=500 ) def get_exam_sessions(self, timetable_id: str, requester_id: str, requester_type: str) -> Dict: """Get exam sessions for a timetable""" try: with DatabaseContextManager() as ctx: # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can access exam sessions", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get timetable timetable = ctx.session.query(ExamTimetable).filter( ExamTimetable.id == timetable_id, ExamTimetable.is_active == True ).first() if not timetable: return custom_response( success=False, data="Exam timetable not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to access sessions for this timetable", status_code=403 ) # Get sessions sessions = ctx.session.query(ExamSession).options( joinedload(ExamSession.course) ).filter( ExamSession.timetable_id == timetable_id ).order_by(ExamSession.exam_date, ExamSession.start_time).all() return custom_response( success=True, data={ 'sessions': [self._format_exam_session(s) for s in sessions] }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching exam sessions: {str(e)}") return custom_response( success=False, data=f"Error fetching sessions: {str(e)}", status_code=500 ) def get_exam_session(self, session_id: str, requester_id: str, requester_type: str) -> Dict: """Get exam session details""" try: with DatabaseContextManager() as ctx: # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can access exam sessions", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get session with timetable session = ctx.session.query(ExamSession).options( joinedload(ExamSession.timetable), joinedload(ExamSession.course) ).filter(ExamSession.id == session_id).first() if not session: return custom_response( success=False, data="Exam session not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if session.timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to access this session", status_code=403 ) return custom_response( success=True, data={ 'session': self._format_exam_session(session, include_details=True) }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching exam session: {str(e)}") return custom_response( success=False, data=f"Error fetching session: {str(e)}", status_code=500 ) def update_exam_session(self, session_id: str, payload: Dict) -> Dict: """Update exam session""" try: with DatabaseContextManager() as ctx: requester_id = payload.get('requester_id') requester_type = payload.get('requester_type') # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can update exam sessions", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get session session = ctx.session.query(ExamSession).options( joinedload(ExamSession.timetable) ).filter(ExamSession.id == session_id).first() if not session: return custom_response( success=False, data="Exam session not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if session.timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to update this session", status_code=403 ) # Update fields updatable_fields = [ 'exam_title', 'exam_code', 'exam_type', 'exam_level', 'exam_module', 'venue', 'room_capacity', 'invigilator_required', 'invigilator_count', 'expected_students', 'max_students', 'instructions', 'materials_required', 'special_requirements', 'notes' ] for field in updatable_fields: if field in payload: setattr(session, field, payload[field]) if 'exam_date' in payload: session.exam_date = datetime.fromisoformat(payload['exam_date'].replace('Z', '+00:00')).date() if 'start_time' in payload: session.start_time = datetime.fromisoformat(payload['start_time'].replace('Z', '+00:00')).time() if 'end_time' in payload: session.end_time = datetime.fromisoformat(payload['end_time'].replace('Z', '+00:00')).time() if 'duration_minutes' in payload: session.duration_minutes = payload['duration_minutes'] session.updated_at = datetime.utcnow() session.updated_by = requester_id ctx.session.commit() return custom_response( success=True, data={ 'message': 'Exam session updated successfully', 'session': self._format_exam_session(session) }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error updating exam session: {str(e)}") return custom_response( success=False, data=f"Error updating session: {str(e)}", status_code=500 ) def delete_exam_session(self, session_id: str, requester_id: str, requester_type: str) -> Dict: """Delete exam session""" try: with DatabaseContextManager() as ctx: # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can delete exam sessions", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Get session session = ctx.session.query(ExamSession).options( joinedload(ExamSession.timetable) ).filter(ExamSession.id == session_id).first() if not session: return custom_response( success=False, data="Exam session not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if session.timetable.department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to delete this session", status_code=403 ) # Delete session ctx.session.delete(session) # Update timetable stats session.timetable.total_exams -= 1 session.timetable.updated_at = datetime.utcnow() session.timetable.updated_by = requester_id ctx.session.commit() return custom_response( success=True, data={'message': 'Exam session deleted successfully'}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error deleting exam session: {str(e)}") return custom_response( success=False, data=f"Error deleting session: {str(e)}", status_code=500 ) def get_department_courses(self, department: str, accreditation_body: str, requester_id: str, requester_type: str) -> Dict: """Get courses for a department filtered by accreditation body""" try: with DatabaseContextManager() as ctx: # Verify requester is supervisor if requester_type != 'supervisor': return custom_response( success=False, data="Only supervisors can access courses", status_code=403 ) supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == requester_id).first() if not supervisor: return custom_response( success=False, data="Supervisor not found", status_code=404 ) # Verify supervisor has access to this department supervisor_departments = [dept.department_name for dept in supervisor.departments if dept.is_active] if department not in supervisor_departments: return custom_response( success=False, data="You don't have permission to access courses in this department", status_code=403 ) # Build query query = ctx.session.query(Course).join(Speciality, Course.speciality_id == Speciality.id).filter( Course.department == department, Course.is_active == True ) if accreditation_body: query = query.filter(Speciality.accreditation_body == accreditation_body) courses = query.order_by(Course.title).all() return custom_response( success=True, data={ 'courses': [self._format_course(c) for c in courses] }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching department courses: {str(e)}") return custom_response( success=False, data=f"Error fetching courses: {str(e)}", status_code=500 ) def _format_timetable(self, timetable: ExamTimetable, include_sessions: bool = False) -> Dict: """Format timetable data for API response""" formatted = { 'id': timetable.id, 'title': timetable.title, 'description': timetable.description, 'department': timetable.department, 'accreditation_body': timetable.accreditation_body, 'exam_period': timetable.exam_period, 'academic_year': timetable.academic_year, 'start_date': timetable.start_date.isoformat() if timetable.start_date else None, 'end_date': timetable.end_date.isoformat() if timetable.end_date else None, 'exam_duration_hours': timetable.exam_duration_hours, 'status': timetable.status, 'is_published': timetable.is_published, 'published_at': timetable.published_at.isoformat() if timetable.published_at else None, 'total_exams': timetable.total_exams, 'total_students': timetable.total_students, 'notes': timetable.notes, 'created_at': timetable.created_at.isoformat() if timetable.created_at else None, 'updated_at': timetable.updated_at.isoformat() if timetable.updated_at else None } if include_sessions and hasattr(timetable, 'exam_sessions'): formatted['sessions'] = [self._format_exam_session(s) for s in timetable.exam_sessions] return formatted def _format_exam_session(self, session: ExamSession, include_details: bool = False) -> Dict: """Format exam session data for API response""" formatted = { 'id': session.id, 'timetable_id': session.timetable_id, 'course_id': session.course_id, 'exam_title': session.exam_title, 'exam_code': session.exam_code, 'exam_type': session.exam_type, 'exam_level': session.exam_level, 'exam_module': session.exam_module, 'exam_date': session.exam_date.isoformat() if session.exam_date else None, 'start_time': session.start_time.strftime('%H:%M') if session.start_time else None, 'end_time': session.end_time.strftime('%H:%M') if session.end_time else None, 'duration_minutes': session.duration_minutes, 'venue': session.venue, 'room_capacity': session.room_capacity, 'invigilator_required': session.invigilator_required, 'invigilator_count': session.invigilator_count, 'expected_students': session.expected_students, 'registered_students': session.registered_students, 'max_students': session.max_students, 'status': session.status, 'is_confirmed': session.is_confirmed, 'confirmed_at': session.confirmed_at.isoformat() if session.confirmed_at else None, 'created_at': session.created_at.isoformat() if session.created_at else None, 'updated_at': session.updated_at.isoformat() if session.updated_at else None } if include_details: formatted.update({ 'instructions': session.instructions, 'materials_required': session.materials_required, 'special_requirements': session.special_requirements, 'notes': session.notes }) if hasattr(session, 'course') and session.course: formatted['course'] = self._format_course(session.course) return formatted def _format_course(self, course: Course) -> Dict: """Format course data for API response""" return { 'id': course.id, 'code': course.code, 'title': course.title, 'description': course.description, 'credits': course.credits, 'department': course.department, 'course_level': course.course_level, 'course_module': course.course_module, 'is_active': course.is_active }