from flask import current_app from sqlalchemy import and_, or_, func, desc from datetime import datetime from typing import Dict, Any, List, Optional import uuid from src.models import DatabaseContextManager from src.models.models import ( Speciality, Course, tutor_course_association, supervisor_course_association, Tutor, Supervisor, SupervisorDepartment, Enrollment, CourseModule, CourseResource, Assignment, TimetableBlock, TeachingSession, Attendance, Reminder, User, Student, NotificationPreference ) from src.utils import custom_response, send_email from src.models.models import User class SpecialityManager: """Enhanced Speciality Management with top-notch styling and comprehensive operations""" def __init__(self): self.table = Speciality def create(self, payload: Dict[str, Any]) -> Dict[str, Any]: """ Create a new speciality with enhanced validation and styling """ try: with DatabaseContextManager() as ctx: # Enhanced validation with detailed error messages if not payload.get('name'): return custom_response( success=False, data="Speciality name is required", status_code=400 ) if not payload.get('department'): return custom_response( success=False, data="Department is required", status_code=400 ) # Check for duplicate names within the same department existing_speciality = ctx.session.query(Speciality).filter( and_( Speciality.name == payload['name'], Speciality.department == payload['department'] ) ).first() if existing_speciality: return custom_response( success=False, data=f"A speciality with name '{payload['name']}' already exists in {payload['department']} department", status_code=409 ) # Create speciality with enhanced data speciality_data = { 'id': str(uuid.uuid4()), 'name': payload['name'].strip(), 'description': payload.get('description', '').strip(), 'department': payload['department'].strip(), 'code': payload.get('code'), 'abbreviation': payload.get('abbreviation'), 'contact_email': payload.get('contact_email'), 'contact_phone': payload.get('contact_phone'), 'website': payload.get('website'), 'location': payload.get('location'), 'accreditation_body': payload.get('accreditation_body'), 'modules': payload.get('modules'), 'speciality_level': payload.get('speciality_level'), 'module_level': payload.get('module_level'), 'is_active': payload.get('is_active', True), 'created_at': datetime.utcnow(), 'updated_at': datetime.utcnow() } speciality = Speciality(**speciality_data) ctx.session.add(speciality) ctx.commit() # Return enhanced response with styling return custom_response( success=True, data={ 'speciality': { 'id': speciality.id, 'name': speciality.name, 'description': speciality.description, 'department': speciality.department, 'code': speciality.code, 'abbreviation': speciality.abbreviation, 'contact_email': speciality.contact_email, 'contact_phone': speciality.contact_phone, 'website': speciality.website, 'location': speciality.location, 'accreditation_body': speciality.accreditation_body, 'modules': speciality.modules, 'speciality_level': speciality.speciality_level, 'module_level': speciality.module_level, 'is_active': speciality.is_active, 'created_at': speciality.created_at.isoformat(), 'updated_at': speciality.updated_at.isoformat(), 'courses_count': 0 } }, status_code=201 ) except Exception as e: current_app.logger.error(f"Error creating speciality: {str(e)}") return custom_response( success=False, data="Failed to create speciality", status_code=500 ) def get_all(self, filters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """ Get all specialities with enhanced filtering, sorting, and styling Filtered by supervisor's department if supervisor_id is provided Supports pagination for large datasets """ try: with DatabaseContextManager() as ctx: # Build query with enhanced filtering query = ctx.session.query(Speciality) # Apply supervisor department filter if filters and filters.get('supervisor_id'): supervisor_id = filters['supervisor_id'] # Get supervisor's departments from the association table supervisor_departments = ctx.session.query(SupervisorDepartment).filter( SupervisorDepartment.supervisor_id == supervisor_id, SupervisorDepartment.is_active == True ).all() if supervisor_departments: # Filter specialities by supervisor's departments supervisor_dept_names = [dept.department_name for dept in supervisor_departments] query = query.filter(Speciality.department.in_(supervisor_dept_names)) else: # If supervisor has no departments, return empty result return custom_response( success=True, data={ 'specialities': [], 'statistics': { 'total_specialities': 0, 'active_specialities': 0, 'total_courses': 0, 'departments': 0 }, 'department_groups': {}, 'filters_applied': filters or {}, 'supervisor_filter_applied': True, 'supervisor_departments': [], 'pagination': { 'page': 1, 'per_page': 10, 'total_pages': 0, 'total_items': 0, 'has_next': False, 'has_prev': False } }, status_code=200 ) # Apply other filters if filters: # Handle multiple departments filter if filters.get('departments'): departments = filters['departments'] if isinstance(departments, list) and len(departments) > 0: query = query.filter(Speciality.department.in_(departments)) elif filters.get('department'): # Single department filter for backward compatibility query = query.filter(Speciality.department == filters['department']) if filters.get('is_active') is not None: query = query.filter(Speciality.is_active == filters['is_active']) if filters.get('search'): search_term = f"%{filters['search']}%" query = query.filter( or_( Speciality.name.ilike(search_term), Speciality.description.ilike(search_term), Speciality.department.ilike(search_term), Speciality.code.ilike(search_term), Speciality.abbreviation.ilike(search_term), Speciality.contact_email.ilike(search_term), Speciality.contact_phone.ilike(search_term), Speciality.location.ilike(search_term), Speciality.accreditation_body.ilike(search_term), Speciality.modules.ilike(search_term), Speciality.speciality_level.ilike(search_term), Speciality.module_level.ilike(search_term) ) ) # Get total count for pagination total_count = query.count() # Apply pagination page = filters.get('page', 1) if filters else 1 per_page = filters.get('per_page', 1000) if filters else 1000 # Increased default from 10 to 1000 # Calculate pagination values total_pages = (total_count + per_page - 1) // per_page offset = (page - 1) * per_page # Apply pagination to query query = query.offset(offset).limit(per_page) # Get specialities with course counts specialities = [] for speciality in query.all(): # Count linked courses courses_count = ctx.session.query(Course).filter( Course.speciality_id == speciality.id ).count() specialities.append({ 'id': speciality.id, 'name': speciality.name, 'description': speciality.description, 'department': speciality.department, 'code': speciality.code, 'abbreviation': speciality.abbreviation, 'contact_email': speciality.contact_email, 'contact_phone': speciality.contact_phone, 'website': speciality.website, 'location': speciality.location, 'accreditation_body': speciality.accreditation_body, 'modules': speciality.modules, 'speciality_level': speciality.speciality_level, 'module_level': speciality.module_level, 'is_active': speciality.is_active, 'sharing_capable': True, # All specialities have sharing capabilities by default 'created_at': speciality.created_at.isoformat(), 'updated_at': speciality.updated_at.isoformat(), 'courses_count': courses_count }) # Enhanced sorting and organization specialities.sort(key=lambda x: (x['department'], x['name'])) # Calculate enhanced statistics total_specialities = total_count active_specialities = ctx.session.query(Speciality).filter(Speciality.is_active == True).count() total_courses = sum(s['courses_count'] for s in specialities) departments = len(set(s['department'] for s in specialities)) # Group by department for better organization department_groups = {} for speciality in specialities: dept = speciality['department'] if dept not in department_groups: department_groups[dept] = [] department_groups[dept].append(speciality) # Get supervisor departments info if filtering was applied supervisor_info = {} if filters and filters.get('supervisor_id'): supervisor_id = filters['supervisor_id'] # Get supervisor details and departments supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first() if supervisor: supervisor_departments = ctx.session.query(SupervisorDepartment).filter( SupervisorDepartment.supervisor_id == supervisor_id, SupervisorDepartment.is_active == True ).all() if supervisor_departments: supervisor_info = { 'supervisor_id': supervisor_id, 'supervisor_name': f"{supervisor.first_name} {supervisor.last_name}", 'supervisor_departments': [dept.department_name for dept in supervisor_departments], 'filter_applied': True } # Build pagination info pagination = { 'page': page, 'per_page': per_page, 'total_pages': total_pages, 'total_items': total_count, 'has_next': page < total_pages, 'has_prev': page > 1 } return custom_response( success=True, data={ 'specialities': specialities, 'statistics': { 'total_specialities': total_specialities, 'active_specialities': active_specialities, 'total_courses': total_courses, 'departments': departments }, 'department_groups': department_groups, 'filters_applied': filters or {}, 'supervisor_info': supervisor_info, 'pagination': pagination }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching specialities: {str(e)}") return custom_response( success=False, data="Failed to fetch specialities", status_code=500 ) def get_by_id(self, speciality_id: str) -> Dict[str, Any]: """ Get a specific speciality by ID with enhanced details and styling """ try: with DatabaseContextManager() as ctx: 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 ) # Get linked courses with enhanced details including tutor information linked_courses = ctx.session.query(Course).filter( Course.speciality_id == speciality_id ).all() courses_data = [] for course in linked_courses: # Get tutor information for this course tutor_info = ctx.session.query( tutor_course_association.c.tutor_id, tutor_course_association.c.is_primary ).filter( tutor_course_association.c.course_id == course.id ).all() tutors_data = [] for tutor_assoc in tutor_info: # Get tutor details from User table tutor_user = ctx.session.query(User).filter( User.id == tutor_assoc.tutor_id, User.is_active == True ).first() if tutor_user: tutors_data.append({ 'id': tutor_user.id, 'first_name': tutor_user.first_name, 'last_name': tutor_user.last_name, 'email': tutor_user.email, 'is_primary': tutor_assoc.is_primary, 'staff_id': getattr(tutor_user, 'staff_id', None) if hasattr(tutor_user, 'staff_id') else None }) # Get supervisor information for this course supervisor_info = ctx.session.query( supervisor_course_association.c.supervisor_id, supervisor_course_association.c.is_primary ).filter( supervisor_course_association.c.course_id == course.id ).all() supervisors_data = [] for supervisor_assoc in supervisor_info: # Get supervisor details from Supervisor table (Supervisor extends User) supervisor = ctx.session.query(Supervisor).filter( Supervisor.id == supervisor_assoc.supervisor_id, Supervisor.is_active == True ).first() if supervisor: supervisors_data.append({ 'id': supervisor.id, 'first_name': supervisor.first_name, 'last_name': supervisor.last_name, 'email': supervisor.email, 'is_primary': supervisor_assoc.is_primary, 'staff_id': getattr(supervisor, 'staff_id', None) if hasattr(supervisor, 'staff_id') else None }) # Determine assignment status assignment_info = None if tutors_data or supervisors_data: # Check if there's a supervisor assigned assigned_supervisor = supervisors_data[0] if supervisors_data else None assigned_tutor = tutors_data[0] if tutors_data else None assignment_info = { 'can_be_claimed': False, 'assigned_instructor': None } if assigned_supervisor: assignment_info['assigned_instructor'] = { 'id': assigned_supervisor['id'], 'name': f"{assigned_supervisor['first_name']} {assigned_supervisor['last_name']}", 'email': assigned_supervisor['email'], 'staff_id': assigned_supervisor['staff_id'], 'type': 'supervisor' } elif assigned_tutor: assignment_info['assigned_instructor'] = { 'id': assigned_tutor['id'], 'name': f"{assigned_tutor['first_name']} {assigned_tutor['last_name']}", 'email': assigned_tutor['email'], 'staff_id': assigned_tutor['staff_id'], 'type': 'tutor' } else: assignment_info = { 'can_be_claimed': True, 'assigned_instructor': None } courses_data.append({ 'id': course.id, 'code': course.code, 'title': course.title, 'credits': course.credits, 'semester': course.semester, 'is_active': course.is_active, 'created_at': course.created_at.isoformat() if course.created_at else None, 'tutors': tutors_data, 'tutors_count': len(tutors_data), 'assigned_supervisors': supervisors_data, 'supervisors_count': len(supervisors_data), 'assignment_info': assignment_info }) # Enhanced speciality data speciality_data = { 'id': speciality.id, 'name': speciality.name, 'description': speciality.description, 'department': speciality.department, 'code': speciality.code, 'abbreviation': speciality.abbreviation, 'contact_email': speciality.contact_email, 'contact_phone': speciality.contact_phone, 'website': speciality.website, 'location': speciality.location, 'accreditation_body': speciality.accreditation_body, 'modules': speciality.modules, 'speciality_level': speciality.speciality_level, 'module_level': speciality.module_level, 'is_active': speciality.is_active, 'created_at': speciality.created_at.isoformat(), 'updated_at': speciality.updated_at.isoformat(), 'courses_count': len(courses_data), 'linked_courses': courses_data, 'metadata': { 'total_credits': sum(c['credits'] or 0 for c in courses_data), 'active_courses': len([c for c in courses_data if c['is_active']]), 'semesters_covered': list(set(c['semester'] for c in courses_data if c['semester'])) } } return custom_response( success=True, data={'speciality': speciality_data}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching speciality: {str(e)}") return custom_response( success=False, data="Failed to fetch speciality", status_code=500 ) def update(self, speciality_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: """ Update a speciality with enhanced validation and styling """ try: with DatabaseContextManager() as ctx: 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 ) # Enhanced validation if 'name' in payload and payload['name']: # Check for duplicate names within the same department existing_speciality = ctx.session.query(Speciality).filter( and_( Speciality.name == payload['name'].strip(), Speciality.department == (payload.get('department', speciality.department)), Speciality.id != speciality_id ) ).first() if existing_speciality: return custom_response( success=False, data=f"A speciality with name '{payload['name']}' already exists in this department", status_code=409 ) speciality.name = payload['name'].strip() if 'description' in payload: speciality.description = (payload.get('description') or '').strip() if 'department' in payload and payload['department']: speciality.department = payload['department'].strip() if 'is_active' in payload: speciality.is_active = payload['is_active'] # New optional fields if 'code' in payload: speciality.code = (payload.get('code') or '').strip() if 'abbreviation' in payload: speciality.abbreviation = (payload.get('abbreviation') or '').strip() if 'contact_email' in payload: speciality.contact_email = (payload.get('contact_email') or '').strip() if 'contact_phone' in payload: speciality.contact_phone = (payload.get('contact_phone') or '').strip() if 'website' in payload: speciality.website = (payload.get('website') or '').strip() if 'location' in payload: speciality.location = (payload.get('location') or '').strip() if 'accreditation_body' in payload: speciality.accreditation_body = (payload.get('accreditation_body') or '').strip() if 'modules' in payload: speciality.modules = (payload.get('modules') or '').strip() if 'speciality_level' in payload: speciality.speciality_level = (payload.get('speciality_level') or '').strip() if 'module_level' in payload: speciality.module_level = (payload.get('module_level') or '').strip() speciality.updated_at = datetime.utcnow() ctx.commit() # Return enhanced response return custom_response( success=True, data={ 'speciality': { 'id': speciality.id, 'name': speciality.name, 'description': speciality.description, 'department': speciality.department, 'code': speciality.code, 'abbreviation': speciality.abbreviation, 'contact_email': speciality.contact_email, 'contact_phone': speciality.contact_phone, 'website': speciality.website, 'location': speciality.location, 'accreditation_body': speciality.accreditation_body, 'modules': speciality.modules, 'speciality_level': speciality.speciality_level, 'module_level': speciality.module_level, 'is_active': speciality.is_active, 'created_at': speciality.created_at.isoformat(), 'updated_at': speciality.updated_at.isoformat() } }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error updating speciality: {str(e)}") return custom_response( success=False, data="Failed to update speciality", status_code=500 ) def toggle_status(self, speciality_id: str, new_status: bool) -> Dict[str, Any]: """ Toggle the active status of a speciality """ try: with DatabaseContextManager() as ctx: 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 ) # Update the status speciality.is_active = new_status speciality.updated_at = datetime.utcnow() ctx.commit() status_text = "activated" if new_status else "deactivated" return custom_response( success=True, data={ 'speciality': { 'id': speciality.id, 'name': speciality.name, 'is_active': speciality.is_active, 'status_message': f"Speciality '{speciality.name}' {status_text} successfully" } }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error toggling speciality status: {str(e)}") return custom_response( success=False, data="Failed to toggle speciality status", status_code=500 ) def get_analytics(self, speciality_id: str) -> Dict[str, Any]: """ Return students and tutors analytics scoped to a speciality by aggregating over linked courses """ try: with DatabaseContextManager() as ctx: # Verify speciality exists speciality = ctx.session.query(Speciality.id).filter(Speciality.id == speciality_id).first() if not speciality: return custom_response(success=False, data="Speciality not found", status_code=404) # Courses under this speciality courses_q = ctx.session.query(Course.id, Course.code, Course.title, Course.is_active).filter( Course.speciality_id == speciality_id ) courses = courses_q.all() course_ids = [c.id for c in courses] # Students per course via speciality relationship students_per_course = [] total_students = 0 if course_ids: # Get students through speciality relationship from src.models.models import Student # Get students in this speciality students_in_speciality = ctx.session.query(Student).filter( Student.speciality_id == speciality_id ).count() total_students = students_in_speciality for c in courses: # For now, we'll use the total students in speciality as a proxy # In a real implementation, you might want to track actual enrollments students_per_course.append({ 'course_id': c.id, 'code': c.code, 'title': c.title, 'students_count': students_in_speciality # All students in speciality can access all courses }) # Tutors per course via association table with enhanced details tutors_per_course = [] total_tutors = 0 active_tutors = 0 fulltime_tutors = 0 total_experience = 0 tutor_count = 0 if course_ids: # Get all tutors for courses in this speciality tutor_ids = ctx.session.query( func.distinct(tutor_course_association.c.tutor_id) ).filter( tutor_course_association.c.course_id.in_(course_ids) ).all() tutor_ids = [t[0] for t in tutor_ids] if tutor_ids: # Get detailed tutor information tutors_detail = ctx.session.query( User.id, User.first_name, User.last_name, User.email, User.is_active ).filter( User.id.in_(tutor_ids) ).all() # Count active tutors active_tutors = len([t for t in tutors_detail if t.is_active]) total_tutors = len(tutors_detail) # Get additional tutor details for experience calculation for tutor_id in tutor_ids: tutor_detail = ctx.session.query( User.id, User.is_active ).filter(User.id == tutor_id).first() if tutor_detail and tutor_detail.is_active: # Try to get years of experience from Tutor table tutor_exp = ctx.session.query( func.coalesce(Tutor.years_of_teaching, 0) ).filter(Tutor.id == tutor_id).scalar() if tutor_exp: total_experience += tutor_exp tutor_count += 1 # Check if full-time (assuming default is True) is_fulltime = ctx.session.query( func.coalesce(Tutor.is_full_time, True) ).filter(Tutor.id == tutor_id).scalar() if is_fulltime: fulltime_tutors += 1 # Build tutors per course data tpc_rows = ctx.session.query( tutor_course_association.c.course_id.label('course_id'), func.count(tutor_course_association.c.tutor_id).label('tutors_count') ).filter( tutor_course_association.c.course_id.in_(course_ids) ).group_by(tutor_course_association.c.course_id).all() tutors_map = {row.course_id: row.tutors_count for row in tpc_rows} for c in courses: tutors_per_course.append({ 'course_id': c.id, 'code': c.code, 'title': c.title, 'tutors_count': int(tutors_map.get(c.id, 0)) }) active_courses = sum(1 for c in courses if c.is_active) average_experience = round(total_experience / tutor_count, 1) if tutor_count > 0 else 0 analytics = { 'overview': { 'courses_count': len(courses), 'active_courses': active_courses, 'total_students': int(total_students), 'total_tutors': int(total_tutors), 'active_tutors': active_tutors, 'fulltime_tutors': fulltime_tutors, 'average_experience': average_experience }, 'students_per_course': students_per_course, 'tutors_per_course': tutors_per_course } return custom_response(success=True, data={'analytics': analytics}, status_code=200) except Exception as e: current_app.logger.error(f"Error fetching speciality analytics: {str(e)}") return custom_response(success=False, data="Failed to fetch analytics", status_code=500) def delete(self, speciality_id: str) -> Dict[str, Any]: """ Delete a speciality and all associated course data with enhanced validation and safety checks """ try: with DatabaseContextManager() as ctx: 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 ) ctx.session.delete(speciality) ctx.commit() return custom_response( success=True, data={ 'deleted_speciality': { 'id': speciality_id, 'name': speciality.name, } }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error deleting speciality: {str(e)}") return custom_response( success=False, data="Failed to delete speciality and associated data", status_code=500 ) def delete_by_department(self, department_name: str) -> Dict[str, Any]: """ Delete all specialities under the specified department. Returns count of deleted items and department name. """ try: with DatabaseContextManager() as ctx: q = ctx.session.query(Speciality).filter(Speciality.department == department_name) to_delete = q.all() if not to_delete: return custom_response( success=True, data={ 'department': department_name, 'deleted_count': 0, 'message': f"No specialities found in department '{department_name}'" }, status_code=200 ) count = 0 for s in to_delete: ctx.session.delete(s) count += 1 ctx.commit() return custom_response( success=True, data={ 'department': department_name, 'deleted_count': count }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error deleting specialities by department: {str(e)}") return custom_response( success=False, data="Failed to delete specialities by department", status_code=500 ) def create_student(self, speciality_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: """ Create a new student and assign them to the specified speciality Args: speciality_id: ID of the speciality to assign the student to payload: Student data including: - student_id: Unique student identifier - email: Student email address - password_hash: Hashed password - first_name: Student's first name - last_name: Student's last name - phone: Student's phone number (optional) - year_of_study: Year of study (default: 1) - program: Program name - gender: Student's gender (optional) - date_of_birth: Date of birth (optional) - nationality: Student's nationality (optional) - guardian_name: Guardian's name (optional) - guardian_contact: Guardian's contact (optional) - emergency_contact: Emergency contact (optional) - address: Student's address (optional) - city: Student's city (optional) - country: Student's country (optional) - postal_code: Postal code (optional) - created_by: ID of the trainer/supervisor creating the student """ try: with DatabaseContextManager() as ctx: # Verify speciality exists speciality = ctx.session.query(Speciality).filter( Speciality.id == speciality_id, Speciality.is_active == True ).first() if not speciality: return custom_response( success=False, data="Speciality not found or inactive", status_code=404 ) # Check if student ID already exists existing_student = ctx.session.query(Student).filter( Student.student_id == payload['student_id'] ).first() if existing_student: return custom_response( success=False, data="Student ID already exists", status_code=400 ) # Check if email already exists existing_email = ctx.session.query(User).filter( User.email == payload['email'] ).first() if existing_email: return custom_response( success=False, data="Email address already in use", status_code=400 ) # Create the student student = Student( id=str(uuid.uuid4()), email=payload['email'], password_hash=payload['password_hash'], first_name=payload['first_name'], last_name=payload['last_name'], phone=payload.get('phone'), user_type='student', student_id=payload['student_id'], year_of_study=payload.get('year_of_study', 1), program=payload['program'], enrollment_date=datetime.utcnow().date(), gender=payload.get('gender'), date_of_birth=payload.get('date_of_birth'), nationality=payload.get('nationality'), guardian_name=payload.get('guardian_name'), guardian_contact=payload.get('guardian_contact'), emergency_contact=payload.get('emergency_contact'), address=payload.get('address'), city=payload.get('city'), country=payload.get('country'), postal_code=payload.get('postal_code'), speciality_id=speciality_id, is_active=True ) ctx.session.add(student) # Create default notification preferences notification_pref = NotificationPreference( id=str(uuid.uuid4()), user_id=student.id, receive_email=True, receive_sms=False, receive_push=True ) ctx.session.add(notification_pref) ctx.session.commit() # Send welcome notification self._send_student_welcome_notification(student, speciality) return custom_response( success=True, data={ 'student': self._student_to_dict(student), 'speciality': { 'id': speciality.id, 'name': speciality.name, 'department': speciality.department } }, status_code=201 ) except Exception as e: current_app.logger.error(f"Error creating student: {str(e)}") return custom_response( success=False, data="Failed to create student", status_code=500 ) def get_speciality_students(self, speciality_id: str, filters: Dict[str, Any] = None) -> Dict[str, Any]: """ Get all students in a speciality with optional filtering Args: speciality_id: ID of the speciality filters: Optional filters (search, year_of_study, status, etc.) """ try: with DatabaseContextManager() as ctx: # Verify speciality exists speciality = ctx.session.query(Speciality).filter( Speciality.id == speciality_id, Speciality.is_active == True ).first() if not speciality: return custom_response( success=False, data="Speciality not found or inactive", status_code=404 ) # Build query query = ctx.session.query(Student).filter( Student.speciality_id == speciality_id, Student.is_active == True ) # Apply filters if filters: if filters.get('search'): search_term = f"%{filters['search']}%" query = query.filter( or_( Student.first_name.ilike(search_term), Student.last_name.ilike(search_term), Student.student_id.ilike(search_term), Student.email.ilike(search_term) ) ) if filters.get('year_of_study'): query = query.filter(Student.year_of_study == filters['year_of_study']) if filters.get('program'): query = query.filter(Student.program.ilike(f"%{filters['program']}%")) # Get total count total_count = query.count() # Apply pagination page = filters.get('page', 1) if filters else 1 per_page = filters.get('per_page', 20) if filters else 20 offset = (page - 1) * per_page students = query.offset(offset).limit(per_page).all() return custom_response( success=True, data={ 'students': [self._student_to_dict(student) for student in students], 'pagination': { 'page': page, 'per_page': per_page, 'total_count': total_count, 'total_pages': (total_count + per_page - 1) // per_page }, 'speciality': { 'id': speciality.id, 'name': speciality.name, 'department': speciality.department } }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching speciality students: {str(e)}") return custom_response( success=False, data="Failed to fetch students", status_code=500 ) def _send_student_welcome_notification(self, student: Student, speciality: Speciality) -> None: """Send welcome notification to newly created student""" try: subject = f"Welcome to {speciality.name} - Student Account Created" message = f"""

