191 lines
8.2 KiB
Python
191 lines
8.2 KiB
Python
"""Résolution des occupations internes d'une salle.
|
|
|
|
Pronote est traité ici comme une provenance optionnelle. Les autres services
|
|
ne doivent appeler que ``active_room_schedules`` ou ``room_is_available``.
|
|
"""
|
|
from datetime import date, datetime, time, timedelta
|
|
import hashlib
|
|
|
|
from ..models.college import RoomSchedule
|
|
from ...extensions import db
|
|
|
|
|
|
def _overlaps(left_start, left_end, right_start, right_end):
|
|
return left_start < right_end and left_end > right_start
|
|
|
|
|
|
def active_room_schedules(room_id, day=None):
|
|
query = RoomSchedule.query.filter_by(room_id=room_id, resolution_status="active")
|
|
rows = query.order_by(RoomSchedule.start_time).all()
|
|
if day is not None:
|
|
monday = day - timedelta(days=day.weekday())
|
|
rows = [row for row in rows if row.week_start == monday and
|
|
row.day_of_week == day.weekday() and
|
|
(not row.valid_from or row.valid_from <= day) and
|
|
(not row.valid_to or day <= row.valid_to)]
|
|
|
|
pronote = [row for row in rows if row.source == "pronote"]
|
|
resolved = []
|
|
for row in rows:
|
|
if row.source == "manual" and not row.protected_from_sync:
|
|
replacement = any(
|
|
_overlaps(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 replacement:
|
|
continue
|
|
resolved.append(row)
|
|
return resolved
|
|
|
|
|
|
def room_occupancy(room_id, day):
|
|
return [(row.start_time, row.end_time, row) for row in active_room_schedules(room_id, day)]
|
|
|
|
|
|
def room_is_available(room_id, start, end):
|
|
"""Vérifie une salle à partir du planning interne résolu."""
|
|
if not isinstance(start, datetime) or not isinstance(end, datetime):
|
|
raise TypeError("La disponibilité d'une salle nécessite deux datetime")
|
|
if end <= start:
|
|
return False
|
|
day = start.date()
|
|
if end.date() != day:
|
|
return False
|
|
start_time, end_time = start.time(), end.time()
|
|
return not any(
|
|
_overlaps(start_time, end_time, occupied_start, occupied_end)
|
|
for occupied_start, occupied_end, _ in room_occupancy(room_id, day)
|
|
)
|
|
|
|
|
|
def available_slots(room_id, day, window_start=time(8), window_end=time(17), duration_minutes=30):
|
|
"""Retourne des intervalles libres déterministes dans une fenêtre."""
|
|
occupied = sorted(room_occupancy(room_id, day), key=lambda item: item[0])
|
|
cursor = window_start
|
|
slots = []
|
|
def minutes(value):
|
|
return value.hour * 60 + value.minute
|
|
def as_time(total):
|
|
return time(total // 60, total % 60)
|
|
for start, end, _ in occupied:
|
|
if minutes(start) > minutes(cursor) and minutes(start) - minutes(cursor) >= duration_minutes:
|
|
slots.append((cursor, start))
|
|
if minutes(end) > minutes(cursor):
|
|
cursor = end
|
|
if minutes(window_end) - minutes(cursor) >= duration_minutes:
|
|
slots.append((cursor, window_end))
|
|
return slots
|
|
|
|
|
|
def sync_pronote_schedules(room_id, entries, *, synced_at=None, week_start=None):
|
|
"""Applique un lot Pronote sans toucher aux créneaux manuels protégés.
|
|
|
|
``entries`` est une liste de dictionnaires normalisés par l'intégration
|
|
Pronote (external_id, week_start, day_of_week, start_time, end_time et
|
|
libellés optionnels). Cette fonction ne contacte jamais Pronote elle-même.
|
|
"""
|
|
synced_at = synced_at or datetime.utcnow()
|
|
synced_weeks = {entry.get("week_start") for entry in entries if entry.get("week_start")}
|
|
if week_start:
|
|
synced_weeks.add(week_start)
|
|
seen = set()
|
|
for entry in entries:
|
|
external_id = str(entry.get("external_id") or "").strip()
|
|
if not external_id:
|
|
continue
|
|
seen.add(external_id)
|
|
row = RoomSchedule.query.filter_by(
|
|
room_id=room_id, source="pronote", external_id=external_id
|
|
).first()
|
|
if not row:
|
|
row = RoomSchedule(room_id=room_id, source="pronote", external_id=external_id)
|
|
db.session.add(row)
|
|
for field in ("week_start", "day_of_week", "start_time", "end_time", "subject", "teacher", "class_name", "course_name", "event_type"):
|
|
if field in entry and entry[field] is not None:
|
|
setattr(row, field, entry[field])
|
|
row.last_synced_at = synced_at
|
|
row.source_updated_at = entry.get("source_updated_at")
|
|
row.resolution_status = "active"
|
|
row.conflict_note = None
|
|
manual_rows = RoomSchedule.query.filter(
|
|
RoomSchedule.room_id == room_id,
|
|
RoomSchedule.source == "manual",
|
|
RoomSchedule.resolution_status == "active",
|
|
RoomSchedule.day_of_week == row.day_of_week,
|
|
RoomSchedule.week_start == row.week_start,
|
|
RoomSchedule.protected_from_sync.is_(False),
|
|
).all()
|
|
protected_rows = RoomSchedule.query.filter(
|
|
RoomSchedule.room_id == room_id,
|
|
RoomSchedule.source == "manual",
|
|
RoomSchedule.resolution_status == "active",
|
|
RoomSchedule.day_of_week == row.day_of_week,
|
|
RoomSchedule.week_start == row.week_start,
|
|
RoomSchedule.protected_from_sync.is_(True),
|
|
).all()
|
|
candidates = []
|
|
for manual in manual_rows:
|
|
if (manual.subject and row.subject and manual.subject != row.subject) or (manual.class_name and row.class_name and manual.class_name != row.class_name):
|
|
continue
|
|
if _overlaps(manual.start_time, manual.end_time, row.start_time, row.end_time):
|
|
candidates.append(manual)
|
|
if len(candidates) == 1:
|
|
candidates[0].resolution_status = "superseded"
|
|
elif len(candidates) > 1:
|
|
note = "Correspondance Pronote ambiguë : validation nécessaire"
|
|
row.conflict_note = note
|
|
for manual in candidates:
|
|
manual.conflict_note = note
|
|
for manual in protected_rows:
|
|
if _overlaps(manual.start_time, manual.end_time, row.start_time, row.end_time):
|
|
note = "Conflit avec un créneau manuel protégé"
|
|
manual.conflict_note = note
|
|
row.conflict_note = note
|
|
existing_query = RoomSchedule.query.filter_by(room_id=room_id, source="pronote")
|
|
if synced_weeks:
|
|
existing_query = existing_query.filter(RoomSchedule.week_start.in_(synced_weeks))
|
|
existing = existing_query.all()
|
|
for row in existing:
|
|
if row.external_id not in seen:
|
|
row.resolution_status = "disabled"
|
|
db.session.commit()
|
|
|
|
|
|
def normalize_pronote_lessons(lessons, week_start):
|
|
"""Transforme la sortie du client Pronote en entrées stables.
|
|
|
|
Le client historique ne fournit pas toujours d'identifiant de cours. Dans
|
|
ce cas, un identifiant déterministe est calculé à partir des attributs
|
|
visibles, ce qui permet de remettre à jour la même ligne sans doublon.
|
|
"""
|
|
entries = []
|
|
for lesson in lessons or []:
|
|
start_raw, end_raw = lesson.get("start_time"), lesson.get("end_time")
|
|
try:
|
|
start_time = time.fromisoformat(str(start_raw))
|
|
end_time = time.fromisoformat(str(end_raw))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
day_of_week = int(lesson.get("day_of_week", 0))
|
|
subject = (lesson.get("subject") or "").strip() or None
|
|
teacher = (lesson.get("teacher") or "").strip() or None
|
|
class_name = (lesson.get("class_name") or "").strip() or None
|
|
stable = lesson.get("external_id") or "|".join(str(value or "") for value in (
|
|
week_start, day_of_week, start_time, end_time, subject, teacher, class_name
|
|
))
|
|
external_id = hashlib.sha256(str(stable).encode("utf-8")).hexdigest()[:64]
|
|
entries.append({
|
|
"external_id": external_id,
|
|
"week_start": week_start,
|
|
"day_of_week": day_of_week,
|
|
"start_time": start_time,
|
|
"end_time": end_time,
|
|
"subject": subject,
|
|
"teacher": teacher,
|
|
"class_name": class_name,
|
|
"course_name": subject or class_name or "Cours",
|
|
"event_type": "cours",
|
|
})
|
|
return entries
|