""" Planning Service - GMAO Collège Service de calcul de planification des interventions """ from datetime import datetime, timedelta, date, time from typing import List, Dict, Optional, Tuple from app_new.extensions import db from app_new.core.models.planning import ( ScheduledTask, AdminTask, WorkSchedule, CollegeClosure, ClosureSchedule, ClosureWorkDay, TechnicianAvailability, PersonalLeave, Training, TrainingParticipant, PlanningDay, PlanningItem ) from app_new.core.models.maintenance import Intervention from app_new.core.models.equipment import Equipment from app_new.core.models.college import Room from sqlalchemy import and_, or_ def get_french_public_holidays(year: int) -> List[Tuple[date, str]]: """ Calcule les jours fériés français pour une année donnée. Inclut les jours fériés fixes et variables (Pâques, Ascension, Pentecôte). Utilise l'algorithme de Gauss pour calculer la date de Pâques. """ holidays = [] # Jours fériés fixes fixed_holidays = [ (1, 1, "Jour de l'an"), (5, 1, "Fête du travail"), (5, 8, "Victoire 1945"), (7, 14, "Fête nationale"), (8, 15, "Assomption"), (11, 1, "Toussaint"), (11, 11, "Armistice 1918"), (12, 25, "Noël"), ] for month, day, name in fixed_holidays: holidays.append((date(year, month, day), name)) # Calcul de Pâques avec l'algorithme de Butcher-Meeus # Formule simplifiée pour l'Église catholique (grégorien) a = year % 19 b = year // 100 c = year % 100 d = b // 4 e = b % 4 f = (b + 8) // 25 g = (b - f + 1) // 3 h = (19 * a + b - d - g + 15) % 30 i = c // 4 k = c % 4 l = (32 + 2 * e + 2 * i - h - k) % 7 m = (a + 11 * h + 22 * l) // 451 month_easter = (h + l - 7 * m + 114) // 31 day_easter = ((h + l - 7 * m + 114) % 31) + 1 easter_sunday = date(year, month_easter, day_easter) # Lundi de Pâques (lendemain de Pâques) holidays.append((easter_sunday + timedelta(days=1), "Lundi de Pâques")) # Ascension (39 jours après Pâques) holidays.append((easter_sunday + timedelta(days=39), "Ascension")) # Lundi de Pentecôte (50 jours après Pâques) holidays.append((easter_sunday + timedelta(days=50), "Lundi de Pentecôte")) return holidays def is_public_holiday(d: date) -> Tuple[bool, str]: """ Vérifie si une date est un jour férié français. Retourne (True, nom du jour férié) ou (False, ''). """ holidays = get_french_public_holidays(d.year) for h_date, h_name in holidays: if h_date == d: return True, h_name return False, '' class DaySchedule: """Structure pour représenter le planning d'une journée.""" def __init__(self, date_obj: date): self.date = date_obj self.is_working = False self.day_name = '' self.working_hours = None self.closure = None self.leave = None self.holiday = None # Nom du jour férié si applicable self.items = [] self.grouped_items = [] class PlanningService: """Service de planification des interventions.""" # Priorités PRIORITY_ADMIN = 1 PRIORITY_CURATIVE_HIGH = 2 PRIORITY_CURATIVE_NORMAL = 3 PRIORITY_PREVENTIVE_PRIORITARY = 2 PRIORITY_PREVENTIVE_NORMAL = 4 PRIORITY_FORMATION = 1 @staticmethod def maintenance_unavailability(d: date, user_id=None): """Retourne le motif qui interdit toute maintenance ce jour-là.""" leave_query = PersonalLeave.query.filter( PersonalLeave.start_date <= d, PersonalLeave.end_date >= d, ) if user_id is not None: leave_query = leave_query.filter(PersonalLeave.user_id == user_id) leave = leave_query.first() if leave: return f"Absence : {leave.leave_type or 'indisponible'}" training_query = Training.query.join(TrainingParticipant).filter( Training.start_date <= d, Training.end_date >= d, TrainingParticipant.is_confirmed.is_(True), ) if user_id is not None: training_query = training_query.filter(TrainingParticipant.user_id == user_id) training = training_query.first() if training: return f"Formation : {training.name}" return None @staticmethod def get_working_hours(d: date, user_id=None) -> Optional[Tuple[time, time, Optional[time], Optional[time]]]: """ Retourne les horaires de travail pour une date donnée. Retourne (start, end, lunch_start, lunch_end) ou None si non travaillé. """ # Vérifier si c'est un jour férié français is_holiday, holiday_name = is_public_holiday(d) if is_holiday: return None if PlanningService.maintenance_unavailability(d, user_id): return None # Vérifier si c'est un jour de vacances closure = CollegeClosure.query.filter( CollegeClosure.start_date <= d, CollegeClosure.end_date >= d ).first() day_of_week = d.weekday() # 0=lundi, 6=dimanche if closure: # Une fermeture scolaire est non travaillée par défaut. Une date # exacte peut toutefois être rouverte avec ses propres horaires. exceptional_day = ClosureWorkDay.query.filter_by( closure_id=closure.id, work_date=d, ).first() if exceptional_day: return ( exceptional_day.start_time, exceptional_day.end_time, exceptional_day.lunch_start, exceptional_day.lunch_end, ) if closure.work_hours_type == 'none': return None if closure.work_hours_type == 'reduced': return (time(9), time(12), None, None) if closure.work_hours_type == 'custom': schedule = ClosureSchedule.query.filter_by( closure_id=closure.id, day_of_week=day_of_week ).first() if schedule: return (schedule.start_time, schedule.end_time, schedule.lunch_start, schedule.lunch_end) return None # Vérifier les congés personnels leave = TechnicianAvailability.query.filter_by(date=d).first() if leave and leave.availability_type in ('conge', 'maladie'): return None # Horaires normaux - chercher le premier schedule avec des horaires valides schedules = WorkSchedule.query.filter_by(day_of_week=day_of_week).all() for schedule in schedules: if schedule.start_time and schedule.end_time: return (schedule.start_time, schedule.end_time, schedule.lunch_start, schedule.lunch_end) # Si un schedule existe mais sans horaires, considérer comme non travaillé if schedules: return None # Week-end non travaillé par défaut if day_of_week >= 5: return None # Horaires par défaut (8h-17h avec pause 12h-13h) return (time(8), time(17), time(12), time(13)) @staticmethod def get_day_schedule(d: date) -> DaySchedule: """Retourne le planning complet pour une journée.""" days_fr = ['Lundi', 'Mardi', 'Mercredi', 'Jeudi', 'Vendredi', 'Samedi', 'Dimanche'] schedule = DaySchedule(d) schedule.day_name = days_fr[d.weekday()] # Vérifier les vacances closure = CollegeClosure.query.filter( CollegeClosure.start_date <= d, CollegeClosure.end_date >= d ).first() # Vérifier les congés personnels leave = TechnicianAvailability.query.filter_by(date=d).first() # Horaires de travail hours = PlanningService.get_working_hours(d) if hours: schedule.is_working = True schedule.working_hours = { 'start': hours[0], 'end': hours[1], 'lunch_start': hours[2], 'lunch_end': hours[3] } else: schedule.is_working = False if closure: schedule.closure = closure.name elif leave: schedule.leave = leave.availability_type else: # Vérifier si c'est un jour férié is_holiday, holiday_name = is_public_holiday(d) if is_holiday: schedule.holiday = holiday_name # Ne pas ajouter de tâches si ce n'est pas un jour de travail return schedule # Récupérer les interventions planifiées pour ce jour scheduled_tasks = ScheduledTask.query.filter_by( scheduled_date=d, status='planned' ).all() for task in scheduled_tasks: # Récupérer le titre depuis la tâche préventive associée task_title = 'Tâche planifiée' if task.preventive_task: task_title = task.preventive_task.name elif task.equipment: task_title = f"Intervention sur {task.equipment.name}" item = { 'id': task.id, 'type': 'preventive', 'title': task_title, 'start_time': task.scheduled_start, 'end_time': task.scheduled_end, 'duration': task.estimated_duration, 'room_id': task.room_id, 'room_name': task.room.name if task.room else (task.equipment.effective_room.name if task.equipment and task.equipment.effective_room else None), 'status': task.status } schedule.items.append(item) # Récupérer les interventions curatives en cours interventions = Intervention.query.filter_by( scheduled_date=d, type='curatif' ).all() for interv in interventions: item = { 'id': interv.id, 'type': 'curative', 'title': interv.title, 'start_time': None, 'end_time': None, 'duration': None, 'room_id': None, 'room_name': interv.room.name if interv.room else None, 'status': interv.status } schedule.items.append(item) # Récupérer les tâches administratives # AdminTask est récurrent (frequency, day_of_week) - on filtre par jour de la semaine admin_tasks = AdminTask.query.filter_by( is_active=True ).all() for admin in admin_tasks: # Vérifier si la tâche correspond au jour actuel if admin.frequency == 'daily': pass # Tous les jours elif admin.frequency == 'weekly' and admin.day_of_week != d.weekday(): continue elif admin.frequency == 'monthly' and admin.day_of_month != d.day: continue item = { 'id': admin.id, 'type': 'admin', 'title': admin.name, 'start_time': admin.start_time, 'end_time': None, # Calculé à partir de start_time + duration 'duration': admin.duration_minutes, 'room_id': None, 'room_name': None, 'status': 'planned' } schedule.items.append(item) # Grouper par salle rooms = {} for item in schedule.items: room_name = item.get('room_name') or 'Sans salle' if room_name not in rooms: rooms[room_name] = [] rooms[room_name].append(item) schedule.grouped_items = [ {'room_name': room_name, 'task_items': items} for room_name, items in sorted(rooms.items()) ] return schedule @staticmethod def is_room_available(room_id: int, start_time: datetime, end_time: datetime) -> bool: """ Vérifie si une salle est disponible pendant une période donnée. Utilise le planning PRONOTE importé. """ from app_new.core.models.college import PronoteSchedule courses = PronoteSchedule.query.filter( PronoteSchedule.room_id == room_id, PronoteSchedule.start_time < end_time, PronoteSchedule.end_time > start_time ).all() return len(courses) == 0 @staticmethod def get_room_occupancy(room_id: int, d: date) -> List[Tuple[time, time]]: """Retourne les périodes occupées d'une salle pour une date.""" from app_new.core.models.college import PronoteSchedule start_dt = datetime.combine(d, time(0)) end_dt = datetime.combine(d, time(23, 59)) courses = PronoteSchedule.query.filter( PronoteSchedule.room_id == room_id, PronoteSchedule.start_time >= start_dt, PronoteSchedule.start_time < end_dt ).order_by(PronoteSchedule.start_time).all() return [(c.start_time.time(), c.end_time.time()) for c in courses] @staticmethod def calculate_planning(start_date: date, end_date: date) -> List[PlanningDay]: """ Calcule le planning sur une période donnée. Prend en compte les priorités, les disponibilités et les regroupements par salle. """ from app_new.core.models.equipment import EquipmentRestriction planning_days = [] current_date = start_date while current_date <= end_date: # Vérifier si le jour est travaillé if PlanningService.get_working_hours(current_date): day_schedule = PlanningService.get_day_schedule(current_date) planning_days.append(day_schedule) current_date += timedelta(days=1) return planning_days @staticmethod def schedule_task(task: ScheduledTask, preferred_date: Optional[date] = None) -> bool: """ Planifie une tâche à une date optimale. Tient compte des disponibilités, restrictions et regroupements. """ from app_new.core.models.equipment import EquipmentRestriction if preferred_date: task.scheduled_date = preferred_date task.status = 'planned' db.session.commit() return True # Trouver la prochaine date disponible current_date = date.today() max_days = 365 # Maximum 1 an for _ in range(max_days): current_date += timedelta(days=1) # Vérifier si c'est un jour travaillé hours = PlanningService.get_working_hours(current_date) if not hours: continue # Vérifier les restrictions de l'équipement if task.equipment: restrictions = EquipmentRestriction.query.filter_by( equipment_id=task.equipment.id ).all() valid = True for restriction in restrictions: if restriction.restriction_type == 'time': # Vérifier si le jour de la semaine correspond day_names = ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'] if restriction.days_of_week: allowed_days = restriction.days_of_week.split(',') if day_names[current_date.weekday()] not in allowed_days: valid = False break elif restriction.restriction_type == 'room_occupancy': # Vérifier si la salle est libre room_id = restriction.room_id if room_id and not PlanningService.is_room_available(room_id, datetime.combine(current_date, time(8, 0)), datetime.combine(current_date, time(17, 0))): valid = False break if not valid: continue # Vérifier la disponibilité de la salle if task.room_id: # TODO: Intégrer avec PRONOTE pass # Planifier la tâche task.scheduled_date = current_date task.status = 'planned' db.session.commit() return True return False @staticmethod def postpone_task(task_id: int, new_date: Optional[date] = None, auto: bool = False) -> bool: """ Reporte une tâche à une date ultérieure. Si auto=True, reporte automatiquement au prochain jour disponible. """ task = ScheduledTask.query.get(task_id) if not task: return False if auto: # Trouver le prochain jour disponible current_date = task.scheduled_date or date.today() current_date += timedelta(days=1) return PlanningService.schedule_task(task, preferred_date=None) else: if new_date: task.scheduled_date = new_date db.session.commit() return True return False