from datetime import datetime, timedelta
from sqlalchemy import func, and_, or_, case
from src.models import DatabaseContextManager
from src.models.models import (
Attendance, TeachingSession, Student, Tutor, Supervisor,
AttendanceStatus, TeachingVerificationMethod,
NotificationPreference, Enrollment, UserType, User, Course
)
from flask import current_app, request
from src.utils import (
ApiABC,
custom_response,
send_email
)
import uuid
from typing import Dict, List
class AttendanceManager:
def __init__(self):
self.table = Attendance
def create(self, payload: Dict) -> Dict:
"""
Create attendance records for a teaching session.
Attendance can only be created after the session has ended.
Args:
payload: {
"session_id": str,
"attendances": [
{
"student_id": str,
"status": "present"|"absent"|"late"|"excused",
"late_minutes": int (optional, required if status is late),
"notes": str (optional),
"verification_method": "manual"|"geo_location"|"biometric"|"video_confirmation"|"qr_code"
}
]
}
Returns:
Response with created attendance records or error
"""
with DatabaseContextManager() as ctx:
# Verify the teaching session exists and has ended
session = ctx.session.query(TeachingSession).filter(
TeachingSession.id == payload['session_id']
).first()
if not session:
return custom_response(
success=False,
data="Teaching session not found",
status_code=404
)
# Check if session has ended
if session.end_time > datetime.utcnow():
return custom_response(
success=False,
data="Cannot record attendance before session ends",
status_code=400
)
# Check if attendance already exists for this session
existing_attendance = ctx.session.query(Attendance).filter(
Attendance.session_id == payload['session_id']
).first()
if existing_attendance:
return custom_response(
success=False,
data="Attendance already recorded for this session",
status_code=400
)
# Verify the current user is the tutor for this session
current_user = self.get_current_user()
if current_user.id != session.tutor_id and current_user.user_type != UserType.admin:
return custom_response(
success=False,
data="Only the session tutor or admin can record attendance",
status_code=403
)
# Process each attendance record
created_records = []
errors = []
for attendance_data in payload['attendances']:
# Validate student exists and is enrolled in the course
student = ctx.session.query(Student).filter(
Student.id == attendance_data['student_id']
).first()
if not student:
errors.append(f"Student {attendance_data['student_id']} not found")
continue
# Check if student is enrolled in the course
enrollment = ctx.session.query(Enrollment).filter(
Enrollment.student_id == attendance_data['student_id'],
Enrollment.course_id == session.course_id,
Enrollment.status == 'active'
).first()
if not enrollment:
errors.append(f"Student {student.first_name} {student.last_name} is not enrolled in this course")
continue
# Validate status
try:
status = AttendanceStatus(attendance_data['status'])
except ValueError:
errors.append(f"Invalid attendance status: {attendance_data['status']}")
continue
# Validate late minutes if status is late
if status == AttendanceStatus.late and 'late_minutes' not in attendance_data:
errors.append(f"late_minutes is required when status is 'late'")
continue
# Validate verification method
try:
verification_method = TeachingVerificationMethod(attendance_data['verification_method'])
except ValueError:
errors.append(f"Invalid verification method: {attendance_data['verification_method']}")
continue
# Create attendance record
attendance = Attendance(
id=str(uuid.uuid4()),
session_id=payload['session_id'],
student_id=attendance_data['student_id'],
tutor_id=session.tutor_id,
status=status,
timestamp=datetime.utcnow(),
notes=attendance_data.get('notes'),
late_minutes=attendance_data.get('late_minutes', 0),
verification_method=verification_method,
device_used=request.user_agent.string,
ip_address=request.remote_addr
)
ctx.session.add(attendance)
created_records.append({
'student_id': attendance_data['student_id'],
'status': status.value
})
if errors:
ctx.session.rollback()
return custom_response(
success=False,
data={
'message': 'Some attendance records could not be created',
'errors': errors,
'successful': created_records
},
status_code=400
)
# Update session status if not already completed
if session.status != 'completed':
session.status = 'completed'
session.tutor_checkout_time = datetime.utcnow()
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Attendance recorded successfully',
'records': created_records,
'session_id': payload['session_id'],
'course_id': session.course_id,
'total_present': len([r for r in created_records if r['status'] == 'present']),
'total_students': len(created_records)
},
status_code=201
)
def bulk_update(self, payload: Dict) -> Dict:
"""
Bulk update attendance records for a session.
Only allowed before attendance is approved by supervisor.
Args:
payload: {
"session_id": str,
"updates": [
{
"attendance_id": str,
"status": "present"|"absent"|"late"|"excused",
"late_minutes": int (optional),
"notes": str (optional)
}
]
}
Returns:
Response with updated records or error
"""
with DatabaseContextManager() as ctx:
# Verify the teaching session exists
session = ctx.session.query(TeachingSession).filter(
TeachingSession.id == payload['session_id']
).first()
if not session:
return custom_response(
success=False,
data="Teaching session not found",
status_code=404
)
# Check if attendance has been approved (no changes allowed after approval)
approved_attendance = ctx.session.query(Attendance).filter(
Attendance.session_id == payload['session_id'],
Attendance.approved_by.isnot(None)
).first()
if approved_attendance:
return custom_response(
success=False,
data="Cannot update attendance after it has been approved",
status_code=400
)
# Verify the current user is the tutor for this session or admin
current_user = self.get_current_user()
if current_user.id != session.tutor_id and current_user.user_type != UserType.admin:
return custom_response(
success=False,
data="Only the session tutor or admin can update attendance",
status_code=403
)
# Process each update
updated_records = []
errors = []
for update in payload['updates']:
attendance = ctx.session.query(Attendance).filter(
Attendance.id == update['attendance_id'],
Attendance.session_id == payload['session_id']
).first()
if not attendance:
errors.append(f"Attendance record {update['attendance_id']} not found")
continue
# Validate status
try:
status = AttendanceStatus(update['status'])
except ValueError:
errors.append(f"Invalid attendance status: {update['status']}")
continue
# Validate late minutes if status is late
if status == AttendanceStatus.late and 'late_minutes' not in update:
errors.append(f"late_minutes is required when status is 'late' for record {update['attendance_id']}")
continue
# Update record
attendance.status = status
attendance.late_minutes = update.get('late_minutes', 0)
attendance.notes = update.get('notes')
attendance.timestamp = datetime.utcnow() # Update timestamp to reflect change
updated_records.append({
'attendance_id': update['attendance_id'],
'status': status.value
})
if errors:
ctx.session.rollback()
return custom_response(
success=False,
data={
'message': 'Some attendance records could not be updated',
'errors': errors,
'successful': updated_records
},
status_code=400
)
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Attendance updated successfully',
'records': updated_records,
'session_id': payload['session_id']
},
status_code=200
)
def approve_attendance(self, session_id: str, payload: Dict) -> Dict:
"""
Approve attendance records for a session (by supervisor).
Args:
session_id: ID of the teaching session
payload: {
"approval_notes": str (optional),
"send_notification": bool (default True)
}
Returns:
Response with approval status
"""
with DatabaseContextManager() as ctx:
# Verify the teaching session exists
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
)
# Verify attendance records exist
attendance_records = ctx.session.query(Attendance).filter(
Attendance.session_id == session_id
).all()
if not attendance_records:
return custom_response(
success=False,
data="No attendance records found for this session",
status_code=404
)
# Check if already approved
if attendance_records[0].approved_by:
return custom_response(
success=False,
data="Attendance already approved",
status_code=400
)
# Verify current user is the course supervisor or admin
current_user = self.get_current_user()
if current_user.id != session.course.supervisor_id and current_user.user_type != UserType.admin:
return custom_response(
success=False,
data="Only the course supervisor or admin can approve attendance",
status_code=403
)
# Update all attendance records
approval_date = datetime.utcnow()
for record in attendance_records:
record.approved_by = current_user.id
record.approval_date = approval_date
record.approval_notes = payload.get('approval_notes')
record.status = AttendanceStatus.present if record.status == AttendanceStatus.pending_approval else record.status
# Update session verification status if not already verified
if not session.is_verified:
session.is_verified = True
session.verified_by = current_user.id
session.verified_at = approval_date
session.verification_method = TeachingVerificationMethod.manual
session.supervisor_notes = payload.get('approval_notes')
# Send notifications if requested
if payload.get('send_notification', True):
self._send_approval_notifications(ctx, session, attendance_records, current_user)
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Attendance approved successfully',
'session_id': session_id,
'approved_by': current_user.id,
'approval_date': approval_date.isoformat(),
'total_records': len(attendance_records),
'present_count': len([r for r in attendance_records if r.status == AttendanceStatus.present])
},
status_code=200
)
def _send_approval_notifications(self, ctx, session: TeachingSession,
attendance_records: List[Attendance],
approver: User) -> None:
"""
Send notifications about attendance approval to tutor and students.
Args:
ctx: Database context
session: TeachingSession object
attendance_records: List of Attendance records
approver: User who approved the attendance
"""
# Notification to tutor
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == session.tutor_id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = f"Attendance Approved: {session.title}"
message = f"""
Attendance Approval Notification
Hello {session.tutor.first_name},
Your attendance records for the following session have been approved by {approver.first_name} {approver.last_name}:
Course: {session.course.code} - {session.course.title}
Session: {session.title}
Date: {session.start_time.strftime('%A, %B %d, %Y')}
Approval Notes: {attendance_records[0].approval_notes or 'None'}
Attendance Summary:
- Present: {len([r for r in attendance_records if r.status == AttendanceStatus.present])}
- Absent: {len([r for r in attendance_records if r.status == AttendanceStatus.absent])}
- Late: {len([r for r in attendance_records if r.status == AttendanceStatus.late])}
- Excused: {len([r for r in attendance_records if r.status == AttendanceStatus.excused])}
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
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 attendance approval notification to tutor: {str(e)}")
# Notifications to students (only those who were marked absent/late)
for record in attendance_records:
if record.status in [AttendanceStatus.absent, AttendanceStatus.late]:
student_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == record.student_id
).first()
if student_prefs and student_prefs.receive_email:
subject = f"Attendance Record: {session.course.code} - {session.title}"
message = f"""
Attendance Record Notification
Hello {record.student.first_name},
Your attendance for the following session has been recorded:
Course: {session.course.code} - {session.course.title}
Session: {session.title}
Date: {session.start_time.strftime('%A, %B %d, %Y')}
Status: {record.status.value.upper()}
{f"
Late Minutes: {record.late_minutes}
" if record.status == AttendanceStatus.late else ""}
{f"
Notes: {record.notes}
" if record.notes else ""}
If you believe this is incorrect, please contact your tutor or course supervisor.
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=record.student.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send attendance notification to student {record.student_id}: {str(e)}")
def get_session_attendance(self, session_id: str) -> Dict:
"""
Get all attendance records for a specific teaching session.
Args:
session_id: ID of the teaching session
Returns:
Response with attendance records and summary
"""
with DatabaseContextManager() as ctx:
# Verify the teaching session exists
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
)
# Get attendance records
attendance_records = ctx.session.query(Attendance).filter(
Attendance.session_id == session_id
).all()
if not attendance_records:
return custom_response(
success=False,
data="No attendance records found for this session",
status_code=404
)
# Get enrollment count for comparison
enrolled_students = ctx.session.query(Enrollment).filter(
Enrollment.course_id == session.course_id,
Enrollment.status == 'active'
).count()
# Calculate summary statistics
status_counts = {
'present': 0,
'absent': 0,
'late': 0,
'excused': 0,
'pending_approval': 0,
'disputed': 0
}
for record in attendance_records:
status_counts[record.status.value] += 1
if record.is_disputed:
status_counts['disputed'] += 1
return custom_response(
success=True,
data={
'session': {
'id': session.id,
'title': session.title,
'course_id': session.course_id,
'course_code': session.course.code,
'start_datetime': session.start_time.isoformat(),
'end_datetime': session.end_time.isoformat(),
'is_verified': session.is_verified,
'verification_method': session.verification_method.value if session.verification_method else None
},
'attendance': [{
'id': record.id,
'student_id': record.student_id,
'student_name': f"{record.student.first_name} {record.student.last_name}",
'student_id_number': record.student.student_id,
'status': record.status.value,
'timestamp': record.timestamp.isoformat(),
'late_minutes': record.late_minutes,
'notes': record.notes,
'is_disputed': record.is_disputed,
'dispute_reason': record.dispute_reason,
'approved_by': record.approved_by,
'approval_date': record.approval_date.isoformat() if record.approval_date else None,
'verification_method': record.verification_method.value if record.verification_method else None
} for record in attendance_records],
'summary': {
'total_enrolled': enrolled_students,
'total_recorded': len(attendance_records),
'status_counts': status_counts,
'attendance_rate': round((status_counts['present'] / enrolled_students) * 100, 1) if enrolled_students else 0
}
},
status_code=200
)
def get_student_attendance(self, student_id: str, course_id: str = None) -> Dict:
"""
Get attendance records for a student, optionally filtered by course.
Args:
student_id: ID of the student
course_id: Optional course ID to filter by
Returns:
Response with attendance records and statistics
"""
with DatabaseContextManager() as ctx:
# 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
)
# Build query
query = ctx.session.query(Attendance).join(
TeachingSession,
Attendance.session_id == TeachingSession.id
).join(
Course,
TeachingSession.course_id == Course.id
).filter(
Attendance.student_id == student_id
)
if course_id:
query = query.filter(TeachingSession.course_id == course_id)
attendance_records = query.order_by(
TeachingSession.start_time.desc()
).all()
# Get all enrolled courses if no course filter
if not course_id:
enrolled_courses = ctx.session.query(Course).join(
Enrollment,
Course.id == Enrollment.course_id
).filter(
Enrollment.student_id == student_id,
Enrollment.status == 'active'
).all()
else:
enrolled_courses = ctx.session.query(Course).filter(
Course.id == course_id
).all()
# Calculate statistics
total_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id.in_([c.id for c in enrolled_courses])
).count()
present_count = len([r for r in attendance_records if r.status == AttendanceStatus.present])
attendance_rate = round((present_count / total_sessions) * 100, 1) if total_sessions else 0
# Group by course if no course filter
course_attendance = []
if not course_id:
for course in enrolled_courses:
course_records = [r for r in attendance_records if r.session.course_id == course.id]
course_present = len([r for r in course_records if r.status == AttendanceStatus.present])
course_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id == course.id
).count()
course_attendance.append({
'course_id': course.id,
'course_code': course.code,
'course_title': course.title,
'total_sessions': course_sessions,
'present_count': course_present,
'attendance_rate': round((course_present / course_sessions) * 100, 1) if course_sessions else 0
})
return custom_response(
success=True,
data={
'student': {
'id': student.id,
'name': f"{student.first_name} {student.last_name}",
'student_id': student.student_id,
'program': student.program,
'year_of_study': student.year_of_study
},
'attendance': [{
'id': record.id,
'session_id': record.session_id,
'session_title': record.session.title,
'course_id': record.session.course_id,
'course_code': record.session.course.code,
'date': record.session.start_time.date().isoformat(),
'status': record.status.value,
'late_minutes': record.late_minutes,
'notes': record.notes,
'is_disputed': record.is_disputed,
'approved': record.approved_by is not None
} for record in attendance_records],
'statistics': {
'total_sessions': total_sessions,
'present_count': present_count,
'attendance_rate': attendance_rate,
'courses': course_attendance
}
},
status_code=200
)
def get_course_attendance(self, course_id: str, page: int = 1, per_page: int = 20) -> Dict:
"""
Get attendance records for a course with pagination.
Args:
course_id: ID of the course
page: Page number (default 1)
per_page: Items per page (default 20)
Returns:
Response with paginated attendance records and summary
"""
with DatabaseContextManager() as ctx:
# Verify course exists
course = ctx.session.query(Course).filter(
Course.id == course_id
).first()
if not course:
return custom_response(
success=False,
data="Course not found",
status_code=404
)
# Get all sessions for the course
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id == course_id
).order_by(TeachingSession.start_time.desc()).all()
if not sessions:
return custom_response(
success=False,
data="No teaching sessions found for this course",
status_code=404
)
# Get all attendance records for these sessions with pagination
query = ctx.session.query(Attendance).join(
TeachingSession,
Attendance.session_id == TeachingSession.id
).filter(
TeachingSession.course_id == course_id
)
total_records = query.count()
attendance_records = query.order_by(
TeachingSession.start_time.desc(),
Attendance.timestamp.desc()
).offset((page - 1) * per_page).limit(per_page).all()
# Calculate summary statistics
status_counts = {
'present': 0,
'absent': 0,
'late': 0,
'excused': 0
}
for record in attendance_records:
status_counts[record.status.value] += 1
# Get enrolled student count
enrolled_students = ctx.session.query(Enrollment).filter(
Enrollment.course_id == course_id,
Enrollment.status == 'active'
).count()
return custom_response(
success=True,
data={
'course': {
'id': course.id,
'code': course.code,
'title': course.title,
'total_sessions': len(sessions),
'enrolled_students': enrolled_students
},
'attendance': [{
'id': record.id,
'session_id': record.session_id,
'session_title': record.session.title,
'session_date': record.session.start_time.date().isoformat(),
'student_id': record.student_id,
'student_name': f"{record.student.first_name} {record.student.last_name}",
'student_id_number': record.student.student_id,
'status': record.status.value,
'late_minutes': record.late_minutes,
'notes': record.notes,
'is_disputed': record.is_disputed,
'approved': record.approved_by is not None,
'verification_method': record.verification_method.value if record.verification_method else None
} for record in attendance_records],
'summary': {
'total_records': total_records,
'status_counts': status_counts,
'attendance_rate': round((status_counts['present'] / (total_records or 1)) * 100, 1)
},
'pagination': {
'page': page,
'per_page': per_page,
'total_pages': (total_records + per_page - 1) // per_page
}
},
status_code=200
)
def dispute_attendance(self, attendance_id: str, payload: Dict) -> Dict:
"""
Dispute an attendance record (by student or tutor).
Args:
attendance_id: ID of the attendance record
payload: {
"reason": str,
"requested_status": "present"|"absent"|"late"|"excused" (optional),
"late_minutes": int (optional, if requested_status is late)
}
Returns:
Response with dispute status
"""
with DatabaseContextManager() as ctx:
# Verify attendance record exists
attendance = ctx.session.query(Attendance).filter(
Attendance.id == attendance_id
).first()
if not attendance:
return custom_response(
success=False,
data="Attendance record not found",
status_code=404
)
# Verify current user is the student or the tutor
current_user = self.get_current_user()
if current_user.id not in [attendance.student_id, attendance.tutor_id]:
return custom_response(
success=False,
data="Only the student or tutor can dispute this attendance",
status_code=403
)
# Check if already disputed
if attendance.is_disputed:
return custom_response(
success=False,
data="Attendance record is already disputed",
status_code=400
)
# Check if attendance has been approved
if attendance.approved_by:
return custom_response(
success=False,
data="Cannot dispute attendance after it has been approved",
status_code=400
)
# Update record with dispute information
attendance.is_disputed = True
attendance.dispute_reason = payload['reason']
attendance.dispute_requested_status = payload.get('requested_status')
attendance.dispute_requested_late_minutes = payload.get('late_minutes')
attendance.dispute_raised_by = current_user.id
attendance.dispute_raised_at = datetime.utcnow()
# Notify supervisor
session = attendance.session
if session.course.supervisor:
supervisor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == session.course.supervisor_id
).first()
if supervisor_prefs and supervisor_prefs.receive_email:
subject = f"Attendance Dispute: {session.title}"
message = f"""
Attendance Dispute Notification
Hello {session.course.supervisor.first_name},
An attendance record has been disputed:
Course: {session.course.code} - {session.course.title}
Session: {session.title}
Date: {session.start_time.strftime('%A, %B %d, %Y')}
Student: {attendance.student.first_name} {attendance.student.last_name}
Current Status: {attendance.status.value.upper()}
Requested Status: {payload.get('requested_status', 'No change requested').upper()}
Reason: {payload['reason']}
Raised By: {current_user.first_name} {current_user.last_name}
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=session.course.supervisor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send dispute notification: {str(e)}")
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Attendance dispute submitted successfully',
'attendance_id': attendance_id,
'session_id': attendance.session_id,
'course_id': attendance.session.course_id,
'supervisor_notified': session.course.supervisor is not None
},
status_code=200
)
def resolve_dispute(self, attendance_id: str, payload: Dict) -> Dict:
"""
Resolve an attendance dispute (by supervisor or admin).
Args:
attendance_id: ID of the attendance record
payload: {
"resolution_status": "present"|"absent"|"late"|"excused",
"late_minutes": int (optional, if resolution_status is late),
"resolution_notes": str (optional),
"send_notification": bool (default True)
}
Returns:
Response with resolution status
"""
with DatabaseContextManager() as ctx:
# Verify attendance record exists and is disputed
attendance = ctx.session.query(Attendance).filter(
Attendance.id == attendance_id
).first()
if not attendance:
return custom_response(
success=False,
data="Attendance record not found",
status_code=404
)
if not attendance.is_disputed:
return custom_response(
success=False,
data="Attendance record is not disputed",
status_code=400
)
# Verify current user is the course supervisor or admin
current_user = self.get_current_user()
if current_user.id != attendance.session.course.supervisor_id and current_user.user_type != UserType.admin:
return custom_response(
success=False,
data="Only the course supervisor or admin can resolve disputes",
status_code=403
)
# Validate resolution status
try:
resolution_status = AttendanceStatus(payload['resolution_status'])
except ValueError:
return custom_response(
success=False,
data="Invalid resolution status",
status_code=400
)
# Validate late minutes if status is late
if resolution_status == AttendanceStatus.late and 'late_minutes' not in payload:
return custom_response(
success=False,
data="late_minutes is required when resolution_status is 'late'",
status_code=400
)
# Update record with resolution
attendance.status = resolution_status
attendance.late_minutes = payload.get('late_minutes', 0)
attendance.is_disputed = False
attendance.dispute_resolved_by = current_user.id
attendance.dispute_resolution_date = datetime.utcnow()
attendance.resolution_notes = payload.get('resolution_notes')
# If not already approved, mark as approved
if not attendance.approved_by:
attendance.approved_by = current_user.id
attendance.approval_date = datetime.utcnow()
# Send notifications if requested
if payload.get('send_notification', True):
self._send_dispute_resolution_notifications(ctx, attendance, current_user)
ctx.session.commit()
return custom_response(
success=True,
data={
'message': 'Attendance dispute resolved successfully',
'attendance_id': attendance_id,
'resolution_status': resolution_status.value,
'resolved_by': current_user.id,
'resolution_date': datetime.utcnow().isoformat()
},
status_code=200
)
def _send_dispute_resolution_notifications(self, ctx, attendance: Attendance, resolver: User) -> None:
"""
Send notifications about dispute resolution to student and tutor.
Args:
ctx: Database context
attendance: Attendance record
resolver: User who resolved the dispute
"""
# Notification to student
student_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == attendance.student_id
).first()
if student_prefs and student_prefs.receive_email:
subject = f"Attendance Dispute Resolved: {attendance.session.course.code}"
message = f"""
Attendance Dispute Resolution
Hello {attendance.student.first_name},
Your attendance dispute has been resolved:
Course: {attendance.session.course.code} - {attendance.session.course.title}
Session: {attendance.session.title}
Date: {attendance.session.start_time.strftime('%A, %B %d, %Y')}
Final Status: {attendance.status.value.upper()}
{f"
Late Minutes: {attendance.late_minutes}
" if attendance.status == AttendanceStatus.late else ""}
Resolution Notes: {attendance.resolution_notes or 'None provided'}
Resolved By: {resolver.first_name} {resolver.last_name}
If you have any further questions, please contact your course supervisor.
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
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)}")
# Notification to tutor
tutor_prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == attendance.tutor_id
).first()
if tutor_prefs and tutor_prefs.receive_email:
subject = f"Attendance Dispute Resolved: {attendance.session.course.code}"
message = f"""
Attendance Dispute Resolution
Hello {attendance.tutor.first_name},
An attendance dispute you were involved with has been resolved:
Course: {attendance.session.course.code} - {attendance.session.course.title}
Session: {attendance.session.title}
Date: {attendance.session.start_time.strftime('%A, %B %d, %Y')}
Student: {attendance.student.first_name} {attendance.student.last_name}
Final Status: {attendance.status.value.upper()}
{f"
Late Minutes: {attendance.late_minutes}
" if attendance.status == AttendanceStatus.late else ""}
Resolution Notes: {attendance.resolution_notes or 'None provided'}
Resolved By: {resolver.first_name} {resolver.last_name}
Best regards,
{current_app.config['APP_NAME']} Team
"""
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
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)}")
def get_supervisor_attendance_report(self, supervisor_id: str, start_date: str = None, end_date: str = None, course_id: str = None) -> Dict:
"""
Get comprehensive attendance report for a supervisor's students
Args:
supervisor_id: ID of the supervisor
start_date: Optional start date (ISO format)
end_date: Optional end date (ISO format)
course_id: Optional course ID to filter by
Returns:
Response with attendance report data
"""
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",
status_code=404
)
# Get all courses managed by this supervisor
courses_query = ctx.session.query(Course).filter(
Course.supervisor_id == supervisor_id,
Course.is_active == True
)
if course_id:
courses_query = courses_query.filter(Course.id == course_id)
courses = courses_query.all()
if not courses:
return custom_response(
success=False,
data="No courses found for this supervisor",
status_code=404
)
course_ids = [course.id for course in courses]
# Get all sessions for these courses
sessions_query = ctx.session.query(TeachingSession).filter(
TeachingSession.course_id.in_(course_ids)
)
# Apply date filters if provided
if start_date:
try:
start_date_obj = datetime.fromisoformat(start_date).date()
sessions_query = sessions_query.filter(TeachingSession.start_time >= start_date_obj)
except ValueError:
return custom_response(
success=False,
data="Invalid start_date format. Use YYYY-MM-DD",
status_code=400
)
if end_date:
try:
end_date_obj = datetime.fromisoformat(end_date).date()
sessions_query = sessions_query.filter(TeachingSession.start_time <= end_date_obj + timedelta(days=1))
except ValueError:
return custom_response(
success=False,
data="Invalid end_date format. Use YYYY-MM-DD",
status_code=400
)
sessions = sessions_query.order_by(TeachingSession.start_time).all()
if not sessions:
return custom_response(
success=False,
data="No sessions found for the specified period",
status_code=404
)
session_ids = [session.id for session in sessions]
# Get all attendance records for these sessions
attendance_records = ctx.session.query(Attendance).filter(
Attendance.session_id.in_(session_ids)
).all()
# Calculate summary statistics
total_attendance = len(attendance_records)
present_count = sum(1 for att in attendance_records if att.status == AttendanceStatus.present)
absent_count = sum(1 for att in attendance_records if att.status == AttendanceStatus.absent)
late_count = sum(1 for att in attendance_records if att.status == AttendanceStatus.late)
excused_count = sum(1 for att in attendance_records if att.status == AttendanceStatus.excused)
attendance_rate = round((present_count / total_attendance) * 100, 1) if total_attendance > 0 else 0
# Generate trend data (last 7 days)
trend_data = []
for i in range(7):
date_obj = datetime.now().date() - timedelta(days=i)
day_sessions = [s for s in sessions if s.start_time == date_obj]
if day_sessions:
day_session_ids = [s.id for s in day_sessions]
day_attendance = [att for att in attendance_records if att.session_id in day_session_ids]
day_present = sum(1 for att in day_attendance if att.status == AttendanceStatus.present)
day_total = len(day_attendance)
day_rate = round((day_present / day_total) * 100, 1) if day_total > 0 else 0
else:
day_rate = 0
trend_data.append({
'date': date_obj.isoformat(),
'rate': day_rate
})
trend_data.reverse()
# Generate distribution data
distribution_data = [
{'status': 'Present', 'value': present_count, 'color': '#10B981'},
{'status': 'Absent', 'value': absent_count, 'color': '#EF4444'},
{'status': 'Late', 'value': late_count, 'color': '#F59E0B'},
{'status': 'Excused', 'value': excused_count, 'color': '#6B7280'}
]
# Generate session details
session_details = []
for session in sessions:
session_attendance = [att for att in attendance_records if att.session_id == session.id]
session_present = sum(1 for att in session_attendance if att.status == AttendanceStatus.present)
session_absent = sum(1 for att in session_attendance if att.status == AttendanceStatus.absent)
session_late = sum(1 for att in session_attendance if att.status == AttendanceStatus.late)
session_details.append({
'date': session.start_time.isoformat(),
'course': session.course.code if session.course else 'N/A',
'title': session.title,
'present': session_present,
'absent': session_absent,
'late': session_late,
'total': len(session_attendance)
})
# Get student-wise attendance
student_attendance = {}
for attendance in attendance_records:
student_id = attendance.student_id
if student_id not in student_attendance:
student_attendance[student_id] = {
'total': 0,
'present': 0,
'absent': 0,
'late': 0,
'excused': 0
}
student_attendance[student_id]['total'] += 1
if attendance.status == AttendanceStatus.present:
student_attendance[student_id]['present'] += 1
elif attendance.status == AttendanceStatus.absent:
student_attendance[student_id]['absent'] += 1
elif attendance.status == AttendanceStatus.late:
student_attendance[student_id]['late'] += 1
elif attendance.status == AttendanceStatus.excused:
student_attendance[student_id]['excused'] += 1
# Convert student attendance to list with student details
student_details = []
for student_id, stats in student_attendance.items():
student = ctx.session.query(Student).filter(Student.id == student_id).first()
if student:
attendance_rate = round((stats['present'] / stats['total']) * 100, 1) if stats['total'] > 0 else 0
student_details.append({
'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,
'attendance_rate': attendance_rate,
'total_sessions': stats['total'],
'present': stats['present'],
'absent': stats['absent'],
'late': stats['late'],
'excused': stats['excused']
})
return custom_response(
success=True,
data={
'summary': {
'attendance_rate': attendance_rate,
'present': present_count,
'absent': absent_count,
'late': late_count,
'excused': excused_count,
'total': total_attendance
},
'trend': trend_data,
'distribution': distribution_data,
'sessions': session_details,
'students': student_details,
'courses': [{
'id': course.id,
'code': course.code,
'title': course.title,
'department': course.department
} for course in courses]
},
status_code=200
)