"""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 ...extensions import db from .planning_service import PlanningService from .room_planning import room_is_available, room_occupancy 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 def _resolved_room_occupancy_batch(day, room_ids): """Charge les occupations de toutes les salles en une requête.""" if not room_ids: return {} from ..models.college import RoomSchedule monday = day - timedelta(days=day.weekday()) rows = RoomSchedule.query.filter( RoomSchedule.room_id.in_(room_ids), RoomSchedule.resolution_status == 'active', RoomSchedule.week_start == monday, RoomSchedule.day_of_week == day.weekday(), ).all() grouped = {room_id: [] for room_id in room_ids} for row in rows: if row.valid_from and row.valid_from > day: continue if row.valid_to and day > row.valid_to: continue grouped.setdefault(row.room_id, []).append(row) resolved = {} for room_id, room_rows in grouped.items(): pronote = [row for row in room_rows if row.source == 'pronote'] selected = [] for row in room_rows: if row.source == 'manual' and not row.protected_from_sync: replaced = any( _overlap(row.start_time, row.end_time, other.start_time, other.end_time) and (not row.class_name or not other.class_name or row.class_name == other.class_name) and (not row.subject or not other.subject or row.subject == other.subject) for other in pronote ) if replaced: continue selected.append((row.start_time, row.end_time, row)) resolved[room_id] = selected return resolved 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, room_occupied=None): if any(_overlap(start, end, item_start, item_end) for item_start, item_end, _ in occupied): return False if candidate.room_id and room_occupied is not None: if any(_overlap(start, end, item_start, item_end) for item_start, item_end, _ in room_occupied.get(candidate.room_id, [])): return False elif 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 = [] room_occupied = _resolved_room_occupancy_batch(day, {candidate.room_id for candidate in candidates if candidate.room_id}) 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é." if any(_overlap(candidate.fixed_start, candidate.fixed_end, start, end) for start, end, _ in occupied): candidate.status = "conflit" candidate.explanation = "Ce rendez-vous fixe chevauche un autre événement fixe ; décision humaine nécessaire." 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 = int(candidate.duration_minutes or 0) if duration <= 0: if candidate.task_type == "external_company": candidate.status = "information" candidate.explanation = "Passage d'entreprise sans plage horaire bloquante." continue candidate.status = "à replanifier" candidate.explanation = "La durée de cette tâche n'est pas configurée." continue 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, room_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 alternative_slots(cls, day: date, target: PlanningCandidate, candidates: Iterable[PlanningCandidate], user_id=None, limit=3): """Retourne quelques créneaux réellement validables pour une tâche.""" windows = cls.work_windows(day, user_id) duration = int(target.duration_minutes or 0) if duration <= 0: return [] occupied = [] room_occupied = _resolved_room_occupancy_batch(day, {item.room_id for item in candidates if item.room_id}) for item in candidates: if item is target or not item.proposed_start or not item.proposed_end: continue occupied.append((item.proposed_start, item.proposed_end, item)) slots = [] for window in windows: cursor = _minutes(window.start) while cursor + duration <= _minutes(window.end): start, end = _as_time(cursor), _as_time(cursor + duration) if cls._slot_available(day, target, start, end, occupied, room_occupied): slots.append((start, end)) if len(slots) >= limit: return slots cursor += 5 return slots @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 from ...contracts.models import ContractVisit 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) task_type = "external_company" if task.company_id else "preventive" title = (task.preventive_task.name if task.preventive_task else task.lot_task.tache if task.lot_task else task.room_name_snapshot or "Maintenance préventive") if task.company: title = f"Accompagnement entreprise — {task.company.name}" candidates.append(PlanningCandidate( source_type="scheduled_task", source_id=task.id, title=title, task_type=task_type, 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=(task.remaining_minutes if task.execution_segments else int(task.estimated_duration or 0)), 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 intervention.effective_estimated_duration or 0), 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, )) admin_query = AdminTask.query.filter_by(is_active=True) if user_id is not None: admin_query = admin_query.filter(db.or_(AdminTask.user_id.is_(None), AdminTask.user_id == user_id)) for admin in admin_query.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 0), fixed_start=admin.start_time, fixed_end=(datetime.combine(day, admin.start_time) + timedelta(minutes=int(admin.duration_minutes or 0))).time() if admin.start_time and admin.duration_minutes else None, status="à planifier", )) # Les passages d'entreprise sont une source interne facultative. Aucun # appel Pronote/ENT n'est effectué ici : seul ContractVisit est lu. visits = ContractVisit.query.filter( ContractVisit.scheduled_date == day, ContractVisit.status.notin_(("realisee", "annulee")), ).all() for visit in visits: if user_id is not None and visit.technician_id not in (None, user_id): continue duration = visit.accompaniment_minutes or 0 fixed_start = visit.start_time if visit.is_fixed_for_planning else None fixed_end = visit.end_time if visit.is_fixed_for_planning else None if visit.accompaniment_mode == "full" and fixed_start and fixed_end: duration = max(0, (_minutes(fixed_end) - _minutes(fixed_start))) company_name = visit.contract.company.name if visit.contract and visit.contract.company else "Entreprise extérieure" candidates.append(PlanningCandidate( source_type="contract_visit", source_id=visit.id, title=f"Passage {company_name}", task_type="external_company", priority=2, constraint="fixed" if fixed_start and fixed_end else "flexible", target_date=day, duration_minutes=duration, assigned_to=visit.technician_id, status=visit.status, fixed_start=fixed_start, fixed_end=fixed_end, explanation=visit.display_mode, )) return cls.propose(day, candidates, user_id=user_id)