diff --git a/app_new/core/services/day_planner.py b/app_new/core/services/day_planner.py new file mode 100644 index 0000000..4b7ebd6 --- /dev/null +++ b/app_new/core/services/day_planner.py @@ -0,0 +1,260 @@ +"""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) diff --git a/app_new/planning/schedules.py b/app_new/planning/schedules.py index 07b4da8..8994eb3 100644 --- a/app_new/planning/schedules.py +++ b/app_new/planning/schedules.py @@ -17,6 +17,32 @@ from datetime import datetime, date, time as dt_time planning_bp = Blueprint('planning', __name__, url_prefix='/planning', template_folder='templates') +@planning_bp.route('/my-day') +@login_required +def my_day(): + """Vue quotidienne : une proposition, jamais une écriture automatique.""" + from datetime import timedelta + from ..core.services.day_planner import DayPlanner + + raw_date = request.args.get('date') + try: + target_date = date.fromisoformat(raw_date) if raw_date else date.today() + except ValueError: + flash('La date demandée est invalide.', 'warning') + target_date = date.today() + candidates = DayPlanner.from_database(target_date, user_id=current_user.id) + windows = DayPlanner.work_windows(target_date, user_id=current_user.id) + return render_template( + 'planning/my_day.html', + target_date=target_date, + candidates=candidates, + work_windows=windows, + previous_date=target_date - timedelta(days=1), + next_date=target_date + timedelta(days=1), + hours_configured=bool(windows), + ) + + @planning_bp.route('/rooms') @login_required def room_schedules(): diff --git a/app_new/templates/_navigation_phase16.html b/app_new/templates/_navigation_phase16.html index 2aaf467..4ae5310 100644 --- a/app_new/templates/_navigation_phase16.html +++ b/app_new/templates/_navigation_phase16.html @@ -11,6 +11,7 @@