from datetime import datetime, time, date, timedelta
from sqlalchemy import and_, or_, func
from sqlalchemy.orm import aliased, joinedload
from src.models import DatabaseContextManager
from src.models.models import (
SupervisorAvailability, Tutor, Course, TeachingSession, TutorAvailability, TimetableBlock,
Timetable, Supervisor, NotificationPreference, supervisor_course_association,
tutor_course_association, SupervisorDepartment, Student, Speciality,
course_department_association, course_speciality_association
)
from flask import current_app, make_response, request
from src.utils import (
custom_response,
send_email
)
from enum import Enum
from typing import Dict, List, Optional
import uuid
from fpdf import FPDF
import random
import os
import json
from src.models.models import User
import io
import base64
from openpyxl import Workbook
from openpyxl.styles import Alignment
from openpyxl.utils import get_column_letter
from sqlalchemy.orm import aliased
class TimetableConflictType(Enum):
"""Types of timetable conflicts"""
OVERLAP = "overlap"
OUTSIDE_AVAILABILITY = "outside_availability"
EXCEEDS_MAX_HOURS = "exceeds_max_hours"
LEAVE_CONFLICT = "leave_conflict"
SESSION_CONFLICT = "session_conflict"
class TimetableBlockManager:
def __init__(self):
self.table = TimetableBlock
def create(self, payload: Dict) -> Dict:
"""Create a timetable block with conflict checking"""
with DatabaseContextManager() as ctx:
# Validate payload
required_fields = ['tutor_id', 'course_id', 'day_of_week', 'start_time', 'end_time', 'block_type']
if not all(field in payload for field in required_fields):
return custom_response(
success=False,
data=f"Missing required fields: {', '.join(required_fields)}",
status_code=400
)
# Parse times
try:
start_time = datetime.strptime(payload['start_time'], '%H:%M').time()
end_time = datetime.strptime(payload['end_time'], '%H:%M').time()
if end_time <= start_time:
return custom_response(
success=False,
data="End time must be after start time",
status_code=400
)
except ValueError as e:
return custom_response(
success=False,
data=f"Invalid time format: {str(e)}",
status_code=400
)
# Check for conflicts
conflict_check = self._check_block_conflicts(ctx, payload)
if conflict_check['has_conflicts']:
return custom_response(
success=False,
data={
'message': 'Timetable block conflicts detected',
'conflicts': conflict_check['conflicts']
},
status_code=409
)
# Create the block
block = TimetableBlock(
id=str(uuid.uuid4()),
tutor_id=payload['tutor_id'],
course_id=payload['course_id'],
timetable_id=payload.get('timetable_id'),
day_of_week=payload['day_of_week'],
start_time=start_time,
end_time=end_time,
room=payload.get('room'),
block_type=payload['block_type'],
recurring=payload.get('recurring', True),
start_date=datetime.strptime(payload['start_date'], '%Y-%m-%d').date() if payload.get('start_date') else None,
end_date=datetime.strptime(payload['end_date'], '%Y-%m-%d').date() if payload.get('end_date') else None,
created_at=datetime.utcnow(),
created_by=payload.get('created_by')
)
ctx.session.add(block)
ctx.session.flush()
# Create teaching sessions if this is a teaching block
if payload['block_type'] in ['lecture', 'tutorial', 'lab'] and payload.get('semester'):
session_result = TimetableGenerator(self)._create_teaching_sessions(
ctx,
block,
payload['semester'],
payload.get('created_by')
)
if not session_result['success']:
ctx.session.rollback()
return custom_response(
success=False,
data=f"Failed to create teaching sessions: {session_result['error']}",
status_code=400
)
ctx.session.commit()
block_dict = self._block_to_dict(block)
block_dict['sessions_created'] = session_result.get('sessions_created', 0) if payload.get('semester') else 0
return custom_response(
success=True,
data=block_dict,
status_code=201
)
def get(self, block_id: str) -> Dict:
"""Get a specific timetable block by ID"""
with DatabaseContextManager() as ctx:
block = ctx.session.query(TimetableBlock).filter(
TimetableBlock.id == block_id
).first()
if not block:
return custom_response(
success=False,
data="Timetable block not found",
status_code=404
)
return custom_response(
success=True,
data=self._block_to_dict(block),
status_code=200
)
def update(self, block_id: str, payload: Dict) -> Dict:
"""
Update a timetable block after validating against conflicts
Args:
block_id: ID of the block to update
payload: Update data (same structure as create but partial)
Returns:
Response with updated block or error
"""
with DatabaseContextManager() as ctx:
block = ctx.session.query(TimetableBlock).filter(
TimetableBlock.id == block_id
).first()
if not block:
return custom_response(
success=False,
data="Timetable block not found",
status_code=404
)
# Check if core fields are being changed
core_fields = ['day_of_week', 'start_time', 'end_time', 'recurring', 'start_date', 'end_date']
needs_validation = any(field in payload for field in core_fields)
# For time fields, parse first
time_updates = {}
if 'start_time' in payload:
try:
time_updates['start_time'] = datetime.strptime(payload['start_time'], '%H:%M').time()
except ValueError:
return custom_response(
success=False,
data="Invalid start_time format. Use HH:MM",
status_code=400
)
if 'end_time' in payload:
try:
time_updates['end_time'] = datetime.strptime(payload['end_time'], '%H:%M').time()
except ValueError:
return custom_response(
success=False,
data="Invalid end_time format. Use HH:MM",
status_code=400
)
# Check time validity if both are being updated
if 'start_time' in payload and 'end_time' in payload:
if time_updates['end_time'] <= time_updates['start_time']:
return custom_response(
success=False,
data="End time must be after start time",
status_code=400
)
# Check for conflicts if changing time/day
if needs_validation:
# Build test block with updated values
test_block = {
'tutor_id': block.tutor_id,
'day_of_week': payload.get('day_of_week', block.day_of_week),
'start_time': payload.get('start_time', block.start_time.strftime('%H:%M')),
'end_time': payload.get('end_time', block.end_time.strftime('%H:%M')),
'recurring': payload.get('recurring', block.recurring),
'start_date': payload.get('start_date', str(block.start_date) if block.start_date else None),
'end_date': payload.get('end_date', str(block.end_date) if block.end_date else None),
'block_type': payload.get('block_type', block.block_type),
'is_update': True # Flag to indicate this is an update operation
}
# For updates, we're more lenient with availability checks
conflict_check = self._check_block_conflicts(
ctx,
test_block,
exclude_block_id=block_id,
is_update=True
)
if conflict_check['has_conflicts']:
# Filter out minor conflicts for updates
critical_conflicts = [
c for c in conflict_check['conflicts']
if c.get('type') in ['overlap', 'leave_conflict'] # Only block critical conflicts
]
if critical_conflicts:
current_app.logger.warning(f"Critical conflicts detected for block {block_id}: {critical_conflicts}")
return custom_response(
success=False,
data={
'message': 'Critical timetable conflicts detected with proposed changes',
'conflicts': critical_conflicts
},
status_code=409
)
else:
# Log non-critical conflicts but allow the update
pass
# Apply updates
if 'day_of_week' in payload:
block.day_of_week = payload['day_of_week']
if 'start_time' in payload:
block.start_time = time_updates['start_time']
if 'end_time' in payload:
block.end_time = time_updates['end_time']
if 'recurring' in payload:
block.recurring = payload['recurring']
if 'start_date' in payload:
block.start_date = datetime.strptime(payload['start_date'], '%Y-%m-%d').date() if payload['start_date'] else None
if 'end_date' in payload:
block.end_date = datetime.strptime(payload['end_date'], '%Y-%m-%d').date() if payload['end_date'] else None
# Handle datetime fields from frontend (ISO format)
if 'start_datetime' in payload and payload['start_datetime']:
try:
start_dt = datetime.fromisoformat(payload['start_datetime'].replace('Z', '+00:00'))
block.start_time = start_dt.time()
# Extract day of week from datetime if not explicitly provided
if 'day_of_week' not in payload:
block.day_of_week = start_dt.weekday() # Monday=0, Sunday=6
except (ValueError, AttributeError):
pass # Ignore invalid datetime format
if 'end_datetime' in payload and payload['end_datetime']:
try:
end_dt = datetime.fromisoformat(payload['end_datetime'].replace('Z', '+00:00'))
block.end_time = end_dt.time()
except (ValueError, AttributeError):
pass # Ignore invalid datetime format
# Update optional fields
if 'course_id' in payload:
block.course_id = payload['course_id']
if 'timetable_id' in payload:
block.timetable_id = payload['timetable_id']
if 'room' in payload:
block.room = payload['room']
# Also propagate room change to associated teaching sessions
try:
sessions_q = ctx.session.query(TeachingSession).filter(
TeachingSession.timetable_id == block.timetable_id,
TeachingSession.day_of_week == block.day_of_week,
TeachingSession.start_time == (time_updates.get('start_time') or block.start_time),
TeachingSession.end_time == (time_updates.get('end_time') or block.end_time),
TeachingSession.course_id == block.course_id,
TeachingSession.session_type == block.block_type
)
for session in sessions_q.all():
session.room = payload['room']
session.updated_at = datetime.utcnow()
except Exception as sync_err:
current_app.logger.error(f"Failed to sync teaching sessions room for block {block.id}: {sync_err}")
if 'block_type' in payload:
block.block_type = payload['block_type']
if 'notes' in payload:
block.notes = payload['notes']
ctx.session.commit()
return custom_response(
success=True,
data=self._block_to_dict(block),
status_code=200
)
def delete(self, block_id: str) -> Dict:
"""Delete a timetable block"""
with DatabaseContextManager() as ctx:
block = ctx.session.query(TimetableBlock).filter(
TimetableBlock.id == block_id
).first()
if not block:
return custom_response(
success=False,
data="Timetable block not found",
status_code=404
)
ctx.session.delete(block)
ctx.session.commit()
return custom_response(
success=True,
data="Timetable block deleted successfully",
status_code=200
)
def get_blocks_for_tutor(self, tutor_id: str, week_start: date = None) -> Dict:
"""
Get timetable blocks for a tutor for a specific week
Args:
tutor_id: ID of the tutor
week_start: Optional start date of the week (defaults to current week)
Returns:
Response with weekly blocks
"""
with DatabaseContextManager() as ctx:
# Determine the week start date
if not week_start:
today = date.today()
week_start = today - timedelta(days=today.weekday())
week_end = week_start + timedelta(days=6)
# Get all blocks for this tutor in the week
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.tutor_id == tutor_id,
or_(
and_(
TimetableBlock.recurring == True,
# For recurring blocks, we just need to match the day of week
),
and_(
TimetableBlock.recurring == False,
TimetableBlock.start_date <= week_end,
TimetableBlock.end_date >= week_start
)
)
).order_by(
TimetableBlock.day_of_week,
TimetableBlock.start_time
).all()
# Organize by day
weekly_blocks = {day: [] for day in range(7)} # 0-6 for Monday-Sunday
for block in blocks:
if block.recurring:
# Add to the corresponding day of week
weekly_blocks[block.day_of_week].append(self._block_to_dict(block))
else:
# Check if this block falls within our week
if block.start_date <= week_end and block.end_date >= week_start:
# Calculate which day of the week this block's start date is
block_day = block.start_date.weekday()
weekly_blocks[block_day].append(self._block_to_dict(block))
# Format for response
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
formatted_blocks = []
for day_num in range(7):
current_date = week_start + timedelta(days=day_num)
formatted_blocks.append({
'day_name': days[day_num],
'date': str(current_date),
'blocks': weekly_blocks[day_num]
})
return custom_response(
success=True,
data={
'week_start': str(week_start),
'week_end': str(week_end),
'tutor_id': tutor_id,
'blocks': formatted_blocks
},
status_code=200
)
def generate_blocks_from_availability(self, tutor_id: str, course_id: str, timetable_id: str) -> Dict:
"""
Generate timetable blocks based on a tutor's approved availability slots
Args:
tutor_id: ID of the tutor
course_id: ID of the course
timetable_id: ID of the timetable
Returns:
Response with created blocks or error
"""
with DatabaseContextManager() as ctx:
# Get tutor's approved availabilities
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.is_approved == True,
TutorAvailability.availability_type == 'teaching'
).all()
if not availabilities:
return custom_response(
success=False,
data="No approved teaching availability slots found for this tutor",
status_code=404
)
created_blocks = []
for availability in availabilities:
# Create a block for each availability slot
block = TimetableBlock(
id=str(uuid.uuid4()),
tutor_id=tutor_id,
course_id=course_id,
timetable_id=timetable_id,
day_of_week=availability.day_of_week,
start_time=availability.start_time,
end_time=availability.end_time,
block_type='lecture', # Default type
recurring=availability.is_recurring,
start_date=availability.valid_from,
end_date=availability.valid_to,
created_at=datetime.utcnow()
)
ctx.session.add(block)
created_blocks.append(block)
ctx.session.commit()
return custom_response(
success=True,
data=[self._block_to_dict(block) for block in created_blocks],
status_code=201
)
def get_blocks_for_department(self, department: str, user_id: str) -> Dict:
"""
Get all timetable blocks for a specific department
Args:
department: Name of the department
user_id: ID of the user requesting the data
Returns:
Response with department blocks
"""
with DatabaseContextManager() as ctx:
# Get the user's department information
user = ctx.session.query(User).filter(User.id == user_id).first()
if not user:
return custom_response(
success=False,
data="User not found",
status_code=404
)
# Get department based on user type
user_department = None
if user.user_type.value == "tutor":
tutor = ctx.session.query(Tutor).filter(Tutor.id == user_id).first()
if tutor and tutor.departments:
# Get primary department
primary_dept = None
for dept in tutor.departments:
if hasattr(dept, 'is_primary'):
# Object format
if dept.is_primary:
primary_dept = dept
break
elif isinstance(dept, dict) and dept.get('is_primary', False):
# Dict format
primary_dept = dept
break
if primary_dept:
if hasattr(primary_dept, 'department_name'):
user_department = primary_dept.department_name
elif isinstance(primary_dept, dict):
user_department = primary_dept.get('department_name')
elif tutor.departments:
# Fallback to first department
first_dept = tutor.departments[0]
if hasattr(first_dept, 'department_name'):
user_department = first_dept.department_name
elif isinstance(first_dept, dict):
user_department = first_dept.get('department_name')
elif user.user_type.value == "supervisor":
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == user_id).first()
if supervisor and supervisor.departments:
# Get primary department
primary_dept = None
for dept in supervisor.departments:
if hasattr(dept, 'is_primary'):
# Object format
if dept.is_primary:
primary_dept = dept
break
elif isinstance(dept, dict) and dept.get('is_primary', False):
# Dict format
primary_dept = dept
break
if primary_dept:
if hasattr(primary_dept, 'department_name'):
user_department = primary_dept.department_name
elif isinstance(primary_dept, dict):
user_department = primary_dept.get('department_name')
elif supervisor.departments:
# Fallback to first department
first_dept = supervisor.departments[0]
if hasattr(first_dept, 'department_name'):
user_department = first_dept.department_name
elif isinstance(first_dept, dict):
user_department = first_dept.get('department_name')
elif user.user_type.value == "student":
student = ctx.session.query(Student).filter(Student.id == user_id).first()
if student:
user_department = student.program # Students might have program instead of department
if not user_department:
return custom_response(
success=False,
data="User department not found",
status_code=404
)
# Get all timetable blocks for courses in this department
# Join with Course to filter by course.department instead of timetable.department
blocks = ctx.session.query(TimetableBlock).join(
Timetable, TimetableBlock.timetable_id == Timetable.id
).join(
Course, TimetableBlock.course_id == Course.id
).filter(
and_(
Course.department == user_department,
Timetable.approval_status == 'approved',
Timetable.is_active == True
)
).order_by(
TimetableBlock.day_of_week,
TimetableBlock.start_time
).all()
current_app.logger.info(f"Found {len(blocks)} approved blocks for department '{user_department}'")
if not blocks:
return custom_response(
success=True,
data=[],
status_code=200
)
# Convert to dictionary format with additional data
block_dicts = []
for block in blocks:
block_dict = self._block_to_dict(block)
# Add course information
if block.course:
block_dict['course_title'] = block.course.title
block_dict['course_department'] = block.course.department
block_dict['course_code'] = block.course.code
# Add tutor information
if block.tutor:
block_dict['tutor_name'] = f"{block.tutor.first_name} {block.tutor.last_name}"
block_dict['tutor_staff_id'] = block.tutor.staff_id
block_dict['tutor_qualification'] = block.tutor.qualification
# Add supervisor tutor information if exists
if hasattr(block, 'supervisor_tutor_id') and block.supervisor_tutor_id:
supervisor_tutor = ctx.session.query(Supervisor).filter(
Supervisor.id == block.supervisor_tutor_id
).first()
if supervisor_tutor:
block_dict['supervisor_tutor_name'] = f"{supervisor_tutor.first_name} {supervisor_tutor.last_name}"
block_dict['supervisor_tutor_staff_id'] = supervisor_tutor.staff_id
# If no regular tutor, use supervisor tutor as the main tutor name
if not block.tutor:
block_dict['tutor_name'] = f"{supervisor_tutor.first_name} {supervisor_tutor.last_name}"
# Add replacement room information if exists
if hasattr(block, 'replacement_room') and block.replacement_room:
block_dict['replacement_room'] = block.replacement_room
# Add timetable information
if block.timetable:
block_dict['timetable_name'] = block.timetable.name
block_dict['timetable_status'] = block.timetable.status
block_dict['timetable_approval_status'] = block.timetable.approval_status
block_dicts.append(block_dict)
return custom_response(
success=True,
data=block_dicts,
status_code=200
)
def _block_to_dict(self, block: TimetableBlock) -> Dict:
"""Convert TimetableBlock model to dictionary"""
# Determine tutor name with fallback to supervisor tutor
tutor_name = None
with DatabaseContextManager() as ctx:
user = ctx.session.query(User).filter(
User.id == block.tutor_id
).first()
tutor_name = f"{user.first_name} {user.last_name}" if user else None
block_dict = {
'id': block.id,
'tutor_id': block.tutor_id,
'tutor_name': tutor_name,
'course_id': block.course_id,
'course_code': block.course.code if block.course else None,
'timetable_id': block.timetable_id,
'day_of_week': block.day_of_week,
'day_name': ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'][block.day_of_week],
'start_time': block.start_time.strftime('%H:%M'),
'end_time': block.end_time.strftime('%H:%M'),
'duration_minutes': (datetime.combine(date.today(), block.end_time) -
datetime.combine(date.today(), block.start_time)).seconds // 60,
'room': block.room,
'block_type': block.block_type,
'recurring': block.recurring,
'start_date': str(block.start_date) if block.start_date else None,
'end_date': str(block.end_date) if block.end_date else None,
'created_at': block.created_at.isoformat() if block.created_at else None,
'is_cancelled': block.is_cancelled,
'cancellation_reason': block.cancellation_reason
}
return block_dict
def _check_block_conflicts(self, ctx, block_data: Dict, exclude_block_id: str = None, is_update: bool = False) -> Dict:
"""
Check for conflicts with existing blocks, tutor availability, and other constraints
Args:
ctx: Database context
block_data: Block data to check
exclude_block_id: Optional ID of block to exclude from conflict checks
Returns:
Dictionary with conflict information
"""
conflicts = []
# Parse times
try:
start_time = datetime.strptime(block_data['start_time'], '%H:%M').time()
end_time = datetime.strptime(block_data['end_time'], '%H:%M').time()
day_of_week = block_data['day_of_week']
tutor_id = block_data['tutor_id']
if end_time <= start_time:
conflicts.append({
'type': TimetableConflictType.OVERLAP.value,
'message': 'End time must be after start time',
'day_of_week': day_of_week,
'start_time': block_data['start_time'],
'end_time': block_data['end_time']
})
return {'has_conflicts': True, 'conflicts': conflicts}
except ValueError as e:
return {'has_conflicts': True, 'conflicts': [{
'type': 'invalid_data',
'message': str(e),
'block': block_data
}]}
# Check tutor exists and is active
tutor = ctx.session.query(Tutor).filter(
Tutor.id == tutor_id,
Tutor.is_active == True
).first()
if not tutor:
current_app.logger.error(f"Tutor {tutor_id} not found or inactive")
return {'has_conflicts': True, 'conflicts': [{
'type': 'invalid_tutor',
'message': 'Tutor not found or inactive'
}]}
# Check if tutor is on leave
if tutor.is_on_leave and tutor.leave_start_date and tutor.leave_end_date:
if not block_data.get('recurring', True):
valid_from = datetime.strptime(block_data['start_date'], '%Y-%m-%d').date() if block_data.get('start_date') else None
valid_to = datetime.strptime(block_data['end_date'], '%Y-%m-%d').date() if block_data.get('end_date') else None
if valid_from and valid_to:
if not (valid_to < tutor.leave_start_date or valid_from > tutor.leave_end_date):
conflicts.append({
'type': TimetableConflictType.LEAVE_CONFLICT.value,
'message': 'Tutor is on leave during this period',
'day_of_week': day_of_week,
'start_time': block_data['start_time'],
'end_time': block_data['end_time'],
'leave_start': str(tutor.leave_start_date),
'leave_end': str(tutor.leave_end_date),
'valid_from': block_data.get('start_date'),
'valid_to': block_data.get('end_date')
})
# Check against tutor's approved availability (more lenient for updates)
if not is_update: # Only strict availability checking for new blocks
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.day_of_week == day_of_week,
TutorAvailability.is_approved == True
).all()
is_available = False
for avail in availabilities:
if avail.start_time <= start_time and avail.end_time >= end_time:
if avail.is_recurring or (
block_data.get('recurring', True) and
block_data.get('start_date') and
block_data.get('end_date') and
avail.valid_from <= datetime.strptime(block_data['start_date'], '%Y-%m-%d').date() and
avail.valid_to >= datetime.strptime(block_data['end_date'], '%Y-%m-%d').date()
):
is_available = True
break
if not is_available:
conflicts.append({
'type': TimetableConflictType.OUTSIDE_AVAILABILITY.value,
'message': 'Block falls outside tutor approved availability',
'day_of_week': day_of_week,
'start_time': block_data['start_time'],
'end_time': block_data['end_time']
})
else:
# For updates, just log availability info but don't block
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.day_of_week == day_of_week,
TutorAvailability.is_approved == True
).all()
# Check for overlapping with existing blocks
existing_blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.tutor_id == tutor_id,
TimetableBlock.day_of_week == day_of_week,
TimetableBlock.id != exclude_block_id if exclude_block_id else True
).all()
for existing in existing_blocks:
if not (end_time <= existing.start_time or start_time >= existing.end_time):
conflicts.append({
'type': TimetableConflictType.OVERLAP.value,
'message': f"Overlaps with existing block ({existing.start_time.strftime('%H:%M')}-{existing.end_time.strftime('%H:%M')})",
'day_of_week': day_of_week,
'start_time': block_data['start_time'],
'end_time': block_data['end_time'],
'conflicting_id': existing.id,
'conflicting_start': existing.start_time.strftime('%H:%M'),
'conflicting_end': existing.end_time.strftime('%H:%M')
})
# Check tutor's max teaching hours if this is a teaching block
if block_data.get('block_type', 'lecture') in ['lecture', 'tutorial', 'lab']:
if not is_update: # For new blocks, check strictly
# Calculate weekly teaching hours if this block is added
weekly_hours = self._calculate_weekly_teaching_hours(ctx, tutor_id, block_data)
if tutor.max_teaching_hours and weekly_hours > tutor.max_teaching_hours:
conflicts.append({
'type': TimetableConflictType.EXCEEDS_MAX_HOURS.value,
'message': f"Would exceed tutor's max teaching hours ({tutor.max_teaching_hours}h/week)",
'day_of_week': day_of_week,
'start_time': block_data['start_time'],
'end_time': block_data['end_time'],
'current_weekly_hours': weekly_hours - self._calculate_block_hours(block_data),
'projected_weekly_hours': weekly_hours,
'max_allowed': tutor.max_teaching_hours
})
else:
# For updates, just log the weekly hours but don't block the move
weekly_hours = self._calculate_weekly_teaching_hours(ctx, tutor_id, exclude_block_id=exclude_block_id)
result = {
'has_conflicts': len(conflicts) > 0,
'conflicts': conflicts
}
if conflicts:
pass
return result
def _calculate_weekly_teaching_hours(self, ctx, tutor_id: str, new_block: Dict = None, exclude_block_id: str = None) -> float:
"""
Calculate total weekly teaching hours for a tutor, optionally including a new block
Args:
ctx: Database context
tutor_id: ID of the tutor
new_block: Optional new block to include in calculation
Returns:
Total weekly teaching hours in hours
"""
# Get all teaching blocks for this tutor (excluding specified block if any)
query = ctx.session.query(TimetableBlock).filter(
TimetableBlock.tutor_id == tutor_id,
TimetableBlock.block_type.in_(['lecture', 'tutorial', 'lab'])
)
if exclude_block_id:
query = query.filter(TimetableBlock.id != exclude_block_id)
blocks = query.all()
# Convert to list of dicts for easier processing
block_dicts = [self._block_to_dict(block) for block in blocks]
# Add the new block if provided
if new_block:
block_dicts.append(new_block)
# Group by day of week and calculate total hours per day
day_hours = {day: 0.0 for day in range(7)}
for block in block_dicts:
duration = (datetime.combine(date.today(), datetime.strptime(block['end_time'], '%H:%M').time()) -
datetime.combine(date.today(), datetime.strptime(block['start_time'], '%H:%M').time()))
day_hours[block['day_of_week']] += duration.total_seconds() / 3600
# Sum up weekly hours
weekly_hours = sum(day_hours.values())
return weekly_hours
def _calculate_block_hours(self, block: Dict) -> float:
"""Calculate duration of a block in hours"""
start = datetime.strptime(block['start_time'], '%H:%M').time()
end = datetime.strptime(block['end_time'], '%H:%M').time()
duration = datetime.combine(date.today(), end) - datetime.combine(date.today(), start)
return duration.total_seconds() / 3600
class TimetableManager:
def __init__(self):
self.table = Timetable
self.block_manager = TimetableBlockManager()
def create(self, payload: Dict) -> Dict:
"""Create a new timetable"""
with DatabaseContextManager() as ctx:
# Validate supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == payload['created_by'],
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Create the timetable
timetable = Timetable(
id=str(uuid.uuid4()),
name=payload['name'],
description=payload.get('description'),
semester=payload['semester'],
academic_year=payload['academic_year'],
created_by=payload['created_by'],
created_at=datetime.utcnow(),
approval_status='draft'
)
ctx.session.add(timetable)
ctx.session.flush()
# Create blocks if provided
if 'blocks' in payload:
created_blocks = []
for block_data in payload['blocks']:
block_data['timetable_id'] = timetable.id
block_data['created_by'] = payload['created_by']
result = self.block_manager.create(block_data)
if not result['success']:
ctx.session.rollback()
return result
created_blocks.append(result['data'])
timetable.blocks_count = len(created_blocks)
ctx.session.commit()
return custom_response(
success=True,
data=self._timetable_to_dict(timetable),
status_code=201
)
def get(self, timetable_id: str, filters: Dict = None) -> Dict:
"""Get a specific timetable by ID with its blocks and optional filtering"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Get blocks for this timetable with optional filtering
query = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id
)
# Apply filters if provided
if filters:
if filters.get('speciality_id'):
query = query.join(Course).filter(Course.speciality_id == filters['speciality_id'])
if filters.get('year_level'):
query = query.join(Course).filter(Course.year_level == filters['year_level'])
if filters.get('term'):
query = query.join(Course).filter(Course.term == filters['term'])
if filters.get('academic_year'):
query = query.join(Course).filter(Course.academic_year == filters['academic_year'])
if filters.get('tutor_id'):
query = query.filter(TimetableBlock.tutor_id == filters['tutor_id'])
if filters.get('search_term'):
search_term = filters['search_term'].lower()
query = query.join(Course).filter(
or_(
Course.code.ilike(f'%{search_term}%'),
Course.title.ilike(f'%{search_term}%')
)
)
blocks = query.all()
timetable_data = self._timetable_to_dict(timetable)
timetable_data['blocks'] = [self.block_manager._block_to_dict(block) for block in blocks]
return custom_response(
success=True,
data=timetable_data,
status_code=200
)
def update(self, timetable_id: str, payload: Dict) -> Dict:
"""
Update a timetable
Args:
timetable_id: ID of the timetable
payload: Update data (partial)
Returns:
Response with updated timetable or error
"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Update fields
if 'name' in payload:
timetable.name = payload['name']
if 'description' in payload:
timetable.description = payload['description']
if 'semester' in payload:
timetable.semester = payload['semester']
if 'academic_year' in payload:
timetable.academic_year = payload['academic_year']
if 'approval_status' in payload:
timetable.approval_status = payload['approval_status']
if payload['approval_status'] == 'approved':
timetable.approved_by = payload.get('approved_by')
timetable.approved_at = datetime.utcnow()
ctx.session.commit()
return custom_response(
success=True,
data=self._timetable_to_dict(timetable),
status_code=200
)
def delete(self, timetable_id: str, requester_id: str) -> Dict:
"""Delete a timetable and all its associated data with email notifications"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Verify requester has permission to delete this timetable
requester = ctx.session.query(Supervisor).filter(
Supervisor.id == requester_id,
Supervisor.is_active == True
).first()
if not requester:
return custom_response(
success=False,
data="Requester not found or inactive",
status_code=404
)
# Check if requester is the creator or has admin privileges
if timetable.created_by != requester_id:
# Check if requester is admin (you can add admin role check here)
# For now, only allow creator to delete
return custom_response(
success=False,
data="Only the timetable creator can delete this timetable",
status_code=403
)
# Get all tutors who will be affected by this deletion
affected_tutors = ctx.session.query(Tutor).join(
TeachingSession,
Tutor.id == TeachingSession.tutor_id
).filter(
TeachingSession.timetable_id == timetable_id
).distinct().all()
# Get all supervisors who will be affected
affected_supervisors = ctx.session.query(Supervisor).join(
TeachingSession,
Supervisor.id == TeachingSession.supervisor_tutor_id
).filter(
TeachingSession.timetable_id == timetable_id
).distinct().all()
# Store timetable info for email notifications
timetable_info = {
'name': timetable.name,
'semester': timetable.semester,
'academic_year': timetable.academic_year,
'department': getattr(timetable, 'department', 'Unknown')
}
# Delete all teaching sessions first
teaching_sessions_deleted = ctx.session.query(TeachingSession).filter(
TeachingSession.timetable_id == timetable_id
).delete()
# Delete all timetable blocks
blocks_deleted = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id
).delete()
# Delete the timetable
ctx.session.delete(timetable)
ctx.session.commit()
# Send email notifications to affected tutors
self._notify_tutors_of_timetable_deletion(affected_tutors, timetable_info)
# Send email notifications to affected supervisors
self._notify_supervisors_of_timetable_deletion(affected_supervisors, timetable_info)
return custom_response(
success=True,
data=f"Timetable and all associated data deleted successfully. {teaching_sessions_deleted} teaching sessions and {blocks_deleted} blocks removed. Email notifications sent to {len(affected_tutors)} tutors and {len(affected_supervisors)} supervisors.",
status_code=200
)
def submit_for_approval(self, timetable_id: str, supervisor_id: str) -> Dict:
"""
Submit a timetable for supervisor approval
Args:
timetable_id: ID of the timetable
supervisor_id: ID of the approving supervisor
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Update status
timetable.approval_status = 'pending'
ctx.session.commit()
# Notify supervisor
self._notify_supervisor_of_pending_approval(timetable, supervisor)
return custom_response(
success=True,
data="Timetable submitted for approval successfully",
status_code=200
)
def approve(self, timetable_id: str, supervisor_id: str) -> Dict:
"""
Approve a timetable (by supervisor)
Args:
timetable_id: ID of the timetable
supervisor_id: ID of the approving supervisor
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Check if already approved
if timetable.approval_status == 'approved':
return custom_response(
success=False,
data="Timetable already approved",
status_code=400
)
# Check if timetable has any blocks
block_count = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id
).count()
if block_count == 0:
return custom_response(
success=False,
data="Cannot approve timetable: No timetable blocks found. Please add blocks before approving.",
status_code=400
)
# Approve the timetable
timetable.approval_status = 'approved'
timetable.approved_by = supervisor_id
timetable.approved_at = datetime.utcnow()
timetable.is_active = True
# Deactivate other timetables for the same semester/year
ctx.session.query(Timetable).filter(
Timetable.semester == timetable.semester,
Timetable.academic_year == timetable.academic_year,
Timetable.id != timetable_id
).update({'is_active': False})
# Create teaching sessions from timetable blocks
teaching_sessions = self._create_teaching_sessions_from_blocks(ctx, timetable_id)
# Check if any teaching sessions were created
if not teaching_sessions:
current_app.logger.warning(f"No teaching sessions created for timetable {timetable_id}")
# Still approve the timetable but log the warning
ctx.session.commit()
return custom_response(
success=True,
data="Timetable approved successfully. No teaching sessions were created (no timetable blocks found).",
status_code=200
)
ctx.session.commit()
# Notify relevant tutors
self._notify_all_tutors_of_approved_timetable(timetable)
return custom_response(
success=True,
data=f"Timetable approved successfully. {len(teaching_sessions)} teaching sessions created. All tutors have been notified via email.",
status_code=200
)
def reject(self, timetable_id: str, supervisor_id: str, reason: str) -> Dict:
"""
Reject a timetable (by supervisor)
Args:
timetable_id: ID of the timetable
supervisor_id: ID of the rejecting supervisor
reason: Reason for rejection
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Reject the timetable
timetable.approval_status = 'rejected'
timetable.approved_by = None
timetable.approved_at = None
timetable.rejection_reason = reason
ctx.session.commit()
# Notify creator
self._notify_creator_of_rejection(timetable, supervisor, reason)
return custom_response(
success=True,
data="Timetable rejected successfully",
status_code=200
)
def approve_and_verify(self, timetable_id: str, supervisor_id: str, notes: str = '') -> Dict:
"""
Approve and verify a timetable in one action
Args:
timetable_id: ID of the timetable
supervisor_id: ID of the approving supervisor
notes: Optional verification notes
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Check if already approved and verified
if timetable.approval_status == 'approved' and timetable.is_verified:
return custom_response(
success=False,
data="Timetable already approved and verified",
status_code=400
)
# Check if timetable has any blocks
block_count = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id
).count()
if block_count == 0:
return custom_response(
success=False,
data="Cannot approve timetable with no blocks",
status_code=400
)
try:
# Step 1: Approve the timetable
timetable.approval_status = 'approved'
timetable.approved_at = datetime.utcnow()
timetable.approved_by = supervisor_id
timetable.updated_at = datetime.utcnow()
# Step 2: Verify the timetable
timetable.is_verified = True
timetable.verified_at = datetime.utcnow()
timetable.verified_by = supervisor_id
timetable.verification_notes = notes
# Step 3: Create teaching sessions from timetable blocks
teaching_sessions = self._create_teaching_sessions_from_blocks(ctx, timetable_id)
if not teaching_sessions:
ctx.session.rollback()
return custom_response(
success=False,
data="Failed to create teaching sessions",
status_code=500
)
# Step 4: Send email alerts to tutors with their teaching schedules
email_result = self._send_timetable_verification_emails(ctx, timetable, teaching_sessions)
ctx.session.commit()
return custom_response(
success=True,
data={
"message": "Timetable approved and verified successfully",
"teaching_sessions_created": len(teaching_sessions),
"emails_sent": email_result.get('emails_sent', 0),
"email_summary": email_result.get('summary', {})
},
status_code=200
)
except Exception as e:
ctx.session.rollback()
current_app.logger.error(f"Error approving and verifying timetable: {e}")
return custom_response(
success=False,
data=f"Failed to approve and verify timetable: {str(e)}",
status_code=500
)
def disapprove(self, timetable_id: str, supervisor_id: str) -> Dict:
"""
Disapprove a timetable (cancel teaching sessions and notify tutors)
Args:
timetable_id: ID of the timetable
supervisor_id: ID of the disapproving supervisor
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Check if timetable is approved
if timetable.approval_status != 'approved':
return custom_response(
success=False,
data="Timetable is not approved, cannot disapprove",
status_code=400
)
# Cancel all teaching sessions associated with this timetable
cancelled_sessions = self._cancel_teaching_sessions_for_timetable(ctx, timetable_id)
# Update timetable status
timetable.approval_status = 'rejected'
timetable.approved_by = None
timetable.approved_at = None
timetable.is_active = False
timetable.rejection_reason = "Disapproved by supervisor"
ctx.session.commit()
# Notify tutors about cancelled sessions
self._notify_tutors_of_disapproved_timetable(timetable, cancelled_sessions)
return custom_response(
success=True,
data=f"Timetable disapproved successfully. {len(cancelled_sessions)} teaching sessions cancelled. All tutors have been notified.",
status_code=200
)
def get_active_timetable(self, semester: str = None, academic_year: str = None) -> Dict:
"""
Get the currently active timetable
Args:
semester: Optional semester filter
academic_year: Optional academic year filter
Returns:
Response with timetable or error
"""
with DatabaseContextManager() as ctx:
query = ctx.session.query(Timetable).filter(
Timetable.is_active == True
)
if semester:
query = query.filter(Timetable.semester == semester)
if academic_year:
query = query.filter(Timetable.academic_year == academic_year)
timetable = query.first()
if not timetable:
return custom_response(
success=False,
data="No active timetable found",
status_code=404
)
# Get all blocks for this timetable
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).all()
timetable_data = self._timetable_to_dict(timetable)
timetable_data['blocks'] = [self.block_manager._block_to_dict(block) for block in blocks]
return custom_response(
success=True,
data=timetable_data,
status_code=200
)
def get_timetables_for_supervisor(self, supervisor_id: str) -> Dict:
"""
Get all timetables created by or pending approval from a supervisor
Args:
supervisor_id: ID of the supervisor
Returns:
Response with list of timetables
"""
with DatabaseContextManager() as ctx:
# Verify supervisor exists
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == supervisor_id,
Supervisor.is_active == True
).first()
if not supervisor:
return custom_response(
success=False,
data="Supervisor not found or inactive",
status_code=404
)
# Get timetables created by this supervisor
created_timetables = ctx.session.query(Timetable).filter(
Timetable.created_by == supervisor_id
).order_by(Timetable.created_at.desc()).all()
# Get timetables pending this supervisor's approval
pending_timetables = ctx.session.query(Timetable).filter(
Timetable.approval_status == 'pending',
).order_by(Timetable.created_at.desc()).all()
# Combine and format timetables
all_timetables = []
for timetable in created_timetables + pending_timetables:
# Calculate statistics
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).all()
total_courses = len(set(block.course_id for block in blocks if block.course_id))
total_tutors = len(set(block.tutor_id for block in blocks if block.tutor_id))
total_students = 0 # This would need to be calculated from enrollments
total_hours = sum(
(datetime.combine(datetime.min.date(), block.end_time) -
datetime.combine(datetime.min.date(), block.start_time)).total_seconds() / 3600
for block in blocks if block.start_time and block.end_time
)
room_assignments = len(set(block.room for block in blocks if block.room))
conflict_count = 0 # This would need conflict detection logic
completion_rate = min(100.0, (len(blocks) / max(1, total_courses)) * 100) if total_courses > 0 else 0.0
# Get department and speciality from the first course if available
department = "General"
speciality = "General"
if blocks:
first_course = ctx.session.query(Course).filter(Course.id == blocks[0].course_id).first()
if first_course:
department = first_course.department or "General"
speciality = first_course.speciality.name if first_course.speciality else "General"
# Convert blocks to dictionary format
blocks_data = [self.block_manager._block_to_dict(block) for block in blocks]
timetable_data = {
'id': timetable.id,
'name': timetable.name,
'description': timetable.description,
'academic_year': timetable.academic_year,
'term': getattr(timetable, 'term', 'Term 1'),
'semester': timetable.semester,
'is_active': timetable.is_active,
'created_at': timetable.created_at.isoformat() if timetable.created_at else None,
'updated_at': getattr(timetable, 'updated_at', timetable.created_at).isoformat() if getattr(timetable, 'updated_at', timetable.created_at) else None,
'supervisor_id': timetable.created_by,
'supervisor_name': f"{timetable.creator.first_name} {timetable.creator.last_name}" if timetable.creator else "Unknown",
'total_courses': total_courses,
'total_students': total_students,
'total_tutors': total_tutors,
'total_hours': total_hours,
'status': getattr(timetable, 'status', 'draft'),
'approval_status': timetable.approval_status,
'last_modified': getattr(timetable, 'updated_at', timetable.created_at).isoformat() if getattr(timetable, 'updated_at', timetable.created_at) else None,
'version': getattr(timetable, 'version', '1.0'),
'department': department,
'speciality': speciality,
'room_assignments': room_assignments,
'conflict_count': conflict_count,
'completion_rate': completion_rate,
'blocks': blocks_data,
'blocks_count': len(blocks)
}
all_timetables.append(timetable_data)
return custom_response(
success=True,
data={
'timetables': all_timetables
},
status_code=200
)
def _timetable_to_dict(self, timetable: Timetable) -> Dict:
"""Convert Timetable model to dictionary"""
with DatabaseContextManager() as ctx:
return {
'id': timetable.id,
'name': timetable.name,
'description': timetable.description,
'semester': timetable.semester,
'academic_year': timetable.academic_year,
'created_by': timetable.created_by,
'creator_name': f"{timetable.creator.first_name} {timetable.creator.last_name}" if timetable.creator else None,
'created_at': timetable.created_at.isoformat() if timetable.created_at else None,
'approved_by': timetable.approved_by,
'approver_name': f"{timetable.approver.first_name} {timetable.approver.last_name}" if timetable.approver else None,
'approved_at': timetable.approved_at.isoformat() if timetable.approved_at else None,
'approval_status': timetable.approval_status,
'is_active': timetable.is_active,
'blocks_count': ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).count() if hasattr(timetable, 'blocks_count') else None
}
def _cancel_teaching_sessions_for_timetable(self, ctx, timetable_id: str) -> List[Dict]:
"""
Cancel all teaching sessions associated with a timetable
Args:
ctx: Database context
timetable_id: ID of the timetable
Returns:
List of cancelled session information
"""
try:
# Get all teaching sessions for this timetable
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.timetable_id == timetable_id,
TeachingSession.status.in_(['scheduled', 'ongoing'])
).all()
cancelled_sessions = []
for session in sessions:
# Update session status to cancelled
session.status = 'cancelled'
session.updated_at = datetime.utcnow()
session.notes = f"Cancelled due to timetable disapproval on {datetime.utcnow().strftime('%Y-%m-%d %H:%M')}"
# Store session info for notification
cancelled_sessions.append({
'id': session.id,
'tutor_id': session.tutor_id,
'course_id': session.course_id,
'title': session.title,
'day_of_week': session.day_of_week,
'start_time': session.start_time.strftime('%H:%M'),
'end_time': session.end_time.strftime('%H:%M'),
'room': session.room,
'session_type': session.session_type
})
current_app.logger.info(f"Cancelled {len(cancelled_sessions)} teaching sessions for timetable {timetable_id}")
return cancelled_sessions
except Exception as e:
current_app.logger.error(f"Error cancelling teaching sessions for timetable {timetable_id}: {str(e)}", exc_info=True)
return []
def _notify_supervisor_of_pending_approval(self, timetable: Timetable, supervisor: Supervisor) -> None:
"""
Send notification to supervisor about pending timetable approval
Args:
timetable: Timetable model instance
supervisor: Supervisor model instance
"""
if not supervisor.notification_preference or not supervisor.notification_preference.receive_email:
return
# Prepare email
subject = f"Timetable Pending Approval: {timetable.name}"
with DatabaseContextManager() as ctx:
# Get timetable stats
blocks_count = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).count()
tutors_count = ctx.session.query(TimetableBlock.tutor_id).filter(
TimetableBlock.timetable_id == timetable.id
).distinct().count()
courses_count = ctx.session.query(TimetableBlock.course_id).filter(
TimetableBlock.timetable_id == timetable.id
).distinct().count()
# Build message
message = f"""
Timetable Approval Request
Hello {supervisor.first_name},
A new timetable has been submitted for your approval:
Timetable Name: {timetable.name}
Semester: {timetable.semester}
Academic Year: {timetable.academic_year}
Blocks: {blocks_count}
Tutors: {tutors_count}
Courses: {courses_count}
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=supervisor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable approval notification: {str(e)}")
def _notify_tutors_of_approved_timetable(self, timetable: Timetable) -> None:
"""
Send notifications to tutors about their assigned blocks in an approved timetable
Args:
timetable: Timetable model instance
"""
with DatabaseContextManager() as ctx:
# Get all tutors with blocks in this timetable
tutors = ctx.session.query(Tutor).join(
TimetableBlock,
Tutor.id == TimetableBlock.tutor_id
).filter(
TimetableBlock.timetable_id == timetable.id
).distinct().all()
for tutor in tutors:
# Get tutor's blocks in this timetable
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id,
TimetableBlock.tutor_id == tutor.id
).all()
# Check notification preferences
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if not prefs or not prefs.receive_email:
continue
# Prepare email
subject = f"Your Teaching Schedule for {timetable.semester} {timetable.academic_year}"
# Group blocks by day
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
grouped_blocks = {day: [] for day in days}
for block in blocks:
day_name = days[block.day_of_week]
grouped_blocks[day_name].append({
'time': f"{block.start_time.strftime('%H:%M')} - {block.end_time.strftime('%H:%M')}",
'course': block.course.code if block.course else 'N/A',
'type': block.block_type,
'room': block.room or 'TBD'
})
def build_schedule_html(grouped_blocks):
schedule_html = []
for day, blocks in grouped_blocks.items():
if blocks:
schedule_html.append(f"{day}
")
schedule_html.append("")
for block in blocks:
schedule_html.append(
f'- {block["time"]}: {block["course"]} ({block["type"]}) - {block["room"]}
'
)
schedule_html.append("
")
return "".join(schedule_html)
schedule_content = build_schedule_html(grouped_blocks)
# Build message
message = f"""
Teaching Schedule Notification
Hello {tutor.first_name},
Your teaching schedule for {timetable.semester} {timetable.academic_year} has been approved:
{schedule_content}
Please review your schedule and contact your supervisor if you have any questions.
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=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable notification to tutor {tutor.id}: {str(e)}")
def send_timetable_emails_to_tutors(self, timetable_id: str, tutor_ids: list = None, requester_id: str = None) -> Dict:
"""
Manually send timetable emails to selected tutors
Args:
timetable_id: ID of the timetable
tutor_ids: List of tutor IDs to send emails to (if None, sends to all tutors)
requester_id: ID of the supervisor requesting the email send
Returns:
Response with success/error message
"""
with DatabaseContextManager() as ctx:
try:
# Get the timetable
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Verify requester is a supervisor
requester = ctx.session.query(Supervisor).filter(
Supervisor.id == requester_id
).first()
if not requester:
return custom_response(
success=False,
data="Unauthorized - Only supervisors can send timetable emails",
status_code=403
)
# Get tutors to notify
if tutor_ids:
# Send to specific tutors
tutors_to_notify = ctx.session.query(Tutor).filter(
Tutor.id.in_(tutor_ids),
Tutor.is_active == True
).all()
else:
# Send to all tutors in the timetable
tutors_to_notify = self._get_tutors_for_timetable(ctx, timetable_id)
if not tutors_to_notify:
return custom_response(
success=False,
data="No tutors found to notify",
status_code=404
)
# Send emails
success_count = 0
failed_count = 0
failed_tutors = []
for tutor in tutors_to_notify:
try:
# Get tutor's blocks and sessions for this timetable
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id,
TimetableBlock.tutor_id == tutor.id
).all()
teaching_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.timetable_id == timetable_id,
TeachingSession.tutor_id == tutor.id
).all()
# Get tutor's assigned courses in this timetable
assigned_courses = ctx.session.query(Course).join(
tutor_course_association,
Course.id == tutor_course_association.c.course_id
).join(
TimetableBlock,
Course.id == TimetableBlock.course_id
).filter(
TimetableBlock.timetable_id == timetable_id,
tutor_course_association.c.tutor_id == tutor.id
).all()
# Check notification preferences
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if prefs and not prefs.receive_email:
continue
# Send email
success = self._send_tutor_timetable_email(tutor, timetable, blocks, teaching_sessions, assigned_courses)
if success:
success_count += 1
else:
failed_count += 1
failed_tutors.append(f"{tutor.first_name} {tutor.last_name}")
except Exception as e:
current_app.logger.error(f"Failed to send email to tutor {tutor.id}: {str(e)}")
failed_count += 1
failed_tutors.append(f"{tutor.first_name} {tutor.last_name}")
# Prepare response message
if failed_count == 0:
message = f"Successfully sent timetable emails to {success_count} tutor(s)"
else:
message = f"Sent emails to {success_count} tutor(s), failed to send to {failed_count} tutor(s)"
if failed_tutors:
message += f". Failed tutors: {', '.join(failed_tutors)}"
return custom_response(
success=True,
data={
"message": message,
"success_count": success_count,
"failed_count": failed_count,
"failed_tutors": failed_tutors
},
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error sending timetable emails: {str(e)}", exc_info=True)
return custom_response(
success=False,
data=f"Failed to send timetable emails: {str(e)}",
status_code=500
)
def _get_tutors_for_timetable(self, ctx, timetable_id: str) -> list:
"""Get all tutors associated with a timetable"""
tutors = set()
# Get tutors with blocks in this timetable
tutors_with_blocks = ctx.session.query(Tutor).join(
TimetableBlock,
Tutor.id == TimetableBlock.tutor_id
).filter(
TimetableBlock.timetable_id == timetable_id
).distinct().all()
tutors.update(tutors_with_blocks)
# Get tutors assigned to courses in this timetable
courses = ctx.session.query(Course).join(
TimetableBlock,
Course.id == TimetableBlock.course_id
).filter(
TimetableBlock.timetable_id == timetable_id
).distinct().all()
for course in courses:
course_tutors = ctx.session.query(Tutor).join(
tutor_course_association,
Tutor.id == tutor_course_association.c.tutor_id
).filter(
tutor_course_association.c.course_id == course.id,
Tutor.is_active == True
).all()
tutors.update(course_tutors)
return list(tutors)
def _send_tutor_timetable_email(self, tutor: Tutor, timetable: Timetable, blocks: list, teaching_sessions: list, assigned_courses: list) -> bool:
"""Send timetable email to a specific tutor"""
try:
# Prepare email
subject = f"Timetable Update: {timetable.semester} {timetable.academic_year}"
# Build course list
course_list = []
for course in assigned_courses:
course_blocks = [b for b in blocks if b.course_id == course.id]
course_sessions = [s for s in teaching_sessions if s.course_id == course.id]
if course_blocks:
course_list.append(f"{course.code}: {course.title} - {len(course_blocks)} block(s) assigned, {len(course_sessions)} teaching session(s) created")
else:
course_list.append(f"{course.code}: {course.title} - No specific blocks assigned yet")
course_content = "".join(course_list) if course_list else "No specific courses assigned yet"
# Group blocks by day for schedule
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
grouped_blocks = {day: [] for day in days}
for block in blocks:
day_name = days[block.day_of_week]
grouped_blocks[day_name].append({
'time': f"{block.start_time.strftime('%H:%M')} - {block.end_time.strftime('%H:%M')}",
'course': block.course.code if block.course else 'N/A',
'type': block.block_type,
'room': block.room or 'TBD'
})
def build_schedule_html(grouped_blocks):
schedule_html = []
for day, blocks in grouped_blocks.items():
if blocks:
schedule_html.append(f"{day}
")
schedule_html.append("")
for block in blocks:
schedule_html.append(
f'- {block["time"]}: {block["course"]} ({block["type"]}) - {block["room"]}
'
)
schedule_html.append("
")
return "".join(schedule_html)
schedule_content = build_schedule_html(grouped_blocks)
has_schedule = any(len(blocks) > 0 for blocks in grouped_blocks.values())
# Build message
message = f"""
📅 Timetable Update
{timetable.semester} {timetable.academic_year}
Hello {tutor.first_name},
Your timetable for {timetable.semester} {timetable.academic_year} has been updated.
Teaching sessions have been automatically created for your assigned courses.
📚 Your Assigned Courses
{f'''
📋 Your Teaching Schedule
{schedule_content}
''' if has_schedule else '''
Your specific teaching schedule will be assigned soon. Please check the timetable portal for updates.
'''}
{f'''
✅ Teaching Sessions Created
{len(teaching_sessions)} teaching session(s) have been automatically created for you. You can now:
- Track student attendance for each session
- Record student assessments and progress
- Manage session materials and resources
- Communicate with students about session updates
''' if teaching_sessions else '''
Teaching sessions will be created as your schedule is finalized.
'''}
Please review the timetable and contact your supervisor if you have any questions or concerns.
Best regards,
{current_app.config['APP_NAME']} Team
"""
# Send email
success, message = send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=subject,
message=message
)
if success:
return True
else:
return False
except Exception as e:
current_app.logger.error(f"Error sending timetable email to tutor {tutor.id}: {str(e)}")
return False
def _notify_all_tutors_of_approved_timetable(self, timetable: Timetable) -> None:
"""
Send notifications to all tutors assigned to courses in the approved timetable
Args:
timetable: Timetable model instance
"""
with DatabaseContextManager() as ctx:
# Get all courses in this timetable
courses = ctx.session.query(Course).join(
TimetableBlock,
Course.id == TimetableBlock.course_id
).filter(
TimetableBlock.timetable_id == timetable.id
).distinct().all()
# Get all tutors assigned to these courses (from tutor_course_association)
all_tutors = set()
# Get tutors with blocks in this timetable
tutors_with_blocks = ctx.session.query(Tutor).join(
TimetableBlock,
Tutor.id == TimetableBlock.tutor_id
).filter(
TimetableBlock.timetable_id == timetable.id
).distinct().all()
all_tutors.update(tutors_with_blocks)
# Get all tutors assigned to courses in this timetable (from tutor_course_association)
for course in courses:
course_tutors = ctx.session.query(Tutor).join(
tutor_course_association,
Tutor.id == tutor_course_association.c.tutor_id
).filter(
tutor_course_association.c.course_id == course.id,
Tutor.is_active == True
).all()
all_tutors.update(course_tutors)
# Convert set to list for processing
tutors_to_notify = list(all_tutors)
for tutor in tutors_to_notify:
# Get tutor's blocks in this timetable (if any)
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id,
TimetableBlock.tutor_id == tutor.id
).all()
# Get tutor's teaching sessions in this timetable
teaching_sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.timetable_id == timetable.id,
TeachingSession.tutor_id == tutor.id
).all()
# Get tutor's assigned courses in this timetable
assigned_courses = ctx.session.query(Course).join(
tutor_course_association,
Course.id == tutor_course_association.c.course_id
).join(
TimetableBlock,
Course.id == TimetableBlock.course_id
).filter(
TimetableBlock.timetable_id == timetable.id,
tutor_course_association.c.tutor_id == tutor.id
).all()
# Check notification preferences
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if not prefs or not prefs.receive_email:
continue
# Prepare email
subject = f"Timetable Approved: {timetable.semester} {timetable.academic_year}"
# Build course list
course_list = []
for course in assigned_courses:
course_blocks = [b for b in blocks if b.course_id == course.id]
course_sessions = [s for s in teaching_sessions if s.course_id == course.id]
if course_blocks:
# Tutor has specific blocks for this course
course_list.append(f"{course.code}: {course.title} - {len(course_blocks)} block(s) assigned, {len(course_sessions)} teaching session(s) created")
else:
# Tutor is assigned to course but no specific blocks yet
course_list.append(f"{course.code}: {course.title} - No specific blocks assigned yet")
course_content = "".join(course_list) if course_list else "No specific courses assigned yet"
# Group blocks by day for schedule
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
grouped_blocks = {day: [] for day in days}
for block in blocks:
day_name = days[block.day_of_week]
grouped_blocks[day_name].append({
'time': f"{block.start_time.strftime('%H:%M')} - {block.end_time.strftime('%H:%M')}",
'course': block.course.code if block.course else 'N/A',
'type': block.block_type,
'room': block.room or 'TBD'
})
def build_schedule_html(grouped_blocks):
schedule_html = []
for day, blocks in grouped_blocks.items():
if blocks:
schedule_html.append(f"{day}
")
schedule_html.append("")
for block in blocks:
schedule_html.append(
f'- {block["time"]}: {block["course"]} ({block["type"]}) - {block["room"]}
'
)
schedule_html.append("
")
return "".join(schedule_html)
schedule_content = build_schedule_html(grouped_blocks)
has_schedule = any(len(blocks) > 0 for blocks in grouped_blocks.values())
# Build message
message = f"""
Timetable Approval Notification
Hello {tutor.first_name},
The timetable for {timetable.semester} {timetable.academic_year} has been approved by your supervisor.
Teaching sessions have been automatically created for your assigned courses.
Your Assigned Courses:
{f'''
Your Teaching Schedule:
{schedule_content}
''' if has_schedule else '''
Your specific teaching schedule will be assigned soon. Please check the timetable portal for updates.
'''}
{f'''
Teaching Sessions Created:
{len(teaching_sessions)} teaching session(s) have been automatically created for you. You can now:
- Track student attendance for each session
- Record student assessments and progress
- Manage session materials and resources
- Communicate with students about session updates
''' if teaching_sessions else '''
Teaching sessions will be created as your schedule is finalized.
'''}
Please review the timetable and contact your supervisor if you have any questions or concerns.
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=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable notification to tutor {tutor.id}: {str(e)}")
def _notify_creator_of_rejection(self, timetable: Timetable, supervisor: Supervisor, reason: str) -> None:
"""
Send notification to timetable creator about rejection
Args:
timetable: Timetable model instance
supervisor: Supervisor who rejected
reason: Reason for rejection
"""
with DatabaseContextManager() as ctx:
creator = ctx.session.query(Supervisor).filter(
Supervisor.id == timetable.created_by
).first()
if not creator or not creator.notification_preference or not creator.notification_preference.receive_email:
return
# Prepare email
subject = f"Timetable Rejected: {timetable.name}"
# Build message
message = f"""
Timetable Rejection Notification
Hello {creator.first_name},
Your timetable has been rejected by {supervisor.first_name} {supervisor.last_name}:
Timetable Name: {timetable.name}
Semester: {timetable.semester}
Academic Year: {timetable.academic_year}
Reason for Rejection: {reason or 'Not specified'}
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=creator.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable rejection notification: {str(e)}")
def _notify_tutors_of_disapproved_timetable(self, timetable: Timetable, cancelled_sessions: List[Dict]) -> None:
"""
Send notifications to tutors about cancelled teaching sessions due to timetable disapproval
Args:
timetable: Timetable model instance
cancelled_sessions: List of cancelled session information
"""
if not cancelled_sessions:
return
with DatabaseContextManager() as ctx:
# Get unique tutor IDs from cancelled sessions
tutor_ids = list(set([session['tutor_id'] for session in cancelled_sessions]))
# Get tutor information
tutors = ctx.session.query(Tutor).filter(
Tutor.id.in_(tutor_ids),
Tutor.is_active == True
).all()
for tutor in tutors:
if not tutor.notification_preference or not tutor.notification_preference.receive_email:
continue
# Get sessions for this specific tutor
tutor_sessions = [s for s in cancelled_sessions if s['tutor_id'] == tutor.id]
# Prepare email
subject = f"Teaching Sessions Cancelled: {timetable.name}"
# Build message
sessions_html = ""
for session in tutor_sessions:
day_names = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
day_name = day_names[session['day_of_week']]
sessions_html += f"""
{session['session_type'].title()}: {session['title'] or 'Teaching Session'}
Day: {day_name}
Time: {session['start_time']} - {session['end_time']}
Room: {session['room']}
"""
message = f"""
Teaching Sessions Cancelled
Hello {tutor.first_name},
The following timetable has been disapproved, and your teaching sessions have been cancelled:
Timetable Name: {timetable.name}
Semester: {timetable.semester}
Academic Year: {timetable.academic_year}
Department: {timetable.department}
Cancelled Sessions ({len(tutor_sessions)}):
{sessions_html}
Important: All cancelled sessions have been removed from your schedule.
Please check with your supervisor for any alternative arrangements.
If you have any questions, please contact your 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=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable disapproval notification to tutor {tutor.id}: {str(e)}")
def _notify_tutors_of_timetable_deletion(self, tutors: List[Tutor], timetable_info: Dict) -> None:
"""
Send notifications to tutors about timetable deletion
Args:
tutors: List of affected tutors
timetable_info: Dictionary containing timetable information
"""
for tutor in tutors:
# Check notification preferences
with DatabaseContextManager() as ctx:
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if not prefs or not prefs.receive_email:
continue
# Prepare email
subject = f"Timetable Deleted: {timetable_info['semester']} {timetable_info['academic_year']}"
# Build message with modern styling
message = f"""
Timetable Deletion Notification
Hello {tutor.first_name} {tutor.last_name},
⚠️ Important Notice
A timetable that contained your teaching sessions has been deleted.
📋 Deleted Timetable Details
Name:
{timetable_info['name']}
Semester:
{timetable_info['semester']}
Academic Year:
{timetable_info['academic_year']}
Department:
{timetable_info['department']}
🚨 Impact on Your Schedule
The following changes have occurred:
- All your teaching sessions for this timetable have been removed
- Your scheduled classes are no longer active
- Student attendance records for these sessions are preserved
- You may need to be reassigned to a new timetable
What You Should Do:
- Contact your supervisor for information about new assignments
- Check for any new timetable notifications
- Update your availability if needed
- Review any pending teaching sessions
"""
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable deletion notification to tutor {tutor.id}: {str(e)}")
def _notify_supervisors_of_timetable_deletion(self, supervisors: List[Supervisor], timetable_info: Dict) -> None:
"""
Send notifications to supervisors about timetable deletion
Args:
supervisors: List of affected supervisors
timetable_info: Dictionary containing timetable information
"""
for supervisor in supervisors:
# Check notification preferences
with DatabaseContextManager() as ctx:
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == supervisor.id
).first()
if not prefs or not prefs.receive_email:
continue
# Prepare email
subject = f"Timetable Deleted: {timetable_info['semester']} {timetable_info['academic_year']} - {timetable_info['department']} Department"
# Build message with modern styling
message = f"""
Department Timetable Deletion Notification
Hello {supervisor.first_name} {supervisor.last_name},
⚠️ Important Administrative Notice
A timetable for your department has been deleted.
📋 Deleted Timetable Details
Name:
{timetable_info['name']}
Semester:
{timetable_info['semester']}
Academic Year:
{timetable_info['academic_year']}
Department:
{timetable_info['department']}
🚨 Impact on Department
The following changes have occurred:
- All teaching sessions for this timetable have been removed
- Tutors in your department may need to be reassigned
- Student schedules may be affected
- A new timetable may need to be created
Recommended Actions:
- Review the impact on your department's teaching schedule
- Communicate with affected tutors about next steps
- Consider creating a new timetable if needed
- Update any related academic planning documents
"""
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=supervisor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable deletion notification to supervisor {supervisor.id}: {str(e)}")
def generate_pdf(self, timetable_id: str) -> Dict:
"""
Generate a beautifully styled PDF version of the timetable
Args:
timetable_id: ID of the timetable to generate PDF for
Returns:
Response with PDF file content or error
"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Get all blocks for this timetable with related data
SupervisorAlias2 = aliased(Supervisor, flat=True)
blocks = ctx.session.query(TimetableBlock)\
.filter(TimetableBlock.timetable_id == timetable_id)\
.join(Course, TimetableBlock.course_id == Course.id)\
.outerjoin(Tutor, TimetableBlock.tutor_id == Tutor.id)\
.outerjoin(SupervisorAlias2, TimetableBlock.supervisor_tutor_id == SupervisorAlias2.id)\
.order_by(
TimetableBlock.day_of_week,
TimetableBlock.start_time
).all()
try:
# Create PDF with enhanced styling
pdf = TimetablePDF()
pdf.add_page()
# Set document title
pdf.set_document_title(timetable.name or "Timetable")
# Generate the timetable grid using the existing method
pdf.create_professional_timetable_grid(blocks)
# Generate PDF content
pdf_output = pdf.output(dest='S')
# Ensure PDF content is bytes
if isinstance(pdf_output, str):
pdf_content = pdf_output.encode('latin1')
else:
pdf_content = bytes(pdf_output)
# Create response
response = make_response(pdf_content)
response.headers['Content-Type'] = 'application/pdf'
response.headers['Content-Disposition'] = f'attachment; filename="{timetable.name or "timetable"}.pdf"'
response.headers['Content-Length'] = str(len(pdf_content))
return response
except Exception as e:
current_app.logger.error(f"Failed to generate PDF: {str(e)}")
return custom_response(
success=False,
data=f"Failed to generate PDF: {str(e)}",
status_code=500
)
def get_timetable_statistics(self, timetable_id: str) -> Dict:
"""
Get comprehensive statistics for a timetable
Args:
timetable_id: ID of the timetable
Returns:
Response with statistics or error
"""
with DatabaseContextManager() as ctx:
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Get blocks and calculate statistics
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id
).all()
stats = {
'total_blocks': len(blocks),
'total_teaching_hours': len(blocks) * 2, # Each block is 2 hours
'courses': {},
'tutors': {},
'daily_distribution': {day: 0 for day in ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']},
'time_distribution': {},
'utilization': {}
}
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
for block in blocks:
# Course statistics
course_code = block.course.code if block.course else 'Unknown'
if course_code not in stats['courses']:
stats['courses'][course_code] = {
'blocks': 0,
'hours': 0,
'tutors': set()
}
stats['courses'][course_code]['blocks'] += 1
stats['courses'][course_code]['hours'] += 2
if block.tutor:
stats['courses'][course_code]['tutors'].add(f"{block.tutor.first_name} {block.tutor.last_name}")
# Tutor statistics
tutor_name = 'Unknown'
if block.tutor:
tutor_name = f"{block.tutor.first_name} {block.tutor.last_name}"
elif hasattr(block, 'supervisor_tutor') and block.supervisor_tutor:
tutor_name = f"{block.supervisor_tutor.first_name} {block.supervisor_tutor.last_name}"
if tutor_name not in stats['tutors']:
stats['tutors'][tutor_name] = {
'blocks': 0,
'hours': 0,
'courses': set()
}
stats['tutors'][tutor_name]['blocks'] += 1
stats['tutors'][tutor_name]['hours'] += 2
if block.course:
stats['tutors'][tutor_name]['courses'].add(block.course.code)
# Daily distribution
day_name = days[block.day_of_week]
stats['daily_distribution'][day_name] += 1
# Time distribution
time_slot = f"{block.start_time.strftime('%H:%M')}-{block.end_time.strftime('%H:%M')}"
if time_slot not in stats['time_distribution']:
stats['time_distribution'][time_slot] = 0
stats['time_distribution'][time_slot] += 1
# Convert sets to counts for JSON serialization
for course_data in stats['courses'].values():
course_data['tutors'] = len(course_data['tutors'])
for tutor_data in stats['tutors'].values():
tutor_data['courses'] = len(tutor_data['courses'])
# Calculate utilization (assuming 4 time slots per day, 5 days a week)
max_possible_blocks = 4 * 5 # 4 time slots * 5 weekdays
stats['utilization'] = {
'percentage': (len(blocks) / max_possible_blocks) * 100 if max_possible_blocks > 0 else 0,
'used_slots': len(blocks),
'available_slots': max_possible_blocks - len(blocks)
}
return custom_response(
success=True,
data=stats,
status_code=200
)
def export_pdf(self, timetable_id: str) -> Dict:
"""Export timetable as PDF using Excel-first approach for better formatting"""
return self.export_pdf_with_filters(timetable_id, {})
def export_pdf_with_filters(self, timetable_id: str, filters: Dict) -> Dict:
"""Export timetable as PDF with filters applied"""
with DatabaseContextManager() as ctx:
# Get timetable with blocks
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Create aliases to avoid column name conflicts
TutorAlias = aliased(Tutor, name='tutor_alias')
SupervisorAlias = aliased(Supervisor, name='supervisor_alias')
query = ctx.session.query(TimetableBlock)\
.filter(TimetableBlock.timetable_id == timetable_id)\
.join(Course, TimetableBlock.course_id == Course.id)\
.outerjoin(TutorAlias, TimetableBlock.tutor_id == TutorAlias.id)\
.outerjoin(SupervisorAlias, TimetableBlock.supervisor_tutor_id == SupervisorAlias.id)
# Get total blocks before filtering for comparison
total_blocks = ctx.session.query(TimetableBlock)\
.filter(TimetableBlock.timetable_id == timetable_id).count()
# Log the filters being applied
# Apply filters
blocks = self._apply_filters_to_blocks(ctx, query, filters, TutorAlias, SupervisorAlias)
# Log the number of blocks after filtering
# Check if any blocks were found after filtering
if not blocks:
return custom_response(
success=False,
data="No timetable blocks found with the specified filters. Please try adjusting your filter criteria.",
status_code=404
)
try:
# Step 1: Generate Excel with proper formatting
excel_data = self._generate_excel_timetable(timetable, blocks)
# Step 2: Convert Excel to PDF using actual timetable data
pdf_content = self._convert_excel_to_pdf_with_blocks(timetable, blocks)
# Generate filename based on filters
filename = self._generate_filename_with_filters(timetable, filters)
# Create response
response = make_response(pdf_content)
response.headers['Content-Type'] = 'application/pdf'
response.headers['Content-Disposition'] = f'attachment; filename="{filename}"'
response.headers['Content-Length'] = str(len(pdf_content))
return response
except Exception as e:
current_app.logger.error(f"Error generating PDF: {str(e)}")
return custom_response(
success=False,
data=f"Error generating PDF: {str(e)}",
status_code=500
)
def _apply_filters_to_blocks(self, ctx, query, filters: Dict, TutorAlias, SupervisorAlias) -> List[TimetableBlock]:
"""Apply filters to the blocks query"""
# Export type filter
export_type = filters.get('export_type', 'full')
# Speciality filter
if filters.get('speciality_id'):
speciality_id = filters['speciality_id']
# Try to filter by speciality_id first, then by name if that fails
try:
# First try to filter by speciality_id (UUID)
query = query.filter(Course.speciality_id == speciality_id)
except Exception:
# If that fails, try to filter by speciality name
try:
from src.models.models import Speciality
query = query.join(Speciality, Course.speciality_id == Speciality.id)\
.filter(Speciality.name == speciality_id)
except Exception as e:
current_app.logger.warning(f"Could not apply speciality filter: {e}")
# If both fail, don't apply the filter
pass
# Year level filter
if filters.get('year_level'):
year_level = filters['year_level']
query = query.filter(Course.course_level == year_level)
# Term filter
if filters.get('term'):
term = filters['term']
query = query.filter(Course.semester == term)
# Academic year filter
if filters.get('academic_year'):
academic_year = filters['academic_year']
query = query.filter(Course.academic_year == academic_year)
# Tutor filter
if filters.get('tutor_id'):
tutor_id = filters['tutor_id']
query = query.filter(
or_(
TimetableBlock.tutor_id == tutor_id,
TimetableBlock.supervisor_tutor_id == tutor_id
)
)
# Search term filter
if filters.get('search_term'):
search_term = f"%{filters['search_term']}%"
query = query.filter(
or_(
Course.code.ilike(search_term),
Course.title.ilike(search_term),
TutorAlias.first_name.ilike(search_term),
TutorAlias.last_name.ilike(search_term),
SupervisorAlias.first_name.ilike(search_term),
SupervisorAlias.last_name.ilike(search_term),
TimetableBlock.room.ilike(search_term)
)
)
# Execute query and return results
blocks = query.all()
return blocks
def _generate_filename_with_filters(self, timetable: Timetable, filters: Dict) -> str:
"""Generate filename based on filters applied"""
base_name = timetable.name or "timetable"
current_date = datetime.now().strftime("%Y-%m-%d")
# Add export type
export_type = filters.get('export_type', 'full')
filename_parts = [base_name, export_type]
# Add speciality if specified
if filters.get('speciality_id'):
speciality_id = filters['speciality_id']
# Try to get speciality name from database
try:
with DatabaseContextManager() as ctx:
from src.models.models import Speciality
# First try to find by ID
speciality = ctx.session.query(Speciality).filter(Speciality.id == speciality_id).first()
if not speciality:
# If not found by ID, try to find by name
speciality = ctx.session.query(Speciality).filter(Speciality.name == speciality_id).first()
if speciality:
speciality_name = speciality.name.replace(' ', '-').replace('/', '-')
filename_parts.append(f"speciality-{speciality_name}")
else:
# If speciality not found, use the provided value
speciality_name = speciality_id.replace(' ', '-').replace('/', '-')
filename_parts.append(f"speciality-{speciality_name}")
except Exception:
# If any error occurs, use the provided value
speciality_name = speciality_id.replace(' ', '-').replace('/', '-')
filename_parts.append(f"speciality-{speciality_name}")
# Add year level if specified
if filters.get('year_level'):
filename_parts.append(filters['year_level'].replace(' ', '-'))
# Add term if specified
if filters.get('term'):
filename_parts.append(filters['term'].replace(' ', '-'))
# Add academic year if specified
if filters.get('academic_year'):
filename_parts.append(filters['academic_year'])
# Add tutor if specified
if filters.get('tutor_id'):
tutor_id = filters['tutor_id']
# Try to get tutor name from database
try:
with DatabaseContextManager() as ctx:
tutor = ctx.session.query(Tutor).filter(Tutor.id == tutor_id).first()
if tutor:
tutor_name = f"{tutor.first_name}-{tutor.last_name}".replace(' ', '-')
filename_parts.append(f"tutor-{tutor_name}")
else:
filename_parts.append(f"tutor-{tutor_id}")
except Exception:
filename_parts.append(f"tutor-{tutor_id}")
# Add search term if specified
if filters.get('search_term'):
search_term = filters['search_term'].replace(' ', '-')[:20] # Limit length
filename_parts.append(f"search-{search_term}")
filename_parts.append(current_date)
return f"{'-'.join(filename_parts)}.pdf"
def _generate_excel_timetable(self, timetable: Timetable, blocks: List[TimetableBlock]) -> bytes:
"""Generate Excel timetable with proper timeblock placement"""
wb = Workbook()
ws = wb.active
ws.title = "Timetable"
# Define time slots (8 AM to 6 PM, 1-hour intervals)
time_slots = []
for hour in range(8, 19):
time_slots.append(f"{hour:02d}:00")
# Define days
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
# Set up column headers
for col, day in enumerate(days, 2):
ws.cell(row=1, column=col, value=day)
# Set up time rows
for row, time_slot in enumerate(time_slots, 2):
ws.cell(row=row, column=1, value=time_slot)
# Process blocks and place them in the grid
for block in blocks:
if not (0 <= block.day_of_week < 5): # Skip weekends
continue
# Calculate column (day + 1 for time column)
col = block.day_of_week + 2
# Calculate rows based on start and end times (Excel rows start at 2)
start_row = self._find_exact_time_slot_row(block.start_time.strftime('%H:%M'), time_slots) + 2
end_row = self._find_exact_time_slot_row(block.end_time.strftime('%H:%M'), time_slots) + 2
# Treat end_row as exclusive; adjust to inclusive for Excel merge
if end_row <= start_row:
continue
merge_end_row = end_row - 1
# Get course information
course_code = block.course.code if block.course else "Unknown"
tutor_name = ""
if block.tutor:
tutor_name = f"{block.tutor.first_name} {block.tutor.last_name}"
elif block.supervisor_tutor:
tutor_name = f"{block.supervisor_tutor.first_name} {block.supervisor_tutor.last_name}"
# Create cell content
cell_content = f"{course_code}\n{block.block_type.title()}\n{tutor_name}"
# Set cell value BEFORE merging to avoid MergedCell write errors
cell = ws.cell(row=start_row, column=col)
cell.value = cell_content
cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
# Merge cells for this block (only if it spans more than one row)
if merge_end_row > start_row:
ws.merge_cells(start_row=start_row, start_column=col,
end_row=merge_end_row, end_column=col)
# No color fill for Excel cells
# Auto-adjust column widths
for col in range(1, len(days) + 2):
ws.column_dimensions[get_column_letter(col)].width = 20
# Auto-adjust row heights
for row in range(1, len(time_slots) + 2):
ws.row_dimensions[row].height = 25
# Save to bytes
excel_buffer = io.BytesIO()
wb.save(excel_buffer)
excel_buffer.seek(0)
return excel_buffer.getvalue()
def _organize_blocks_for_excel(self, blocks: List[TimetableBlock], time_slots: List[str]) -> List[Dict]:
"""Organize blocks to avoid overlaps by making them adjacent"""
organized_blocks = []
# Group blocks by day
blocks_by_day = {}
for block in blocks:
day = block.day_of_week
if day not in blocks_by_day:
blocks_by_day[day] = []
blocks_by_day[day].append(block)
# Process each day
for day, day_blocks in blocks_by_day.items():
# Sort blocks by start time
sorted_blocks = sorted(day_blocks, key=lambda x: x.start_time)
# Organize overlapping blocks into adjacent columns
# Fix: Use correct day column mapping (0=Monday, 1=Tuesday, etc.)
day_column = day + 1 # +1 because column 1 is for time
for block in sorted_blocks:
# Calculate time positions
start_time_str = block.start_time.strftime('%H:%M')
end_time_str = block.end_time.strftime('%H:%M')
start_row = self._find_time_slot_row(start_time_str, time_slots)
end_row = self._find_time_slot_row(end_time_str, time_slots)
if start_row == -1 or end_row == -1:
continue
# Check for overlaps and adjust column if needed
current_column = day_column
while self._has_overlap_in_column(organized_blocks, current_column, start_row, end_row):
current_column += 1
# Get course and tutor information
course_code = block.course.code if block.course else "Unknown"
course_title = block.course.title if block.course else "Unknown Course"
tutor_name = ""
if block.tutor:
tutor_name = f"{block.tutor.first_name} {block.tutor.last_name}"
elif block.supervisor_tutor:
tutor_name = f"{block.supervisor_tutor.first_name} {block.supervisor_tutor.last_name}"
# Create course info with unit name and tutor
course_info = f"{course_code}\n{course_title[:25]}\n{tutor_name[:25]}"
# No color assignment
organized_blocks.append({
'day_col': current_column,
'start_row': start_row,
'end_row': end_row,
'course_info': course_info,
'block': block
})
# Ensure we have at least some blocks to display
if not organized_blocks:
current_app.logger.warning("No blocks were organized! Creating fallback blocks to ensure visibility.")
# Create fallback blocks for each day to ensure something is visible
for day in range(5): # Monday to Friday
day_column = day + 1
organized_blocks.append({
'day_col': day_column,
'start_row': 2,
'end_row': 3,
'course_info': f"Sample Block\nDay {day + 1}\nTBD",
'block': None
})
else:
# Log the first few blocks for verification
for i, block_data in enumerate(organized_blocks[:5]):
pass
return organized_blocks
def _calculate_duration_hours(self, start_time: str, end_time: str) -> float:
"""Calculate duration in hours between two time strings"""
try:
# Parse start time
start_parts = start_time.split(':')
start_hour = int(start_parts[0])
start_minute = int(start_parts[1]) if len(start_parts) > 1 else 0
# Parse end time
end_parts = end_time.split(':')
end_hour = int(end_parts[0])
end_minute = int(end_parts[1]) if len(end_parts) > 1 else 0
# Calculate duration
duration_minutes = (end_hour * 60 + end_minute) - (start_hour * 60 + start_minute)
duration_hours = duration_minutes / 60.0
return max(0, duration_hours)
except (ValueError, IndexError) as e:
return 1.0 # Default to 1 hour
def _find_exact_time_slot_row(self, time_str: str, time_slots: List[str]) -> int:
"""Find exact time slot index (0-based). Fallback to matching hour or nearest slot.
Returns 0-based index into time_slots, or -1 if it cannot be determined.
"""
try:
# Normalize input
input_time = time_str.strip()
# Exact string match
for i, slot in enumerate(time_slots):
if slot == input_time:
return i
# Parse hour from input (supports "HH" or "HH:MM")
if ':' in input_time:
hour_part = input_time.split(':')[0]
else:
hour_part = input_time
hour = int(hour_part)
# Match by same hour exactly
for i, slot in enumerate(time_slots):
if int(slot.split(':')[0]) == hour:
return i
# First slot with hour greater than or equal
for i, slot in enumerate(time_slots):
if int(slot.split(':')[0]) >= hour:
return i
# Otherwise, clamp to last slot
return len(time_slots) - 1 if time_slots else -1
except (ValueError, IndexError):
return -1
def _find_time_slot_row(self, time_str: str, time_slots: List[str]) -> int:
"""Find the row number for a given time"""
try:
# Parse the time string to get hour and minutes
if ':' in time_str:
parts = time_str.split(':')
hour = int(parts[0])
minutes = int(parts[1]) if len(parts) > 1 else 0
else:
# If it's just a number, assume it's the hour
hour = int(time_str)
minutes = 0
# Find the exact time slot match first
for i, slot in enumerate(time_slots):
if slot == time_str:
row = i + 2 # +2 because data starts at row 2 (after header)
return row
# If no exact match, find the corresponding hour slot
for i, slot in enumerate(time_slots):
slot_hour = int(slot.split(':')[0])
if slot_hour == hour:
row = i + 2 # +2 because data starts at row 2 (after header)
return row
# If hour not found, find the closest slot
for i, slot in enumerate(time_slots):
slot_hour = int(slot.split(':')[0])
if slot_hour >= hour:
row = i + 2 # +2 because data starts at row 2 (after header)
return row
# If hour is beyond our time slots, use the last slot
row = len(time_slots) + 1
return row
except (ValueError, IndexError) as e:
return -1
def _has_overlap_in_column(self, existing_blocks: List[Dict], column: int, start_row: int, end_row: int) -> bool:
"""Check if a time block overlaps with existing blocks in a column"""
for block in existing_blocks:
if block['day_col'] == column:
# Check for time overlap
if not (end_row <= block['start_row'] or start_row >= block['end_row']):
return True
return False
def _get_course_color(self, course_code: str) -> str:
"""Get a consistent color for a course"""
colors = [
"E6F3FF", # Light blue
"E6FFE6", # Light green
"FFF2E6", # Light orange
"F0E6FF", # Light purple
"FFE6E6", # Light red
"FFFFE6", # Light yellow
"E6FFFF", # Light cyan
"FFE6FF", # Light magenta
]
# Handle empty or None course codes
if not course_code:
course_code = "DEFAULT"
# Use hash of course code to get consistent color
hash_value = abs(hash(course_code)) % len(colors)
return colors[hash_value]
def _convert_excel_to_pdf(self, excel_data: bytes) -> bytes:
"""Convert Excel data to PDF using FPDF with actual timetable content"""
# Create PDF in landscape orientation for better timetable layout
pdf = FPDF(orientation='L', format='A4')
pdf.add_page()
# Set margins
pdf.set_margins(10, 10, 10)
# Add title
pdf.set_font('Arial', 'B', 18)
pdf.cell(0, 15, 'Academic Timetable', ln=True, align='C')
pdf.ln(5)
# Add subtitle
pdf.set_font('Arial', 'B', 14)
pdf.cell(0, 10, 'Generated using Excel-first approach for optimal formatting', ln=True, align='C')
pdf.ln(10)
# Create the actual timetable grid
self._create_pdf_timetable_grid(pdf, excel_data)
return pdf.output(dest='S').encode('latin1') if isinstance(pdf.output(dest='S'), str) else bytes(pdf.output(dest='S'))
def _create_pdf_timetable_grid(self, pdf: FPDF, excel_data: bytes):
"""Create the actual timetable grid in PDF based on Excel data"""
# Define time slots (8 AM to 6 PM, 1-hour intervals)
time_slots = []
for hour in range(8, 19): # 8 AM to 6 PM
time_slots.append(f"{hour:02d}:00")
# Define days
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
# Calculate cell dimensions
page_width = pdf.w - 20 # Account for margins
page_height = pdf.h - 40 # Account for margins and title
# Column width (time column + 5 day columns)
col_width = page_width / 6
# Row height
row_height = page_height / (len(time_slots) + 1) # +1 for header row
# Set font for grid
pdf.set_font('Arial', '', 8)
# Draw header row
# Time column header
pdf.cell(col_width, row_height, 'Time', border=1, align='C')
# Day column headers
for day in days:
pdf.cell(col_width, row_height, day, border=1, align='C')
pdf.ln()
# Draw time rows
for time_slot in time_slots:
# Time column
pdf.cell(col_width, row_height, time_slot, border=1, align='C')
# Day columns (empty for now, will be filled with actual data)
for _ in range(5):
pdf.cell(col_width, row_height, '', border=1, align='C')
pdf.ln()
# Now add the actual timetable blocks
self._add_timetable_blocks_to_pdf(pdf, col_width, row_height, time_slots)
def _add_timetable_blocks_to_pdf(self, pdf: FPDF, col_width: float, row_height: float, time_slots: List[str]):
"""Add actual timetable blocks to the PDF grid"""
# This would need to be implemented to read the Excel data and place blocks
# For now, we'll create a simple representation
# Example: Add some sample blocks to demonstrate the layout
sample_blocks = [
{'day': 0, 'start_time': '09:00', 'end_time': '11:00', 'course': 'MATH101', 'tutor': 'Dr. Smith'},
{'day': 1, 'start_time': '10:00', 'end_time': '12:00', 'course': 'PHYS101', 'tutor': 'Dr. Johnson'},
{'day': 2, 'start_time': '14:00', 'end_time': '16:00', 'course': 'CHEM101', 'tutor': 'Dr. Brown'},
]
for block in sample_blocks:
# Calculate positions
start_row = self._find_time_slot_row(block['start_time'], time_slots)
end_row = self._find_time_slot_row(block['end_time'], time_slots)
if start_row != -1 and end_row != -1:
# Calculate PDF coordinates
x = 10 + (block['day'] + 1) * col_width # +1 for time column
y = 40 + start_row * row_height # 40 for title and header
width = col_width
height = (end_row - start_row) * row_height
# Draw the block
pdf.rect(x, y, width, height, 'D') # Just draw border, no fill
# Add text
pdf.set_xy(x, y)
pdf.set_font('Arial', 'B', 7)
pdf.cell(width, height/2, block['course'], align='C')
pdf.set_font('Arial', '', 6)
pdf.cell(width, height/2, block['tutor'], align='C')
def _convert_excel_to_pdf_with_blocks(self, timetable: Timetable, blocks: List[TimetableBlock]) -> bytes:
"""Convert timetable blocks directly to PDF with proper formatting"""
# Create PDF in portrait orientation with A3 size for maximum space
pdf = FPDF(orientation='P', format='A3')
pdf.add_page()
# Set zero margins for maximum space
pdf.set_margins(0, 0, 0)
# Start grid from top of page to maximize space
pdf.set_y(5) # Start grid at very top of page
# Create the actual timetable grid with real data
self._create_pdf_timetable_grid_with_blocks(pdf, blocks)
# Ensure we don't exceed page height
if pdf.get_y() > pdf.h - 20:
current_app.logger.warning("Content exceeds page height, adjusting layout")
return pdf.output(dest='S').encode('latin1') if isinstance(pdf.output(dest='S'), str) else bytes(pdf.output(dest='S'))
def _create_pdf_timetable_grid_with_blocks(self, pdf: FPDF, blocks: List[TimetableBlock]):
"""Create the actual timetable grid in PDF using real timetable blocks"""
# Define time slots (8 AM to 6 PM, 1-hour intervals)
time_slots = []
for hour in range(8, 19): # 8 AM to 6 PM
time_slots.append(f"{hour:02d}:00")
# Define days (Monday to Friday only)
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
# Calculate cell dimensions to fit everything on one page with zero margins
page_width = pdf.w # Full A3 width
page_height = pdf.h - 10 # Full A3 height minus small top margin
# Column width (time column + 5 day columns)
col_width = page_width / 6
# Row height - calculate based on number of time slots to fit on page
row_height = min(page_height / (len(time_slots) + 1), 45) # Increased height to 45mm for much bigger blocks
# Organize blocks by day and time slot for easy lookup
block_grid = {}
for block in blocks:
if 0 <= block.day_of_week < 5: # Only Monday-Friday
day = block.day_of_week
start_time_str = block.start_time.strftime('%H:%M')
end_time_str = block.end_time.strftime('%H:%M')
start_row = self._find_exact_time_slot_row(start_time_str, time_slots)
end_row = self._find_exact_time_slot_row(end_time_str, time_slots)
# Get course information
course_code = block.course.code if block.course else "Unknown"
tutor_name = ""
if block.tutor:
tutor_name = f"{block.tutor.first_name} {block.tutor.last_name}"
elif block.supervisor_tutor:
tutor_name = f"{block.supervisor_tutor.first_name} {block.supervisor_tutor.last_name}"
block_info = f"{course_code}\n{block.block_type.title()}\n{tutor_name}"
if start_row != -1 and end_row != -1 and start_row < end_row:
# Create key for each time slot this block spans
for row in range(start_row, end_row):
key = f"{day}_{row}"
if key not in block_grid:
block_grid[key] = []
block_grid[key].append(block_info)
# Set font for grid
pdf.set_font('Arial', 'B', 12) # Increased font size for headers
# Draw header row with borders to make it stand out
pdf.cell(col_width, row_height, 'Time', border=1, align='C')
for day in days:
pdf.cell(col_width, row_height, day, border=1, align='C')
pdf.ln()
# Draw time rows with blocks integrated
for row_idx, time_slot in enumerate(time_slots):
# Time column with border to make it stand out
pdf.set_font('Arial', 'B', 10) # Bold font for time slots
pdf.cell(col_width, row_height, time_slot, border=1, align='C')
# Day columns with blocks
for day_idx in range(5):
key = f"{day_idx}_{row_idx}"
if key in block_grid:
# Draw cell with block content and prominent border
block_content = block_grid[key][0] # Take first block if multiple
pdf.set_font('Arial', 'B', 12) # Much larger font size for blocks
# Use multi_cell to handle line breaks automatically with thicker border
current_x = pdf.get_x()
current_y = pdf.get_y()
pdf.multi_cell(col_width, row_height/3, block_content, border=1, align='C')
# Reset position for next cell
pdf.set_xy(current_x + col_width, current_y)
pdf.set_font('Arial', 'B', 10) # Keep bold font for consistency
else:
# Draw empty cell with light border to maintain grid structure
pdf.cell(col_width, row_height, '', border=1, align='C')
pdf.ln()
# Method removed - blocks are now integrated directly into the grid drawing
def _hex_to_rgb(self, hex_color: str) -> tuple:
"""Convert hex color to RGB tuple"""
if not hex_color:
return (230, 243, 255) # Default light blue
hex_color = hex_color.lstrip('#')
# Ensure we have exactly 6 characters
if len(hex_color) != 6:
return (230, 243, 255) # Default light blue
try:
return tuple(int(hex_color[i:i+2], 16) for i in (0, 2, 4))
except ValueError:
return (230, 243, 255) # Default light blue
def export_excel(self, timetable_id: str) -> Dict:
"""Export timetable as Excel file with proper formatting"""
with DatabaseContextManager() as ctx:
# Get timetable with blocks
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
# Get all blocks for this timetable with explicit joins
from sqlalchemy.orm import aliased
SupervisorAlias3 = aliased(Supervisor, flat=True)
blocks = ctx.session.query(TimetableBlock)\
.filter(TimetableBlock.timetable_id == timetable_id)\
.join(Course, TimetableBlock.course_id == Course.id)\
.outerjoin(Tutor, TimetableBlock.tutor_id == Tutor.id)\
.outerjoin(SupervisorAlias3, TimetableBlock.supervisor_tutor_id == SupervisorAlias3.id)\
.all()
try:
# Generate Excel with proper formatting
excel_data = self._generate_excel_timetable(timetable, blocks)
# Create response
response = make_response(excel_data)
response.headers['Content-Type'] = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
response.headers['Content-Disposition'] = f'attachment; filename="{timetable.name or "timetable"}.xlsx"'
response.headers['Content-Length'] = str(len(excel_data))
return response
except Exception as e:
current_app.logger.error(f"Error generating Excel: {str(e)}")
return custom_response(
success=False,
data=f"Error generating Excel: {str(e)}",
status_code=500
)
def get_timetables_by_user(self, user_id: str) -> Dict:
"""
Get all timetables created by a specific user
Args:
user_id: ID of the user
Returns:
Response with list of timetables
"""
with DatabaseContextManager() as ctx:
# Get timetables created by this user
user_timetables = ctx.session.query(Timetable).filter(
Timetable.created_by == user_id
).order_by(Timetable.created_at.desc()).all()
# Format timetables
all_timetables = []
for timetable in user_timetables:
# Calculate statistics
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).all()
total_courses = len(set(block.course_id for block in blocks if block.course_id))
total_tutors = len(set(block.tutor_id for block in blocks if block.tutor_id))
total_students = 0 # This would need to be calculated from enrollments
total_hours = sum(
(datetime.combine(datetime.min.date(), block.end_time) -
datetime.combine(datetime.min.date(), block.start_time)).total_seconds() / 3600
for block in blocks if block.start_time and block.end_time
)
room_assignments = len(set(block.room for block in blocks if block.room))
conflict_count = 0 # This would need conflict detection logic
completion_rate = min(100.0, (len(blocks) / max(1, total_courses)) * 100) if total_courses > 0 else 0.0
# Get department and speciality from the first course if available
department = "General"
speciality = "General"
if blocks:
first_course = ctx.session.query(Course).filter(Course.id == blocks[0].course_id).first()
if first_course:
department = first_course.department or "General"
speciality = first_course.speciality.name if first_course.speciality else "General"
# Convert blocks to dictionary format
blocks_data = [self.block_manager._block_to_dict(block) for block in blocks]
timetable_data = {
'id': timetable.id,
'name': timetable.name,
'description': timetable.description,
'academic_year': timetable.academic_year,
'term': getattr(timetable, 'term', 'Term 1'),
'semester': timetable.semester,
'is_active': timetable.is_active,
'created_at': timetable.created_at.isoformat() if timetable.created_at else None,
'updated_at': getattr(timetable, 'updated_at', timetable.created_at).isoformat() if getattr(timetable, 'updated_at', timetable.created_at) else None,
'supervisor_id': timetable.created_by,
'supervisor_name': f"{timetable.creator.first_name} {timetable.creator.last_name}" if timetable.creator else "Unknown",
'total_courses': total_courses,
'total_students': total_students,
'total_tutors': total_tutors,
'total_hours': total_hours,
'status': getattr(timetable, 'status', 'draft'),
'approval_status': timetable.approval_status,
'last_modified': getattr(timetable, 'updated_at', timetable.created_at).isoformat() if getattr(timetable, 'updated_at', timetable.created_at) else None,
'version': getattr(timetable, 'version', '1.0'),
'department': department,
'speciality': speciality,
'room_assignments': room_assignments,
'conflict_count': conflict_count,
'completion_rate': completion_rate,
'blocks': blocks_data,
'blocks_count': len(blocks)
}
current_app.logger.info(f"[SUPERVISOR DEBUG] Added timetable {timetable.id} to response with {len(blocks_data)} blocks")
all_timetables.append(timetable_data)
# Debug logging for final response
current_app.logger.info(f"[SUPERVISOR DEBUG] Final response: {len(all_timetables)} timetables")
for i, tt in enumerate(all_timetables):
current_app.logger.info(f"[SUPERVISOR DEBUG] Timetable {i}: {tt['id']} - {tt['name']} - blocks_count: {tt.get('blocks_count', 'NOT_FOUND')}")
if 'blocks' in tt:
current_app.logger.info(f"[SUPERVISOR DEBUG] Timetable {i} blocks array length: {len(tt['blocks'])}")
else:
current_app.logger.error(f"[SUPERVISOR DEBUG] Timetable {i} MISSING 'blocks' field!")
return custom_response(
success=True,
data={
'timetables': all_timetables
},
status_code=200
)
def _send_timetable_verification_emails(self, ctx, timetable, teaching_sessions):
"""
Send email alerts to tutors with their teaching schedules after timetable verification
Args:
ctx: Database context
timetable: Timetable object
teaching_sessions: List of created teaching sessions
Returns:
Dict with email sending results
"""
try:
# Group teaching sessions by teacher (both tutors and supervisors)
tutor_sessions = {}
supervisor_sessions = {}
for session in teaching_sessions:
# Group by tutor
if session.tutor_id:
if session.tutor_id not in tutor_sessions:
tutor_sessions[session.tutor_id] = []
tutor_sessions[session.tutor_id].append(session)
# Group by supervisor
if session.supervisor_tutor_id:
if session.supervisor_tutor_id not in supervisor_sessions:
supervisor_sessions[session.supervisor_tutor_id] = []
supervisor_sessions[session.supervisor_tutor_id].append(session)
emails_sent = 0
email_summary = {
'total_tutors': len(tutor_sessions),
'total_supervisors': len(supervisor_sessions),
'successful_emails': 0,
'failed_emails': 0,
'users_without_email': 0
}
# Send emails to tutors
for tutor_id, sessions in tutor_sessions.items():
try:
tutor = ctx.session.query(Tutor).filter(Tutor.id == tutor_id).first()
if not tutor:
current_app.logger.warning(f"Tutor {tutor_id} not found")
continue
if not tutor.email:
current_app.logger.warning(f"Tutor {tutor.first_name} {tutor.last_name} has no email address")
email_summary['users_without_email'] += 1
continue
# Check notification preferences
notification_pref = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor_id
).first()
if notification_pref and not notification_pref.receive_email:
continue
# Generate email content for tutors
email_content = self._generate_timetable_verification_email(timetable, sessions, tutor)
# Send email
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=f"Your Teaching Schedule - {timetable.name} ({timetable.semester} {timetable.academic_year})",
message=email_content
)
emails_sent += 1
email_summary['successful_emails'] += 1
current_app.logger.info(f"Email sent successfully to tutor {tutor.email}")
except Exception as email_error:
current_app.logger.error(f"Failed to send email to tutor {tutor.email}: {email_error}")
email_summary['failed_emails'] += 1
except Exception as tutor_error:
current_app.logger.error(f"Error processing tutor {tutor_id}: {tutor_error}")
email_summary['failed_emails'] += 1
continue
# Send emails to supervisors
for supervisor_id, sessions in supervisor_sessions.items():
try:
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if not supervisor:
current_app.logger.warning(f"Supervisor {supervisor_id} not found")
continue
if not supervisor.email:
current_app.logger.warning(f"Supervisor {supervisor.first_name} {supervisor.last_name} has no email address")
email_summary['users_without_email'] += 1
continue
# Check notification preferences
notification_pref = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == supervisor_id
).first()
if notification_pref and not notification_pref.receive_email:
continue
# Generate enhanced email content for supervisors
email_content = self._generate_supervisor_verification_email(timetable, sessions, supervisor, len(teaching_sessions))
# Send email
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=supervisor.email,
subject=f"Timetable Verification - {timetable.name} ({timetable.semester} {timetable.academic_year})",
message=email_content
)
emails_sent += 1
email_summary['successful_emails'] += 1
current_app.logger.info(f"Email sent successfully to supervisor {supervisor.email}")
except Exception as email_error:
current_app.logger.error(f"Failed to send email to supervisor {supervisor.email}: {email_error}")
email_summary['failed_emails'] += 1
except Exception as supervisor_error:
current_app.logger.error(f"Error processing supervisor {supervisor_id}: {supervisor_error}")
email_summary['failed_emails'] += 1
continue
return {
'emails_sent': emails_sent,
'summary': email_summary
}
except Exception as e:
current_app.logger.error(f"Error sending timetable verification emails: {e}")
return {
'emails_sent': 0,
'summary': {
'error': str(e),
'total_teachers': 0,
'successful_emails': 0,
'failed_emails': 0,
'teachers_without_email': 0
}
}
def _generate_timetable_verification_email(self, timetable, sessions, teacher):
"""
Generate email content for timetable verification notification
Args:
timetable: Timetable object
sessions: List of teaching sessions for this teacher
teacher: Teacher (tutor or supervisor) object
Returns:
HTML formatted email content
"""
try:
# Group sessions by day
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
day_sessions = {}
for session in sessions:
day_name = days[session.day_of_week]
if day_name not in day_sessions:
day_sessions[day_name] = []
day_sessions[day_name].append(session)
# Build schedule HTML
schedule_html = []
for day in days:
if day in day_sessions and day_sessions[day]:
schedule_html.append(f"{day}
")
schedule_html.append("")
for session in day_sessions[day]:
# Get course information
course_code = getattr(session, 'course_code', 'N/A')
course_title = getattr(session, 'course_title', 'N/A')
schedule_html.append(f"""
{session.start_time.strftime('%H:%M')} - {session.end_time.strftime('%H:%M')}
Course: {course_code} - {course_title}
Type: {session.session_type.title()} |
Room: {session.room or 'TBD'}
""")
schedule_html.append("
Dear {teacher.first_name} {teacher.last_name},
Your teaching schedule for {timetable.name} ({timetable.semester} {timetable.academic_year}) has been verified and is now active.
📋 Schedule Summary
Total Sessions: {len(sessions)}
Semester: {timetable.semester}
Academic Year: {timetable.academic_year}
Status: ✅ Verified and Active
📅 Your Weekly Teaching Schedule
{''.join(schedule_html)}
ℹ️ Important Information
- All sessions are now active and students can view them
- Please ensure you're available at the scheduled times
- Contact your supervisor if you need to make any changes
- You can track attendance and record student progress
If you have any questions about your schedule, please contact your supervisor.
Best regards,
{current_app.config.get('APP_NAME', 'Academic Management System')} Team
"""
return email_content
except Exception as e:
current_app.logger.error(f"Error generating email content: {e}")
# Return a simple fallback email
return f"""
Teaching Schedule Notification
Dear {teacher.first_name} {teacher.last_name},
Your teaching schedule for {timetable.name} has been verified.
Total sessions: {len(sessions)}
Please log in to view your complete schedule.
"""
def _generate_supervisor_verification_email(self, timetable, sessions, supervisor, total_sessions):
"""
Generate enhanced email content for supervisor timetable verification notification
Args:
timetable: Timetable object
sessions: List of teaching sessions for this supervisor
supervisor: Supervisor object
total_sessions: Total number of sessions in the timetable
Returns:
HTML formatted email content
"""
try:
# Group sessions by day
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
day_sessions = {}
for session in sessions:
day_name = days[session.day_of_week]
if day_name not in day_sessions:
day_sessions[day_name] = []
day_sessions[day_name].append(session)
# Build schedule HTML
schedule_html = []
for day in days:
if day in day_sessions and day_sessions[day]:
schedule_html.append(f"{day}
")
schedule_html.append("")
for session in day_sessions[day]:
# Get course information
course_code = getattr(session, 'course_code', 'N/A')
course_title = getattr(session, 'course_title', 'N/A')
schedule_html.append(f"""
{session.start_time.strftime('%H:%M')} - {session.end_time.strftime('%H:%M')}
Course: {course_code} - {course_title}
Type: {session.session_type.title()} |
Room: {session.room or 'TBD'}
""")
schedule_html.append("
Dear {supervisor.first_name} {supervisor.last_name},
The timetable {timetable.name} ({timetable.semester} {timetable.academic_year}) has been verified and is now active.
📊 Timetable Summary
Timetable Name: {timetable.name}
Semester: {timetable.semester}
Academic Year: {timetable.academic_year}
Total Sessions: {total_sessions}
Your Sessions: {len(sessions)}
Status: ✅ Verified and Active
👨💼 Supervisor Responsibilities
- Monitor teaching session attendance and progress
- Review and approve tutor assessments
- Ensure quality standards are maintained
- Provide guidance and support to tutors
- Track student performance and outcomes
📅 Your Teaching Schedule
{''.join(schedule_html)}
ℹ️ System Access
- All tutors have been notified of their teaching schedules
- You can monitor attendance and progress in real-time
- Access detailed reports and analytics
- Review and approve tutor submissions
- Generate performance reports for management
🎯 Next Steps
- Review the complete timetable in the system
- Ensure all tutors are aware of their schedules
- Set up monitoring and reporting procedures
- Prepare for the upcoming academic period
"""
return email_content
except Exception as e:
current_app.logger.error(f"Error generating supervisor email content: {e}")
return f"""
Timetable Verification Notification
Dear {supervisor.first_name} {supervisor.last_name},
The timetable {timetable.name} ({timetable.semester} {timetable.academic_year}) has been verified and is now active.
As a supervisor, you have {len(sessions)} teaching sessions assigned.
Please check the system for detailed schedule information and monitoring capabilities.
Best regards,
Academic Management System
"""
def verify_timetable(self, timetable_id, payload):
"""Verify a timetable and create teaching sessions"""
with DatabaseContextManager() as ctx:
# try:
timetable = ctx.session.query(Timetable).filter(Timetable.id == timetable_id).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
if timetable.approval_status != 'approved':
return custom_response(
success=False,
data="Timetable must be approved before verification",
status_code=400
)
# Update timetable verification status
timetable.is_verified = True
timetable.verified_at = datetime.utcnow()
timetable.verified_by = payload.get('verifier_id')
timetable.verification_notes = payload.get('notes', '')
timetable.updated_at = datetime.utcnow()
# Create teaching sessions from timetable blocks
teaching_sessions = self._create_teaching_sessions_from_blocks(ctx, timetable_id)
if not teaching_sessions:
return custom_response(
success=False,
data="Failed to create teaching sessions",
status_code=500
)
# Send email alerts to tutors with their teaching schedules
email_result = self._send_timetable_verification_emails(ctx, timetable, teaching_sessions)
ctx.session.commit()
return custom_response(
success=True,
data={
"message": "Timetable verified successfully",
"teaching_sessions_created": len(teaching_sessions),
"emails_sent": email_result.get('emails_sent', 0),
"email_summary": email_result.get('summary', {})
},
status_code=200
)
# except Exception as e:
# ctx.session.rollback()
# current_app.logger.error(f"Error verifying timetable: {e}")
# return custom_response(
# success=False,
# data=f"Failed to verify timetable: {str(e)}",
# status_code=500
# )
def _create_teaching_sessions_from_blocks(self, ctx, timetable_id):
"""Create teaching sessions from timetable blocks"""
try:
# First, get all blocks for this timetable without joins to see what we have
all_blocks = ctx.session.query(TimetableBlock)\
.filter(TimetableBlock.timetable_id == timetable_id)\
.all()
if not all_blocks:
return []
# Now get blocks with valid course and teacher relationships
SupervisorAlias = aliased(Supervisor, flat=True)
blocks = ctx.session.query(TimetableBlock)\
.filter(TimetableBlock.timetable_id == timetable_id)\
.join(Course, TimetableBlock.course_id == Course.id)\
.outerjoin(Tutor, TimetableBlock.tutor_id == Tutor.id)\
.outerjoin(SupervisorAlias, TimetableBlock.supervisor_tutor_id == SupervisorAlias.id)\
.options(
joinedload(TimetableBlock.course),
joinedload(TimetableBlock.tutor),
joinedload(TimetableBlock.supervisor_tutor)
)\
.all()
teaching_sessions = []
for block in blocks:
try:
# Create teaching session for each block
teaching_session = TeachingSession(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
tutor_id=block.tutor_id,
supervisor_tutor_id=block.supervisor_tutor_id,
course_id=block.course_id,
room=block.room or 'TBD',
day_of_week=block.day_of_week,
start_time=block.start_time,
end_time=block.end_time,
session_type=block.block_type,
recurring=block.recurring,
status='scheduled',
notes=f"Created from verified timetable block {block.id}",
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
ctx.session.add(teaching_session)
ctx.session.commit()
teaching_sessions.append(teaching_session)
except Exception as block_error:
current_app.logger.error(f"Error creating teaching session for block {block.id}: {block_error}")
continue
if teaching_sessions:
ctx.session.flush()
current_app.logger.info(f"Created {len(teaching_sessions)} teaching sessions from {len(blocks)} blocks")
return teaching_sessions
except Exception as e:
current_app.logger.error(f"Error creating teaching sessions from blocks: {e}")
raise e
def share_timetable_email(self, timetable_id, payload):
"""Share timetable via email to tutors"""
with DatabaseContextManager() as ctx:
try:
timetable = ctx.session.query(Timetable).filter_by(id=timetable_id).first()
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
if not timetable.is_verified:
return custom_response(
success=False,
data="Timetable must be verified before sharing",
status_code=400
)
# Get teaching sessions for this timetable
teaching_sessions = ctx.session.query(TeachingSession).filter_by(timetable_id=timetable_id).all()
# Group sessions by tutor
tutor_sessions = {}
for session in teaching_sessions:
if session.tutor_id not in tutor_sessions:
tutor_sessions[session.tutor_id] = []
tutor_sessions[session.tutor_id].append(session)
# Send emails to tutors
emails_sent = self._send_timetable_emails(timetable, tutor_sessions)
# Update timetable email status
timetable.email_sent = True
timetable.email_sent_at = datetime.utcnow()
timetable.email_recipients = json.dumps(list(tutor_sessions.keys()))
timetable.updated_at = datetime.utcnow()
ctx.session.commit()
return custom_response(
success=True,
data={
"message": f"Timetable shared successfully via email to {emails_sent} tutors"
},
status_code=200
)
except Exception as e:
ctx.session.rollback()
current_app.logger.error(f"Error sharing timetable email: {e}")
return custom_response(
success=False,
data=f"Failed to share timetable email: {str(e)}",
status_code=500
)
def _send_timetable_emails(self, timetable, tutor_sessions):
"""Send timetable emails to tutors"""
try:
emails_sent = 0
for tutor_id, sessions in tutor_sessions.items():
with DatabaseContextManager() as ctx:
tutor = ctx.session.query(Tutor).filter(Tutor.id == tutor_id).first()
if not tutor or not tutor.email:
continue
# Check notification preferences
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == tutor.id
).first()
if not prefs or not prefs.receive_email:
continue
# Generate email content
email_content = self._generate_timetable_email(timetable, sessions, tutor)
# Send email using the existing email infrastructure
try:
send_email(
sender_email=current_app.config['MAIL_SENDER'],
sender_password=current_app.config['MAIL_PASSWORD'],
receiver_email=tutor.email,
subject=f"Your Teaching Schedule for {timetable.semester} {timetable.academic_year}",
message=email_content
)
emails_sent += 1
current_app.logger.info(f"Timetable email sent successfully to {tutor.email}")
except Exception as e:
current_app.logger.error(f"Failed to send timetable email to {tutor.email}: {str(e)}")
return emails_sent
except Exception as e:
current_app.logger.error(f"Error sending emails: {e}")
return 0
def _generate_timetable_email(self, timetable, sessions, tutor):
"""Generate email content for timetable sharing"""
email_content = f"""
Dear {tutor.first_name} {tutor.last_name},
Your timetable for {timetable.name} ({timetable.term} {timetable.academic_year}) has been verified and is now active.
Teaching Schedule:
"""
# Group sessions by day
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
day_sessions = {}
for session in sessions:
day_name = days[session.day_of_week]
if day_name not in day_sessions:
day_sessions[day_name] = []
day_sessions[day_name].append(session)
# Add sessions to email
for day, day_sessions_list in day_sessions.items():
if day_sessions_list:
email_content += f"\n{day}:\n"
for session in day_sessions_list:
course = Course.query.get(session.course_id)
email_content += f" • {session.start_time.strftime('%H:%M')} - {session.end_time.strftime('%H:%M')}: {course.code if course else 'Unknown Course'} ({session.session_type}) - Room {session.room}\n"
email_content += f"""
You can now:
• Track student attendance for your sessions
• Record student assessments and progress
• Access the full timetable via the portal
• View detailed session information
If you have any questions, please contact your supervisor.
Best regards,
Academic Management System
---
This is an automated message. Please do not reply to this email.
"""
return email_content
def get_teaching_sessions(self, timetable_id):
"""Get teaching sessions for a timetable"""
try:
with DatabaseContextManager() as ctx:
sessions = ctx.session.query(
TeachingSession
).filter(TeachingSession.timetable_id==timetable_id).all()
session_data = []
for session in sessions:
tutor = ctx.session.query(User).filter(User.id==session.tutor_id).first()
course = ctx.session.query(Course).filter(Course.id==session.course_id).first()
session_data.append({
"id": session.id,
"tutor_name": f"{tutor.first_name} {tutor.last_name}" if tutor else "Unknown",
"tutor_email": tutor.email if tutor else "",
"course_code": course.code if course else "Unknown",
"course_title": course.title if course else "Unknown",
"room": session.room,
"day_of_week": session.day_of_week,
"start_time": session.start_time.strftime('%H:%M'),
"end_time": session.end_time.strftime('%H:%M'),
"session_type": session.session_type,
"status": session.status,
"notes": session.notes
})
return custom_response(
success=True,
data=session_data,
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error getting teaching sessions: {e}")
return custom_response(
success=False,
data=f"Failed to get teaching sessions: {str(e)}",
status_code=500
)
def create_teaching_sessions(self, timetable_id):
"""Create teaching sessions from timetable blocks"""
try:
# Check if timetable is verified
timetable = Timetable.query.get(timetable_id)
if not timetable:
return custom_response(
success=False,
data="Timetable not found",
status_code=404
)
if not timetable.is_verified:
return custom_response(
success=False,
data="Timetable must be verified before creating teaching sessions",
status_code=400
)
# Create teaching sessions
teaching_sessions = self._create_teaching_sessions_from_blocks(timetable_id)
return custom_response(
success=True,
data={
"message": f"Created {len(teaching_sessions)} teaching sessions",
"sessions_created": len(teaching_sessions)
},
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error creating teaching sessions: {e}")
return custom_response(
success=False,
data=f"Failed to create teaching sessions: {str(e)}",
status_code=500
)
def get_verified_timetables(self) -> Dict:
"""
Get all verified timetables for admin view
Returns:
Response with list of verified timetables
"""
with DatabaseContextManager() as ctx:
try:
# Get all verified timetables
verified_timetables = ctx.session.query(Timetable).filter(
Timetable.is_verified == True,
Timetable.is_active == True
).order_by(Timetable.created_at.desc()).all()
if not verified_timetables:
return custom_response(
success=True,
data=[],
status_code=200
)
# Format timetables
formatted_timetables = []
for timetable in verified_timetables:
timetable_data = self._timetable_to_dict(timetable)
formatted_timetables.append(timetable_data)
return custom_response(
success=True,
data=formatted_timetables,
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error fetching verified timetables: {str(e)}")
return custom_response(
success=False,
data=f"Error fetching verified timetables: {str(e)}",
status_code=500
)
def get_all_timetables_for_admin(self) -> Dict:
"""
Get all timetables for admin management view with comprehensive information
Returns:
Response with list of all timetables with detailed information
"""
with DatabaseContextManager() as ctx:
try:
current_app.logger.info("Fetching all timetables for admin view")
# Get all timetables with related data
timetables = ctx.session.query(Timetable).order_by(Timetable.created_at.desc()).all()
if not timetables:
return custom_response(
success=True,
data={
'timetables': []
},
status_code=200
)
# Format timetables with comprehensive information
formatted_timetables = []
for timetable in timetables:
# Get supervisor information
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == timetable.created_by
).first()
supervisor_name = f"{supervisor.first_name} {supervisor.last_name}" if supervisor else "Unknown"
# Get timetable blocks count and statistics
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).all()
# Calculate statistics
total_courses = len(set(block.course_id for block in blocks))
total_tutors = len(set(block.tutor_id for block in blocks))
# Calculate total hours
total_hours = 0
for block in blocks:
if block.start_time and block.end_time:
start_dt = datetime.combine(date.today(), block.start_time)
end_dt = datetime.combine(date.today(), block.end_time)
duration = (end_dt - start_dt).total_seconds() / 3600 # Convert to hours
total_hours += duration
# Get course information for students count (approximate)
courses = ctx.session.query(Course).join(
TimetableBlock, Course.id == TimetableBlock.course_id
).filter(
TimetableBlock.timetable_id == timetable.id
).distinct().all()
total_students = sum(getattr(course, 'student_count', 0) for course in courses)
# Calculate completion rate based on blocks
completion_rate = min(100, (len(blocks) / max(1, total_courses * 5)) * 100) # Assume 5 blocks per course
# Count room assignments
room_assignments = len([block for block in blocks if block.room])
# Count conflicts (simplified)
conflict_count = 0 # This would need more complex logic to calculate actual conflicts
# Determine status based on approval and verification
if timetable.approval_status == 'approved' and getattr(timetable, 'is_verified', False):
status = 'active'
elif timetable.approval_status == 'pending':
status = 'pending'
elif timetable.approval_status == 'rejected':
status = 'rejected'
else:
status = 'draft'
# Create comprehensive timetable data
timetable_data = {
'id': timetable.id,
'name': timetable.name,
'description': getattr(timetable, 'description', ''),
'academic_year': timetable.academic_year,
'term': getattr(timetable, 'term', ''),
'semester': timetable.semester,
'is_active': getattr(timetable, 'is_active', False),
'created_at': timetable.created_at.isoformat() if timetable.created_at else None,
'updated_at': getattr(timetable, 'updated_at', timetable.created_at).isoformat() if getattr(timetable, 'updated_at', timetable.created_at) else None,
'supervisor_id': timetable.created_by,
'supervisor_name': supervisor_name,
'total_courses': total_courses,
'total_students': total_students,
'total_tutors': total_tutors,
'total_hours': round(total_hours, 1),
'status': status,
'approval_status': timetable.approval_status,
'last_modified': getattr(timetable, 'updated_at', timetable.created_at).isoformat() if getattr(timetable, 'updated_at', timetable.created_at) else None,
'version': getattr(timetable, 'version', '1.0'),
'department': getattr(supervisor, 'department', 'Unknown') if supervisor else 'Unknown',
'speciality': getattr(timetable, 'speciality', ''),
'room_assignments': room_assignments,
'conflict_count': conflict_count,
'completion_rate': round(completion_rate, 1),
'is_verified': getattr(timetable, 'is_verified', False),
'email_sent': getattr(timetable, 'email_sent', False)
}
formatted_timetables.append(timetable_data)
current_app.logger.info(f"Successfully formatted {len(formatted_timetables)} timetables for admin view")
return custom_response(
success=True,
data={
'timetables': formatted_timetables
},
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error fetching all timetables for admin: {str(e)}", exc_info=True)
return custom_response(
success=False,
data=f"Error fetching all timetables: {str(e)}",
status_code=500
)
def get_all_active_timetable_blocks(self) -> Dict:
"""
Get all blocks from active timetables for admin view
Returns:
Response with list of blocks from all active timetables
"""
with DatabaseContextManager() as ctx:
try:
# Get query parameters for filtering
speciality_id = request.args.get('speciality_id')
year_level = request.args.get('year_level')
term = request.args.get('term')
academic_year = request.args.get('academic_year')
tutor_id = request.args.get('tutor_id')
search_term = request.args.get('search_term')
current_app.logger.info(f"Fetching all active timetable blocks with filters: speciality_id={speciality_id}, year_level={year_level}, term={term}, academic_year={academic_year}, tutor_id={tutor_id}, search_term={search_term}")
# Build query for all active timetables (admin view shows all)
query = ctx.session.query(TimetableBlock).join(
Timetable, TimetableBlock.timetable_id == Timetable.id
).join(
Course, TimetableBlock.course_id == Course.id
).filter(
Timetable.is_active == True
)
# Apply filters
if speciality_id:
current_app.logger.info(f"Applying speciality filter: speciality_id={speciality_id} (type: {type(speciality_id)})")
# Check how many blocks have this speciality_id before filtering
blocks_before_filter = query.count()
current_app.logger.info(f"Blocks before speciality filter: {blocks_before_filter}")
# Try to convert speciality_id to int if it's a string
try:
speciality_id_int = int(speciality_id)
query = query.filter(Course.speciality_id == speciality_id_int)
current_app.logger.info(f"Using integer speciality_id: {speciality_id_int}")
except (ValueError, TypeError):
# If conversion fails, use as string
query = query.filter(Course.speciality_id == speciality_id)
current_app.logger.info(f"Using string speciality_id: {speciality_id}")
# Check how many blocks remain after filtering
blocks_after_filter = query.count()
current_app.logger.info(f"Blocks after speciality filter: {blocks_after_filter}")
# Also log some sample speciality_ids to debug
sample_courses = ctx.session.query(Course.speciality_id).limit(5).all()
current_app.logger.info(f"Sample speciality_ids in courses: {[course[0] for course in sample_courses]}")
# Note: year_level field doesn't exist in TimetableBlock model
# if year_level:
# query = query.filter(TimetableBlock.year_level == year_level)
if term:
query = query.filter(Timetable.term == term)
if academic_year:
query = query.filter(Timetable.academic_year == academic_year)
if tutor_id:
query = query.filter(TimetableBlock.tutor_id == tutor_id)
if search_term:
# Search in course title, room, or tutor name (via User relationship)
search_filter = or_(
Course.title.ilike(f'%{search_term}%'),
Course.code.ilike(f'%{search_term}%'),
TimetableBlock.room.ilike(f'%{search_term}%')
)
query = query.filter(search_filter)
# Execute query
blocks = query.order_by(
TimetableBlock.day_of_week,
TimetableBlock.start_time
).all()
current_app.logger.info(f"Found {len(blocks)} active timetable blocks")
# Format blocks
formatted_blocks = []
for block in blocks:
try:
block_data = self.block_manager._block_to_dict(block)
formatted_blocks.append(block_data)
except Exception as block_error:
current_app.logger.error(f"Error formatting block {block.id}: {str(block_error)}")
continue
current_app.logger.info(f"Formatted {len(formatted_blocks)} blocks successfully")
return custom_response(
success=True,
data=formatted_blocks,
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error fetching verified timetable blocks: {str(e)}")
return custom_response(
success=False,
data=f"Error fetching verified timetable blocks: {str(e)}",
status_code=500
)
def debug_verified_count(self) -> Dict:
"""
Debug method to check verified timetables and blocks count
Returns:
Response with counts of verified timetables and blocks
"""
with DatabaseContextManager() as ctx:
try:
# Count verified timetables
verified_timetables_count = ctx.session.query(Timetable).filter(
Timetable.is_verified == True,
Timetable.is_active == True
).count()
# Count all timetables
total_timetables_count = ctx.session.query(Timetable).count()
# Count timetables by status
draft_timetables_count = ctx.session.query(Timetable).filter(Timetable.is_verified == False).count()
verified_timetables_count_total = ctx.session.query(Timetable).filter(Timetable.is_verified == True).count()
inactive_timetables_count = ctx.session.query(Timetable).filter(Timetable.is_active == False).count()
# Count verified blocks
verified_blocks_count = ctx.session.query(TimetableBlock).join(
Timetable, TimetableBlock.timetable_id == Timetable.id
).filter(
Timetable.is_verified == True,
Timetable.is_active == True
).count()
# Count all blocks
total_blocks_count = ctx.session.query(TimetableBlock).count()
# Get sample verified timetables
sample_timetables = ctx.session.query(Timetable).filter(
Timetable.is_verified == True,
Timetable.is_active == True
).limit(3).all()
sample_timetables_data = []
for timetable in sample_timetables:
blocks_count = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).count()
sample_timetables_data.append({
'id': timetable.id,
'name': timetable.name,
'department': timetable.department,
'term': timetable.term,
'academic_year': timetable.academic_year,
'is_verified': timetable.is_verified,
'is_active': timetable.is_active,
'blocks_count': blocks_count,
'created_at': timetable.created_at.isoformat() if timetable.created_at else None,
'verified_at': timetable.verified_at.isoformat() if timetable.verified_at else None
})
return custom_response(
success=True,
data={
'verified_timetables_count': verified_timetables_count,
'total_timetables_count': total_timetables_count,
'verified_blocks_count': verified_blocks_count,
'total_blocks_count': total_blocks_count,
'sample_timetables': sample_timetables_data,
'timetable_status_breakdown': {
'draft_timetables': draft_timetables_count,
'verified_timetables_total': verified_timetables_count_total,
'inactive_timetables': inactive_timetables_count
}
},
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error in debug_verified_count: {str(e)}")
return custom_response(
success=False,
data=f"Error in debug: {str(e)}",
status_code=500
)
def get_student_timetable_blocks(self, student_id: str) -> Dict:
"""
Get timetable blocks filtered by student's enrolled specialities
Args:
student_id: The ID of the student
Returns:
Response with list of timetable blocks for the student's speciality
"""
with DatabaseContextManager() as ctx:
try:
# First, get the student and their speciality
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
)
if not student.speciality_id:
return custom_response(
success=False,
data="Student has no speciality assigned",
status_code=400
)
current_app.logger.info(f"Fetching timetable blocks for student {student_id} with speciality {student.speciality_id}")
# Build query for timetable blocks filtered by student's speciality
query = ctx.session.query(TimetableBlock).join(
Timetable, TimetableBlock.timetable_id == Timetable.id
).join(
Course, TimetableBlock.course_id == Course.id
).filter(
Timetable.is_active == True,
Course.speciality_id == student.speciality_id
)
# Execute query
blocks = query.order_by(
TimetableBlock.day_of_week,
TimetableBlock.start_time
).all()
current_app.logger.info(f"Found {len(blocks)} timetable blocks for student {student_id} speciality {student.speciality_id}")
# Format blocks
formatted_blocks = []
for block in blocks:
try:
block_data = self.block_manager._block_to_dict(block)
formatted_blocks.append(block_data)
except Exception as block_error:
current_app.logger.error(f"Error formatting block {block.id}: {str(block_error)}")
continue
current_app.logger.info(f"Formatted {len(formatted_blocks)} blocks successfully for student {student_id}")
return custom_response(
success=True,
data=formatted_blocks,
status_code=200
)
except Exception as e:
current_app.logger.error(f"Error fetching student timetable blocks: {str(e)}")
return custom_response(
success=False,
data=f"Error fetching student timetable blocks: {str(e)}",
status_code=500
)
class TimetablePDF(FPDF):
def __init__(self):
super().__init__(orientation='L', unit='mm', format='A3') # A3 landscape for more space
self.set_auto_page_break(auto=False) # Disable auto page break to use full height
self.set_margins(10, 10, 10) # Reasonable margins
# Color palette inspired by academic documents with green theme
self.header_color = (34, 139, 34) # Forest green for headers
self.primary_color = (0, 128, 0) # Dark green for accents
self.secondary_color = (50, 205, 50) # Lime green for highlights
self.light_green = (144, 238, 144) # Light green for backgrounds
self.medium_green = (107, 142, 35) # Olive green for borders
self.dark_text = (0, 100, 0) # Dark green text
self.white = (255, 255, 255) # White
# Add fonts (assuming they're available)
try:
current_dir = os.path.dirname(os.path.abspath(__file__))
font_dir = os.path.join(current_dir, '..', '..', 'static', 'fonts')
self.add_font('AfacadFlux', '', os.path.join(font_dir, 'AfacadFlux-Regular.ttf'), uni=True)
self.add_font('AfacadFlux', 'B', os.path.join(font_dir, 'AfacadFlux-Bold.ttf'), uni=True)
self.add_font('AfacadFlux', 'I', os.path.join(font_dir, 'AfacadFlux-Regular.ttf'), uni=True) # Use regular for italic
except:
# Fallback to standard fonts if custom fonts not available
self.set_font('Arial', '', 10)
def header(self):
# No header - completely removed
pass
def footer(self):
# No footer - completely removed
pass
def set_document_title(self, title, subtitle=None):
"""Set the document title with attractive styling"""
# Title on the left
self.set_font('AfacadFlux', 'B', 12)
self.set_text_color(self.primary_color[0], self.primary_color[1], self.primary_color[2])
self.cell(0, 15, title, 0, 0, 'L') # 0 at end means no line break
if subtitle:
# Subtitle on the right (same row)
self.set_font('AfacadFlux', '', 10)
self.set_text_color(self.dark_text[0], self.dark_text[1], self.dark_text[2])
self.cell(0, 15, subtitle, 0, 1, 'R') # 1 at end means line break after subtitle
else:
# If no subtitle, add line break after title
self.ln()
def create_professional_timetable_grid(self, blocks, year_level=None):
"""Create a professional-looking timetable grid based on the sample"""
# Group blocks by day and time for easy lookup
schedule_grid = {}
# Organize blocks by day and time
for block in blocks:
day_of_week = block.day_of_week
start_time = block.start_time.strftime('%H:%M') if hasattr(block.start_time, 'strftime') else str(block.start_time)
end_time = block.end_time.strftime('%H:%M') if hasattr(block.end_time, 'strftime') else str(block.end_time)
if day_of_week not in schedule_grid:
schedule_grid[day_of_week] = {}
time_key = f"{start_time}-{end_time}"
# Determine tutor information with fallback to supervisor tutor
tutor_name = 'Staff'
tutor_email = 'N/A'
tutor_phone = 'N/A'
if block.tutor:
tutor_name = f"{block.tutor.first_name} {block.tutor.last_name}"
tutor_email = block.tutor.email if hasattr(block.tutor, 'email') else 'N/A'
tutor_phone = block.tutor.phone if hasattr(block.tutor, 'phone') else 'N/A'
elif hasattr(block, 'supervisor_tutor') and block.supervisor_tutor:
tutor_name = f"{block.supervisor_tutor.first_name} {block.supervisor_tutor.last_name}"
tutor_email = block.supervisor_tutor.email if hasattr(block.supervisor_tutor, 'email') else 'N/A'
tutor_phone = block.supervisor_tutor.phone if hasattr(block.supervisor_tutor, 'phone') else 'N/A'
schedule_grid[day_of_week][time_key] = {
'course_code': block.course.code if block.course else 'Unknown',
'course_title': block.course.title if block.course else 'Unknown Course',
'tutor_name': tutor_name,
'tutor_email': tutor_email,
'tutor_phone': tutor_phone,
'room': block.room or 'TBA',
'start_time': start_time,
'end_time': end_time,
'block_type': block.block_type
}
# Define days (only 5 weekdays)
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
# Create time slots with 2-hour continuous blocks from 8 AM to 6 PM
time_slots = [
'8AM to 10AM', '10AM to 12PM', '12PM to 2PM', '2PM to 4PM', '4PM to 6PM'
]
# Grid dimensions - optimized for A3 landscape with 2-hour continuous blocks
time_col_width = 40 # Increased width for A3 paper
day_col_width = (407 - time_col_width) / 5 # 407mm is A3 width in landscape minus margins (420-10-10)
cell_height = 60 # Increased height to accommodate all content and match time slot height
# Draw header row (days) with professional styling
self.set_fill_color(255, 255, 255) # Plain white background
self.set_draw_color(180, 180, 180) # Gray borders
self.set_text_color(0, 0, 0) # Black text
self.set_font('AfacadFlux', 'B', 12)
self.set_line_width(0.3)
# Time column header
self.cell(time_col_width, 20, 'Time', 1, 0, 'C', True) # Increased height for A3
# Day headers
for day in days:
self.cell(day_col_width, 20, day.capitalize(), 1, 0, 'C', True) # Increased height for A3
self.ln()
# Draw time slots with plain white backgrounds and gray borders
for time_slot in time_slots:
# Time column with plain white background and gray borders
self.set_fill_color(255, 255, 255) # Plain white background
self.set_text_color(self.dark_text[0], self.dark_text[1], self.dark_text[2])
self.set_font('AfacadFlux', 'B', 12) # Increased font size for A3
self.cell(time_col_width, cell_height, time_slot, 1, 0, 'C', True)
# Day columns
for day_idx, day_name in enumerate(days):
# Check if there's a block at this time slot
block_found = False
block_data = None
# Look for blocks that match this time slot
for time_key, data in schedule_grid.get(day_idx, {}).items():
# Proper time matching for 2-hour continuous blocks with AM/PM format
try:
# Parse the block's start time
block_start_str = data['start_time']
if hasattr(block_start_str, 'strftime'):
block_start_str = block_start_str.strftime('%H:%M')
# Parse the current time slot (e.g., '8AM to 10AM')
slot_start = time_slot.split(' to ')[0].strip() # Get start time (e.g., '8AM')
slot_hour = int(slot_start.replace('AM', '').replace('PM', ''))
# Handle 12 PM case and convert to 24-hour format
if 'PM' in slot_start and slot_hour != 12:
slot_hour += 12 # Convert PM hours to 24-hour
elif 'AM' in slot_start and slot_hour == 12:
slot_hour = 0 # Convert 12 AM to 0
# Parse block start time
if ':' in str(block_start_str):
block_hour = int(str(block_start_str).split(':')[0])
else:
block_hour = int(str(block_start_str))
# Check if this time slot matches the block's start time
if slot_hour == block_hour:
block_found = True
block_data = data
break
except (ValueError, AttributeError) as e:
# If parsing fails, skip this block
continue
if block_found and block_data:
# Draw block with course information using medium gray top and lighter gray bottom
self.set_fill_color(255, 255, 255) # Plain white background for blocks
self.set_draw_color(180, 180, 180) # Gray borders for blocks
# Create a multi-cell for the block content with medium gray top and lighter gray bottom
x = self.get_x()
y = self.get_y()
# Top section - Medium gray background for course code only
self.set_fill_color(128, 128, 128) # Medium gray for top section
# Course code in medium gray background
self.set_font('AfacadFlux', 'B', 12) # Increased font size for better readability
self.set_text_color(255, 255, 255) # White text on medium gray background
self.cell(day_col_width, 18, block_data['course_code'], 0, 2, 'C', True) # Center aligned, reduced height
# Bottom section - Very light gray background for detailed information
self.set_fill_color(249, 250, 251) # Very light gray (bg-gray-50 equivalent)
# Room information in lighter gray background
self.set_font('AfacadFlux', 'I', 11) # Increased font size for better readability
self.set_text_color(0, 0, 0) # Black text on lighter gray background
room_info = block_data['room']
if len(room_info) > 30:
room_info = room_info[:27] + '...'
self.cell(day_col_width, 8, room_info, 0, 2, 'L', True) # Left aligned
# Lecturer name in lighter gray background
self.set_font('AfacadFlux', 'B', 10) # Increased font size for better readability
self.set_text_color(0, 0, 0) # Black text on lighter gray background
tutor_name = block_data['tutor_name']
if len(tutor_name) > 25:
tutor_name = tutor_name[:22] + '...'
self.cell(day_col_width, 8, tutor_name, 0, 2, 'L', True) # Left aligned
# Lecturer contact info in lighter gray background
self.set_font('AfacadFlux', '', 8) # Increased font size for better readability
self.set_text_color(0, 0, 0) # Black text on lighter gray background
tutor_email = block_data.get('tutor_email', 'N/A')
if len(tutor_email) > 30:
tutor_email = tutor_email[:27] + '...'
self.cell(day_col_width, 6, tutor_email, 0, 2, 'L', True) # Left aligned
# Course title in lighter gray background
self.set_font('AfacadFlux', '', 9) # Increased font size for better readability
self.set_text_color(0, 0, 0) # Black text on lighter gray background
course_title = block_data['course_title'].title() if block_data['course_title'] else 'Unknown Course' # Capitalized
if len(course_title) > 30:
course_title = course_title[:27] + '...'
self.cell(day_col_width, 7, course_title, 0, 2, 'L', True) # Left aligned
# Time slot in lighter gray background
self.set_font('AfacadFlux', 'B', 9) # Increased font size for better readability
self.set_text_color(0, 0, 0) # Black text on lighter gray background
# Display the actual block time instead of the time_slot
block_time = f"{block_data['start_time']}-{block_data['end_time']}"
self.cell(day_col_width, 6, block_time, 0, 2, 'L', True) # Left aligned
# Reset position
self.set_xy(x + day_col_width, y)
else:
# Empty cell with plain white background and gray borders
self.set_fill_color(255, 255, 255) # Plain white background
self.set_draw_color(180, 180, 180) # Gray borders
self.cell(day_col_width, cell_height, '', 1, 0, 'C', True)
self.ln()
# Add a key/legend for block types
self.ln(10) # Increased spacing before legend for A3
self.set_font('AfacadFlux', 'B', 12) # Increased font size for A3
self.set_text_color(self.dark_text[0], self.dark_text[1], self.dark_text[2])
self.cell(0, 8, 'LEGEND:', 0, 1, 'L') # Increased height for A3
self.set_font('AfacadFlux', '', 11) # Increased font size for A3
self.cell(0, 6, 'Lec = Lecture, Lab = Laboratory, Tut = Tutorial', 0, 1, 'L') # Increased height for A3
class TimetableGenerator:
def __init__(self):
self.block_manager = TimetableBlockManager()
self.MAX_WEEKLY_HOURS = 20 # Maximum 20 hours per week per tutor/supervisor
self.STANDARD_BLOCK_HOURS = 2 # Standard 2-hour blocks
self.MAX_COURSE_HOURS_PER_DAY = 2 # ABSOLUTE MAXIMUM: No course can exceed 2 hours per day
self.MAX_COURSE_HOURS_PER_WEEK = 6 # Maximum 6 hours for practical courses
self.MAX_NON_PRACTICAL_HOURS_PER_WEEK = 4 # Maximum 4 hours for non-practical courses
self.TEACHING_START_TIME = time(8, 0) # 8AM start
self.TEACHING_END_TIME = time(18, 0) # 6PM end
self.SEMESTER_DATE_RANGES = {
'Semester 1': '2025-01-01 - 2025-06-01',
'Semester 2': '2025-07-01 - 2025-12-31',
}
self.YEAR_LEVELS = ['First Year', 'Second Year', 'Third Year', 'Fourth Year']
def generate_timetable(self, timetable_name: str, semester: str, academic_year: str, created_by: str,
year_level: str = None, supervisor_id: str = None) -> Dict:
with DatabaseContextManager() as ctx:
try:
department = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id
).first()
# Create main timetable record
timetable = Timetable(
id=str(uuid.uuid4()),
name=timetable_name,
semester=semester,
academic_year=academic_year,
created_by=created_by,
created_at=datetime.now(),
department=department.department_name,
term=semester
)
ctx.session.add(timetable)
ctx.session.flush()
# Get all active tutors and supervisors
tutors = ctx.session.query(Tutor).filter(
Tutor.is_active == True
).all()
supervisors = ctx.session.query(Supervisor).filter(
Supervisor.is_active == True
).all()
if not tutors and not supervisors:
return custom_response(
success=False,
data="No active tutors or supervisors found",
status_code=400
)
# Get courses for this semester
semester_courses = self._get_courses_by_semester(ctx, semester, created_by)
if not semester_courses.get(semester):
return custom_response(
success=False,
data=f"No courses found for {semester}",
status_code=400
)
final_courses = semester_courses[semester]
# Validate course completeness (now flexible - warnings instead of errors)
validation_result = self._validate_course_completeness(ctx, semester, final_courses, supervisor_id)
if not validation_result['success']:
return custom_response(
success=False,
data=validation_result['data'],
status_code=validation_result['status_code']
)
# Store warnings for later inclusion in response
validation_warnings = validation_result.get('details', {}).get('warnings', [])
# Generate blocks using the enhanced strategy
generation_result = self._generate_blocks_enhanced_strategy(
ctx, timetable.id, tutors, supervisors, final_courses, semester, supervisor_id
)
if not generation_result['success']:
return custom_response(
success=False,
data=generation_result['data'],
status_code=generation_result['status_code']
)
ctx.session.commit()
# Get the complete timetable
complete_timetable = self._get_complete_timetable(ctx, timetable.id)
if not complete_timetable['success']:
return custom_response(
success=False,
data=complete_timetable['data'],
status_code=complete_timetable['status_code']
)
# Collect all warnings from validation and generation
all_warnings = validation_warnings.copy()
generation_warnings = generation_result.get('data', {}).get('warnings', [])
all_warnings.extend(generation_warnings)
# Prepare response data with warnings if any
response_data = complete_timetable['data']
if all_warnings:
response_data['warnings'] = all_warnings
response_data['message'] = f"Timetable generated successfully with {len(all_warnings)} warning(s)."
return custom_response(
success=True,
data=response_data,
status_code=201
)
except Exception as e:
ctx.session.rollback()
current_app.logger.error(f"Error generating timetable: {e}")
return custom_response(
success=False,
data=f"Failed to generate timetable: {str(e)}",
status_code=500
)
def _get_courses_by_semester(self, ctx, semester: str, supervisor_id: str = None) -> Dict:
"""Get courses grouped by semester - filtered by supervisor's department and shared courses"""
# Get supervisor's departments
supervisor_departments = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
supervisor_dept_names = [dept.department_name for dept in supervisor_departments]
# Build query for courses - include department-specific and shared courses
query = ctx.session.query(Course).filter(
and_(
Course.is_active == True,
Course.semester == semester,
or_(
# Department-specific courses
Course.supervisor_id == supervisor_id,
# Shared courses that are linked to specialities from supervisor's departments
and_(
Course.is_shared_course == True,
Course.id.in_(
ctx.session.query(course_speciality_association.c.course_id).join(
Speciality, course_speciality_association.c.speciality_id == Speciality.id
).filter(
Speciality.department.in_(supervisor_dept_names),
course_speciality_association.c.is_active == True
)
)
),
# Shared courses that are directly linked to supervisor's departments
and_(
Course.is_shared_course == True,
Course.id.in_(
ctx.session.query(course_department_association.c.course_id).filter(
course_department_association.c.department_name.in_(supervisor_dept_names),
course_department_association.c.is_active == True
)
)
)
)
)
)
courses = query.all()
current_app.logger.info(f"[COURSE FILTERING] Found {len(courses)} courses for semester {semester} and supervisor {supervisor_id}")
current_app.logger.info(f"[COURSE FILTERING] Supervisor departments: {supervisor_dept_names}")
semester_courses = {}
# Always include all courses under the requested semester for better coverage
# This ensures we don't miss courses due to semester mismatches
semester_courses[semester] = []
for course in courses:
# Calculate required blocks based on whether course has practical components
if hasattr(course, 'has_practical') and course.has_practical:
required_blocks = 3 # 3 blocks for practical courses (2 hours each = 6 hours total)
else:
required_blocks = 2 # Standard 2 blocks per course (2 hours each = 4 hours total)
# Get shared specialities and departments for this course
shared_specialities = []
shared_departments = []
sharing_capable_specialities = []
if course.is_shared_course:
# Get shared specialities (for interdisciplinary courses)
shared_specialities = [{
'id': s.id,
'name': s.name,
'department': s.department
} for s in course.shared_specialities]
# Get sharing-capable specialities (for speciality-level courses)
# Note: Currently using the same association table, but could be enhanced with a separate relationship
sharing_capable_specialities = [{
'id': s.id,
'name': s.name,
'department': s.department
} for s in course.shared_specialities] # Same as shared_specialities for now
# Get shared departments
shared_departments = [{
'department_name': cd.department_name,
'is_primary_department': cd.is_primary_department
} for cd in course.shared_departments if cd.is_active]
# Log shared course details for debugging
current_app.logger.info(f"[SHARED COURSE] Course {course.code} - Type: {getattr(course, 'shared_course_type', 'N/A')}, Level: {getattr(course, 'sharing_level', 'N/A')}")
current_app.logger.info(f"[SHARED COURSE] Specialities: {len(shared_specialities)}, Departments: {len(shared_departments)}")
semester_courses[semester].append({
'id': course.id,
'code': course.code,
'title': course.title,
'credits': course.credits,
'department': course.department,
'year_level': course.course_level,
'has_practical': getattr(course, 'has_practical', False),
'is_shared_course': getattr(course, 'is_shared_course', False),
'shared_course_type': getattr(course, 'shared_course_type', 'department_specific'),
'sharing_level': getattr(course, 'sharing_level', 'single'),
'shared_specialities': shared_specialities,
'sharing_capable_specialities': sharing_capable_specialities,
'shared_departments': shared_departments,
'required_blocks': required_blocks,
'assigned_blocks': 0,
'tutor_id': None, # Will be populated from tutor_course_association
'supervisor_id': course.supervisor_id # Get actual supervisor_id from course
})
return semester_courses
def _validate_course_completeness(self, ctx, semester: str, generated_courses: List[Dict], supervisor_id: str) -> Dict:
"""
Validate course completeness for timetable generation (now optional/flexible)
Args:
ctx: Database context
semester: Semester name
generated_courses: List of courses that will be included in timetable
supervisor_id: ID of the supervisor creating the timetable
Returns:
Validation result with success status and warnings/info
"""
try:
# Get ALL courses for this semester from the supervisor's department
all_semester_courses = ctx.session.query(Course).filter(
and_(
Course.is_active == True,
Course.supervisor_id == supervisor_id,
Course.semester == semester
)
).all()
# Count total courses for this semester
total_courses_in_semester = len(all_semester_courses)
courses_in_timetable = len(generated_courses)
current_app.logger.info(f"[COURSE VALIDATION] Semester: {semester}, Total courses: {total_courses_in_semester}, Courses in timetable: {courses_in_timetable}")
# Check for missing courses (now just warnings, not errors)
missing_courses = []
warnings = []
if courses_in_timetable != total_courses_in_semester:
generated_course_ids = {course['id'] for course in generated_courses}
for course in all_semester_courses:
if course.id not in generated_course_ids:
missing_courses.append({
'id': course.id,
'code': course.code,
'title': course.title,
'credits': course.credits
})
if missing_courses:
missing_course_list = ", ".join([f"{course['code']} ({course['title']})" for course in missing_courses])
warning_message = f"Note: {len(missing_courses)} course(s) from {semester} are not included in this timetable generation: {missing_course_list}"
warnings.append(warning_message)
current_app.logger.warning(f"[COURSE VALIDATION] {warning_message}")
# Check for courses without proper assignments (still an error)
courses_without_assignments = []
for course in generated_courses:
if not course.get('tutor_id') and not course.get('supervisor_id'):
courses_without_assignments.append({
'id': course['id'],
'code': course['code'],
'title': course['title']
})
if courses_without_assignments:
unassigned_course_list = ", ".join([f"{course['code']} ({course['title']})" for course in courses_without_assignments])
error_message = f"Cannot generate timetable: Some courses are not assigned to tutors or supervisors. Unassigned courses: {unassigned_course_list}"
current_app.logger.error(f"[COURSE VALIDATION] {error_message}")
return {
'success': False,
'data': error_message,
'status_code': 400,
'details': {
'unassigned_courses': courses_without_assignments
}
}
# Success with optional warnings
success_message = f"Course validation completed: {courses_in_timetable} courses will be included in the timetable"
if total_courses_in_semester > courses_in_timetable:
success_message += f" (out of {total_courses_in_semester} total courses for {semester})"
current_app.logger.info(f"[COURSE VALIDATION] {success_message}")
return {
'success': True,
'data': success_message,
'status_code': 200,
'details': {
'total_courses': total_courses_in_semester,
'courses_in_timetable': courses_in_timetable,
'missing_courses': missing_courses,
'warnings': warnings
}
}
except Exception as e:
current_app.logger.error(f"[COURSE VALIDATION] Error during course completeness validation: {str(e)}", exc_info=True)
return {
'success': False,
'data': f"Error validating course completeness: {str(e)}",
'status_code': 500
}
def _normalize_semester_name(self, semester: str) -> str:
"""Normalize semester names for better matching"""
if not semester:
return ""
semester_lower = semester.lower().strip()
# Map common semester variations
semester_mapping = {
'term 1': 'term 1',
'term 2': 'term 2',
'term 3': 'term 3',
'semester 1': 'term 1',
'semester 2': 'term 2',
'semester 3': 'term 3',
'first term': 'term 1',
'second term': 'term 2',
'third term': 'term 3',
'year 1': 'year 1',
'year 2': 'year 2',
'year 3': 'year 3',
'year 4': 'year 4',
'level 3': 'level 3',
'level 4': 'level 4',
'level 5': 'level 5',
'level 6': 'level 6'
}
return semester_mapping.get(semester_lower, semester_lower)
def _generate_blocks_for_semester(self, ctx, timetable_id: str, tutors: List[Tutor],
supervisors: List[Supervisor], courses: List[Dict],
supervisor_id: str = None) -> Dict:
"""Generate timetable blocks for a semester"""
try:
# Get availability slots for tutors and supervisors
tutor_availability = self._get_availability_with_2hour_slots(ctx, tutors, supervisors, supervisor_id)
if not tutor_availability:
return custom_response(
success=False,
data="No availability slots found for tutors or supervisors",
status_code=400
)
# CRITICAL: Validate that availability data actually contains slots
total_availability_slots = sum(len(slots) for slots in tutor_availability.values())
if total_availability_slots == 0:
return custom_response(
success=False,
data="No availability slots found. Please ensure tutors and supervisors have set their availability with approved time slots.",
status_code=400
)
tutors_with_availability = sum(1 for tutor in tutors if tutor_availability.get(tutor.id, []))
supervisors_with_availability = sum(1 for supervisor in supervisors if tutor_availability.get(supervisor.id, []))
for tutor in tutors:
slots = tutor_availability.get(tutor.id, [])
for supervisor in supervisors:
slots = tutor_availability.get(supervisor.id, [])
if tutors_with_availability + supervisors_with_availability == 0:
return {
'success': False,
'data': "No teachers have set availability. Please ensure tutors and supervisors set their availability before generating timetables.",
'status_code': 400
}
created_blocks = []
failed_courses = []
all_tutor_associations = ctx.session.query(tutor_course_association).all()
all_supervisor_associations = ctx.session.query(supervisor_course_association).all()
course_teacher_map = {}
for assoc in all_tutor_associations:
if assoc.course_id not in course_teacher_map:
course_teacher_map[assoc.course_id] = []
course_teacher_map[assoc.course_id].append(('tutor', assoc.tutor_id))
for assoc in all_supervisor_associations:
if assoc.course_id not in course_teacher_map:
course_teacher_map[assoc.course_id] = []
course_teacher_map[assoc.course_id].append(('supervisor', assoc.supervisor_id))
# Sort courses by priority (higher credits first)
sorted_courses = sorted(courses, key=lambda x: x.get('credits', 0), reverse=True)
for course in sorted_courses:
# Find suitable tutors for this course
suitable_tutors = self._find_suitable_tutors(ctx, course['id'], tutors, supervisor_id)
suitable_supervisors = self._find_suitable_supervisors(ctx, course['id'], supervisors, supervisor_id)
all_teachers = suitable_tutors + suitable_supervisors
if not all_teachers:
failed_courses.append({
'course': course,
'reason': 'No tutors or supervisors assigned to this course'
})
continue
# Check if any of the teachers have availability
teachers_with_availability = []
availability = tutor_availability.get(teacher.id, [])
for teacher in all_teachers:
if teacher.id in availability and availability[teacher.id]:
teachers_with_availability.append(teacher)
else:
pass
if not teachers_with_availability:
failed_courses.append({
'course': course,
'reason': 'Tutors assigned but no approved availability'
})
continue
# For regular courses, create 2 blocks on different days
if course.get('course_type') != 'practical':
blocks_created = self._schedule_regular_course(ctx, timetable_id, course, teachers_with_availability[0], availability, created_blocks)
if blocks_created:
pass
else:
current_app.logger.warning(f"Failed to create blocks for course {course['code']}")
failed_courses.append({
'course': course,
'reason': 'Failed to schedule blocks despite having available teachers'
})
else:
blocks_created = self._schedule_practical_course(ctx, timetable_id, course, teachers_with_availability[0], availability, created_blocks)
if not blocks_created:
failed_courses.append({
'course': course,
'reason': 'Failed to schedule practical blocks despite having available teachers'
})
courses_with_tutors = []
courses_without_tutors = []
for course in courses:
assigned_teachers = course_teacher_map.get(course['id'], [])
if assigned_teachers:
courses_with_tutors.append(course)
else:
courses_without_tutors.append(course)
validation_result = self._validate_timetable_conflicts(ctx, timetable_id)
for course in courses:
course_blocks = [b for b in created_blocks if b.get('course_id') == course['id']]
# Summary of failed courses with action items
if failed_courses:
# Group failed courses by reason for better analysis
failed_by_reason = {}
for failed_course in failed_courses:
reason = failed_course.get('reason', 'Unknown reason')
if reason not in failed_by_reason:
failed_by_reason[reason] = []
failed_by_reason[reason].append(failed_course)
return {
'success': True,
'data': {
'message': f'Successfully generated {len(created_blocks)} blocks using simple strategy. Teaching sessions will be created upon verification.',
'blocks_created': len(created_blocks),
'failed_courses': failed_courses,
'timetable_id': timetable_id,
'strategy': 'simple_genius_strategy',
'validation': validation_result,
'summary': {
'total_courses': len(courses),
'successful_courses': len([c for c in courses if any(b.get('course_id') == c['id'] for b in created_blocks)]),
'failed_courses_count': len(failed_courses),
'total_blocks': len(created_blocks),
'validation_passed': validation_result['is_valid'],
'next_step': 'Verify timetable to create teaching sessions and notify tutors'
}
},
'status_code': 201
}
except Exception as e:
current_app.logger.error(f"Error generating blocks: {e}")
return {
'success': False,
'data': f"Failed to generate blocks: {str(e)}",
'status_code': 500
}
def _get_availability_with_2hour_slots(self, ctx, tutors: List[Tutor], supervisors: List[Supervisor], supervisor_id: str = None) -> Dict:
"""Get availability slots in 2-hour blocks for tutors and supervisors"""
availability = {}
# Get supervisor departments for filtering
supervisor_departments = []
if supervisor_id:
supervisor_dept_records = ctx.session.query(SupervisorDepartment).filter(
SupervisorDepartment.supervisor_id == supervisor_id,
SupervisorDepartment.is_active == True
).all()
supervisor_departments = [dept.department_name for dept in supervisor_dept_records]
# Get tutor availability (filtered by department)
for tutor in tutors:
# Filter tutors by department if supervisor departments are specified
if supervisor_departments:
tutor_departments = []
if hasattr(tutor, 'departments') and tutor.departments:
tutor_departments = [dept.department_name for dept in tutor.departments if dept.is_active]
# Check if tutor has any department in common with supervisor
if not any(dept in supervisor_departments for dept in tutor_departments):
continue
# Get tutor name from users table since Tutor model doesn't have first_name/last_name
try:
user = ctx.session.query(User).filter(User.id == tutor.id).first()
tutor_name = f"{user.first_name} {user.last_name}" if user else f"Tutor {tutor.id}"
except Exception as e:
pass
tutor_name = f"Tutor {tutor.id}"
tutor_availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor.id,
TutorAvailability.is_approved == True,
TutorAvailability.is_cancelled == False
).all()
availability[tutor.id] = self._convert_to_2hour_slots(tutor_availabilities)
# Get supervisor availability (filtered by department)
for supervisor in supervisors:
# Filter supervisors by department if supervisor departments are specified
if supervisor_departments:
supervisor_departments_list = []
if hasattr(supervisor, 'departments') and supervisor.departments:
supervisor_departments_list = [dept.department_name for dept in supervisor.departments if dept.is_active]
# Check if supervisor has any department in common with the creating supervisor
if not any(dept in supervisor_departments for dept in supervisor_departments_list):
continue
# Get supervisor name from users table since Supervisor model doesn't have first_name/last_name
try:
user = ctx.session.query(User).filter(User.id == supervisor.id).first()
supervisor_name = f"{user.first_name} {user.last_name}" if user else f"Supervisor {supervisor.id}"
except Exception as e:
pass
supervisor_name = f"Supervisor {supervisor.id}"
supervisor_availabilities = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.supervisor_id == supervisor.id,
SupervisorAvailability.is_approved == True,
SupervisorAvailability.is_cancelled == False
).all()
availability[supervisor.id] = self._convert_to_2hour_slots(supervisor_availabilities)
return availability
def _convert_to_2hour_slots(self, availabilities) -> List[Dict]:
"""Convert availability slots to 2-hour blocks"""
slots = []
for availability in availabilities:
start_time = availability.start_time
end_time = availability.end_time
# Convert to 2-hour blocks
current_time = start_time
while current_time < end_time:
block_end = datetime.combine(date.today(), current_time) + timedelta(hours=2)
block_end_time = block_end.time()
if block_end_time <= end_time:
slots.append({
'day_of_week': availability.day_of_week,
'start_time': current_time.strftime('%H:%M'),
'end_time': block_end_time.strftime('%H:%M'),
'availability_id': availability.id
})
current_time = block_end_time
return slots
def _find_suitable_tutors(self, ctx, course_id: str, tutors: List[Tutor], supervisor_id: str = None) -> List[Tutor]:
"""Find suitable tutors for a course based on tutor-course associations"""
suitable_tutors = []
# First, let's check if there are any tutor-course associations at all
all_associations = ctx.session.query(tutor_course_association).all()
# Log all associations for debugging
for assoc in all_associations:
pass
for tutor in tutors:
# Get tutor name from users table since Tutor model doesn't have first_name/last_name
try:
user = ctx.session.query(User).filter(User.id == tutor.id).first()
tutor_name = f"{user.first_name} {user.last_name}" if user else f"Tutor {tutor.id}"
except Exception as e:
current_app.logger.warning(f"Could not get user info for tutor {tutor.id}: {e}")
tutor_name = f"Tutor {tutor.id}"
# Check if tutor can teach this course by looking at tutor_course_association table
tutor_course_assoc = ctx.session.query(tutor_course_association).filter(
tutor_course_association.c.tutor_id == tutor.id,
tutor_course_association.c.course_id == course_id
).first()
# Also log all associations for this tutor to debug
all_tutor_assocs = ctx.session.query(tutor_course_association).filter(
tutor_course_association.c.tutor_id == tutor.id
).all()
# Only consider tutors who are actually assigned to this course
if tutor_course_assoc:
suitable_tutors.append(tutor)
else:
pass
# If no suitable tutors found, log more details about the course
if not suitable_tutors:
current_app.logger.warning(f"No suitable tutors found for course {course_id}")
current_app.logger.warning(f"This could mean:")
current_app.logger.warning(f"1. No tutors are assigned to this course in tutor_course_association table")
current_app.logger.warning(f"2. The course_id {course_id} doesn't exist")
current_app.logger.warning(f"3. The tutor_course_association table is empty")
return suitable_tutors
def _find_suitable_supervisors(self, ctx, course_id: str, supervisors: List[Supervisor], supervisor_id: str = None) -> List[Supervisor]:
"""Find suitable supervisors for a course based on supervisor_id column in courses table"""
suitable_supervisors = []
# Check the courses table for supervisor_id instead of supervisor_course_association
course = ctx.session.query(Course).filter(Course.id == course_id).first()
if not course:
current_app.logger.warning(f"Course {course_id} not found in courses table")
return suitable_supervisors
# If course has a supervisor_id, find that supervisor
if course.supervisor_id:
for supervisor in supervisors:
# Get supervisor name from users table since Supervisor model doesn't have first_name/last_name
try:
user = ctx.session.query(User).filter(User.id == supervisor.id).first()
supervisor_name = f"{user.first_name} {user.last_name}" if user else f"Supervisor {supervisor.id}"
except Exception as e:
current_app.logger.warning(f"Could not get user info for supervisor {supervisor.id}: {e}")
supervisor_name = f"Supervisor {supervisor.id}"
# Check if this supervisor is assigned to this course via supervisor_id column
if supervisor.id == course.supervisor_id:
suitable_supervisors.append(supervisor)
else:
pass
else:
pass
# If no suitable supervisors found, log more details about the course
if not suitable_supervisors:
current_app.logger.warning(f"No suitable supervisors found for course {course_id}")
current_app.logger.warning(f"This could mean:")
current_app.logger.warning(f"1. The course.supervisor_id is NULL or empty")
current_app.logger.warning(f"2. The assigned supervisor is not in the active supervisors list")
current_app.logger.warning(f"3. The supervisor_id doesn't match any supervisor in the system")
return suitable_supervisors
def _can_assign_slot(self, teacher_id: str, slot: Dict, course: Dict,
teacher_weekly_hours: Dict, teacher_daily_schedule: Dict,
course_daily_hours: Dict, day: int) -> bool:
"""Check if a slot can be assigned to a teacher for a course"""
# Check teacher weekly hours limit - STRICTLY enforce 20 hours maximum per week
current_weekly_hours = teacher_weekly_hours.get(teacher_id, 0)
if current_weekly_hours + 2 > 20: # 20 hours maximum per week
return False
# STRICTLY enforce course daily hours limit - maximum 2 hours per course per day (no 4-hour blocks)
course_max_hours_per_day = 2 # Hard limit: no course can exceed 2 hours per day
current_course_hours_today = course_daily_hours.get(course['id'], {}).get(day, 0)
if current_course_hours_today + 2 > course_max_hours_per_day:
return False
# For practical courses, ensure they only appear on the same day (continuous 6-hour block)
if course.get('has_practical'):
# Check if this course already has blocks on different days
existing_days = set()
for d in range(5): # Monday to Friday
day_schedule = teacher_daily_schedule.get(teacher_id, {}).get(d, [])
for existing_slot in day_schedule:
if existing_slot.get('course_id') == course['id']:
existing_days.add(d)
# If course already has blocks on different days, don't allow new day
if existing_days and day not in existing_days:
return False
# For practical courses, ensure we don't exceed 6 hours total
course_total_hours = sum(course_daily_hours.get(course['id'], {}).values())
if course_total_hours + 2 > 6: # Maximum 6 hours for practical courses
return False
else:
# For non-practical courses, ensure we don't exceed 4 hours total
course_total_hours = sum(course_daily_hours.get(course['id'], {}).values())
if course_total_hours + 2 > 4: # Maximum 4 hours for non-practical courses
return False
# Check teacher daily schedule conflicts
day_schedule = teacher_daily_schedule.get(teacher_id, {}).get(day, [])
for existing_slot in day_schedule:
# Check for time overlap
existing_start = datetime.strptime(existing_slot['start_time'], '%H:%M').time()
existing_end = datetime.strptime(existing_slot['end_time'], '%H:%M').time()
new_start = datetime.strptime(slot['start_time'], '%H:%M').time()
new_end = datetime.strptime(slot['end_time'], '%H:%M').time()
if (new_start < existing_end and new_end > existing_start):
return False
return True
def _create_timetable_block(self, ctx, timetable_id: str, course: Dict,
teacher, slot: Dict) -> Optional[Dict]:
"""Create a timetable block"""
try:
# Safely get teacher ID
teacher_id = None
if hasattr(teacher, 'id'):
teacher_id = teacher.id
elif hasattr(teacher, 'tutor_id'):
teacher_id = teacher.tutor_id
elif hasattr(teacher, 'supervisor_id'):
teacher_id = teacher.supervisor_id
else:
current_app.logger.error(f"Cannot determine teacher ID for teacher: {teacher}")
return None
# Determine block type based on practical status
block_type = 'lab' if course.get('has_practical') else 'lecture'
block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course['id'],
day_of_week=slot['day_of_week'],
start_time=datetime.strptime(slot['start_time'], '%H:%M').time(),
end_time=datetime.strptime(slot['end_time'], '%H:%M').time(),
block_type=block_type,
recurring=True,
created_at=datetime.utcnow(),
created_by=teacher_id
)
# Set tutor or supervisor based on teacher type
if hasattr(teacher, 'tutor_id') or hasattr(teacher, 'first_name'): # Assume it's a tutor
block.tutor_id = teacher_id
else:
block.supervisor_tutor_id = teacher_id
ctx.session.add(block)
ctx.session.flush()
return self.block_manager._block_to_dict(block)
except Exception as e:
current_app.logger.error(f"Error creating block: {e}")
current_app.logger.error(f"Teacher object: {teacher}")
current_app.logger.error(f"Course: {course}")
current_app.logger.error(f"Slot: {slot}")
return None
def _create_teaching_sessions(self, ctx, timetable_id: str, blocks: List[Dict]) -> Dict:
"""Create teaching sessions from timetable blocks"""
sessions_created = 0
for block in blocks:
try:
# Create teaching session for each block
# This would typically create sessions for the entire semester
# For now, we'll create a sample session
course_code = block.get('course_code', 'Unknown Course')
session = TeachingSession(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
title=f"{course_code} Session",
description=f"Regular teaching session for {course_code}",
session_type='lecture',
status='scheduled',
created_at=datetime.now(),
updated_at=datetime.now()
)
ctx.session.add(session)
sessions_created += 1
except Exception as e:
current_app.logger.error(f"Error creating teaching session: {e}")
current_app.logger.error(f"Block data: {block}")
continue
ctx.session.flush()
return {
'success': True,
'sessions_created': sessions_created
}
def _get_complete_timetable(self, ctx, timetable_id: str) -> Dict:
"""Get the complete generated timetable"""
timetable = ctx.session.query(Timetable).filter(
Timetable.id == timetable_id
).first()
if not timetable:
return {
'success': False,
'data': "Timetable not found",
'status_code': 404
}
# Get all blocks
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id
).all()
# Get statistics
stats = {
'total_blocks': len(blocks),
'courses_scheduled': len(set(block.course_id for block in blocks)),
'tutors_involved': len(set(block.tutor_id for block in blocks)),
'weekly_hours': len(blocks) * self.STANDARD_BLOCK_HOURS
}
return {
'success': True,
'data': {
'timetable': {
'id': timetable.id,
'name': timetable.name,
'semester': timetable.semester,
'academic_year': timetable.academic_year,
'created_at': timetable.created_at.isoformat(),
'blocks_count': len(blocks),
'statistics': stats
},
'blocks': [self.block_manager._block_to_dict(block) for block in blocks]
},
'status_code': 201
}
def _ensure_practical_course_continuity(self, ctx, course: Dict, teacher, day: int,
teacher_daily_schedule: Dict, course_daily_hours: Dict) -> bool:
"""
Ensure practical courses are scheduled in continuous blocks on the same day
Args:
course: Course data
teacher: Teacher (tutor or supervisor)
day: Day of the week
teacher_daily_schedule: Current teacher's daily schedule
course_daily_hours: Current course's daily hours
Returns:
True if the course can be scheduled on this day, False otherwise
"""
if not course.get('has_practical'):
return True # Non-practical courses don't need continuity checks
# For practical courses, check if we can create a continuous 6-hour block
current_day_hours = course_daily_hours.get(course['id'], {}).get(day, 0)
# If this is the first block for this course on this day, check if we can fit 3 blocks
if current_day_hours == 0:
# Check if teacher has enough consecutive availability on this day
day_schedule = teacher_daily_schedule.get(teacher.id, {}).get(day, [])
# Count available 2-hour slots on this day
available_slots = len(day_schedule)
# For practical courses, we need 3 consecutive 2-hour slots
if available_slots < 3:
return False
return True
# If this course already has blocks on this day, ensure we don't exceed 6 hours
elif current_day_hours >= 6:
return False
# If this course has blocks on this day but less than 6 hours, allow continuation
else:
return True
def _get_optimal_day_distribution(self, course: Dict, available_days: List[int],
course_daily_hours: Dict, used_days: set) -> List[int]:
"""
Get optimal day distribution for a course to ensure blocks are spread across different days
Args:
course: Course data
available_days: List of available days
course_daily_hours: Current course daily hours
used_days: Days already used by this course
Returns:
List of days in optimal order for scheduling
"""
if not course.get('has_practical'):
# For non-practical courses, prioritize days we haven't used yet
unused_days = [day for day in available_days if day not in used_days]
used_days_list = [day for day in available_days if day in used_days]
# Randomize both lists for variety
import random
random.shuffle(unused_days)
random.shuffle(used_days_list)
# Return unused days first, then used days
return unused_days + used_days_list
else:
# For practical courses, if we have blocks on a day, prioritize that day
# Otherwise, prioritize days with enough availability for 3 blocks
if used_days:
# Continue on the same day for continuity
used_day = list(used_days)[0]
if used_day in available_days:
return [used_day] + [day for day in available_days if day != used_day]
# For new practical courses, randomize available days
import random
randomized_days = available_days.copy()
random.shuffle(randomized_days)
return randomized_days
def _should_avoid_day_clustering(self, course: Dict, day: int, course_daily_hours: Dict,
used_days: set, current_blocks_created: int) -> bool:
"""
Check if we should avoid clustering blocks on the same day for non-practical courses
Args:
course: Course data
day: Day to check
course_daily_hours: Current course daily hours
used_days: Days already used by this course
current_blocks_created: Number of blocks already created for this course
Returns:
True if we should avoid this day to prevent clustering
"""
if course.get('has_practical'):
return False # Practical courses need to cluster on the same day
# For non-practical courses, encourage spreading across different days
if current_blocks_created == 0:
return False # First block can go anywhere
# If we already have blocks on this day, try to avoid adding more
if day in used_days:
# Allow up to 2 blocks per day for non-practical courses
current_day_hours = course_daily_hours.get(course['id'], {}).get(day, 0)
if current_day_hours >= 2: # Already have 2 hours on this day
return True # Avoid clustering
# If we have multiple days available, prefer unused days
if len(used_days) > 0 and day in used_days:
# Check if there are unused days available
total_days = 5 # Monday to Friday
unused_days = set(range(total_days)) - used_days
if unused_days:
return True # Prefer unused days
return False
def _enforce_strict_daily_limit(self, course: Dict, day: int, course_daily_hours: Dict) -> bool:
"""
Strictly enforce the 2-hour daily limit for ALL courses
Args:
course: Course data
day: Day to check
course_daily_hours: Current course daily hours
Returns:
True if the course can be scheduled on this day, False if it would exceed 2 hours
"""
current_day_hours = course_daily_hours.get(course['id'], {}).get(day, 0)
# ABSOLUTE RULE: No course can exceed 2 hours per day
if current_day_hours >= 2:
return False
return True
def generate_simple_timetable(self, timetable_name: str, semester: str, academic_year: str, created_by: str,
year_level: str = None, supervisor_id: str = None) -> Dict:
"""
Generate a simple timetable using the genius strategy:
- Regular courses: 2 timeblocks on different days with randomized times
- Practical courses: 1 continuous 6-hour block (3 x 2-hour slots) on one day per week
Args:
timetable_name: Name of the timetable
semester: Semester (e.g., 'Term 1', 'Term 2')
academic_year: Academic year (e.g., '2024-2025')
created_by: ID of user creating the timetable
year_level: Year level (ignored - only semester is used for filtering)
supervisor_id: Supervisor ID for supervisor-specific timetables
Returns:
Response with generated timetable or error
"""
with DatabaseContextManager() as ctx:
try:
# Create main timetable record
timetable = Timetable(
id=str(uuid.uuid4()),
name=timetable_name,
semester=semester,
academic_year=academic_year,
created_by=created_by,
created_at=datetime.now(),
)
ctx.session.add(timetable)
ctx.session.flush()
# Get all active tutors and supervisors
tutors = ctx.session.query(Tutor).filter(
Tutor.is_active == True
).all()
supervisors = ctx.session.query(Supervisor).filter(
Supervisor.is_active == True
).all()
if not tutors and not supervisors:
return custom_response(
success=False,
data="No active tutors or supervisors found",
status_code=400
)
# Get courses for this semester
semester_courses = self._get_courses_by_semester(ctx, semester, created_by)
if not semester_courses.get(semester):
return custom_response(
success=False,
data=f"No courses found for {semester}",
status_code=400
)
# Generate blocks using the enhanced strategy
generation_result = self._generate_blocks_enhanced_strategy(
ctx, timetable.id, tutors, supervisors, semester_courses[semester], semester, supervisor_id
)
if not generation_result['success']:
return custom_response(
success=False,
data=generation_result['data'],
status_code=generation_result['status_code']
)
ctx.session.commit()
# Get the complete timetable
complete_timetable = self._get_complete_timetable(ctx, timetable.id)
if not complete_timetable['success']:
return custom_response(
success=False,
data=complete_timetable['data'],
status_code=complete_timetable['status_code']
)
except Exception as e:
ctx.session.rollback()
current_app.logger.error(f"Error generating timetable: {e}")
return custom_response(
success=False,
data=f"Failed to generate timetable: {str(e)}",
status_code=500
)
def _generate_blocks_enhanced_strategy(self, ctx, timetable_id: str, tutors: List[Tutor],
supervisors: List[Supervisor], courses: List[Dict],
semester: str, supervisor_id: str = None) -> Dict:
"""
Enhanced timetable generation algorithm with speciality-level constraints:
1. Speciality-level scheduling (max 2 courses per speciality at same time)
2. Course collision prevention within specialities
3. Shared course locking across departments/specialities
4. Collision minimization algorithm
5. All courses scheduling (not just 4 out of 10+)
6. Supervisor course assignments
7. Tutor time conflicts (no overlapping classes)
8. Tutor consecutive class restrictions
9. Booked supervisor sessions
10. Tutor-course associations
"""
try:
# Step 1: Get all tutor-course associations
tutor_course_assignments = self._get_tutor_course_assignments(ctx, supervisor_id)
# Step 2: Get supervisor course assignments
supervisor_course_assignments = self._get_supervisor_course_assignments(ctx, supervisor_id)
# Step 3: Get existing booked sessions (supervisor sessions)
booked_sessions = self._get_booked_supervisor_sessions(ctx)
# Step 4: Get tutor availability with conflict checking
availability = self._get_enhanced_availability(ctx, tutors, supervisors, booked_sessions, supervisor_id)
# Step 5: Initialize speciality-level constraint tracking
speciality_schedule_matrix = self._initialize_speciality_schedule_matrix(ctx, courses)
# Step 6: Initialize tutor-level conflict prevention matrix
tutor_schedule_matrix = self._initialize_tutor_schedule_matrix(ctx, tutors, supervisors)
# Step 7: Process shared courses first to lock their time slots
shared_courses, regular_courses = self._categorize_courses_by_sharing(courses)
locked_time_slots = self._process_shared_courses_first(ctx, shared_courses, speciality_schedule_matrix)
if not availability:
return {
'success': False,
'data': "No availability slots found for tutors or supervisors",
'status_code': 400
}
# Step 7: Create availability matrix for conflict checking
availability_matrix = self._create_availability_matrix(availability, booked_sessions)
created_blocks = []
failed_courses = []
# Step 8: Process shared courses first (they have locked time slots)
for course in shared_courses:
course_tutors = tutor_course_assignments.get(course['id'], [])
course_supervisors = supervisor_course_assignments.get(course['id'], [])
all_teachers = course_tutors + course_supervisors
if not all_teachers:
failed_courses.append({
'course': course,
'reason': 'No tutors or supervisors assigned to shared course'
})
continue
# Find teachers with availability
available_teachers = [t for t in all_teachers if t.id in availability and availability[t.id]]
if not available_teachers:
failed_courses.append({
'course': course,
'reason': 'Assigned teachers have no availability for shared course'
})
continue
# Schedule shared course with locked time slots
blocks_created = self._schedule_shared_course_with_constraints(
ctx, timetable_id, course, available_teachers,
availability, speciality_schedule_matrix, tutor_schedule_matrix, locked_time_slots,
created_blocks, semester, supervisor_id
)
if not blocks_created:
failed_courses.append({
'course': course,
'reason': 'Failed to schedule shared course with locked time slots'
})
# Step 9: Process regular courses with speciality constraints
for i, course in enumerate(regular_courses, 1):
# Find assigned tutors for this course
course_tutors = tutor_course_assignments.get(course['id'], [])
course_supervisors = supervisor_course_assignments.get(course['id'], [])
if not course_tutors and not course_supervisors:
failed_courses.append({
'course': course,
'reason': 'No tutors or supervisors assigned to this course'
})
continue
# Combine tutors and supervisors
all_teachers = course_tutors + course_supervisors
# Find teachers with availability
available_teachers = []
teachers_without_availability = []
for teacher in all_teachers:
if teacher.id in availability and availability[teacher.id]:
available_teachers.append(teacher)
else:
teachers_without_availability.append({
'id': teacher.id,
'name': f"{teacher.first_name} {teacher.last_name}",
'has_availability_record': teacher.id in availability,
'availability_count': len(availability.get(teacher.id, []))
})
if not available_teachers:
# Try to create default availability for teachers without availability records
current_app.logger.warning(f"[COURSE ASSIGNMENT] Course {course['code']} ({course['id']}) - No teachers with availability. Attempting to create default availability.")
for teacher_info in teachers_without_availability:
teacher_id = teacher_info['id']
# Create default availability (Monday-Friday, 8AM-6PM)
default_availability = self._create_default_availability_for_teacher(teacher_id)
if default_availability:
availability[teacher_id] = default_availability
available_teachers.append(next(t for t in all_teachers if t.id == teacher_id))
current_app.logger.info(f"[DEFAULT AVAILABILITY] Created default availability for teacher {teacher_id}")
if not available_teachers:
current_app.logger.error(f"[COURSE ASSIGNMENT] Course {course['code']} ({course['id']}) - Failed to create default availability. Teachers assigned: {teachers_without_availability}")
failed_courses.append({
'course': course,
'reason': 'Assigned teachers have no availability and default availability creation failed',
'teachers_without_availability': teachers_without_availability
})
continue
# Sort teachers by current workload (least loaded first) for better distribution
teacher_workloads = {}
for teacher in available_teachers:
# Count how many courses this teacher is already assigned to
workload = sum(1 for block in created_blocks if
(block.get('tutor_id') == teacher.id or
block.get('supervisor_tutor_id') == teacher.id))
teacher_workloads[teacher.id] = workload
# Sort by workload (least loaded first), then by availability
available_teachers.sort(key=lambda t: (teacher_workloads[t.id], -len(availability[t.id])))
current_app.logger.info(f"[TEACHER ASSIGNMENT] Course {course['id']} - Teacher workloads: {[(t.id, teacher_workloads[t.id], len(availability[t.id])) for t in available_teachers[:3]]}")
# Add randomization to avoid predictable teacher selection
import random
# Keep the first 3 least loaded teachers, then randomize the rest
if len(available_teachers) > 3:
top_teachers = available_teachers[:3]
remaining_teachers = available_teachers[3:]
random.shuffle(remaining_teachers)
available_teachers = top_teachers + remaining_teachers
# Try to schedule the course with speciality constraints
blocks_created = self._schedule_course_with_speciality_constraints(
ctx, timetable_id, course, available_teachers,
availability, speciality_schedule_matrix, tutor_schedule_matrix, created_blocks,
semester, supervisor_id
)
if blocks_created:
pass # Successfully scheduled
else:
failed_courses.append({
'course': course,
'reason': 'No suitable time slots available'
})
# Validate the entire timetable for conflicts
validation_result = self._validate_timetable_conflicts(ctx, timetable_id)
if not validation_result['is_valid']:
current_app.logger.error(f"Timetable validation failed: Found {validation_result['conflict_count']} conflicts")
# Continue with generation but log the conflicts
# Summary of failed courses with action items
if failed_courses:
# Group failed courses by reason for better analysis
failed_by_reason = {}
for failed_course in failed_courses:
reason = failed_course.get('reason', 'Unknown reason')
if reason not in failed_by_reason:
failed_by_reason[reason] = []
failed_by_reason[reason].append(failed_course)
# Additional validation: Check course scheduling success (now flexible)
successful_courses = len([c for c in courses if any(b.get('course_id') == c['id'] for b in created_blocks)])
total_courses = len(courses)
# Generate warnings for failed courses instead of failing the entire generation
warnings = []
if successful_courses != total_courses:
failed_course_list = ", ".join([f"{fc['course']['code']} ({fc['reason']})" for fc in failed_courses])
warning_message = f"Note: {total_courses - successful_courses} course(s) could not be scheduled: {failed_course_list}"
warnings.append(warning_message)
current_app.logger.warning(f"[TIMETABLE GENERATION] {warning_message}")
# Still return success but with warnings
return {
'success': True,
'data': {
'timetable_id': timetable_id,
'blocks_created': len(created_blocks),
'courses_scheduled': successful_courses,
'total_courses': total_courses,
'warnings': warnings,
'message': f"Timetable generated successfully with {successful_courses} out of {total_courses} courses scheduled."
},
'status_code': 201
}
return {
'success': True,
'data': {
'message': f'Successfully generated {len(created_blocks)} blocks using enhanced strategy. All {total_courses} courses have been scheduled. Teaching sessions will be created upon verification.',
'blocks_created': len(created_blocks),
'failed_courses': failed_courses,
'timetable_id': timetable_id,
'strategy': 'enhanced_strategy',
'validation': validation_result,
'summary': {
'total_courses': len(courses),
'successful_courses': len([c for c in courses if any(b.get('course_id') == c['id'] for b in created_blocks)]),
'failed_courses_count': len(failed_courses),
'total_blocks': len(created_blocks),
'validation_passed': validation_result['is_valid'],
'next_step': 'Verify timetable to create teaching sessions and notify tutors'
}
},
'status_code': 201
}
except Exception as e:
current_app.logger.error(f"Error in enhanced strategy generation: {e}")
return {
'success': False,
'data': f"Failed to generate timetable using enhanced strategy: {str(e)}",
'status_code': 500
}
def _initialize_speciality_schedule_matrix(self, ctx, courses: List[Dict]) -> Dict:
"""
Initialize a matrix to track speciality-level scheduling constraints.
Returns:
Dict with structure: {
'speciality_id': {
'day_of_week': {
'time_slot': {
'courses': [course_ids],
'max_courses': 2,
'locked': False
}
}
}
}
"""
try:
speciality_matrix = {}
# Get all specialities from courses
course_speciality_map = {}
for course in courses:
course_id = course['id']
# Get specialities for this course
specialities = ctx.session.query(Speciality).join(
course_speciality_association,
Speciality.id == course_speciality_association.c.speciality_id
).filter(
course_speciality_association.c.course_id == course_id,
course_speciality_association.c.is_active == True,
Speciality.is_active == True
).all()
course_speciality_map[course_id] = specialities
# Initialize matrix for each speciality
for speciality in specialities:
if speciality.id not in speciality_matrix:
speciality_matrix[speciality.id] = {
'name': speciality.name,
'department': speciality.department,
'schedule': {}
}
# Initialize time slots for each speciality
time_slots = self._generate_time_slots()
days_of_week = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
for speciality_id in speciality_matrix:
for day in days_of_week:
speciality_matrix[speciality_id]['schedule'][day] = {}
for time_slot in time_slots:
speciality_matrix[speciality_id]['schedule'][day][time_slot] = {
'courses': [],
'max_courses': 2, # Maximum 2 courses per speciality at same time
'locked': False,
'shared_course_id': None
}
current_app.logger.info(f"[SPECIALITY MATRIX] Initialized matrix for {len(speciality_matrix)} specialities")
return speciality_matrix
except Exception as e:
current_app.logger.error(f"Error initializing speciality schedule matrix: {str(e)}")
return {}
def _categorize_courses_by_sharing(self, courses: List[Dict]) -> tuple:
"""
Categorize courses into shared and regular courses.
Returns:
tuple: (shared_courses, regular_courses)
"""
shared_courses = []
regular_courses = []
for course in courses:
if course.get('is_shared_course', False):
shared_courses.append(course)
else:
regular_courses.append(course)
current_app.logger.info(f"[COURSE CATEGORIZATION] Shared courses: {len(shared_courses)}, Regular courses: {len(regular_courses)}")
return shared_courses, regular_courses
def _process_shared_courses_first(self, ctx, shared_courses: List[Dict], speciality_matrix: Dict) -> Dict:
"""
Process shared courses first and lock their time slots across all specialities.
Returns:
Dict of locked time slots: {
'course_id': {
'day': 'Monday',
'time_slot': '08:00-10:00',
'specialities': [speciality_ids]
}
}
"""
try:
locked_slots = {}
for course in shared_courses:
course_id = course['id']
course_specialities = course.get('shared_specialities', [])
if not course_specialities:
continue
# Find optimal time slot for this shared course
optimal_slot = self._find_optimal_shared_course_slot(
ctx, course, course_specialities, speciality_matrix
)
if optimal_slot:
locked_slots[course_id] = optimal_slot
# Lock the time slot across all specialities for this course
for speciality_info in course_specialities:
speciality_id = speciality_info['id']
if speciality_id in speciality_matrix:
day = optimal_slot['day']
time_slot = optimal_slot['time_slot']
speciality_matrix[speciality_id]['schedule'][day][time_slot]['locked'] = True
speciality_matrix[speciality_id]['schedule'][day][time_slot]['shared_course_id'] = course_id
speciality_matrix[speciality_id]['schedule'][day][time_slot]['courses'].append(course_id)
current_app.logger.info(f"[SHARED COURSE LOCK] Course {course['code']} locked at {optimal_slot['day']} {optimal_slot['time_slot']}")
return locked_slots
except Exception as e:
current_app.logger.error(f"Error processing shared courses: {str(e)}")
return {}
def _find_optimal_shared_course_slot(self, ctx, course: Dict, course_specialities: List[Dict], speciality_matrix: Dict) -> Dict:
"""
Find the optimal time slot for a shared course that minimizes conflicts.
Returns:
Dict with optimal slot info or None if no suitable slot found
"""
try:
# Get time slots
time_slots = self._generate_time_slots()
days_of_week = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
best_slot = None
min_conflicts = float('inf')
for day in days_of_week:
for time_slot in time_slots:
conflicts = 0
can_schedule = True
# Check if this slot is available in all specialities
for speciality_info in course_specialities:
speciality_id = speciality_info['id']
if speciality_id not in speciality_matrix:
continue
slot_info = speciality_matrix[speciality_id]['schedule'][day][time_slot]
# Check if slot is already locked or at capacity
if slot_info['locked'] or len(slot_info['courses']) >= slot_info['max_courses']:
can_schedule = False
break
# Count existing courses in this slot
conflicts += len(slot_info['courses'])
if can_schedule and conflicts < min_conflicts:
min_conflicts = conflicts
best_slot = {
'day': day,
'time_slot': time_slot,
'conflicts': conflicts,
'specialities': [s['id'] for s in course_specialities]
}
return best_slot
except Exception as e:
current_app.logger.error(f"Error finding optimal shared course slot: {str(e)}")
return None
def _schedule_shared_course_with_constraints(self, ctx, timetable_id: str, course: Dict,
available_teachers: List, availability: Dict,
speciality_matrix: Dict, tutor_matrix: Dict, locked_slots: Dict,
created_blocks: List, semester: str, supervisor_id: str) -> bool:
"""
Schedule a shared course using its locked time slots.
Returns:
bool: True if successfully scheduled, False otherwise
"""
try:
course_id = course['id']
if course_id not in locked_slots:
current_app.logger.warning(f"[SHARED COURSE] No locked slot found for course {course['code']}")
return False
locked_slot = locked_slots[course_id]
day = locked_slot['day']
time_slot = locked_slot['time_slot']
# Find teacher available at this specific time with tutor conflict checking
suitable_teacher = None
for teacher in available_teachers:
teacher_id = teacher.id
teacher_slots = availability.get(teacher_id, [])
# Check if teacher has availability at this specific time
has_availability = False
for slot in teacher_slots:
if (slot['day_of_week'] == day and
slot['start_time'] == time_slot.split('-')[0] and
slot['end_time'] == time_slot.split('-')[1]):
has_availability = True
break
if has_availability:
# Check for tutor conflicts
tutor_conflict = self._check_tutor_conflict(
teacher_id, day, time_slot.split('-')[0], time_slot.split('-')[1],
tutor_matrix, course_id
)
if not tutor_conflict['has_conflict']:
suitable_teacher = teacher
break
else:
current_app.logger.debug(f"[SHARED COURSE] Teacher {teacher_id} has conflict at {day} {time_slot}: {tutor_conflict['message']}")
if not suitable_teacher:
current_app.logger.warning(f"[SHARED COURSE] No teacher available without conflicts for course {course['code']} at locked time {day} {time_slot}")
return False
# Create the timetable block
slot_data = {
'day_of_week': day,
'start_time': time_slot.split('-')[0],
'end_time': time_slot.split('-')[1]
}
block = self._create_timetable_block(ctx, timetable_id, course, suitable_teacher, slot_data)
if block:
created_blocks.append(block)
# Update tutor matrix to prevent future conflicts
self._update_tutor_matrix(
suitable_teacher.id, day, time_slot.split('-')[0], time_slot.split('-')[1],
course_id, tutor_matrix, 'add'
)
current_app.logger.info(f"[SHARED COURSE] Successfully scheduled {course['code']} at {day} {time_slot}")
return True
return False
except Exception as e:
current_app.logger.error(f"Error scheduling shared course: {str(e)}")
return False
def _schedule_course_with_speciality_constraints(self, ctx, timetable_id: str, course: Dict,
available_teachers: List, availability: Dict,
speciality_matrix: Dict, tutor_matrix: Dict, created_blocks: List,
semester: str, supervisor_id: str) -> bool:
"""
Schedule a regular course with speciality-level constraints.
Returns:
bool: True if successfully scheduled, False otherwise
"""
try:
course_id = course['id']
# Get specialities for this course
course_specialities = ctx.session.query(Speciality).join(
course_speciality_association,
Speciality.id == course_speciality_association.c.speciality_id
).filter(
course_speciality_association.c.course_id == course_id,
course_speciality_association.c.is_active == True,
Speciality.is_active == True
).all()
if not course_specialities:
# Course has no speciality constraints, use regular scheduling
return self._schedule_course_enhanced(
ctx, timetable_id, course, available_teachers,
availability, {}, created_blocks, semester, supervisor_id
)
# Find suitable time slots considering both speciality and tutor constraints
suitable_slots = self._find_tutor_compatible_slots(
course, available_teachers, availability, tutor_matrix,
speciality_matrix, created_blocks
)
if not suitable_slots:
current_app.logger.warning(f"[SPECIALITY CONSTRAINT] No suitable slots found for course {course['code']}")
return False
# Schedule the course using the best available slot
blocks_created = 0
required_blocks = course.get('required_blocks', 2)
for slot_info in suitable_slots[:required_blocks]:
teacher = slot_info['teacher']
slot_data = slot_info['slot']
# Check if this slot violates speciality constraints
if not self._check_speciality_constraints_violation(
course_id, course_specialities, slot_data, speciality_matrix
):
block = self._create_timetable_block(ctx, timetable_id, course, teacher, slot_data)
if block:
created_blocks.append(block)
blocks_created += 1
# Update speciality matrix
self._update_speciality_matrix(
course_id, course_specialities, slot_data, speciality_matrix
)
# Update tutor matrix to prevent future conflicts
self._update_tutor_matrix(
teacher.id, slot_data['day_of_week'], slot_data['start_time'], slot_data['end_time'],
course_id, tutor_matrix, 'add'
)
current_app.logger.info(f"[TUTOR CONSTRAINT] Scheduled {course['code']} at {slot_data['day_of_week']} {slot_data['start_time']}-{slot_data['end_time']}")
return blocks_created >= required_blocks
except Exception as e:
current_app.logger.error(f"Error scheduling course with speciality constraints: {str(e)}")
return False
def _find_speciality_compatible_slots(self, course: Dict, course_specialities: List,
available_teachers: List, availability: Dict,
speciality_matrix: Dict, created_blocks: List) -> List[Dict]:
"""
Find time slots that are compatible with speciality constraints.
Returns:
List of compatible slot information
"""
try:
compatible_slots = []
for teacher in available_teachers:
teacher_slots = availability.get(teacher.id, [])
for slot in teacher_slots:
day = slot['day_of_week']
start_time = slot['start_time']
end_time = slot['end_time']
time_slot = f"{start_time}-{end_time}"
# Check if this slot is compatible with speciality constraints
is_compatible = True
conflicts = 0
for speciality in course_specialities:
speciality_id = speciality.id
if speciality_id not in speciality_matrix:
continue
slot_info = speciality_matrix[speciality_id]['schedule'][day][time_slot]
# Check if slot is locked or at capacity
if slot_info['locked'] or len(slot_info['courses']) >= slot_info['max_courses']:
is_compatible = False
break
conflicts += len(slot_info['courses'])
if is_compatible:
compatible_slots.append({
'teacher': teacher,
'slot': slot,
'conflicts': conflicts,
'speciality_count': len(course_specialities)
})
# Sort by conflicts (fewer conflicts first) and speciality count
compatible_slots.sort(key=lambda x: (x['conflicts'], x['speciality_count']))
return compatible_slots
except Exception as e:
current_app.logger.error(f"Error finding speciality compatible slots: {str(e)}")
return []
def _check_speciality_constraints_violation(self, course_id: str, course_specialities: List,
slot_data: Dict, speciality_matrix: Dict) -> bool:
"""
Check if scheduling a course at a specific slot would violate speciality constraints.
Returns:
bool: True if violation detected, False otherwise
"""
try:
day = slot_data['day_of_week']
start_time = slot_data['start_time']
end_time = slot_data['end_time']
time_slot = f"{start_time}-{end_time}"
for speciality in course_specialities:
speciality_id = speciality.id
if speciality_id not in speciality_matrix:
continue
slot_info = speciality_matrix[speciality_id]['schedule'][day][time_slot]
# Check if adding this course would exceed the limit
if len(slot_info['courses']) >= slot_info['max_courses']:
return True
# Check if slot is locked for a different shared course
if slot_info['locked'] and slot_info['shared_course_id'] != course_id:
return True
return False
except Exception as e:
current_app.logger.error(f"Error checking speciality constraints violation: {str(e)}")
return True
def _update_speciality_matrix(self, course_id: str, course_specialities: List,
slot_data: Dict, speciality_matrix: Dict) -> None:
"""
Update the speciality matrix after scheduling a course.
"""
try:
day = slot_data['day_of_week']
start_time = slot_data['start_time']
end_time = slot_data['end_time']
time_slot = f"{start_time}-{end_time}"
for speciality in course_specialities:
speciality_id = speciality.id
if speciality_id in speciality_matrix:
slot_info = speciality_matrix[speciality_id]['schedule'][day][time_slot]
slot_info['courses'].append(course_id)
except Exception as e:
current_app.logger.error(f"Error updating speciality matrix: {str(e)}")
def _generate_time_slots(self) -> List[str]:
"""
Generate 2-hour time slots for the timetable.
Returns:
List of time slot strings in format "HH:MM-HH:MM"
"""
time_slots = []
start_hour = 8 # 8 AM
end_hour = 18 # 6 PM
for hour in range(start_hour, end_hour, 2):
start_time = f"{hour:02d}:00"
end_time = f"{hour + 2:02d}:00"
time_slots.append(f"{start_time}-{end_time}")
return time_slots
def _initialize_tutor_schedule_matrix(self, ctx, tutors: List[Tutor], supervisors: List[Supervisor]) -> Dict:
"""
Initialize a matrix to track tutor-level scheduling constraints to prevent overlapping classes.
Returns:
Dict with structure: {
'tutor_id': {
'day_of_week': {
'time_slot': {
'courses': [course_ids],
'max_courses': 1, # Only 1 course per tutor per time slot
'locked': False,
'blocked_by': None
}
}
}
}
"""
try:
tutor_matrix = {}
# Initialize matrix for all tutors and supervisors
all_teachers = tutors + supervisors
for teacher in all_teachers:
teacher_id = teacher.id
tutor_matrix[teacher_id] = {
'name': f"{teacher.first_name} {teacher.last_name}" if hasattr(teacher, 'first_name') else f"Teacher {teacher_id}",
'type': 'tutor' if isinstance(teacher, Tutor) else 'supervisor',
'schedule': {}
}
# Initialize time slots for each teacher
time_slots = self._generate_time_slots()
days_of_week = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
for teacher_id in tutor_matrix:
for day in days_of_week:
tutor_matrix[teacher_id]['schedule'][day] = {}
for time_slot in time_slots:
tutor_matrix[teacher_id]['schedule'][day][time_slot] = {
'courses': [],
'max_courses': 1, # Critical: Only 1 course per tutor per time slot
'locked': False,
'blocked_by': None,
'conflict_reason': None
}
current_app.logger.info(f"[TUTOR MATRIX] Initialized matrix for {len(tutor_matrix)} teachers")
return tutor_matrix
except Exception as e:
current_app.logger.error(f"Error initializing tutor schedule matrix: {str(e)}")
return {}
def _check_tutor_conflict(self, tutor_id: str, day: str, start_time: str, end_time: str,
tutor_matrix: Dict, course_id: str = None) -> Dict:
"""
Check if scheduling a course for a tutor would create a conflict.
Args:
tutor_id: ID of the tutor
day: Day of the week
start_time: Start time (HH:MM format)
end_time: End time (HH:MM format)
tutor_matrix: Tutor schedule matrix
course_id: ID of the course being scheduled (optional)
Returns:
Dict with conflict information
"""
try:
if tutor_id not in tutor_matrix:
return {
'has_conflict': False,
'conflict_type': None,
'conflicting_courses': [],
'message': 'Tutor not found in matrix'
}
# Convert times to time slot format
time_slot = f"{start_time}-{end_time}"
# Check if the exact time slot is available
if day in tutor_matrix[tutor_id]['schedule'] and time_slot in tutor_matrix[tutor_id]['schedule'][day]:
slot_info = tutor_matrix[tutor_id]['schedule'][day][time_slot]
# Check if slot is locked or has courses
if slot_info['locked'] or len(slot_info['courses']) >= slot_info['max_courses']:
conflicting_courses = slot_info['courses'].copy()
if course_id and course_id in conflicting_courses:
conflicting_courses.remove(course_id) # Don't count the same course
if conflicting_courses:
return {
'has_conflict': True,
'conflict_type': 'tutor_time_overlap',
'conflicting_courses': conflicting_courses,
'message': f'Tutor {tutor_matrix[tutor_id]["name"]} already has a class at {day} {time_slot}',
'blocked_by': slot_info['blocked_by'],
'conflict_reason': slot_info['conflict_reason']
}
# Check for overlapping time slots (more comprehensive check)
overlapping_conflicts = self._check_overlapping_time_slots(
tutor_id, day, start_time, end_time, tutor_matrix, course_id
)
if overlapping_conflicts['has_conflict']:
return overlapping_conflicts
return {
'has_conflict': False,
'conflict_type': None,
'conflicting_courses': [],
'message': 'No conflicts detected'
}
except Exception as e:
current_app.logger.error(f"Error checking tutor conflict: {str(e)}")
return {
'has_conflict': True,
'conflict_type': 'error',
'conflicting_courses': [],
'message': f'Error checking conflict: {str(e)}'
}
def _check_overlapping_time_slots(self, tutor_id: str, day: str, start_time: str, end_time: str,
tutor_matrix: Dict, course_id: str = None) -> Dict:
"""
Check for overlapping time slots more comprehensively.
Returns:
Dict with overlap conflict information
"""
try:
if tutor_id not in tutor_matrix or day not in tutor_matrix[tutor_id]['schedule']:
return {'has_conflict': False}
# Convert times to datetime objects for comparison
from datetime import datetime
new_start = datetime.strptime(start_time, '%H:%M').time()
new_end = datetime.strptime(end_time, '%H:%M').time()
conflicting_courses = []
# Check all time slots for this tutor on this day
for time_slot, slot_info in tutor_matrix[tutor_id]['schedule'][day].items():
if len(slot_info['courses']) > 0: # Slot has courses
slot_start_str, slot_end_str = time_slot.split('-')
slot_start = datetime.strptime(slot_start_str, '%H:%M').time()
slot_end = datetime.strptime(slot_end_str, '%H:%M').time()
# Check if times overlap
if not (new_end <= slot_start or slot_end <= new_start):
# Times overlap
for course in slot_info['courses']:
if course_id and course == course_id:
continue # Don't count the same course
conflicting_courses.append({
'course_id': course,
'time_slot': time_slot,
'overlap_start': max(new_start, slot_start).strftime('%H:%M'),
'overlap_end': min(new_end, slot_end).strftime('%H:%M')
})
if conflicting_courses:
return {
'has_conflict': True,
'conflict_type': 'tutor_time_overlap',
'conflicting_courses': conflicting_courses,
'message': f'Tutor has overlapping classes on {day}',
'overlap_details': conflicting_courses
}
return {'has_conflict': False}
except Exception as e:
current_app.logger.error(f"Error checking overlapping time slots: {str(e)}")
return {'has_conflict': True}
def _update_tutor_matrix(self, tutor_id: str, day: str, start_time: str, end_time: str,
course_id: str, tutor_matrix: Dict, action: str = 'add') -> bool:
"""
Update the tutor schedule matrix after scheduling/unscheduling a course.
Args:
tutor_id: ID of the tutor
day: Day of the week
start_time: Start time (HH:MM format)
end_time: End time (HH:MM format)
course_id: ID of the course
tutor_matrix: Tutor schedule matrix
action: 'add' or 'remove'
Returns:
bool: True if successful, False otherwise
"""
try:
if tutor_id not in tutor_matrix:
return False
time_slot = f"{start_time}-{end_time}"
if day not in tutor_matrix[tutor_id]['schedule']:
return False
if time_slot not in tutor_matrix[tutor_id]['schedule'][day]:
return False
slot_info = tutor_matrix[tutor_id]['schedule'][day][time_slot]
if action == 'add':
if course_id not in slot_info['courses']:
slot_info['courses'].append(course_id)
current_app.logger.info(f"[TUTOR MATRIX] Added course {course_id} to tutor {tutor_id} at {day} {time_slot}")
elif action == 'remove':
if course_id in slot_info['courses']:
slot_info['courses'].remove(course_id)
current_app.logger.info(f"[TUTOR MATRIX] Removed course {course_id} from tutor {tutor_id} at {day} {time_slot}")
return True
except Exception as e:
current_app.logger.error(f"Error updating tutor matrix: {str(e)}")
return False
def _find_tutor_compatible_slots(self, course: Dict, available_teachers: List,
availability: Dict, tutor_matrix: Dict,
speciality_matrix: Dict, created_blocks: List) -> List[Dict]:
"""
Find time slots that are compatible with both speciality and tutor constraints.
Returns:
List of compatible slot information with tutor conflict checking
"""
try:
compatible_slots = []
for teacher in available_teachers:
teacher_id = teacher.id
teacher_slots = availability.get(teacher_id, [])
for slot in teacher_slots:
day = slot['day_of_week']
start_time = slot['start_time']
end_time = slot['end_time']
time_slot = f"{start_time}-{end_time}"
# Check tutor conflict first (most critical)
tutor_conflict = self._check_tutor_conflict(
teacher_id, day, start_time, end_time, tutor_matrix, course['id']
)
if tutor_conflict['has_conflict']:
current_app.logger.debug(f"[TUTOR CONFLICT] Teacher {teacher_id} has conflict at {day} {time_slot}: {tutor_conflict['message']}")
continue
# Check speciality constraints
course_specialities = course.get('shared_specialities', [])
speciality_conflicts = 0
is_speciality_compatible = True
for speciality_info in course_specialities:
speciality_id = speciality_info['id']
if speciality_id in speciality_matrix:
slot_info = speciality_matrix[speciality_id]['schedule'][day][time_slot]
if slot_info['locked'] or len(slot_info['courses']) >= slot_info['max_courses']:
is_speciality_compatible = False
break
speciality_conflicts += len(slot_info['courses'])
if is_speciality_compatible:
compatible_slots.append({
'teacher': teacher,
'slot': slot,
'tutor_conflicts': 0, # No tutor conflicts
'speciality_conflicts': speciality_conflicts,
'speciality_count': len(course_specialities),
'priority_score': self._calculate_slot_priority_score(
teacher_id, day, start_time, end_time,
tutor_matrix, speciality_matrix, course_specialities
)
})
# Sort by priority score (higher is better) and conflicts (lower is better)
compatible_slots.sort(key=lambda x: (x['priority_score'], -x['tutor_conflicts'], -x['speciality_conflicts']))
return compatible_slots
except Exception as e:
current_app.logger.error(f"Error finding tutor compatible slots: {str(e)}")
return []
def _calculate_slot_priority_score(self, tutor_id: str, day: str, start_time: str, end_time: str,
tutor_matrix: Dict, speciality_matrix: Dict, course_specialities: List) -> float:
"""
Calculate a priority score for a time slot based on various factors.
Returns:
float: Priority score (higher is better)
"""
try:
score = 0.0
# Base score
score += 100.0
# Penalty for tutor workload (prefer tutors with fewer classes)
if tutor_id in tutor_matrix:
tutor_total_classes = 0
for day_schedule in tutor_matrix[tutor_id]['schedule'].values():
for slot_info in day_schedule.values():
tutor_total_classes += len(slot_info['courses'])
# Reduce score based on workload (more classes = lower score)
score -= (tutor_total_classes * 5.0)
# Bonus for speciality compatibility
time_slot = f"{start_time}-{end_time}"
for speciality_info in course_specialities:
speciality_id = speciality_info['id']
if speciality_id in speciality_matrix and day in speciality_matrix[speciality_id]['schedule']:
if time_slot in speciality_matrix[speciality_id]['schedule'][day]:
slot_info = speciality_matrix[speciality_id]['schedule'][day][time_slot]
# Bonus for empty slots, penalty for crowded slots
if len(slot_info['courses']) == 0:
score += 20.0
else:
score -= (len(slot_info['courses']) * 10.0)
# Time-based preferences (prefer morning slots)
start_hour = int(start_time.split(':')[0])
if start_hour <= 10: # Morning slots
score += 10.0
elif start_hour >= 16: # Late afternoon
score -= 5.0
return score
except Exception as e:
current_app.logger.error(f"Error calculating slot priority score: {str(e)}")
return 0.0
def _schedule_regular_course(self, ctx, timetable_id: str, course: Dict, teacher,
availability: Dict, created_blocks: List, semester: str = None, supervisor_id: str = None) -> bool:
"""
Schedule a regular course with 2 timeblocks on different days with randomized times
Args:
ctx: Database context
timetable_id: ID of the timetable
course: Course data
teacher: Teacher (tutor or supervisor)
availability: Teacher availability data
created_blocks: List of already created blocks
Returns:
True if successfully scheduled, False otherwise
"""
try:
# Get teacher's availability
teacher_availability = availability.get(teacher.id, [])
if not teacher_availability:
return False
# Group availability by day
availability_by_day = {}
for slot in teacher_availability:
day = slot['day_of_week']
if day not in availability_by_day:
availability_by_day[day] = []
availability_by_day[day].append(slot)
# Get available days (need at least 2 different days)
available_days = list(availability_by_day.keys())
if len(available_days) < 2:
return False
# Randomize day selection for variety
import random
random.shuffle(available_days)
# Try to find available slots on different days
blocks_created = []
days_used = set()
for day in available_days:
if len(blocks_created) >= 2:
break
if day in days_used:
continue
# Find available slots for this teacher on this day
available_slots = self._find_available_slots_for_teacher(ctx, teacher.id, day, 2.0)
if available_slots:
# Randomize slot selection
random.shuffle(available_slots)
for slot in available_slots:
# Check for speciality conflicts and shared course constraints
speciality_check = self._check_speciality_conflicts(
ctx, timetable_id, course['id'], day, slot['start_time'], slot['end_time']
)
shared_check = self._check_shared_course_constraints(
ctx, course['id'], day, slot['start_time'], slot['end_time']
)
# Handle common courses specially
if course.get('is_shared_course', False) and course.get('shared_course_type', '') == 'common':
# Convert course dict to Course model for the common course handler
course_model = ctx.session.query(Course).filter(Course.id == course['id']).first()
if course_model:
common_course_info = self._handle_common_course_scheduling(ctx, course_model, semester, supervisor_id)
if common_course_info.get('is_common_course', False):
current_app.logger.info(f"[COMMON COURSE] Course {course['code']} requires special scheduling: {common_course_info.get('scheduling_recommendation', '')}")
# Only proceed if no speciality conflicts and shared course constraints are satisfied
if not speciality_check['has_conflict'] and not shared_check['constraints']:
# Create the block
block = self._create_timetable_block(ctx, timetable_id, course, teacher, slot)
if block:
# Add conflict information to the block
block['speciality_conflicts'] = speciality_check['conflicts']
block['shared_constraints'] = shared_check['constraints']
created_blocks.append(block)
blocks_created.append(block)
days_used.add(day)
break
else:
pass # Failed to create block
else:
# Log conflicts for manual review
if speciality_check['has_conflict']:
current_app.logger.warning(f"Speciality conflict detected for course {course['code']}: {speciality_check['conflicts']}")
if shared_check['constraints']:
current_app.logger.warning(f"Shared course constraint detected for course {course['code']}: {shared_check['constraints']}")
continue # Conflicts detected
if len(blocks_created) >= 2:
break
if len(blocks_created) == 2:
return True
else:
# Remove any blocks that were created
for block in blocks_created:
if block in created_blocks:
created_blocks.remove(block)
return False
except Exception as e:
current_app.logger.error(f"Error scheduling regular course {course['code']}: {e}")
return False
def _schedule_practical_course(self, ctx, timetable_id: str, course: Dict, teacher,
availability: Dict, created_blocks: List, semester: str = None, supervisor_id: str = None) -> bool:
"""
Schedule a practical course with 1 continuous 6-hour block (3 x 2-hour slots) on one day per week
Args:
ctx: Database context
timetable_id: ID of the timetable
course: Course data
teacher: Teacher (tutor or supervisor)
availability: Teacher availability data
created_blocks: List of already created blocks
Returns:
True if successfully scheduled, False otherwise
"""
try:
# Get teacher's availability
teacher_availability = availability.get(teacher.id, [])
if not teacher_availability:
return False
# Group availability by day
availability_by_day = {}
for slot in teacher_availability:
day = slot['day_of_week']
if day not in availability_by_day:
availability_by_day[day] = []
availability_by_day[day].append(slot)
# Find a day with at least 3 consecutive 2-hour slots
import random
available_days = list(availability_by_day.keys())
random.shuffle(available_days) # Randomize day selection
for day in available_days:
# Find available slots for this teacher on this day
available_slots = self._find_available_slots_for_teacher(ctx, teacher.id, day, 2.0)
if len(available_slots) >= 3:
# Randomize slot selection within the day
random.shuffle(available_slots)
# Take the first 3 slots for the 6-hour continuous block
selected_slots = available_slots[:3]
# Verify slots are consecutive (same day) and check for conflicts
blocks_created = []
all_conflicts = []
for i, slot in enumerate(selected_slots):
# Check for speciality conflicts and shared course constraints
speciality_check = self._check_speciality_conflicts(
ctx, timetable_id, course['id'], day, slot['start_time'], slot['end_time']
)
shared_check = self._check_shared_course_constraints(
ctx, course['id'], day, slot['start_time'], slot['end_time']
)
# Handle common courses specially
if getattr(course, 'is_shared_course', False) and getattr(course, 'shared_course_type', '') == 'common':
common_course_info = self._handle_common_course_scheduling(ctx, course, semester, supervisor_id)
if common_course_info.get('is_common_course', False):
current_app.logger.info(f"[COMMON COURSE] Course {course['code']} requires special scheduling: {common_course_info.get('scheduling_recommendation', '')}")
# Check for tutor schedule conflicts
conflict_check = self._check_tutor_schedule_conflicts(
ctx, teacher.id, day, slot['start_time'], slot['end_time']
)
# Only proceed if no conflicts
if (not speciality_check['has_conflict'] and
not shared_check['constraints'] and
not conflict_check['has_conflicts']):
# Create the block
block = self._create_timetable_block(ctx, timetable_id, course, teacher, slot)
if block:
# Add conflict information to the block
block['speciality_conflicts'] = speciality_check['conflicts']
block['shared_constraints'] = shared_check['constraints']
block['tutor_conflicts'] = conflict_check['conflicts']
blocks_created.append(block)
else:
current_app.logger.error(f"Failed to create block {i+1}/3 for practical course {course['code']}")
# Remove any blocks that were created
for created_block in blocks_created:
if created_block in created_blocks:
created_blocks.remove(created_block)
return False
else:
# Log conflicts for manual review
if speciality_check['has_conflict']:
all_conflicts.extend(speciality_check['conflicts'])
current_app.logger.warning(f"Speciality conflict detected for practical course {course['code']} slot {i+1}: {speciality_check['conflicts']}")
if shared_check['constraints']:
all_conflicts.extend(shared_check['constraints'])
current_app.logger.warning(f"Shared course constraint detected for practical course {course['code']} slot {i+1}: {shared_check['constraints']}")
if conflict_check['has_conflicts']:
all_conflicts.extend(conflict_check['conflicts'])
current_app.logger.warning(f"Tutor conflict detected for practical course {course['code']} slot {i+1}: {conflict_check['conflicts']}")
continue
# If we successfully created all 3 blocks, add them to the main list
if len(blocks_created) == 3:
created_blocks.extend(blocks_created)
return True
else:
current_app.logger.warning(f"Only created {len(blocks_created)}/3 blocks for practical course {course['code']} on day {day}")
# Remove any blocks that were created
for created_block in blocks_created:
if created_block in created_blocks:
created_blocks.remove(created_block)
continue
else:
continue
current_app.logger.warning(f"No suitable day found for practical course {course['code']} - need day with at least 3 available slots")
return False
except Exception as e:
current_app.logger.error(f"Error scheduling practical course {course['code']}: {e}")
return False
def _check_tutor_schedule_conflicts(self, ctx, teacher_id: str, day_of_week: int,
start_time: str, end_time: str,
exclude_block_id: str = None) -> Dict:
"""
Comprehensive check for tutor schedule conflicts:
1. No overlapping time slots for the same tutor
2. Block must be within tutor's approved availability
3. No multiple classes at the same time
Args:
ctx: Database context
teacher_id: ID of the tutor/supervisor
day_of_week: Day of week (0-6, Monday=0)
start_time: Start time in HH:MM format
end_time: End time in HH:MM format
exclude_block_id: ID of block to exclude from conflict checks
Returns:
Dict with conflict information
"""
conflicts = []
try:
# Parse times
start_time_obj = datetime.strptime(start_time, '%H:%M').time()
end_time_obj = datetime.strptime(end_time, '%H:%M').time()
if end_time_obj <= start_time_obj:
conflicts.append({
'type': 'invalid_time',
'message': 'End time must be after start time',
'day_of_week': day_of_week,
'start_time': start_time,
'end_time': end_time
})
return {'has_conflicts': True, 'conflicts': conflicts}
# 1. Check if tutor has approved availability on this day and time
tutor_availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == teacher_id,
TutorAvailability.day_of_week == day_of_week,
TutorAvailability.is_approved == True,
TutorAvailability.is_cancelled == False
).all()
supervisor_availabilities = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.supervisor_id == teacher_id,
SupervisorAvailability.day_of_week == day_of_week,
SupervisorAvailability.is_approved == True,
SupervisorAvailability.is_cancelled == False
).all()
all_availabilities = tutor_availabilities + supervisor_availabilities
if not all_availabilities:
conflicts.append({
'type': 'no_availability',
'message': f'Tutor has no approved availability on day {day_of_week}',
'day_of_week': day_of_week,
'start_time': start_time,
'end_time': end_time
})
return {'has_conflicts': True, 'conflicts': conflicts}
# Check if the proposed time slot fits within any availability slot
slot_fits = False
for availability in all_availabilities:
if (availability.start_time <= start_time_obj and
availability.end_time >= end_time_obj):
slot_fits = True
break
if not slot_fits:
conflicts.append({
'type': 'outside_availability',
'message': f'Proposed time slot {start_time}-{end_time} does not fit within any approved availability slot',
'day_of_week': day_of_week,
'start_time': start_time,
'end_time': end_time,
'available_slots': [f"{a.start_time.strftime('%H:%M')}-{a.end_time.strftime('%H:%M')}" for a in all_availabilities]
})
return {'has_conflicts': True, 'conflicts': conflicts}
# 2. Check for overlapping with existing blocks (same tutor, same day)
existing_blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.day_of_week == day_of_week,
TimetableBlock.id != exclude_block_id if exclude_block_id else True
).all()
# Filter blocks for this specific tutor
tutor_blocks = []
for block in existing_blocks:
if (hasattr(block, 'tutor_id') and block.tutor_id == teacher_id) or \
(hasattr(block, 'supervisor_tutor_id') and block.supervisor_tutor_id == teacher_id):
tutor_blocks.append(block)
# Check for time overlaps
for existing_block in tutor_blocks:
existing_start = existing_block.start_time
existing_end = existing_block.end_time
# Check if times overlap
if not (end_time_obj <= existing_start or start_time_obj >= existing_end):
conflicts.append({
'type': 'time_overlap',
'message': f'Time slot overlaps with existing block {existing_block.id} ({existing_start.strftime("%H:%M")}-{existing_end.strftime("%H:%M")})',
'day_of_week': day_of_week,
'start_time': start_time,
'end_time': end_time,
'conflicting_block_id': existing_block.id,
'conflicting_start': existing_start.strftime('%H:%M'),
'conflicting_end': existing_end.strftime('%H:%M')
})
# 3. Check weekly teaching hours limit
try:
weekly_hours = self.block_manager._calculate_weekly_teaching_hours(ctx, teacher_id, exclude_block_id=exclude_block_id)
block_hours = (datetime.combine(date.today(), end_time_obj) -
datetime.combine(date.today(), start_time_obj)).total_seconds() / 3600
# Get teacher's max hours (default to 20 if not set)
teacher = ctx.session.query(Tutor).filter(Tutor.id == teacher_id).first()
if not teacher:
teacher = ctx.session.query(Supervisor).filter(Supervisor.id == teacher_id).first()
max_hours = getattr(teacher, 'max_teaching_hours', 28) or 28
if weekly_hours + block_hours > max_hours:
conflicts.append({
'type': 'exceeds_weekly_hours',
'message': f'Would exceed weekly teaching hours limit ({weekly_hours + block_hours:.1f}/{max_hours})',
'day_of_week': day_of_week,
'start_time': start_time,
'end_time': end_time,
'current_weekly_hours': weekly_hours,
'block_hours': block_hours,
'max_allowed': max_hours
})
except Exception as e:
current_app.logger.warning(f"Block manager weekly hours calculation failed for teacher {teacher_id}: {e}")
# Try fallback method
try:
weekly_hours = self._calculate_teacher_weekly_hours(ctx, teacher_id, exclude_block_id)
block_hours = (datetime.combine(date.today(), end_time_obj) -
datetime.combine(date.today(), start_time_obj)).total_seconds() / 3600
# Get teacher's max hours (default to 20 if not set)
teacher = ctx.session.query(Tutor).filter(Tutor.id == teacher_id).first()
if not teacher:
teacher = ctx.session.query(Supervisor).filter(Supervisor.id == teacher_id).first()
max_hours = getattr(teacher, 'max_teaching_hours', 28) or 28
if weekly_hours + block_hours > max_hours:
conflicts.append({
'type': 'exceeds_weekly_hours',
'message': f'Would exceed weekly teaching hours limit ({weekly_hours + block_hours:.1f}/{max_hours})',
'day_of_week': day_of_week,
'start_time': start_time,
'end_time': end_time,
'current_weekly_hours': weekly_hours,
'block_hours': block_hours,
'max_allowed': max_hours
})
except Exception as fallback_error:
current_app.logger.warning(f"Fallback weekly hours calculation also failed for teacher {teacher_id}: {fallback_error}")
# Continue without weekly hours check rather than failing completely
result = {
'has_conflicts': len(conflicts) > 0,
'conflicts': conflicts
}
if conflicts:
pass
return result
except Exception as e:
current_app.logger.error(f"Error checking tutor schedule conflicts: {e}")
return {
'has_conflicts': True,
'conflicts': [{
'type': 'error',
'message': f'Error checking conflicts: {str(e)}'
}]
}
def _find_available_slots_for_teacher(self, ctx, teacher_id: str, day_of_week: int,
block_duration_hours: float = 2.0) -> List[Dict]:
"""
Find available time slots for a teacher on a specific day
Args:
ctx: Database context
teacher_id: ID of the tutor/supervisor
day_of_week: Day of week (0-6, Monday=0)
block_duration_hours: Duration of the block in hours
Returns:
List of available time slots
"""
available_slots = []
try:
# Get tutor availabilities
tutor_availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == teacher_id,
TutorAvailability.day_of_week == day_of_week,
TutorAvailability.is_approved == True,
TutorAvailability.is_cancelled == False
).all()
# Get supervisor availabilities
supervisor_availabilities = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.supervisor_id == teacher_id,
SupervisorAvailability.day_of_week == day_of_week,
SupervisorAvailability.is_approved == True,
SupervisorAvailability.is_cancelled == False
).all()
all_availabilities = tutor_availabilities + supervisor_availabilities
if not all_availabilities:
return available_slots
# Get existing blocks for this teacher on this day
existing_blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.day_of_week == day_of_week
).all()
# Filter blocks for this specific teacher
teacher_blocks = []
for block in existing_blocks:
if (hasattr(block, 'tutor_id') and block.tutor_id == teacher_id) or \
(hasattr(block, 'supervisor_tutor_id') and block.supervisor_tutor_id == teacher_id):
teacher_blocks.append(block)
# For each availability slot, find available sub-slots
for availability in all_availabilities:
start_time = availability.start_time
end_time = availability.end_time
# Convert to datetime for easier manipulation
start_dt = datetime.combine(date.today(), start_time)
end_dt = datetime.combine(date.today(), end_time)
# Create 2-hour slots within this availability
current_time = start_dt
while current_time + timedelta(hours=block_duration_hours) <= end_dt:
slot_start = current_time.time()
slot_end = (current_time + timedelta(hours=block_duration_hours)).time()
# Check if this slot conflicts with existing blocks
slot_available = True
for block in teacher_blocks:
if not (slot_end <= block.start_time or slot_start >= block.end_time):
slot_available = False
break
if slot_available:
available_slots.append({
'day_of_week': day_of_week,
'start_time': slot_start.strftime('%H:%M'),
'end_time': slot_end.strftime('%H:%M'),
'availability_id': availability.id
})
current_time += timedelta(hours=block_duration_hours)
return available_slots
except Exception as e:
current_app.logger.error(f"Error finding available slots for teacher {teacher_id}: {e}")
return []
def _check_speciality_conflicts(self, ctx, timetable_id: str, course_id: str,
day_of_week: int, start_time: time, end_time: time) -> Dict:
"""
Check if scheduling a course would create a speciality conflict.
Args:
ctx: Database context
timetable_id: ID of the timetable
course_id: ID of the course to be scheduled
day_of_week: Day of the week (0-6)
start_time: Start time of the slot
end_time: End time of the slot
Returns:
Dict with conflict information
"""
try:
# Get the course and its speciality
course = ctx.session.query(Course).filter(Course.id == course_id).first()
if not course or not course.speciality_id:
return {'has_conflict': False, 'conflicts': []}
# Get all existing blocks for the same day and time slot
existing_blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id,
TimetableBlock.day_of_week == day_of_week,
TimetableBlock.id != course_id # Exclude the current course
).all()
conflicts = []
for block in existing_blocks:
# Check if time slots overlap
if not (end_time <= block.start_time or block.end_time <= start_time):
# Get the course for this block
block_course = ctx.session.query(Course).filter(Course.id == block.course_id).first()
if block_course and block_course.speciality_id == course.speciality_id:
conflicts.append({
'type': 'speciality_conflict',
'message': f'Course {course.code} conflicts with {block_course.code} - both share speciality {course.speciality.name}',
'conflicting_course': {
'id': block_course.id,
'code': block_course.code,
'title': block_course.title
},
'speciality': {
'id': course.speciality_id,
'name': course.speciality.name
},
'time_slot': {
'day': day_of_week,
'start_time': start_time.strftime('%H:%M'),
'end_time': end_time.strftime('%H:%M')
}
})
return {
'has_conflict': len(conflicts) > 0,
'conflicts': conflicts
}
except Exception as e:
current_app.logger.error(f"Error checking speciality conflicts: {e}")
return {'has_conflict': False, 'conflicts': []}
def _check_shared_course_constraints(self, ctx, course_id: str, day_of_week: int,
start_time: time, end_time: time) -> Dict:
"""
Check if a shared course can be scheduled at the specified time across all departments.
Args:
ctx: Database context
course_id: ID of the course to be scheduled
day_of_week: Day of the week (0-6)
start_time: Start time of the slot
end_time: End time of the slot
Returns:
Dict with constraint information
"""
try:
# Get the course and check if it's a shared course
course = ctx.session.query(Course).filter(Course.id == course_id).first()
if not course or not getattr(course, 'is_shared_course', False):
return {'is_shared': False, 'constraints': []}
# Get course type for specific handling
course_type = getattr(course, 'shared_course_type', 'department_specific')
current_app.logger.info(f"[SHARED COURSE] Checking constraints for {course.code} (type: {course_type})")
# Get all departments this course is shared with
shared_departments = ctx.session.query(course_department_association).filter(
course_department_association.c.course_id == course_id,
course_department_association.c.is_active == True
).all()
if not shared_departments:
current_app.logger.warning(f"[SHARED COURSE] Course {course.code} is marked as shared but has no department associations")
return {'is_shared': False, 'constraints': []}
constraints = []
# Special handling for common courses
if course_type == 'common':
current_app.logger.info(f"[COMMON COURSE] Course {course.code} is a common course shared across {len(shared_departments)} departments")
# Common courses need to be scheduled at the same time across all departments
# Check if the time slot is available in ALL shared departments
for dept_assoc in shared_departments:
dept_name = dept_assoc.department_name
# Get all timetables for this department
dept_timetables = ctx.session.query(Timetable).filter(
Timetable.department == dept_name
).all()
for timetable in dept_timetables:
# Check if there are any conflicting blocks in this department's timetable
conflicting_blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id,
TimetableBlock.day_of_week == day_of_week,
TimetableBlock.course_id != course_id # Exclude the current course
).all()
for block in conflicting_blocks:
# Check if time slots overlap
if not (end_time <= block.start_time or block.end_time <= start_time):
constraints.append({
'type': 'common_course_constraint',
'message': f'Common course {course.code} conflicts with {block.course.code} in department {dept_name}',
'department': dept_name,
'timetable_id': timetable.id,
'conflicting_block': {
'id': block.id,
'course_id': block.course_id,
'start_time': block.start_time.strftime('%H:%M'),
'end_time': block.end_time.strftime('%H:%M')
}
})
else:
# Regular shared course handling
current_app.logger.info(f"[SHARED COURSE] Course {course.code} is a {course_type} course")
# Check if the time slot is available across all shared departments
for dept_assoc in shared_departments:
dept_name = dept_assoc.department_name
# Get all timetables for this department
dept_timetables = ctx.session.query(Timetable).filter(
Timetable.department == dept_name
).all()
for timetable in dept_timetables:
# Check if there are any conflicting blocks in this department's timetable
conflicting_blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id,
TimetableBlock.day_of_week == day_of_week,
TimetableBlock.course_id != course_id # Exclude the current course
).all()
for block in conflicting_blocks:
# Check if time slots overlap
if not (end_time <= block.start_time or block.end_time <= start_time):
constraints.append({
'type': 'shared_course_constraint',
'message': f'Shared course {course.code} conflicts with {block.course.code} in department {dept_name}',
'department': dept_name,
'timetable_id': timetable.id,
'conflicting_block': {
'id': block.id,
'course_id': block.course_id,
'start_time': block.start_time.strftime('%H:%M'),
'end_time': block.end_time.strftime('%H:%M')
}
})
return {
'is_shared': True,
'constraints': constraints,
'shared_departments': [dept.department_name for dept in shared_departments]
}
except Exception as e:
current_app.logger.error(f"Error checking shared course constraints: {e}")
return {'is_shared': False, 'constraints': []}
def _find_optimal_time_slot(self, ctx, timetable_id: str, course_id: str,
teacher_availability: List, created_blocks: List) -> Dict:
"""
Find the optimal time slot for a course considering speciality conflicts and shared course constraints.
Args:
ctx: Database context
timetable_id: ID of the timetable
course_id: ID of the course to be scheduled
teacher_availability: List of available time slots for the teacher
created_blocks: List of already created blocks
Returns:
Dict with optimal time slot information
"""
try:
course = ctx.session.query(Course).filter(Course.id == course_id).first()
if not course:
return {'found': False, 'reason': 'Course not found'}
# Get course speciality and shared course info
speciality_id = course.speciality_id
is_shared_course = getattr(course, 'is_shared_course', False)
# Score each available slot based on constraints
slot_scores = []
for slot in teacher_availability:
day_of_week = slot['day_of_week']
start_time = slot['start_time']
end_time = slot['end_time']
score = 100 # Start with perfect score
conflicts = []
# Check speciality conflicts
speciality_check = self._check_speciality_conflicts(
ctx, timetable_id, course_id, day_of_week, start_time, end_time
)
if speciality_check['has_conflict']:
score -= 50 # Heavy penalty for speciality conflicts
conflicts.extend(speciality_check['conflicts'])
# Check shared course constraints
shared_check = self._check_shared_course_constraints(
ctx, course_id, day_of_week, start_time, end_time
)
if shared_check['is_shared'] and shared_check['constraints']:
score -= 30 # Penalty for shared course constraints
conflicts.extend(shared_check['constraints'])
# Check for existing blocks in the same time slot
for block in created_blocks:
if (block.get('day_of_week') == day_of_week and
not (end_time <= block.get('start_time') or block.get('end_time') <= start_time)):
score -= 40 # Penalty for time conflicts
conflicts.append({
'type': 'time_conflict',
'message': f'Time slot conflicts with existing block',
'conflicting_block': block
})
# Prefer slots with higher scores
slot_scores.append({
'slot': slot,
'score': score,
'conflicts': conflicts,
'day_of_week': day_of_week,
'start_time': start_time,
'end_time': end_time
})
# Sort by score (highest first)
slot_scores.sort(key=lambda x: x['score'], reverse=True)
# Find the best slot with no conflicts
for slot_info in slot_scores:
if slot_info['score'] >= 100: # No conflicts
return {
'found': True,
'slot': slot_info['slot'],
'score': slot_info['score'],
'conflicts': []
}
# If no perfect slot, return the best available one
if slot_scores:
best_slot = slot_scores[0]
return {
'found': True,
'slot': best_slot['slot'],
'score': best_slot['score'],
'conflicts': best_slot['conflicts'],
'warning': 'Scheduled with conflicts - manual review recommended'
}
return {'found': False, 'reason': 'No suitable time slots available'}
except Exception as e:
current_app.logger.error(f"Error finding optimal time slot: {e}")
return {'found': False, 'reason': f'Error: {str(e)}'}
def _validate_timetable_conflicts(self, ctx, timetable_id: str) -> Dict:
"""
Validate the entire timetable for conflicts after generation
Args:
ctx: Database context
timetable_id: ID of the timetable to validate
Returns:
Dict with validation results and any conflicts found
"""
try:
# Get all blocks for this timetable
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id
).order_by(
TimetableBlock.day_of_week,
TimetableBlock.start_time
).all()
if not blocks:
return {
'is_valid': True,
'conflicts': [],
'message': 'No blocks found in timetable'
}
# Group blocks by day and time to check for overlaps
day_blocks = {}
tutor_conflicts = {}
for block in blocks:
day = block.day_of_week
if day not in day_blocks:
day_blocks[day] = []
day_blocks[day].append(block)
# Check each day for conflicts
for day, day_blocks_list in day_blocks.items():
# Sort blocks by start time
day_blocks_list.sort(key=lambda x: x.start_time)
# Check for overlapping blocks
for i, block1 in enumerate(day_blocks_list):
for j, block2 in enumerate(day_blocks_list[i+1:], i+1):
# Check if blocks overlap in time
if not (block1.end_time <= block2.start_time or block2.end_time <= block1.start_time):
# Check if same tutor is involved
tutor1_id = block1.tutor_id or block1.supervisor_tutor_id
tutor2_id = block2.tutor_id or block2.supervisor_tutor_id
if tutor1_id and tutor2_id and tutor1_id == tutor2_id:
# Same tutor has overlapping blocks - this is a conflict
conflict_key = f"tutor_{tutor1_id}_day_{day}"
if conflict_key not in tutor_conflicts:
tutor_conflicts[conflict_key] = []
tutor_conflicts[conflict_key].append({
'type': 'tutor_time_overlap',
'message': f'Tutor has overlapping classes on day {day}',
'day_of_week': day,
'day_name': ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'][day],
'block1': {
'id': block1.id,
'course_id': block1.course_id,
'start_time': block1.start_time.strftime('%H:%M'),
'end_time': block1.end_time.strftime('%H:%M'),
'block_type': block1.block_type
},
'block2': {
'id': block2.id,
'course_id': block2.course_id,
'start_time': block2.start_time.strftime('%H:%M'),
'end_time': block2.end_time.strftime('%H:%M'),
'block_type': block2.block_type
}
})
# Check for availability violations
availability_violations = []
for block in blocks:
tutor_id = block.tutor_id or block.supervisor_tutor_id
if not tutor_id:
continue
# Check if block is within tutor's approved availability
if hasattr(block, 'tutor_id') and block.tutor_id:
availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor_id,
TutorAvailability.day_of_week == block.day_of_week,
TutorAvailability.is_approved == True,
TutorAvailability.is_cancelled == False
).all()
else:
availabilities = ctx.session.query(SupervisorAvailability).filter(
SupervisorAvailability.supervisor_id == tutor_id,
SupervisorAvailability.day_of_week == block.day_of_week,
SupervisorAvailability.is_approved == True,
SupervisorAvailability.is_cancelled == False
).all()
# Check if block fits within any availability slot
block_fits = False
if availabilities:
for availability in availabilities:
# More flexible availability checking - allow blocks that start within availability window
# and don't exceed the end time by more than 30 minutes
availability_start = availability.start_time
availability_end = availability.end_time
block_start = block.start_time
block_end = block.end_time
# Check if block starts within availability and ends within reasonable time
if (availability_start <= block_start and
block_end <= availability_end):
block_fits = True
break
# Allow some flexibility for 2-hour blocks that might slightly overlap
# if the overlap is minimal (less than 30 minutes)
if (availability_start <= block_start and
block_start < availability_end):
# Calculate overlap
overlap_minutes = (datetime.combine(date.today(), block_end) -
datetime.combine(date.today(), availability_end)).seconds // 60
if overlap_minutes <= 30: # Allow up to 30 minutes overlap
block_fits = True
break
else:
# If no specific availability found, check if it's a supervisor
# Supervisors might not have explicit availability records
if hasattr(block, 'supervisor_tutor_id') and block.supervisor_tutor_id:
block_fits = True # Assume supervisors are available during standard hours
if not block_fits and availabilities: # Only report violation if we have availability records
availability_violations.append({
'type': 'availability_violation',
'message': f'Block {block.id} is outside tutor availability',
'block_id': block.id,
'tutor_id': tutor_id,
'day_of_week': block.day_of_week,
'day_name': ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'][block.day_of_week],
'block_time': f"{block.start_time.strftime('%H:%M')}-{block.end_time.strftime('%H:%M')}",
'available_slots': [f"{a.start_time.strftime('%H:%M')}-{a.end_time.strftime('%H:%M')}" for a in availabilities]
})
# Combine all conflicts
all_conflicts = list(tutor_conflicts.values()) + availability_violations
# Flatten the conflicts list
flattened_conflicts = []
for conflict_group in all_conflicts:
if isinstance(conflict_group, list):
flattened_conflicts.extend(conflict_group)
else:
flattened_conflicts.append(conflict_group)
is_valid = len(flattened_conflicts) == 0
result = {
'is_valid': is_valid,
'conflicts': flattened_conflicts,
'total_blocks': len(blocks),
'conflict_count': len(flattened_conflicts),
'message': f'Timetable validation complete. Found {len(flattened_conflicts)} conflicts.' if not is_valid else 'Timetable is valid with no conflicts.'
}
if is_valid:
pass
else:
current_app.logger.warning(f"Timetable {timetable_id} validation failed - {len(flattened_conflicts)} conflicts found")
for conflict in flattened_conflicts:
current_app.logger.warning(f"Conflict: {conflict}")
return result
except Exception as e:
current_app.logger.error(f"Error validating timetable {timetable_id}: {e}")
return {
'is_valid': False,
'conflicts': [{
'type': 'validation_error',
'message': f'Error during validation: {str(e)}'
}],
'message': f'Error during timetable validation: {str(e)}'
}
def _calculate_teacher_weekly_hours(self, ctx, teacher_id: str, exclude_block_id: str = None) -> float:
"""
Calculate total weekly teaching hours for a teacher (fallback method)
Args:
ctx: Database context
teacher_id: ID of the teacher
exclude_block_id: Optional ID of block to exclude from calculation
Returns:
Total weekly teaching hours in hours
"""
try:
# Get all teaching blocks for this teacher (excluding specified block if any)
query = ctx.session.query(TimetableBlock).filter(
TimetableBlock.block_type.in_(['lecture', 'tutorial', 'lab'])
)
# Filter by teacher ID (could be tutor_id or supervisor_tutor_id)
query = query.filter(
or_(
TimetableBlock.tutor_id == teacher_id,
TimetableBlock.supervisor_tutor_id == teacher_id
)
)
if exclude_block_id:
query = query.filter(TimetableBlock.id != exclude_block_id)
blocks = query.all()
# Calculate total hours
total_hours = 0.0
for block in blocks:
if block.start_time and block.end_time:
duration = (datetime.combine(date.today(), block.end_time) -
datetime.combine(date.today(), block.start_time))
total_hours += duration.total_seconds() / 3600
return total_hours
except Exception as e:
current_app.logger.error(f"Error calculating weekly hours for teacher {teacher_id}: {e}")
return 0.0
def _validate_day_placement(self, blocks: List[TimetableBlock]):
"""Validate that blocks are being placed on the correct days"""
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
for block in blocks:
day_index = block.day_of_week
if 0 <= day_index < len(days):
expected_day = days[day_index]
else:
current_app.logger.warning(f"Block has invalid day index: {day_index}")
def _check_database_state(self, ctx, semester: str):
"""Check the database state to understand why courses might be missing"""
# Check total courses
total_courses = ctx.session.query(Course).filter(Course.is_active == True).count()
# Check courses by semester
courses_by_semester = {}
all_courses = ctx.session.query(Course).filter(Course.is_active == True).all()
for course in all_courses:
sem = course.semester or 'No Semester'
if sem not in courses_by_semester:
courses_by_semester[sem] = []
courses_by_semester[sem].append(course.code)
for sem, codes in courses_by_semester.items():
pass
# Check tutor-course associations
total_tutor_associations = ctx.session.query(tutor_course_association).count()
# Check supervisor-course associations
total_supervisor_associations = ctx.session.query(supervisor_course_association).count()
# Check which courses have tutors assigned
courses_with_tutors = set()
tutor_assocs = ctx.session.query(tutor_course_association).all()
for assoc in tutor_assocs:
courses_with_tutors.add(assoc.course_id)
# Check which courses have supervisors assigned
courses_with_supervisors = set()
supervisor_assocs = ctx.session.query(supervisor_course_association).all()
for assoc in supervisor_assocs:
courses_with_supervisors.add(assoc.course_id)
# Check total unique courses with any teacher assignment
all_assigned_courses = courses_with_tutors.union(courses_with_supervisors)
return {
'total_courses': total_courses,
'courses_by_semester': courses_by_semester,
'total_tutor_associations': total_tutor_associations,
'total_supervisor_associations': total_supervisor_associations,
'courses_with_tutors': len(courses_with_tutors),
'courses_with_supervisors': len(courses_with_supervisors),
'total_assigned_courses': len(all_assigned_courses)
}
def _auto_assign_tutors_to_courses(self, ctx, courses: List[Dict], tutors: List[Tutor], supervisors: List[Supervisor]) -> Dict:
"""
Automatically assign tutors to courses that don't have any tutors assigned.
This ensures all courses can be included in the timetable generation.
Args:
ctx: Database context
courses: List of courses to process
tutors: List of available tutors
supervisors: List of available supervisors
Returns:
Dict with success status and assignment count
"""
try:
assigned_count = 0
failed_assignments = []
# Get all available teachers (tutors + supervisors)
all_teachers = tutors + supervisors
if not all_teachers:
return {
'success': False,
'message': 'No tutors or supervisors available for assignment',
'assigned_count': 0
}
# Shuffle teachers for random distribution
import random
random.shuffle(all_teachers)
for course in courses:
# Check if course already has tutors assigned
existing_tutors = ctx.session.query(tutor_course_association).filter(
tutor_course_association.c.course_id == course['id']
).all()
# Check if course has a supervisor assigned
course_obj = ctx.session.query(Course).filter(Course.id == course['id']).first()
has_supervisor = course_obj and course_obj.supervisor_id
if existing_tutors or has_supervisor:
continue
# Try to assign a tutor to this course
assignment_success = False
for teacher in all_teachers:
# Skip if teacher is already assigned to too many courses
teacher_courses = ctx.session.query(tutor_course_association).filter(
tutor_course_association.c.tutor_id == teacher.id
).count()
if teacher_courses >= 5: # Limit to 5 courses per teacher
continue
# Check if teacher has availability
if hasattr(teacher, 'availabilities') and teacher.availabilities:
# Check availability for both tutors and supervisors
availability_count = len([a for a in teacher.availabilities if a.is_approved and not a.is_cancelled])
else:
availability_count = 0
if availability_count == 0:
continue
# Assign teacher to course
try:
if hasattr(teacher, 'staff_id'): # This is a tutor
# Create tutor-course association
stmt = tutor_course_association.insert().values(
tutor_id=teacher.id,
course_id=course['id'],
is_primary=False
)
ctx.session.execute(stmt)
else: # This is a supervisor
# Update course supervisor_id
course_obj.supervisor_id = teacher.id
assigned_count += 1
assignment_success = True
break
except Exception as e:
current_app.logger.error(f"Failed to assign teacher {teacher.id} to course {course['code']}: {e}")
continue
if not assignment_success:
failed_assignments.append({
'course': course['code'],
'reason': 'No suitable teacher found with availability'
})
# Commit all assignments
ctx.session.commit()
if failed_assignments:
pass # Failed assignments logged in response
return {
'success': True,
'message': f'Successfully assigned {assigned_count} teachers to courses',
'assigned_count': assigned_count,
'failed_assignments': failed_assignments
}
except Exception as e:
ctx.session.rollback()
current_app.logger.error(f"Error in auto-assignment: {e}")
return {
'success': False,
'message': f'Auto-assignment failed: {str(e)}',
'assigned_count': 0
}
def _slots_are_consecutive(self, slot1: Dict, slot2: Dict) -> bool:
"""Check if two slots are consecutive"""
if slot1['day_of_week'] != slot2['day_of_week']:
return False
# Check if slots are adjacent in time
slot1_end = slot1['end_time'].hour * 60 + slot1['end_time'].minute
slot2_start = slot2['start_time'].hour * 60 + slot2['start_time'].minute
slot2_end = slot2['end_time'].hour * 60 + slot2['end_time'].minute
slot1_start = slot1['start_time'].hour * 60 + slot1['start_time'].minute
return slot1_end == slot2_start or slot2_end == slot1_start
def _get_tutor_course_assignments(self, ctx, supervisor_id: str = None) -> Dict[str, List[Tutor]]:
"""Get all tutor-course assignments from tutor_course_association table, filtered by supervisor's department"""
assignments = {}
# Get supervisor's departments if specified
supervisor_departments = []
if supervisor_id:
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if supervisor:
supervisor_departments = supervisor.get_all_departments()
# Query the association table
tutor_course_pairs = ctx.session.query(tutor_course_association).all()
for pair in tutor_course_pairs:
course_id = pair.course_id
tutor = ctx.session.query(Tutor).filter(Tutor.id == pair.tutor_id).first()
# Get course to check department
course = ctx.session.query(Course).filter(Course.id == course_id).first()
# Filter by department if supervisor departments are specified
if supervisor_departments and course and course.department not in supervisor_departments:
continue
if tutor and tutor.is_active:
if course_id not in assignments:
assignments[course_id] = []
assignments[course_id].append(tutor)
return assignments
def _get_supervisor_course_assignments(self, ctx, supervisor_id: str = None) -> Dict[str, List[Supervisor]]:
"""Get all supervisor-course assignments, filtered by supervisor's department"""
assignments = {}
# Get supervisor's departments if specified
supervisor_departments = []
if supervisor_id:
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if supervisor:
supervisor_departments = supervisor.get_all_departments()
# Get courses with supervisor assignments
query = ctx.session.query(Course).filter(
Course.supervisor_id.isnot(None),
Course.is_active == True
)
# Filter by department if supervisor departments are specified
if supervisor_departments:
query = query.filter(Course.department.in_(supervisor_departments))
courses_with_supervisors = query.all()
for course in courses_with_supervisors:
supervisor = ctx.session.query(Supervisor).filter(
Supervisor.id == course.supervisor_id,
Supervisor.is_active == True
).first()
if supervisor:
if course.id not in assignments:
assignments[course.id] = []
assignments[course.id].append(supervisor)
return assignments
def _get_booked_supervisor_sessions(self, ctx) -> List[Dict]:
"""Get all booked supervisor sessions that should be considered as conflicts"""
from src.models.models import TeachingSession
booked_sessions = []
# Get all teaching sessions that are already scheduled (not cancelled)
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.status.in_(['scheduled', 'ongoing', 'completed'])
).all()
for session in sessions:
booked_sessions.append({
'tutor_id': session.tutor_id,
'supervisor_id': session.supervisor_tutor_id,
'day_of_week': session.day_of_week,
'start_time': session.start_time,
'end_time': session.end_time,
'course_id': session.course_id
})
return booked_sessions
def _get_enhanced_availability(self, ctx, tutors: List[Tutor], supervisors: List[Supervisor],
booked_sessions: List[Dict], supervisor_id: str = None) -> Dict[str, List[Dict]]:
"""Get enhanced availability considering booked sessions and conflicts, filtered by department"""
from src.models.models import TutorAvailability
availability = {}
# Get supervisor's departments if specified
supervisor_departments = []
if supervisor_id:
supervisor = ctx.session.query(Supervisor).filter(Supervisor.id == supervisor_id).first()
if supervisor:
supervisor_departments = supervisor.get_all_departments()
# Get availability for tutors (filtered by department)
tutors_with_availability = 0
tutors_filtered_by_department = 0
tutors_without_availability_records = 0
for tutor in tutors:
if not tutor.is_active:
continue
# Filter tutors by department if supervisor departments are specified
if supervisor_departments:
tutor_departments = []
if hasattr(tutor, 'departments') and tutor.departments:
tutor_departments = [dept.department_name for dept in tutor.departments if dept.is_active]
# Check if tutor has any department in common with supervisor
if not any(dept in supervisor_departments for dept in tutor_departments):
tutors_filtered_by_department += 1
current_app.logger.debug(f"[AVAILABILITY FILTER] Tutor {tutor.id} filtered by department. Tutor depts: {tutor_departments}, Supervisor depts: {supervisor_departments}")
continue
tutor_availabilities = ctx.session.query(TutorAvailability).filter(
TutorAvailability.tutor_id == tutor.id,
TutorAvailability.is_approved == True,
TutorAvailability.is_cancelled == False
).all()
if tutor_availabilities:
availability[tutor.id] = self._process_availability_slots(tutor_availabilities, booked_sessions, tutor.id)
tutors_with_availability += 1
else:
tutors_without_availability_records += 1
current_app.logger.debug(f"[AVAILABILITY FILTER] Tutor {tutor.id} has no approved availability records")
current_app.logger.info(f"[AVAILABILITY SUMMARY] Tutors with availability: {tutors_with_availability}, Filtered by department: {tutors_filtered_by_department}, No availability records: {tutors_without_availability_records}")
# Get availability for supervisors (they can teach too)
for supervisor in supervisors:
if not supervisor.is_active:
continue
# For now, assume supervisors are available during standard hours
# You can extend this to use actual supervisor availability if needed
availability[supervisor.id] = self._get_standard_availability_slots(booked_sessions, supervisor.id)
return availability
def _process_availability_slots(self, availabilities: List, booked_sessions: List[Dict], teacher_id: str) -> List[Dict]:
"""Process availability slots and remove conflicts with booked sessions"""
import random
slots = []
for availability in availabilities:
# Convert availability to 2-hour slots
day_slots = self._convert_availability_to_slots(
availability.day_of_week,
availability.start_time,
availability.end_time
)
# Remove conflicts with booked sessions
for slot in day_slots:
if not self._has_conflict_with_booked_sessions(slot, booked_sessions, teacher_id):
slots.append(slot)
# Randomize the order of slots to avoid predictable patterns
random.shuffle(slots)
return slots
def _convert_availability_to_slots(self, day_of_week: int, start_time: time, end_time: time) -> List[Dict]:
"""Convert availability to 2-hour slots with improved flexibility"""
slots = []
# Convert to minutes for easier calculation
start_minutes = start_time.hour * 60 + start_time.minute
end_minutes = end_time.hour * 60 + end_time.minute
# Ensure we have at least 2 hours of availability
if end_minutes - start_minutes < 120:
current_app.logger.warning(f"[SLOT CONVERSION] Availability too short: {start_time}-{end_time} (less than 2 hours)")
return slots
# Create 2-hour slots (120 minutes) with some overlap flexibility
current_minutes = start_minutes
while current_minutes + 120 <= end_minutes:
slot_start = time(current_minutes // 60, current_minutes % 60)
slot_end = time((current_minutes + 120) // 60, (current_minutes + 120) % 60)
slots.append({
'day_of_week': day_of_week,
'start_time': slot_start,
'end_time': slot_end
})
# Move by 30 minutes for overlap flexibility (instead of 120)
current_minutes += 30
current_app.logger.info(f"[SLOT CONVERSION] Created {len(slots)} slots for day {day_of_week} from {start_time} to {end_time}")
return slots
def _has_conflict_with_booked_sessions(self, slot: Dict, booked_sessions: List[Dict], teacher_id: str) -> bool:
"""Check if a slot conflicts with booked sessions"""
for session in booked_sessions:
if session.get('tutor_id') == teacher_id or session.get('supervisor_id') == teacher_id:
if session['day_of_week'] == slot['day_of_week']:
# Check for time overlap
if self._times_overlap(slot['start_time'], slot['end_time'],
session['start_time'], session['end_time']):
return True
return False
def _times_overlap(self, start1: time, end1: time, start2: time, end2: time) -> bool:
"""Check if two time ranges overlap"""
return start1 < end2 and start2 < end1
def _get_standard_availability_slots(self, booked_sessions: List[Dict], supervisor_id: str) -> List[Dict]:
"""Get standard availability slots for supervisors (8 AM to 6 PM)"""
import random
slots = []
# Standard teaching hours: 8 AM to 6 PM
for day in range(5): # Monday to Friday
start_minutes = 8 * 60 # 8 AM
end_minutes = 18 * 60 # 6 PM
current_minutes = start_minutes
while current_minutes + 120 <= end_minutes:
slot_start = time(current_minutes // 60, current_minutes % 60)
slot_end = time((current_minutes + 120) // 60, (current_minutes + 120) % 60)
slot = {
'day_of_week': day,
'start_time': slot_start,
'end_time': slot_end
}
# Remove conflicts with booked sessions
if not self._has_conflict_with_booked_sessions(slot, booked_sessions, supervisor_id):
slots.append(slot)
current_minutes += 120
# Randomize the order of slots to avoid predictable patterns
random.shuffle(slots)
return slots
def _create_availability_matrix(self, availability: Dict[str, List[Dict]], booked_sessions: List[Dict]) -> Dict:
"""Create a matrix to track which time slots are occupied"""
matrix = {}
# Initialize matrix for all teachers
for teacher_id in availability:
matrix[teacher_id] = {}
for day in range(5):
matrix[teacher_id][day] = set()
# Add booked sessions to matrix
for session in booked_sessions:
teacher_id = session.get('tutor_id') or session.get('supervisor_id')
if teacher_id:
if teacher_id not in matrix:
matrix[teacher_id] = {}
if session['day_of_week'] not in matrix[teacher_id]:
matrix[teacher_id][session['day_of_week']] = set()
# Add time range to matrix
start_minutes = session['start_time'].hour * 60 + session['start_time'].minute
end_minutes = session['end_time'].hour * 60 + session['end_time'].minute
for minute in range(start_minutes, end_minutes):
matrix[teacher_id][session['day_of_week']].add(minute)
return matrix
def _schedule_course_enhanced(self, ctx, timetable_id: str, course: Dict, available_teachers: List,
availability: Dict, availability_matrix: Dict, created_blocks: List,
semester: str, supervisor_id: str) -> bool:
"""Enhanced course scheduling with conflict checking and shared course support"""
required_blocks = course.get('required_blocks', 2)
course_id = course['id']
# Check if this is a shared course
is_shared_course = course.get('is_shared_course', False)
shared_specialities = course.get('shared_specialities', [])
sharing_capable_specialities = course.get('sharing_capable_specialities', [])
shared_departments = course.get('shared_departments', [])
sharing_level = course.get('sharing_level', 'single')
# Try each teacher until we find one that can schedule the course
for teacher in available_teachers:
teacher_id = teacher.id
if teacher_id not in availability:
continue
# Get available slots for this teacher
teacher_slots = availability[teacher_id]
current_app.logger.info(f"[COURSE SCHEDULING] Teacher {teacher_id} has {len(teacher_slots)} available slots for course {course_id}")
# Try to find non-consecutive slots
scheduled_slots = self._find_non_consecutive_slots(
teacher_slots, required_blocks, availability_matrix, teacher_id
)
current_app.logger.info(f"[COURSE SCHEDULING] Found {len(scheduled_slots)} slots for course {course_id} with teacher {teacher_id}")
if scheduled_slots:
# Create blocks for these slots
blocks_created = []
# Create blocks for each scheduled slot
for slot in scheduled_slots:
# Determine if teacher is a tutor or supervisor
is_tutor = hasattr(teacher, 'staff_id') or (hasattr(teacher, 'user_type') and teacher.user_type == 'tutor')
is_supervisor = hasattr(teacher, 'department') or (hasattr(teacher, 'user_type') and teacher.user_type == 'supervisor')
# Create the main block
main_block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course_id,
tutor_id=teacher_id if is_tutor else None,
supervisor_tutor_id=teacher_id if is_supervisor else None,
day_of_week=slot['day_of_week'],
start_time=slot['start_time'],
end_time=slot['end_time'],
block_type='lecture',
recurring=True,
created_at=datetime.now()
)
ctx.session.add(main_block)
blocks_created.append(main_block)
# If this is a shared course, create synchronized blocks for all sharing entities
if is_shared_course:
# Handle different sharing levels and course types
shared_course_type = course.get('shared_course_type', 'department_specific')
# For interdisciplinary courses (regardless of sharing level)
if shared_course_type == 'interdisciplinary':
# Create blocks for shared specialities (interdisciplinary courses)
for speciality in shared_specialities:
speciality_block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course_id,
tutor_id=teacher_id if is_tutor else None,
supervisor_tutor_id=teacher_id if is_supervisor else None,
day_of_week=slot['day_of_week'],
start_time=slot['start_time'],
end_time=slot['end_time'],
block_type='lecture',
recurring=True,
created_at=datetime.now(),
# Add metadata to identify this as a shared block
room=f"Shared-{speciality.get('name', 'Speciality')}",
notes=f"Interdisciplinary course block for speciality: {speciality.get('name', 'Unknown')} in department: {speciality.get('department', 'Unknown')}"
)
ctx.session.add(speciality_block)
blocks_created.append(speciality_block)
# Handle sharing levels (can be combined with course types)
if sharing_level == 'institution':
# Create blocks for shared departments (institutional level)
for department in shared_departments:
department_block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course_id,
tutor_id=teacher_id if is_tutor else None,
supervisor_tutor_id=teacher_id if is_supervisor else None,
day_of_week=slot['day_of_week'],
start_time=slot['start_time'],
end_time=slot['end_time'],
block_type='lecture',
recurring=True,
created_at=datetime.now(),
# Add metadata to identify this as a shared block
room=f"Shared-{department.get('department_name', 'Department')}",
notes=f"Institutional course block for department: {department.get('department_name', 'Unknown')}"
)
ctx.session.add(department_block)
blocks_created.append(department_block)
elif sharing_level == 'speciality':
# Create blocks for sharing-capable specialities (speciality level)
for speciality in sharing_capable_specialities:
speciality_block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course_id,
tutor_id=teacher_id if is_tutor else None,
supervisor_tutor_id=teacher_id if is_supervisor else None,
day_of_week=slot['day_of_week'],
start_time=slot['start_time'],
end_time=slot['end_time'],
block_type='lecture',
recurring=True,
created_at=datetime.now(),
# Add metadata to identify this as a shared block
room=f"Shared-{speciality.get('name', 'Speciality')}",
notes=f"Speciality-level course block for sharing-capable speciality: {speciality.get('name', 'Unknown')} in department: {speciality.get('department', 'Unknown')}"
)
ctx.session.add(speciality_block)
blocks_created.append(speciality_block)
elif sharing_level == 'department':
# Create blocks for departmental specialities (department level)
# For departmental level, we create blocks for specialities within the course's department
for speciality in shared_specialities:
speciality_block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course_id,
tutor_id=teacher_id if is_tutor else None,
supervisor_tutor_id=teacher_id if is_supervisor else None,
day_of_week=slot['day_of_week'],
start_time=slot['start_time'],
end_time=slot['end_time'],
block_type='lecture',
recurring=True,
created_at=datetime.now(),
# Add metadata to identify this as a shared block
room=f"Shared-{speciality.get('name', 'Speciality')}",
notes=f"Departmental course block for speciality: {speciality.get('name', 'Unknown')} in department: {speciality.get('department', 'Unknown')}"
)
ctx.session.add(speciality_block)
blocks_created.append(speciality_block)
elif sharing_level == 'external':
# Create blocks for external sharing (cross-institutional)
# This could be shared with external institutions or partners
external_block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course_id,
tutor_id=teacher_id if is_tutor else None,
supervisor_tutor_id=teacher_id if is_supervisor else None,
day_of_week=slot['day_of_week'],
start_time=slot['start_time'],
end_time=slot['end_time'],
block_type='lecture',
recurring=True,
created_at=datetime.now(),
# Add metadata to identify this as an external shared block
room=f"External-Shared",
notes=f"External sharing course block - shared with external institutions or partners"
)
ctx.session.add(external_block)
blocks_created.append(external_block)
else:
# Fallback for other sharing levels or legacy support
# This handles cases where sharing_level is not explicitly set or is 'single'
# Create blocks for shared specialities if they exist
if shared_specialities:
for speciality in shared_specialities:
speciality_block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course_id,
tutor_id=teacher_id if is_tutor else None,
supervisor_tutor_id=teacher_id if is_supervisor else None,
day_of_week=slot['day_of_week'],
start_time=slot['start_time'],
end_time=slot['end_time'],
block_type='lecture',
recurring=True,
created_at=datetime.now(),
# Add metadata to identify this as a shared block
room=f"Shared-{speciality.get('name', 'Speciality')}",
notes=f"Shared course block for speciality: {speciality.get('name', 'Unknown')} in department: {speciality.get('department', 'Unknown')}"
)
ctx.session.add(speciality_block)
blocks_created.append(speciality_block)
# Create blocks for shared departments (if not already covered by specialities)
if shared_departments:
for department in shared_departments:
# Check if this department is already covered by specialities
department_covered = any(spec.get('department') == department.get('department_name') for spec in shared_specialities)
if not department_covered:
department_block = TimetableBlock(
id=str(uuid.uuid4()),
timetable_id=timetable_id,
course_id=course_id,
tutor_id=teacher_id if is_tutor else None,
supervisor_tutor_id=teacher_id if is_supervisor else None,
day_of_week=slot['day_of_week'],
start_time=slot['start_time'],
end_time=slot['end_time'],
block_type='lecture',
recurring=True,
created_at=datetime.now(),
# Add metadata to identify this as a shared block
room=f"Shared-{department.get('department_name', 'Department')}",
notes=f"Shared course block for department: {department.get('department_name', 'Unknown')}"
)
ctx.session.add(department_block)
blocks_created.append(department_block)
# Update availability matrix
self._update_availability_matrix(availability_matrix, teacher_id, scheduled_slots)
# Remove scheduled slots from teacher's availability to prevent reuse
self._remove_scheduled_slots_from_availability(availability, teacher_id, scheduled_slots)
# Add to created blocks list
for block in blocks_created:
created_blocks.append({
'id': block.id,
'course_id': block.course_id,
'tutor_id': block.tutor_id,
'supervisor_tutor_id': block.supervisor_tutor_id,
'day_of_week': block.day_of_week,
'start_time': block.start_time,
'end_time': block.end_time,
'room': getattr(block, 'room', None),
'notes': getattr(block, 'notes', None)
})
# Log shared course information
if is_shared_course:
speciality_count = len(shared_specialities)
department_count = len([d for d in shared_departments if not any(spec.get('department') == d for spec in shared_specialities)])
current_app.logger.info(f"[SHARED COURSE SCHEDULING] Successfully scheduled shared course {course_id} with {len(scheduled_slots)} time slots, {speciality_count} speciality blocks, and {department_count} department blocks")
else:
current_app.logger.info(f"[COURSE SCHEDULING] Successfully scheduled course {course_id} with {len(scheduled_slots)} blocks")
return True
return False
def _remove_scheduled_slots_from_availability(self, availability: Dict, teacher_id: str, scheduled_slots: List[Dict]):
"""Remove scheduled slots from teacher's availability to prevent reuse"""
if teacher_id not in availability:
return
teacher_slots = availability[teacher_id]
slots_to_remove = []
for scheduled_slot in scheduled_slots:
for i, available_slot in enumerate(teacher_slots):
if (available_slot['day_of_week'] == scheduled_slot['day_of_week'] and
available_slot['start_time'] == scheduled_slot['start_time'] and
available_slot['end_time'] == scheduled_slot['end_time']):
slots_to_remove.append(i)
break
# Remove slots in reverse order to maintain indices
for i in sorted(slots_to_remove, reverse=True):
teacher_slots.pop(i)
current_app.logger.info(f"[AVAILABILITY UPDATE] Removed {len(slots_to_remove)} slots from teacher {teacher_id}. Remaining slots: {len(teacher_slots)}")
def _create_default_availability_for_teacher(self, teacher_id: str) -> List[Dict]:
"""Create default availability slots for a teacher (Monday-Friday, 8AM-6PM)"""
default_slots = []
# Create slots for Monday to Friday (0-4)
for day in range(5): # Monday = 0, Friday = 4
# Create 2-hour slots from 8AM to 6PM
start_hour = 8
end_hour = 18
current_hour = start_hour
while current_hour + 2 <= end_hour:
slot_start = time(current_hour, 0)
slot_end = time(current_hour + 2, 0)
default_slots.append({
'day_of_week': day,
'start_time': slot_start,
'end_time': slot_end
})
current_hour += 2
current_app.logger.info(f"[DEFAULT AVAILABILITY] Created {len(default_slots)} default slots for teacher {teacher_id}")
return default_slots
def _handle_common_course_scheduling(self, ctx, course: Course, semester: str, supervisor_id: str) -> Dict:
"""
Handle special scheduling logic for common courses.
Common courses need to be scheduled across multiple departments simultaneously.
Args:
ctx: Database context
course: Course model instance
semester: Semester string
supervisor_id: Supervisor ID
Returns:
Dict with scheduling information
"""
try:
if not getattr(course, 'is_shared_course', False) or getattr(course, 'shared_course_type', '') != 'common':
return {'is_common_course': False}
current_app.logger.info(f"[COMMON COURSE] Handling common course scheduling for {course.code}")
# Get all departments this common course is shared with
shared_departments = ctx.session.query(course_department_association).filter(
course_department_association.c.course_id == course.id,
course_department_association.c.is_active == True
).all()
if not shared_departments:
current_app.logger.warning(f"[COMMON COURSE] Course {course.code} is marked as common but has no department associations")
return {'is_common_course': False, 'error': 'No department associations'}
# Get all timetables for the shared departments
department_timetables = {}
for dept_assoc in shared_departments:
dept_name = dept_assoc.department_name
timetables = ctx.session.query(Timetable).filter(
Timetable.department == dept_name,
Timetable.semester == semester
).all()
department_timetables[dept_name] = timetables
# Find common available time slots across all departments
common_slots = self._find_common_available_slots(ctx, department_timetables, course.id)
return {
'is_common_course': True,
'shared_departments': [dept.department_name for dept in shared_departments],
'department_timetables': {dept: len(timetables) for dept, timetables in department_timetables.items()},
'common_available_slots': len(common_slots),
'scheduling_recommendation': 'Schedule simultaneously across all departments' if common_slots else 'No common slots available'
}
except Exception as e:
current_app.logger.error(f"Error handling common course scheduling for {course.code}: {str(e)}")
return {'is_common_course': False, 'error': str(e)}
def _find_common_available_slots(self, ctx, department_timetables: Dict, course_id: str) -> List[Dict]:
"""
Find time slots that are available across all departments for a common course.
Args:
ctx: Database context
department_timetables: Dictionary mapping department names to their timetables
course_id: Course ID
Returns:
List of common available time slots
"""
try:
# Get all possible time slots (Monday-Friday, 8AM-6PM)
all_possible_slots = []
for day in range(5): # Monday to Friday
for hour in range(8, 18, 2): # 8AM to 6PM in 2-hour blocks
slot = {
'day_of_week': day,
'start_time': time(hour, 0),
'end_time': time(hour + 2, 0)
}
all_possible_slots.append(slot)
common_slots = []
# Check each possible slot against all department timetables
for slot in all_possible_slots:
slot_available_in_all_departments = True
for dept_name, timetables in department_timetables.items():
for timetable in timetables:
# Check if this slot is already occupied in this department's timetable
conflicting_blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id,
TimetableBlock.day_of_week == slot['day_of_week'],
TimetableBlock.course_id != course_id
).all()
for block in conflicting_blocks:
# Check if time slots overlap
if not (slot['end_time'] <= block.start_time or block.end_time <= slot['start_time']):
slot_available_in_all_departments = False
break
if not slot_available_in_all_departments:
break
if not slot_available_in_all_departments:
break
if slot_available_in_all_departments:
common_slots.append(slot)
current_app.logger.info(f"[COMMON SLOTS] Found {len(common_slots)} common available slots for course {course_id}")
return common_slots
except Exception as e:
current_app.logger.error(f"Error finding common available slots: {str(e)}")
return []
def _find_non_consecutive_slots(self, available_slots: List[Dict], required_blocks: int,
availability_matrix: Dict, teacher_id: str) -> List[Dict]:
"""Find non-consecutive slots for a course with randomization"""
import random
if len(available_slots) < required_blocks:
current_app.logger.warning(f"[SLOT FINDING] Not enough slots: {len(available_slots)} < {required_blocks}")
return []
# Group slots by day
slots_by_day = {}
for slot in available_slots:
day = slot['day_of_week']
if day not in slots_by_day:
slots_by_day[day] = []
slots_by_day[day].append(slot)
# Randomize the order of days for variety
available_days = list(slots_by_day.keys())
random.shuffle(available_days)
current_app.logger.info(f"[SLOT FINDING] Available days: {available_days}, Required blocks: {required_blocks}")
for day, slots in slots_by_day.items():
current_app.logger.info(f"[SLOT FINDING] Day {day}: {len(slots)} slots")
# Try to find slots on different days first (randomized order)
selected_slots = []
used_days = set()
# First pass: Try to get one slot from each available day (randomized)
for day in available_days:
if len(selected_slots) >= required_blocks:
break
if day in used_days:
continue
day_slots = slots_by_day[day]
# Randomize the order of slots within each day
random.shuffle(day_slots)
# Find a slot that doesn't conflict with existing blocks
for slot in day_slots:
if not self._slot_conflicts_with_matrix(slot, availability_matrix, teacher_id):
selected_slots.append(slot)
used_days.add(day)
break
# Second pass: If we need more slots, try same day but non-consecutive (randomized)
if len(selected_slots) < required_blocks:
# Randomize the order again for the second pass
random.shuffle(available_days)
for day in available_days:
if len(selected_slots) >= required_blocks:
break
if day not in slots_by_day:
continue
day_slots = slots_by_day[day]
# Randomize the order of slots within each day
random.shuffle(day_slots)
# Find non-consecutive slots
for slot in day_slots:
if not self._slot_conflicts_with_matrix(slot, availability_matrix, teacher_id):
# Check if this slot is consecutive with any selected slot
is_consecutive = False
for selected_slot in selected_slots:
if selected_slot['day_of_week'] == day:
if self._slots_are_consecutive(slot, selected_slot):
is_consecutive = True
break
if not is_consecutive:
selected_slots.append(slot)
break
# If we still don't have enough slots, try a more flexible approach
if len(selected_slots) < required_blocks:
current_app.logger.warning(f"[SLOT FINDING] Only found {len(selected_slots)} slots, trying flexible approach")
# Try to get any available slots, even if consecutive
for day in available_days:
if len(selected_slots) >= required_blocks:
break
day_slots = slots_by_day[day]
random.shuffle(day_slots)
for slot in day_slots:
if len(selected_slots) >= required_blocks:
break
if not self._slot_conflicts_with_matrix(slot, availability_matrix, teacher_id):
# Check if we already have this exact slot
slot_exists = any(
existing['day_of_week'] == slot['day_of_week'] and
existing['start_time'] == slot['start_time'] and
existing['end_time'] == slot['end_time']
for existing in selected_slots
)
if not slot_exists:
selected_slots.append(slot)
# Final randomization: Shuffle the selected slots to avoid predictable patterns
random.shuffle(selected_slots)
current_app.logger.info(f"[SLOT FINDING] Final result: {len(selected_slots)} slots found")
return selected_slots[:required_blocks]
def _slot_conflicts_with_matrix(self, slot: Dict, availability_matrix: Dict, teacher_id: str) -> bool:
"""Check if a slot conflicts with the availability matrix"""
if teacher_id not in availability_matrix:
return False
day = slot['day_of_week']
if day not in availability_matrix[teacher_id]:
return False
start_minutes = slot['start_time'].hour * 60 + slot['start_time'].minute
end_minutes = slot['end_time'].hour * 60 + slot['end_time'].minute
for minute in range(start_minutes, end_minutes):
if minute in availability_matrix[teacher_id][day]:
return True
return False
def _update_availability_matrix(self, availability_matrix: Dict, teacher_id: str, slots: List[Dict]):
"""Update availability matrix with newly scheduled slots"""
if teacher_id not in availability_matrix:
availability_matrix[teacher_id] = {}
for slot in slots:
day = slot['day_of_week']
if day not in availability_matrix[teacher_id]:
availability_matrix[teacher_id][day] = set()
start_minutes = slot['start_time'].hour * 60 + slot['start_time'].minute
end_minutes = slot['end_time'].hour * 60 + slot['end_time'].minute
for minute in range(start_minutes, end_minutes):
availability_matrix[teacher_id][day].add(minute)
def _validate_enhanced_timetable(self, ctx, timetable_id: str) -> Dict:
"""Validate the enhanced timetable for conflicts"""
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable_id
).all()
conflicts = []
# Check for tutor conflicts
for i, block1 in enumerate(blocks):
for j, block2 in enumerate(blocks[i+1:], i+1):
if block1.tutor_id and block2.tutor_id and block1.tutor_id == block2.tutor_id:
if block1.day_of_week == block2.day_of_week:
if self._times_overlap(block1.start_time, block1.end_time,
block2.start_time, block2.end_time):
conflicts.append({
'type': 'tutor_conflict',
'tutor_id': block1.tutor_id,
'blocks': [block1.id, block2.id],
'message': f'Tutor has overlapping classes'
})
# Check for supervisor conflicts
for i, block1 in enumerate(blocks):
for j, block2 in enumerate(blocks[i+1:], i+1):
if block1.supervisor_tutor_id and block2.supervisor_tutor_id and block1.supervisor_tutor_id == block2.supervisor_tutor_id:
if block1.day_of_week == block2.day_of_week:
if self._times_overlap(block1.start_time, block1.end_time,
block2.start_time, block2.end_time):
conflicts.append({
'type': 'supervisor_conflict',
'supervisor_id': block1.supervisor_tutor_id,
'blocks': [block1.id, block2.id],
'message': f'Supervisor has overlapping classes'
})
return {
'is_valid': len(conflicts) == 0,
'conflict_count': len(conflicts),
'conflicts': conflicts
}
def c(self, department: str) -> Dict:
"""
Get approved timetables for a specific department
Args:
department: Department name
Returns:
Response with list of approved timetables for the department
"""
with DatabaseContextManager() as ctx:
# Get all supervisors for this department
timetables = ctx.session.query(Timetable).join(
SupervisorDepartment, Timetable.approved_by == SupervisorDepartment.supervisor_id
).filter(
SupervisorDepartment.department_name == department,
Timetable.approval_status == 'approved'
).order_by(Timetable.created_at.desc()).all()
if not timetables:
return custom_response(
success=False,
data=f"No approved timetables found for department '{department}'",
status_code=404
)
# Format timetables with blocks
formatted_timetables = []
for timetable in timetables:
# Get all blocks for this timetable
blocks = ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).all()
timetable_data = self._timetable_to_dict(timetable)
timetable_data['blocks'] = [self.block_manager._block_to_dict(block) for block in blocks]
formatted_timetables.append(timetable_data)
return custom_response(
success=True,
data=formatted_timetables,
status_code=200
)
def get_all_departments_with_timetables(self) -> Dict:
"""
Debug method to get all departments that have timetables
Returns:
Response with list of departments and their timetable counts
"""
with DatabaseContextManager() as ctx:
# Get all unique departments that have timetables
departments = ctx.session.query(Timetable.department).distinct().all()
department_stats = []
for dept in departments:
if dept[0]: # Skip None/empty departments
dept_name = dept[0]
total_count = ctx.session.query(Timetable).filter(
Timetable.department == dept_name
).count()
approved_count = ctx.session.query(Timetable).filter(
Timetable.department == dept_name,
Timetable.approval_status == 'approved',
Timetable.is_active == True
).count()
department_stats.append({
'department': dept_name,
'total_timetables': total_count,
'approved_timetables': approved_count
})
return custom_response(
success=True,
data={
'departments': department_stats,
'total_departments': len(department_stats)
},
status_code=200
)
def _timetable_to_dict(self, timetable: Timetable) -> Dict:
"""Convert Timetable model to dictionary"""
with DatabaseContextManager() as ctx:
return {
'id': timetable.id,
'name': timetable.name,
'description': timetable.description,
'semester': timetable.semester,
'academic_year': timetable.academic_year,
'created_by': timetable.created_by,
'creator_name': f"{timetable.creator.first_name} {timetable.creator.last_name}" if timetable.creator else None,
'created_at': timetable.created_at.isoformat() if timetable.created_at else None,
'approved_by': timetable.approved_by,
'approver_name': f"{timetable.approver.first_name} {timetable.approver.last_name}" if timetable.approver else None,
'approved_at': timetable.approved_at.isoformat() if timetable.approved_at else None,
'approval_status': timetable.approval_status,
'is_active': timetable.is_active,
'blocks_count': ctx.session.query(TimetableBlock).filter(
TimetableBlock.timetable_id == timetable.id
).count() if hasattr(timetable, 'blocks_count') else None
}
def _notify_supervisors_of_approved_timetable(self, timetable: Timetable) -> None:
"""
Send notifications to all supervisors about the approved timetable
Args:
timetable: Timetable model instance
"""
with DatabaseContextManager() as ctx:
# Get all supervisors
supervisors = ctx.session.query(Supervisor).filter(
Supervisor.is_active == True
).all()
for supervisor in supervisors:
# Check notification preferences
prefs = ctx.session.query(NotificationPreference).filter(
NotificationPreference.user_id == supervisor.id
).first()
if not prefs or not prefs.receive_email:
continue
# Get supervisor's department tutors and their sessions
department_tutors = ctx.session.query(Tutor).filter(
Tutor.department == (supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Not assigned'),
Tutor.is_active == True
).all()
# Get teaching sessions for department tutors
tutor_sessions = {}
total_sessions = 0
for tutor in department_tutors:
sessions = ctx.session.query(TeachingSession).filter(
TeachingSession.timetable_id == timetable.id,
TeachingSession.tutor_id == tutor.id
).all()
if sessions:
tutor_sessions[tutor] = sessions
total_sessions += len(sessions)
# Prepare email
subject = f"Timetable Approved: {timetable.semester} {timetable.academic_year} - {(supervisor.get_primary_department() if hasattr(supervisor, 'get_primary_department') else 'Unknown')} Department"
# Build message with modern styling
message = f"""
Department Timetable Approved
Hello {supervisor.first_name} {supervisor.last_name},
✅ Timetable Successfully Approved!
The timetable for your department has been approved and teaching sessions have been created.
📊 Department Overview
{len(department_tutors)}
Active Tutors
{total_sessions}
Teaching Sessions
{len(tutor_sessions)}
Tutors with Sessions
👥 Department Tutors & Sessions
Next Steps:
- Monitor tutor performance and attendance
- Review teaching session reports
- Address any scheduling conflicts or issues
- Provide support to tutors as needed
"""
try:
send_email(
sender_email=current_app.config["MAIL_SENDER"],
sender_password=current_app.config["MAIL_PASSWORD"],
receiver_email=supervisor.email,
subject=subject,
message=message
)
except Exception as e:
current_app.logger.error(f"Failed to send timetable notification to supervisor {supervisor.id}: {str(e)}")