Welcome to Kisiwa Tech!

Your student account has been created

Account Details

Hello {student.first_name},

Your student account has been successfully created and you have been enrolled in the {speciality.name} program.

Your Information

Student ID: {student.student_id}

Email: {student.email}

Program: {student.program}

Year of Study: {student.year_of_study}

Speciality: {speciality.name}

Department: {speciality.department}

Next Steps

Access Student Portal

If you have any questions or need assistance, please contact your academic advisor or the student services department.

Best regards,
Kisiwa Tech Academic Team

""" send_email( sender_email='kisiwa@mutabletech.co.ke', sender_password='z}Y{WN$jV=Sv', receiver_email=student.email, subject=subject, message=message ) except Exception as e: current_app.logger.error(f"Failed to send welcome notification: {str(e)}") def _student_to_dict(self, student: Student) -> Dict[str, Any]: """Convert Student object to dictionary""" return { 'id': student.id, 'student_id': student.student_id, 'email': student.email, 'first_name': student.first_name, 'last_name': student.last_name, 'phone': student.phone, 'year_of_study': student.year_of_study, 'program': student.program, 'enrollment_date': student.enrollment_date.isoformat() if student.enrollment_date else None, 'gender': student.gender, 'date_of_birth': student.date_of_birth.isoformat() if student.date_of_birth else None, 'nationality': student.nationality, '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, 'speciality_id': student.speciality_id, 'is_active': student.is_active, 'created_at': student.created_at.isoformat() if student.created_at else None, 'updated_at': student.updated_at.isoformat() if student.updated_at else None } def bulk_operations(self, operations: List[Dict[str, Any]]) -> Dict[str, Any]: """ Perform bulk operations on specialities with enhanced styling """ try: with DatabaseContextManager() as ctx: results = [] for operation in operations: op_type = operation.get('type') speciality_id = operation.get('id') if op_type == 'activate': speciality = ctx.session.query(Speciality).filter(Speciality.id == speciality_id).first() if speciality: speciality.is_active = True speciality.updated_at = datetime.utcnow() results.append({ 'id': speciality_id, 'operation': 'activate', 'status': 'success', 'message': f"Speciality '{speciality.name}' activated" }) else: results.append({ 'id': speciality_id, 'operation': 'activate', 'status': 'error', 'message': 'Speciality not found' }) elif op_type == 'deactivate': speciality = ctx.session.query(Speciality).filter(Speciality.id == speciality_id).first() if speciality: speciality.is_active = False speciality.updated_at = datetime.utcnow() results.append({ 'id': speciality_id, 'operation': 'deactivate', 'status': 'success', 'message': f"Speciality '{speciality.name}' deactivated" }) else: results.append({ 'id': speciality_id, 'operation': 'deactivate', 'status': 'error', 'message': 'Speciality not found' }) ctx.commit() success_count = len([r for r in results if r['status'] == 'success']) error_count = len([r for r in results if r['status'] == 'error']) return custom_response( success=True, data={ 'results': results, 'summary': { 'total_operations': len(operations), 'successful': success_count, 'failed': error_count } }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error in bulk operations: {str(e)}") return custom_response( success=False, data="Failed to perform bulk operations", status_code=500 ) def get_statistics(self) -> Dict[str, Any]: """ Get comprehensive statistics for specialities with enhanced styling """ try: with DatabaseContextManager() as ctx: # Basic counts total_specialities = ctx.session.query(Speciality).count() active_specialities = ctx.session.query(Speciality).filter(Speciality.is_active == True).count() # Department statistics dept_stats = ctx.session.query( Speciality.department, func.count(Speciality.id).label('count'), func.count(Course.id).label('courses_count') ).outerjoin(Course, Speciality.id == Course.speciality_id).group_by(Speciality.department).all() # Course distribution course_distribution = ctx.session.query( Speciality.id, Speciality.name, func.count(Course.id).label('courses_count') ).outerjoin(Course, Speciality.id == Course.speciality_id).group_by(Speciality.id, Speciality.name).all() # Recent activity recent_specialities = ctx.session.query(Speciality).order_by( desc(Speciality.created_at) ).limit(5).all() # Enhanced statistics statistics = { 'overview': { 'total_specialities': total_specialities, 'active_specialities': active_specialities, 'inactive_specialities': total_specialities - active_specialities, 'total_departments': len(set(stat.department for stat in dept_stats)) }, 'department_breakdown': [ { 'department': stat.department, 'specialities_count': stat.count, 'courses_count': stat.courses_count or 0 } for stat in dept_stats ], 'top_specialities': [ { 'name': dist.name, 'courses_count': dist.courses_count or 0 } for dist in sorted(course_distribution, key=lambda x: x.courses_count or 0, reverse=True)[:10] ], 'recent_activity': [ { 'id': spec.id, 'name': spec.name, 'department': spec.department, 'created_at': spec.created_at.isoformat() } for spec in recent_specialities ] } return custom_response( success=True, data={'statistics': statistics}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error fetching speciality statistics: {str(e)}") return custom_response( success=False, data="Failed to fetch statistics", status_code=500 ) # Global instance speciality_manager = SpecialityManager()