diff --git a/app_new/core/services/day_planner.py b/app_new/core/services/day_planner.py
index 4b7ebd6..acf0ab3 100644
--- a/app_new/core/services/day_planner.py
+++ b/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(
- candidate.room_id,
- datetime.combine(day, start),
- datetime.combine(day, end),
- ):
+ 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
@@ -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"),
- task_type="preventive", priority=2 if task.lot_task_id else 4,
+ 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=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,
diff --git a/app_new/planning/schedules.py b/app_new/planning/schedules.py
index 8994eb3..3cdfee4 100644
--- a/app_new/planning/schedules.py
+++ b/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():
diff --git a/app_new/templates/planning/my_day.html b/app_new/templates/planning/my_day.html
index 7c4d9c5..170bcc7 100644
--- a/app_new/templates/planning/my_day.html
+++ b/app_new/templates/planning/my_day.html
@@ -23,6 +23,12 @@
{% endif %}
{% if candidates %}
+
+ {{ candidates|length }} tâche(s)
+ {{ candidates|selectattr('status', 'equalto', 'à replanifier')|list|length }} à replanifier
+ {{ candidates|selectattr('task_type', 'equalto', 'emergency')|list|length }} urgence(s)
+ {{ candidates|selectattr('status', 'equalto', 'conflit')|list|length }} conflit(s)
+
{% for item in candidates %}
@@ -45,6 +51,23 @@
{{ item.explanation }}
{% endif %}
+ {% set item_alternatives = alternatives.get((item.source_type, item.source_id), []) %}
+ {% if item_alternatives and item.status != 'conflit' %}
+
+ Replanifier cette tâche
+
+
+ {% endif %}
{% endfor %}
diff --git a/tests/integration/test_phase16_navigation.py b/tests/integration/test_phase16_navigation.py
index 8be346b..b79f34f 100644
--- a/tests/integration/test_phase16_navigation.py
+++ b/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
diff --git a/tests/unit/test_day_planner.py b/tests/unit/test_day_planner.py
index 5c27bd7..eff180e 100644
--- a/tests/unit/test_day_planner.py
+++ b/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