"""Proposition déterministe d'une journée de travail. Ce service ne persiste jamais une proposition. Il transforme les sources existantes (tâches planifiées, interventions et tâches administratives) en objets homogènes, puis cherche des créneaux compatibles avec les horaires de travail et les occupations internes des salles. """ from dataclasses import dataclass, field from datetime import date, datetime, time, timedelta from typing import Iterable, Optional from .planning_service import PlanningService from .room_planning import room_is_available TYPE_LABELS = { "preventive": "Maintenance préventive", "curative": "Intervention", "admin": "Administratif", "external_company": "Entreprise extérieure", "emergency": "Urgence", } CONSTRAINT_LABELS = { "fixed": "Horaire fixe", "urgent": "Urgent", "deadline": "À réaliser avant une échéance", "flexible": "Déplaçable", } @dataclass class PlanningCandidate: source_type: str source_id: int title: str description: str = "" task_type: str = "preventive" priority: int = 3 constraint: str = "flexible" target_date: Optional[date] = None earliest_start: Optional[time] = None latest_end: Optional[time] = None duration_minutes: int = 30 room_id: Optional[int] = None room_name: Optional[str] = None zone_name: Optional[str] = None building_name: Optional[str] = None assigned_to: Optional[int] = None status: str = "à planifier" fixed_start: Optional[time] = None fixed_end: Optional[time] = None explanation: str = "" proposed_start: Optional[time] = field(default=None, compare=False) proposed_end: Optional[time] = field(default=None, compare=False) @property def type_label(self): return TYPE_LABELS.get(self.task_type, self.task_type) @property def constraint_label(self): return CONSTRAINT_LABELS.get(self.constraint, self.constraint) @property def location_label(self): parts = [self.building_name, self.zone_name, self.room_name] return " > ".join(part for part in parts if part) or "Localisation non renseignée" @dataclass class WorkWindow: start: time end: time def _minutes(value: time) -> int: return value.hour * 60 + value.minute def _as_time(value: int) -> time: return time(value // 60, value % 60) def _overlap(a_start: time, a_end: time, b_start: time, b_end: time) -> bool: return a_start < b_end and a_end > b_start class DayPlanner: """Construit une proposition sans appeler aucune intégration externe.""" @staticmethod def work_windows(day: date, user_id: Optional[int] = None) -> list[WorkWindow]: # Le planificateur ne fabrique pas silencieusement une journée type. # Les anciennes vues peuvent conserver leur comportement historique, # mais « Ma journée » exige une configuration explicite des horaires. from ..models.planning import WorkSchedule, WorkScheduleTemplate query = WorkSchedule.query.filter_by(day_of_week=day.weekday(), is_active=True) if user_id is not None: query = query.filter(WorkSchedule.user_id == user_id) configured = query.first() is not None if not configured and user_id is not None: configured = WorkScheduleTemplate.query.filter_by(user_id=user_id, is_active=True).first() is not None if not configured: return [] hours = PlanningService.get_working_hours(day, user_id=user_id) if not hours: return [] start, end, lunch_start, lunch_end = hours windows = [WorkWindow(start, end)] if lunch_start and lunch_end and lunch_start < lunch_end: windows = [] if start < lunch_start: windows.append(WorkWindow(start, lunch_start)) if lunch_end < end: windows.append(WorkWindow(lunch_end, end)) return windows @staticmethod def _slot_available(day, candidate, start, end, occupied): if any(_overlap(start, end, item_start, item_end) for item_start, item_end, _ in occupied): return False if candidate.room_id and not room_is_available( candidate.room_id, datetime.combine(day, start), datetime.combine(day, end), ): return False if candidate.earliest_start and start < candidate.earliest_start: return False if candidate.latest_end and end > candidate.latest_end: return False return True @classmethod def propose(cls, day: date, candidates: Iterable[PlanningCandidate], user_id=None): candidates = list(candidates) windows = cls.work_windows(day, user_id) if not windows: for candidate in candidates: candidate.explanation = "Aucun horaire de travail disponible pour cette journée." return candidates fixed = [c for c in candidates if c.fixed_start and c.fixed_end] flexible = [c for c in candidates if c not in fixed] occupied = [] for candidate in sorted(fixed, key=lambda item: (_minutes(item.fixed_start), item.source_type, item.source_id)): candidate.proposed_start = candidate.fixed_start candidate.proposed_end = candidate.fixed_end candidate.status = "planifié" candidate.explanation = "Horaire fixe conservé." occupied.append((candidate.fixed_start, candidate.fixed_end, candidate)) # Les urgences, échéances et priorités passent avant le simple # regroupement géographique. Le tri reste stable et déterministe. flexible.sort(key=lambda item: ( 0 if item.constraint == "urgent" else 1, 0 if item.constraint == "deadline" else 1, item.priority, item.latest_end or time(23, 59), item.building_name or "", item.zone_name or "", item.room_name or "", item.source_type, item.source_id, )) previous = None for candidate in flexible: duration = max(int(candidate.duration_minutes or 1), 1) chosen = None for window in windows: cursor = _minutes(window.start) window_end = _minutes(window.end) while cursor + duration <= window_end: start, end = _as_time(cursor), _as_time(cursor + duration) if cls._slot_available(day, candidate, start, end, occupied): chosen = (start, end) break cursor += 5 if chosen: break if chosen: candidate.proposed_start, candidate.proposed_end = chosen candidate.status = "proposé" if previous and previous.building_name == candidate.building_name and previous.zone_name == candidate.zone_name: reason = "Vous êtes déjà dans la même zone" elif candidate.room_id: reason = "salle disponible et durée compatible" else: reason = "créneau de travail disponible" candidate.explanation = reason + (" ; échéance prioritaire" if candidate.constraint == "deadline" else "") occupied.append((chosen[0], chosen[1], candidate)) previous = candidate else: candidate.status = "à replanifier" candidate.explanation = "Aucun créneau compatible aujourd'hui." return sorted(candidates, key=lambda item: (_minutes(item.proposed_start) if item.proposed_start else 9999, item.source_type, item.source_id)) @classmethod def from_database(cls, day: date, user_id=None): """Charge les sources accessibles sans créer de nouvelle table.""" from ..models.planning import ScheduledTask, AdminTask from ..models.maintenance import Intervention candidates = [] tasks = ScheduledTask.query.filter( ScheduledTask.scheduled_date == day, ScheduledTask.status.in_(("planned", "in_progress", "suspended", "postponed")), ScheduledTask.intervention_id.is_(None), ).all() for task in tasks: if user_id and task.assigned_to_id not in (None, user_id): continue room = task.room or (task.equipment.effective_room if task.equipment else None) candidates.append(PlanningCandidate( source_type="scheduled_task", source_id=task.id, title=(task.preventive_task.name if task.preventive_task else task.lot_task.tache if task.lot_task else "Maintenance préventive"), task_type="preventive", priority=2 if task.lot_task_id else 4, constraint="fixed" if task.scheduled_start and task.scheduled_end else "flexible", target_date=day, duration_minutes=int(task.estimated_duration or 30), room_id=task.room_id or (room.id if room else None), room_name=room.name if room else None, zone_name=room.zone.name if room and room.zone else None, building_name=room.building.name if room and room.building else None, assigned_to=task.assigned_to_id, status=task.status, fixed_start=task.scheduled_start, fixed_end=task.scheduled_end, )) interventions = Intervention.query.filter( Intervention.scheduled_date == day, Intervention.is_deleted.is_(False), ~Intervention.status.in_(("terminee", "cloturee", "annulee", "ignoree")), ).all() for intervention in interventions: if user_id and intervention.technician_id not in (None, user_id) and intervention.assigned_to_id not in (None, user_id): continue room = intervention.room candidates.append(PlanningCandidate( source_type="intervention", source_id=intervention.id, title=intervention.title, description=intervention.description or "", task_type="emergency" if intervention.priority == "urgente" else "curative", priority=1 if intervention.priority == "urgente" else 3, constraint="urgent" if intervention.priority == "urgente" else ("fixed" if intervention.scheduled_start and intervention.scheduled_end else "flexible"), target_date=day, duration_minutes=int(intervention.estimated_duration or 30), room_id=intervention.room_id, room_name=room.name if room else None, zone_name=room.zone.name if room and room.zone else None, building_name=room.building.name if room and room.building else None, assigned_to=intervention.technician_id or intervention.assigned_to_id, status=intervention.status, fixed_start=intervention.scheduled_start, fixed_end=intervention.scheduled_end, )) for admin in AdminTask.query.filter_by(is_active=True).all(): if admin.frequency == "weekly" and admin.day_of_week != day.weekday(): continue if admin.frequency == "monthly" and admin.day_of_month != day.day: continue candidates.append(PlanningCandidate( source_type="admin_task", source_id=admin.id, title=admin.name, description=admin.description or "", task_type="admin", priority=3, constraint="fixed" if admin.start_time else "flexible", target_date=day, duration_minutes=int(admin.duration_minutes or 30), fixed_start=admin.start_time, fixed_end=(datetime.combine(day, admin.start_time) + timedelta(minutes=int(admin.duration_minutes or 30))).time() if admin.start_time else None, status="à planifier", )) return cls.propose(day, candidates, user_id=user_id)