feat(planning): add minimal daily rescheduling

app_new/core/services/day_planner.py

@@ -10,7 +10,7 @@ from datetime import date, datetime, time, timedelta
 from typing import Iterable, Optional
 
 from .planning_service import PlanningService
-from .room_planning import room_is_available
+from .room_planning import room_is_available, room_occupancy
 
 
 TYPE_LABELS = {
@@ -85,6 +85,44 @@ 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."""
 
@@ -116,14 +154,13 @@ class DayPlanner:
         return windows
 
     @staticmethod
-    def _slot_available(day, candidate, start, end, occupied):
+    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 not room_is_available(
+        if candidate.room_id and room_occupied is not None:
-            candidate.room_id,
+            if any(_overlap(start, end, item_start, item_end) for item_start, item_end, _ in room_occupied.get(candidate.room_id, [])):
-            datetime.combine(day, start),
+                return False
-            datetime.combine(day, end),
+        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
@@ -143,11 +180,15 @@ class DayPlanner:
         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
@@ -172,7 +213,7 @@ class DayPlanner:
                 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):
+                    if cls._slot_available(day, candidate, start, end, occupied, room_occupied):
                         chosen = (start, end)
                         break
                     cursor += 5
@@ -195,6 +236,29 @@ class DayPlanner:
                 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 = max(int(target.duration_minutes or 1), 1)
+        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."""
@@ -211,10 +275,14 @@ class DayPlanner:
             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=(task.preventive_task.name if task.preventive_task else task.lot_task.tache if task.lot_task else "Maintenance préventive"),
+                title=title,
-                task_type="preventive", priority=2 if task.lot_task_id else 4,
+                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=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,

app_new/planning/schedules.py

@@ -32,6 +32,13 @@ def my_day():
         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)
+    alternatives = {
+        (item.source_type, item.source_id): DayPlanner.alternative_slots(
+            target_date, item, candidates, user_id=current_user.id
+        )
+        for item in candidates
+        if item.status in ('à replanifier', 'proposé', 'conflit')
+    }
     return render_template(
         'planning/my_day.html',
         target_date=target_date,
@@ -40,9 +47,74 @@ def my_day():
         previous_date=target_date - timedelta(days=1),
         next_date=target_date + timedelta(days=1),
         hours_configured=bool(windows),
+        alternatives=alternatives,
     )
 
 
+@planning_bp.route('/my-day/reschedule', methods=['POST'])
+@login_required
+def my_day_reschedule():
+    """Valide un seul créneau choisi par l'utilisateur."""
+    from ..core.services.day_planner import DayPlanner
+    from ..core.models.planning import ScheduledTask
+    source_type = request.form.get('source_type', '')
+    source_id = request.form.get('source_id', type=int)
+    raw_date = request.form.get('date', '')
+    raw_slot = request.form.get('slot', '')
+    raw_start, raw_end = (raw_slot.split('|', 1) if '|' in raw_slot else ('', ''))
+    try:
+        target_date = date.fromisoformat(raw_date)
+        start_time = dt_time.fromisoformat(raw_start)
+        end_time = dt_time.fromisoformat(raw_end)
+    except (TypeError, ValueError):
+        flash('Le créneau choisi est invalide.', 'danger')
+        return redirect(url_for('planning.my_day'))
+    if not source_id or end_time <= start_time:
+        flash('Le créneau choisi est invalide.', 'danger')
+        return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+    candidates = DayPlanner.from_database(target_date, user_id=current_user.id)
+    candidate = next((item for item in candidates if item.source_type == source_type and item.source_id == source_id), None)
+    if not candidate or candidate.status == 'terminée':
+        flash('Cette tâche n’est plus replanifiable.', 'warning')
+        return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+    duration = int(candidate.duration_minutes or 1)
+    if (end_time.hour * 60 + end_time.minute) - (start_time.hour * 60 + start_time.minute) != duration:
+        flash('La durée choisie ne correspond pas à la durée de la tâche.', 'danger')
+        return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+    windows = DayPlanner.work_windows(target_date, user_id=current_user.id)
+    if not any(window.start <= start_time and end_time <= window.end for window in windows):
+        flash('Ce créneau est en dehors des horaires de travail ou pendant la pause.', 'danger')
+        return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+    others = [item for item in candidates if item is not candidate and item.proposed_start and item.proposed_end]
+    if any(item.proposed_start < end_time and item.proposed_end > start_time for item in others):
+        flash('Ce créneau est déjà pris par une autre tâche fixe ou proposée.', 'danger')
+        return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+    if candidate.room_id:
+        from ..core.services.room_planning import room_is_available
+        if not room_is_available(candidate.room_id, datetime.combine(target_date, start_time), datetime.combine(target_date, end_time)):
+            flash('La salle est occupée sur ce créneau.', 'danger')
+            return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+    if source_type == 'scheduled_task':
+        task = ScheduledTask.query.get(source_id)
+        if not task or (task.assigned_to_id not in (None, current_user.id)):
+            flash('Cette tâche n’est pas accessible.', 'danger')
+            return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+        task.scheduled_date, task.scheduled_start, task.scheduled_end = target_date, start_time, end_time
+        task.status = 'planned'
+    elif source_type == 'intervention':
+        task = Intervention.query.get(source_id)
+        if not task or (task.technician_id not in (None, current_user.id) and task.assigned_to_id not in (None, current_user.id)):
+            flash('Cette intervention n’est pas accessible.', 'danger')
+            return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+        task.scheduled_date, task.scheduled_start, task.scheduled_end = target_date, start_time, end_time
+    else:
+        flash('Cette catégorie ne peut pas encore être replanifiée depuis Ma journée.', 'info')
+        return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+    db.session.commit()
+    flash('Le nouveau créneau a été enregistré.', 'success')
+    return redirect(url_for('planning.my_day', date=target_date.isoformat()))
+
+
 @planning_bp.route('/rooms')
 @login_required
 def room_schedules():

