college-infos/pronote_sync/sync/planner.py

"""Planification de la synchronisation CalDAV.

Ce module compare les données Pronote normalisées aux événements distants
marqués comme gérés par ``pronote-sync`` et produit un plan de synchronisation
CalDAV (ajouts, mises à jour, suppressions) pour chaque catégorie d'événement :
cours, devoirs et événements scolaires.

Le plan est calculé de manière pure et déterministe : deux entrées identiques
produisent un plan identique, et un événement dont la signature sémantique
n'a pas changé n'apparaît dans aucune liste du plan (idempotence).
"""

from __future__ import annotations

from typing import Any

from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.sync import CalDAVSyncPlan
from pronote_sync.sync.serialization import (
    component_to_signature,
    homework_to_vevent,
    lesson_to_vevent,
    school_event_to_vevent,
)


def compute_plan(
    pronote_data: PronoteData,
    remote_managed: list[tuple[str, str, Any]],
) -> tuple[CalDAVSyncPlan, dict[str, str]]:
    """Calcule le plan de synchronisation CalDAV et le mapping des UID distants.

    L'appariement entre les événements locaux et distants se fait sur l'UID
    canonique (forme normalisée, identique pour une même source Pronote,
    suffixe temporel retiré) tandis que les mutations (suppressions, mises à
    jour) ciblent l'UID brut tel que stocké sur le serveur. Le mapping
    ``canonical_uid -> raw_uid`` retourné permet à l'exécuteur de cibler le
    bon objet distant lors des mises à jour.

    :param pronote_data: Données Pronote normalisées (cours, devoirs, événements).
    :param remote_managed: Liste de tuples (raw_uid, canonical_uid, vevent)
        pour les événements distants marqués comme gérés par pronote-sync.
    :return: Tuple (plan de synchronisation, mapping canonical_uid -> raw_uid).
        Les listes ``*_to_remove`` contiennent l'UID brut distant, les autres
        listes contiennent les modèles Pronote locaux.
    :rtype: tuple[CalDAVSyncPlan, dict[str, str]]
    """
    #: canonical_uid -> signature sémantique du VEVENT distant (pour l'appariement).
    remote_signatures_by_canonical: dict[str, str] = {}
    #: canonical_uid -> UID brut distant (pour cibler le bon objet lors des mutations).
    remote_raw_by_canonical: dict[str, str] = {}
    for raw_uid, canonical_uid, vevent in remote_managed:
        remote_signatures_by_canonical[canonical_uid] = component_to_signature(vevent)
        remote_raw_by_canonical[canonical_uid] = raw_uid

    lessons_to_add: list[Lesson] = []
    lessons_to_update: list[Lesson] = []
    lessons_to_remove: list[str] = []

    local_lessons_by_uid: dict[str, Lesson] = {lesson.id: lesson for lesson in pronote_data.lessons}
    for lesson in pronote_data.lessons:
        local_canonical = lesson.id
        local_sig = component_to_signature(lesson_to_vevent(lesson))
        if local_canonical not in remote_signatures_by_canonical:
            lessons_to_add.append(lesson)
        elif remote_signatures_by_canonical[local_canonical] != local_sig:
            lessons_to_update.append(lesson)

    homeworks_to_add: list[Homework] = []
    homeworks_to_update: list[Homework] = []
    homeworks_to_remove: list[str] = []

    local_homeworks_by_uid: dict[str, Homework] = {
        f"homework-{homework.id}": homework for homework in pronote_data.homeworks
    }
    for homework in pronote_data.homeworks:
        local_canonical = f"homework-{homework.id}"
        local_sig = component_to_signature(homework_to_vevent(homework))
        if local_canonical not in remote_signatures_by_canonical:
            homeworks_to_add.append(homework)
        elif remote_signatures_by_canonical[local_canonical] != local_sig:
            homeworks_to_update.append(homework)

    school_events_to_add: list[SchoolEvent] = []
    school_events_to_update: list[SchoolEvent] = []
    school_events_to_remove: list[str] = []

    local_school_events_by_uid: dict[str, SchoolEvent] = {
        f"school-event-{event.label}-{event.from_date.isoformat()}": event
        for event in pronote_data.school_events
    }
    for school_event in pronote_data.school_events:
        local_canonical = f"school-event-{school_event.label}-{school_event.from_date.isoformat()}"
        local_sig = component_to_signature(school_event_to_vevent(school_event))
        if local_canonical not in remote_signatures_by_canonical:
            school_events_to_add.append(school_event)
        elif remote_signatures_by_canonical[local_canonical] != local_sig:
            school_events_to_update.append(school_event)

    # Détection des événements distants orphelins : un UID canonique distant
    # absent des données locales est supprimé en ciblant l'UID brut stocké sur
    # le serveur. L'acheminement vers la bonne liste de suppression se fait sur
    # le préfixe de l'UID canonique.
    for canonical_uid in remote_signatures_by_canonical:
        raw_uid = remote_raw_by_canonical[canonical_uid]
        if canonical_uid.startswith("homework-"):
            if canonical_uid not in local_homeworks_by_uid:
                homeworks_to_remove.append(raw_uid)
        elif canonical_uid.startswith("school-event-"):
            if canonical_uid not in local_school_events_by_uid:
                school_events_to_remove.append(raw_uid)
        elif canonical_uid not in local_lessons_by_uid:
            lessons_to_remove.append(raw_uid)

    return (
        CalDAVSyncPlan(
            lessons_to_add=lessons_to_add,
            lessons_to_update=lessons_to_update,
            lessons_to_remove=lessons_to_remove,
            homeworks_to_add=homeworks_to_add,
            homeworks_to_update=homeworks_to_update,
            homeworks_to_remove=homeworks_to_remove,
            school_events_to_add=school_events_to_add,
            school_events_to_update=school_events_to_update,
            school_events_to_remove=school_events_to_remove,
        ),
        remote_raw_by_canonical,
    )