app_new/templates/planning/my_day.html

@@ -23,6 +23,12 @@
   {% endif %}
 
   {% if candidates %}
+  <div class="d-flex flex-wrap gap-2 mb-3" aria-label="Résumé de la journée">
+    <span class="badge text-bg-primary">{{ candidates|length }} tâche(s)</span>
+    <span class="badge text-bg-warning">{{ candidates|selectattr('status', 'equalto', 'à replanifier')|list|length }} à replanifier</span>
+    <span class="badge text-bg-danger">{{ candidates|selectattr('task_type', 'equalto', 'emergency')|list|length }} urgence(s)</span>
+    <span class="badge text-bg-danger">{{ candidates|selectattr('status', 'equalto', 'conflit')|list|length }} conflit(s)</span>
+  </div>
   <div class="timeline d-grid gap-3" aria-label="Tâches proposées">
     {% for item in candidates %}
     <article class="card shadow-sm border-start border-{{ 'danger' if item.task_type == 'emergency' else 'primary' }} border-4">
@@ -45,6 +51,23 @@
           </div>
         </div>
         {% if item.explanation %}<p class="mb-0 mt-2 small"><i class="bi bi-info-circle"></i> {{ item.explanation }}</p>{% endif %}
+        {% set item_alternatives = alternatives.get((item.source_type, item.source_id), []) %}
+        {% if item_alternatives and item.status != 'conflit' %}
+        <details class="mt-2">
+          <summary class="small text-primary" style="cursor:pointer">Replanifier cette tâche</summary>
+          <form method="post" action="{{ url_for('planning.my_day_reschedule') }}" class="row g-2 mt-1 align-items-end">
+            <input type="hidden" name="source_type" value="{{ item.source_type }}">
+            <input type="hidden" name="source_id" value="{{ item.source_id }}">
+            <input type="hidden" name="date" value="{{ target_date.isoformat() }}">
+            <div class="col-8 col-sm-5"><label class="form-label small" for="slot-{{ item.source_type }}-{{ item.source_id }}">Créneau compatible</label>
+              <select class="form-select form-select-sm" id="slot-{{ item.source_type }}-{{ item.source_id }}" name="slot" required>
+                {% for start, end in item_alternatives %}<option value="{{ start.strftime('%H:%M') }}|{{ end.strftime('%H:%M') }}">{{ start.strftime('%H:%M') }} – {{ end.strftime('%H:%M') }}</option>{% endfor %}
+              </select>
+            </div>
+            <div class="col-auto"><button class="btn btn-sm btn-outline-primary" type="submit">Enregistrer</button></div>
+          </form>
+        </details>
+        {% endif %}
       </div>
     </article>
     {% endfor %}

tests/integration/test_phase16_navigation.py

@@ -95,3 +95,11 @@ def test_new_menu_destinations_do_not_return_accidental_404_or_500(authenticated
     for path in paths:
         response = authenticated_client.get(path)
         assert response.status_code not in (404, 500), path
+
+
+def test_my_day_replanning_rejects_invalid_slot_without_write(authenticated_client):
+    response = authenticated_client.post(
+        "/planning/my-day/reschedule",
+        data={"source_type": "scheduled_task", "source_id": "999999", "date": "2026-08-24", "slot": "bad"},
+    )
+    assert response.status_code == 302

tests/unit/test_day_planner.py

@@ -47,3 +47,98 @@ def test_no_work_window_explains_missing_schedule(monkeypatch):
     monkeypatch.setattr(DayPlanner, "work_windows", staticmethod(lambda day, user_id=None: []))
     result = DayPlanner.propose(date(2026, 8, 24), [PlanningCandidate("admin", 1, "Tâche")])
     assert "horaire de travail" in result[0].explanation
+
+
+def _full_day(monkeypatch):
+    monkeypatch.setattr(DayPlanner, "work_windows", staticmethod(lambda day, user_id=None: [WorkWindow(time(8), time(12)), WorkWindow(time(13, 30), time(17))]))
+
+
+def test_urgent_candidate_is_ordered_before_flexible(monkeypatch):
+    _full_day(monkeypatch)
+    rows = DayPlanner.propose(date(2026, 8, 24), [
+        PlanningCandidate("admin", 1, "Flexible", priority=4, duration_minutes=20),
+        PlanningCandidate("intervention", 2, "Fuite", task_type="emergency", constraint="urgent", priority=1, duration_minutes=20),
+    ])
+    assert rows[0].source_id == 2
+    assert "créneau" in rows[0].explanation or rows[0].status == "proposé"
+
+
+def test_deadline_candidate_does_not_finish_after_latest_end(monkeypatch):
+    _full_day(monkeypatch)
+    candidate = PlanningCandidate("admin", 1, "Commande", constraint="deadline", duration_minutes=60, latest_end=time(10))
+    rows = DayPlanner.propose(date(2026, 8, 24), [candidate])
+    assert rows[0].proposed_end <= time(10)
+
+
+def test_deadline_candidate_is_marked_when_no_slot_before_deadline(monkeypatch):
+    _full_day(monkeypatch)
+    candidate = PlanningCandidate("admin", 1, "Commande", constraint="deadline", duration_minutes=30, earliest_start=time(11), latest_end=time(11, 15))
+    rows = DayPlanner.propose(date(2026, 8, 24), [candidate])
+    assert rows[0].status == "à replanifier"
+
+
+def test_fixed_overlap_is_visible_as_conflict(monkeypatch):
+    _full_day(monkeypatch)
+    rows = DayPlanner.propose(date(2026, 8, 24), [
+        PlanningCandidate("company", 1, "Entreprise", task_type="external_company", constraint="fixed", fixed_start=time(9), fixed_end=time(10)),
+        PlanningCandidate("intervention", 2, "Urgence", task_type="emergency", constraint="fixed", fixed_start=time(9, 30), fixed_end=time(10, 15)),
+    ])
+    assert rows[1].status == "conflit"
+    assert "chevauche" in rows[1].explanation
+
+
+def test_geographic_labels_are_kept_for_explanation(monkeypatch):
+    _full_day(monkeypatch)
+    rows = DayPlanner.propose(date(2026, 8, 24), [
+        PlanningCandidate("scheduled_task", 1, "Salle", building_name="A", zone_name="RDC", room_name="101"),
+        PlanningCandidate("scheduled_task", 2, "Zone", building_name="A", zone_name="RDC", room_name="102"),
+    ])
+    assert rows[0].location_label == "A > RDC > 101"
+    assert rows[1].location_label == "A > RDC > 102"
+
+
+def test_unknown_location_is_explicit(monkeypatch):
+    _full_day(monkeypatch)
+    candidate = PlanningCandidate("admin", 1, "Dossier")
+    assert DayPlanner.propose(date(2026, 8, 24), [candidate])[0].location_label == "Localisation non renseignée"
+
+
+def test_room_occupation_rejects_slot(monkeypatch):
+    _full_day(monkeypatch)
+    monkeypatch.setattr("app_new.core.services.day_planner._resolved_room_occupancy_batch", lambda day, room_ids: {12: [(time(8), time(17), object())]})
+    candidate = PlanningCandidate("scheduled_task", 1, "Salle occupée", room_id=12, duration_minutes=30)
+    rows = DayPlanner.propose(date(2026, 8, 24), [candidate])
+    assert rows[0].status == "à replanifier"
+
+
+def test_room_without_id_can_use_work_window(monkeypatch):
+    _full_day(monkeypatch)
+    candidate = PlanningCandidate("scheduled_task", 1, "Sans salle", duration_minutes=30)
+    rows = DayPlanner.propose(date(2026, 8, 24), [candidate])
+    assert rows[0].status == "proposé"
+
+
+def test_alternative_slots_exclude_fixed_event(monkeypatch):
+    _full_day(monkeypatch)
+    fixed = PlanningCandidate("company", 1, "Fixe", constraint="fixed", fixed_start=time(8), fixed_end=time(9))
+    target = PlanningCandidate("admin", 2, "À replacer", duration_minutes=30)
+    proposed = DayPlanner.propose(date(2026, 8, 24), [fixed, target])
+    slots = DayPlanner.alternative_slots(date(2026, 8, 24), target, proposed)
+    assert slots and slots[0][0] >= time(9)
+
+
+def test_alternative_slots_are_limited_to_three(monkeypatch):
+    _full_day(monkeypatch)
+    target = PlanningCandidate("admin", 2, "À replacer", duration_minutes=15)
+    slots = DayPlanner.alternative_slots(date(2026, 8, 24), target, [target], limit=3)
+    assert len(slots) == 3
+
+
+def test_explicit_user_parameter_is_forwarded(monkeypatch):
+    seen = {}
+    def windows(day, user_id=None):
+        seen["user_id"] = user_id
+        return [WorkWindow(time(8), time(9))]
+    monkeypatch.setattr(DayPlanner, "work_windows", staticmethod(windows))
+    DayPlanner.propose(date(2026, 8, 24), [PlanningCandidate("admin", 1, "Tâche")], user_id=42)
+    assert seen["user_id"] == 42