diff --git a/.env.example b/.env.example index 981af19..12a63c5 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,7 @@ # --- Pronote --- PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024 +PRONOTE_URL=https://college.ent/pronote/eleve.html +PRONOTE_ACCOUNT_TYPE=parent PRONOTE_USERNAME=parent.dupont PRONOTE_PASSWORD=your_secure_password PRONOTE_ENT=monbureaunumerique diff --git a/pronote_sync/config/settings.py b/pronote_sync/config/settings.py index f4e865f..898885d 100644 --- a/pronote_sync/config/settings.py +++ b/pronote_sync/config/settings.py @@ -27,6 +27,8 @@ class PronoteSettings(BaseSettings): username: str | None = None password: SecretStr | None = None ent: str | None = None + pronote_url: str | None = None + account_type: Literal["student", "parent"] = "parent" agenda_source: Literal["auto", "ical", "pronotepy"] = "auto" homework_source: Literal["auto", "ical", "pronotepy"] = "auto" messages_source: Literal["pronotepy"] = "pronotepy" diff --git a/pronote_sync/sources/pronote/client.py b/pronote_sync/sources/pronote/client.py index 784c9d5..547d4e9 100644 --- a/pronote_sync/sources/pronote/client.py +++ b/pronote_sync/sources/pronote/client.py @@ -3,16 +3,20 @@ Ce module fournit l'encapsulation du client ``pronotepy`` pour la source Pronote : récupération des messages des professeurs, des informations et sondages, ainsi que des cours et devoirs en mode repli lorsque le flux -iCal échoue. Toutes les erreurs sont journalisées avec des secrets masqués. +iCal échoue. Les erreurs des méthodes dégradées (messages, informations) +sont journalisées avec des secrets masqués ; les erreurs de récupération +des cours et des devoirs se propagent pour déclencher le repli iCal. """ from __future__ import annotations import logging from datetime import date -from typing import Protocol +from typing import Any, Protocol import pronotepy +import pronotepy.ent as pronotepy_ent +import requests from pronote_sync.config.settings import PronoteSettings from pronote_sync.models.agenda import Lesson, LessonStatus @@ -23,6 +27,69 @@ from pronote_sync.utils.redaction import redact_exception logger = logging.getLogger(__name__) +def _get_ent_callable(name: str) -> Any: + """Retourne le callable ``pronotepy`` associé à un nom d'ENT. + + L'accès par :func:`getattr` évite les erreurs ``attr-defined`` de mypy + sur les attributs non exportés explicitement par ``pronotepy.ent``. + + :param name: Nom de l'attribut dans ``pronotepy.ent``. + :return: Callable ``pronotepy`` associé. + :rtype: Any + """ + return getattr(pronotepy_ent, name) + + +_ENT_NAMES: list[str] = [ + "monbureaunumerique", + "ent_elyco", + "bordeaux", + "ent_creuse", + "occitanie_montpellier", + "paris_classe_numerique", + "ile_de_france", + "ent_hdf", + "ac_orleans_tours", + "ac_poitiers", + "ac_rennes", + "laclasse_educonnect", + "ent77", + "ent_ecollege78", + "ent_essonne", + "val_doise", + "val_de_marne", + "ent_var", + "atrium_sud", + "laclasse_lyon", + "eclat_bfc", + "cas_arsene76", + "cas_ent27", + "cas_kosmos", + "ent_creuse_educonnect", + "ent_mayotte", + "ent_somme", + "ent_94", + "extranet_colleges_somme", + "ac_reunion", +] + +_ENT_RESOLVERS: dict[str, Any] = {name: _get_ent_callable(name) for name in _ENT_NAMES} + + +def _resolve_ent(ent_name: str) -> Any: + """Résout un nom d'ENT en callable ``pronotepy``. + + :param ent_name: Nom de l'ENT tel que configuré (ex. ``"bordeaux"``). + :return: Callable ``pronotepy`` associé à l'ENT. + :raises ValueError: Si le nom d'ENT n'est pas reconnu. + """ + resolver = _ENT_RESOLVERS.get(ent_name) + if resolver is None: + supported = ", ".join(sorted(_ENT_RESOLVERS.keys())) + raise ValueError(f"ENT inconnu : {ent_name!r}. ENT supportés : {supported}") + return resolver + + class PronoteClientProtocol(Protocol): """Interface du client Pronote consommée par la logique de repli.""" @@ -42,13 +109,23 @@ class PronoteClientProtocol(Protocol): """ ... - def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]: - """Récupère les cours et les devoirs via ``pronotepy`` (repli iCal). + def get_lessons(self, start: date, end: date) -> list[Lesson]: + """Récupère les cours via ``pronotepy`` (repli iCal). :param start: Date de début de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse). - :return: Tuple ``(cours, devoirs)``. - :rtype: tuple[list[Lesson], list[Homework]] + :return: Liste des cours. + :rtype: list[Lesson] + """ + ... + + def get_homeworks(self, start: date, end: date) -> list[Homework]: + """Récupère les devoirs via ``pronotepy``. + + :param start: Date de début de la fenêtre (incluse). + :param end: Date de fin de la fenêtre (incluse). + :return: Liste des devoirs. + :rtype: list[Homework] """ ... @@ -56,11 +133,13 @@ class PronoteClientProtocol(Protocol): class PronoteClient: """Client d'accès à Pronote via ``pronotepy``. - Encapsule ``pronotepy.Client`` avec une connexion paresseuse : la - connexion n'est établie qu'à la première méthode de récupération - appelée. Les erreurs ``pronotepy.PronoteAPIError`` sont journalisées - avec des secrets masqués et les méthodes de récupération retournent - alors une valeur vide au lieu de propager l'exception. + Encapsule ``pronotepy.Client`` ou ``pronotepy.ParentClient`` selon le + type de compte, avec une connexion paresseuse : la connexion n'est + établie qu'à la première méthode de récupération appelée. Les erreurs + des méthodes dégradées (``get_messages()``, ``get_informations()``) + sont journalisées avec des secrets masqués et retournent une valeur + vide ; ``get_lessons()`` et ``get_homeworks()`` laissent les + exceptions se propager pour déclencher le repli iCal. """ def __init__(self, settings: PronoteSettings) -> None: @@ -75,24 +154,39 @@ class PronoteClient: """Crée et connecte le client ``pronotepy`` (connexion paresseuse). Le client est créé une seule fois puis réutilisé pour les appels - suivants. L'erreur de connexion est relancée sans journalisation, - la méthode publique appelante étant responsable de la journaliser. + suivants. Le nom d'ENT est résolu via :func:`_resolve_ent` et le + type de compte (``student`` ou ``parent``) détermine la classe de + client utilisée. L'erreur de connexion est relancée sans + journalisation, la méthode publique appelante étant responsable + de la journaliser. :return: Le client ``pronotepy`` connecté. :rtype: pronotepy.Client - :raises ValueError: Si ``username``, ``password`` ou ``ent`` est manquant. + :raises ValueError: Si ``pronote_url``, ``username``, ``password`` + ou ``ent`` est manquant, ou si l'ENT est inconnu. :raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue. """ if self._client is None: + pronote_url = self._settings.pronote_url username = self._settings.username password = self._settings.password ent = self._settings.ent - if username is None or password is None or ent is None: - raise ValueError("username, password et ent sont requis pour pronotepy") - try: - self._client = pronotepy.Client(username, password.get_secret_value(), ent) - except pronotepy.PronoteAPIError: - raise + if pronote_url is None or username is None or password is None or ent is None: + raise ValueError( + "pronote_url, username, password et ent sont requis pour pronotepy" + ) + resolver = _resolve_ent(ent) + client_class: type[pronotepy.Client] = ( + pronotepy.ParentClient + if self._settings.account_type == "parent" + else pronotepy.Client + ) + self._client = client_class( + pronote_url=pronote_url, + username=username, + password=password.get_secret_value(), + ent=resolver, + ) return self._client def get_messages(self) -> list[Message]: @@ -122,7 +216,13 @@ class PronoteClient: ) ) return messages - except (pronotepy.PronoteAPIError, ValueError) as exc: + except ( + pronotepy.PronoteAPIError, + ValueError, + requests.RequestException, + ConnectionError, + TimeoutError, + ) as exc: logger.error( "Échec de la récupération des messages Pronote : %s", redact_exception(exc), @@ -154,62 +254,86 @@ class PronoteClient: ) ) return messages - except (pronotepy.PronoteAPIError, ValueError) as exc: + except ( + pronotepy.PronoteAPIError, + ValueError, + requests.RequestException, + ConnectionError, + TimeoutError, + ) as exc: logger.error( "Échec de la récupération des informations Pronote : %s", redact_exception(exc), ) return [] - def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]: - """Récupère les cours et les devoirs via ``pronotepy``. + def get_lessons(self, start: date, end: date) -> list[Lesson]: + """Récupère les cours via ``pronotepy`` (repli iCal). - À utiliser uniquement si les sources iCal sont indisponibles ou en - repli automatique. Les cours annulés sont mappés sur le statut - ``CANCELLED`` ; **pronotepy** ne fournissant ni la date de + Les exceptions ne sont pas attrapées : elles se propagent afin que + l'appelant puisse détecter l'échec et déclencher le repli (ou une + erreur explicite). + + :param start: Date de début de la fenêtre (incluse). + :param end: Date de fin de la fenêtre (incluse). + :return: Liste des cours. + :rtype: list[Lesson] + :raises pronotepy.PronoteAPIError: Si l'API Pronote échoue. + :raises ValueError: Si la configuration ou l'ENT est invalide. + :raises requests.RequestException: Si une requête réseau échoue. + :raises ConnectionError: Si la connexion réseau échoue. + :raises TimeoutError: Si la requête réseau expire. + """ + client = self._connect() + lessons: list[Lesson] = [] + for lesson in client.lessons(start, end): + content = lesson.content + lessons.append( + Lesson( + id=lesson.id, + start=lesson.start, + end=lesson.end, + subject=lesson.subject.name if lesson.subject is not None else "", + teachers=tuple(lesson.teacher_names or ()), + rooms=tuple(lesson.classrooms or ()), + group=lesson.group_name, + status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL), + content=content.description if content is not None else None, + ) + ) + return lessons + + def get_homeworks(self, start: date, end: date) -> list[Homework]: + """Récupère les devoirs via ``pronotepy``. + + Les exceptions ne sont pas attrapées : elles se propagent afin que + l'appelant puisse détecter l'échec et déclencher le repli (ou une + erreur explicite). **pronotepy** ne fournissant ni la date de distribution ni les professeurs des devoirs, ces champs restent vides. :param start: Date de début de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse). - :return: Tuple ``(cours, devoirs)`` ; vide en cas d'erreur. - :rtype: tuple[list[Lesson], list[Homework]] + :return: Liste des devoirs. + :rtype: list[Homework] + :raises pronotepy.PronoteAPIError: Si l'API Pronote échoue. + :raises ValueError: Si la configuration ou l'ENT est invalide. + :raises requests.RequestException: Si une requête réseau échoue. + :raises ConnectionError: Si la connexion réseau échoue. + :raises TimeoutError: Si la requête réseau expire. """ - try: - client = self._connect() - lessons: list[Lesson] = [] - for lesson in client.lessons(start, end): - content = lesson.content - lessons.append( - Lesson( - id=lesson.id, - start=lesson.start, - end=lesson.end, - subject=lesson.subject.name if lesson.subject is not None else "", - teachers=tuple(lesson.teacher_names or ()), - rooms=tuple(lesson.classrooms or ()), - group=lesson.group_name, - status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL), - content=content.description if content is not None else None, - ) + client = self._connect() + homeworks: list[Homework] = [] + for hw in client.homework(start, end): + homeworks.append( + Homework( + id=hw.id, + subject=hw.subject.name, + teachers=(), + assigned_on=None, + due_on=hw.date, + text=hw.description, + html=hw.description, ) - homeworks: list[Homework] = [] - for hw in client.homework(start, end): - homeworks.append( - Homework( - id=hw.id, - subject=hw.subject.name, - teachers=(), - assigned_on=None, - due_on=hw.date, - text=hw.description, - html=hw.description, - ) - ) - return lessons, homeworks - except (pronotepy.PronoteAPIError, ValueError) as exc: - logger.error( - "Échec de la récupération de l'agenda via pronotepy : %s", - redact_exception(exc), ) - return [], [] + return homeworks diff --git a/pronote_sync/sources/pronote/fallback.py b/pronote_sync/sources/pronote/fallback.py index 46736f2..693288b 100644 --- a/pronote_sync/sources/pronote/fallback.py +++ b/pronote_sync/sources/pronote/fallback.py @@ -3,11 +3,14 @@ Ce module fournit l'enum :class:`AgendaSource`, le protocole :class:`PronoteFetcherProtocol` consommé par le pipeline ainsi que la classe :class:`PronoteFetcher` qui sélectionne la source selon la -configuration (``PRONOTE_AGENDA_SOURCE`` / ``PRONOTE_HOMEWORK_SOURCE``) -avec repli automatique iCal → pronotepy en mode ``auto``. Les messages +configuration (``PRONOTE_AGENDA_SOURCE`` / ``PRONOTE_HOMEWORK_SOURCE``). +La logique applique un repli **unique** : une source primaire est +essayée en premier et, en cas d'échec, une seule source de repli (jamais +réciproque ni itératif) est essayée si elle est configurée. Les messages et informations proviennent toujours de pronotepy. Toutes les erreurs sont journalisées avec des secrets masqués via -:func:`~pronote_sync.utils.redaction.redact_exception`. +:func:`~pronote_sync.utils.redaction.redact_exception` ; les exceptions +d'origine ne sont jamais chaînées (``from None``). """ from __future__ import annotations @@ -15,7 +18,7 @@ from __future__ import annotations import logging from datetime import date, timedelta from enum import StrEnum -from typing import Protocol +from typing import Literal, Protocol from pronote_sync.config.settings import Settings from pronote_sync.errors import PipelineCriticalError @@ -28,6 +31,8 @@ from pronote_sync.utils.redaction import redact_exception logger = logging.getLogger(__name__) +_SourceName = Literal["ical", "pronotepy"] + class AgendaSource(StrEnum): """Source configurée pour la récupération de l'agenda et des devoirs.""" @@ -77,12 +82,13 @@ class PronoteFetcherProtocol(Protocol): class PronoteFetcher: - """Récupère les données Pronote via iCal ou pronotepy avec repli. + """Récupère les données Pronote via iCal ou pronotepy avec repli unique. Unifie les sources iCal et pronotepy selon la source configurée - (``agenda_source`` / ``homework_source``) : en mode ``AUTO``, le flux - iCal est essayé en premier et pronotepy sert de repli. Les messages et - informations proviennent toujours de pronotepy. + (``agenda_source`` / ``homework_source``) : la source primaire est + essayée en premier et, si elle échoue, une seule source de repli est + essayée lorsqu'elle est configurée. Les messages et informations + proviennent toujours de pronotepy. """ def __init__(self, settings: Settings, pronote_client: PronoteClientProtocol) -> None: @@ -105,6 +111,29 @@ class PronoteFetcher: end = today + timedelta(days=self._settings.app.sync_future_days) return start, end + def _is_ical_configured(self) -> bool: + """Vérifie que la source iCal est configurée. + + :return: ``True`` si ``ical_url`` est défini, ``False`` sinon. + :rtype: bool + """ + return self._settings.pronote.ical_url is not None + + def _is_pronotepy_configured(self) -> bool: + """Vérifie que la source pronotepy est entièrement configurée. + + :return: ``True`` si ``pronote_url``, ``username``, ``password`` + et ``ent`` sont tous définis, ``False`` sinon. + :rtype: bool + """ + pronote = self._settings.pronote + return ( + pronote.pronote_url is not None + and pronote.username is not None + and pronote.password is not None + and pronote.ent is not None + ) + def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]: """Récupère l'agenda depuis le flux iCal. @@ -129,70 +158,101 @@ class PronoteFetcher: :return: Tuple ``(cours, événements scolaires)``. :rtype: tuple[list[Lesson], list[SchoolEvent]] + :raises pronotepy.PronoteAPIError: Si l'API Pronote échoue. + :raises ValueError: Si la configuration ou l'ENT est invalide. + :raises requests.RequestException: Si une requête réseau échoue. + :raises ConnectionError: Si la connexion réseau échoue. + :raises TimeoutError: Si la requête réseau expire. """ start, end = self._fetch_window() - lessons, _ = self._pronote_client.get_agenda_fallback(start, end) + lessons = self._pronote_client.get_lessons(start, end) return lessons, [] + def _agenda_sources(self) -> tuple[_SourceName, _SourceName | None]: + """Sélectionne la source primaire et le repli unique pour l'agenda. + + En mode ``AUTO``, iCal est primaire si ``ical_url`` est configuré + (repli pronotepy si la configuration pronotepy est complète), + sinon pronotepy sans repli. Les modes explicites ``ICAL`` et + ``PRONOTEPY`` désignent la source primaire et tentent l'autre + source en repli unique si elle est configurée. + + :return: Tuple ``(source primaire, source de repli ou ``None``)``. + :rtype: tuple[_SourceName, _SourceName | None] + :raises PipelineCriticalError: Si aucune source n'est configurée en mode ``AUTO``. + """ + source = AgendaSource(self._settings.pronote.agenda_source) + if source is AgendaSource.ICAL: + return "ical", "pronotepy" if self._is_pronotepy_configured() else None + if source is AgendaSource.PRONOTEPY: + return "pronotepy", "ical" if self._is_ical_configured() else None + if self._is_ical_configured(): + return "ical", "pronotepy" if self._is_pronotepy_configured() else None + if self._is_pronotepy_configured(): + return "pronotepy", None + raise PipelineCriticalError( + "Impossible de récupérer l'agenda : ni la source iCal ni pronotepy n'est configurée" + ) from None + + def _fetch_agenda_source(self, name: _SourceName) -> tuple[list[Lesson], list[SchoolEvent]]: + """Récupère l'agenda depuis la source nommée. + + :param name: Nom de la source (``"ical"`` ou ``"pronotepy"``). + :return: Tuple ``(cours, événements scolaires)``. + :rtype: tuple[list[Lesson], list[SchoolEvent]] + """ + if name == "ical": + return self._fetch_agenda_ical() + return self._fetch_agenda_pronotepy() + def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]: """Récupère les cours et les événements scolaires selon la source configurée. - En mode ``AUTO``, iCal est essayé en premier et pronotepy sert de - repli ; si les deux sources échouent, une erreur critique est levée. + La source primaire est essayée en premier ; si elle échoue, la + source de repli unique (l'autre source, si configurée) est + essayée. Si la source primaire et le repli échouent — ou si + aucune source n'est configurée en mode ``AUTO`` — une erreur + critique est levée. :return: Tuple ``(cours, événements scolaires)``. :rtype: tuple[list[Lesson], list[SchoolEvent]] - :raises PipelineCriticalError: Si toutes les sources configurées échouent. + :raises PipelineCriticalError: Si toutes les sources tentées échouent. """ - source = AgendaSource(self._settings.pronote.agenda_source) - if source is AgendaSource.ICAL: - try: - return self._fetch_agenda_ical() - except Exception as exc: - logger.error( - "Échec de la récupération iCal pour l'agenda : %s", - redact_exception(exc), - ) - raise PipelineCriticalError( - "Impossible de récupérer l'agenda : la source iCal a échoué" - ) from exc - if source is AgendaSource.PRONOTEPY: - try: - return self._fetch_agenda_pronotepy() - except Exception as exc: - logger.error( - "Échec de la récupération pronotepy pour l'agenda : %s", - redact_exception(exc), - ) - raise PipelineCriticalError( - "Impossible de récupérer l'agenda : la source pronotepy a échoué" - ) from exc - - # Mode AUTO : essayer iCal d'abord, puis replier sur pronotepy. + primary, fallback = self._agenda_sources() try: - return self._fetch_agenda_ical() - except Exception as exc: - logger.warning( - "Échec de la récupération iCal pour l'agenda : %s", - redact_exception(exc), - ) - logger.info("Repli sur pronotepy pour l'agenda.") - try: - lessons, school_events = self._fetch_agenda_pronotepy() + return self._fetch_agenda_source(primary) except Exception as exc: logger.error( - "Échec de la récupération pronotepy pour l'agenda : %s", + "Échec de la récupération %s pour l'agenda : %s", + primary, redact_exception(exc), ) - raise PipelineCriticalError( - "Impossible de récupérer l'agenda : les sources iCal et pronotepy ont échoué" - ) from exc - if not lessons: - logger.warning( - "Le repli pronotepy pour l'agenda a retourné un résultat vide après l'échec " - "d'iCal : impossible de distinguer une absence de cours d'un échec silencieux." - ) - return lessons, school_events + if fallback is None: + raise PipelineCriticalError( + f"Impossible de récupérer l'agenda : la source {primary} a échoué" + ) from None + logger.info("Repli sur %s pour l'agenda.", fallback) + try: + lessons, school_events = self._fetch_agenda_source(fallback) + except Exception as exc: + logger.error( + "Échec de la récupération %s pour l'agenda : %s", + fallback, + redact_exception(exc), + ) + raise PipelineCriticalError( + f"Impossible de récupérer l'agenda : les sources {primary}" + f" et {fallback} ont échoué" + ) from None + if not lessons: + logger.warning( + "Le repli %s pour l'agenda a retourné un résultat vide après l'échec " + "de %s : impossible de distinguer une absence de cours d'un échec " + "silencieux.", + fallback, + primary, + ) + return lessons, school_events def _fetch_homework_ical(self, target_date: date) -> list[Homework]: """Récupère les devoirs depuis le flux iCal pour la date cible. @@ -207,76 +267,112 @@ class PronoteFetcher: lessons, _ = self._fetch_agenda_ical() return collect_homeworks(lessons, target_date) - def _fetch_homework_pronotepy(self) -> list[Homework]: - """Récupère les devoirs depuis pronotepy. + def _fetch_homework_pronotepy(self, target_date: date) -> list[Homework]: + """Récupère les devoirs depuis pronotepy pour la date cible. - :return: Liste des devoirs. - :rtype: list[Homework] - """ - start, end = self._fetch_window() - _, homeworks = self._pronote_client.get_agenda_fallback(start, end) - return homeworks - - def fetch_homework(self, target_date: date) -> list[Homework]: - """Récupère les devoirs selon la source configurée. - - En mode ``AUTO``, iCal est essayé en premier et pronotepy sert de - repli ; si les deux sources échouent, une erreur critique est levée. + Les devoirs sont filtrés sur la date d'échéance : seuls ceux dont + ``due_on`` correspond à ``target_date`` sont conservés. :param target_date: Date cible pour laquelle collecter les devoirs. :return: Liste des devoirs. :rtype: list[Homework] - :raises PipelineCriticalError: Si toutes les sources configurées échouent. + :raises pronotepy.PronoteAPIError: Si l'API Pronote échoue. + :raises ValueError: Si la configuration ou l'ENT est invalide. + :raises requests.RequestException: Si une requête réseau échoue. + :raises ConnectionError: Si la connexion réseau échoue. + :raises TimeoutError: Si la requête réseau expire. + """ + start, end = self._fetch_window() + homeworks = self._pronote_client.get_homeworks(start, end) + return [hw for hw in homeworks if hw.due_on == target_date] + + def _homework_sources(self) -> tuple[_SourceName, _SourceName | None]: + """Sélectionne la source primaire et le repli unique pour les devoirs. + + En mode ``AUTO``, iCal est primaire si ``ical_url`` est configuré + (repli pronotepy si la configuration pronotepy est complète), + sinon pronotepy sans repli. Les modes explicites ``ICAL`` et + ``PRONOTEPY`` désignent la source primaire et tentent l'autre + source en repli unique si elle est configurée. + + :return: Tuple ``(source primaire, source de repli ou ``None``)``. + :rtype: tuple[_SourceName, _SourceName | None] + :raises PipelineCriticalError: Si aucune source n'est configurée en mode ``AUTO``. """ source = AgendaSource(self._settings.pronote.homework_source) if source is AgendaSource.ICAL: - try: - return self._fetch_homework_ical(target_date) - except Exception as exc: - logger.error( - "Échec de la récupération iCal pour les devoirs : %s", - redact_exception(exc), - ) - raise PipelineCriticalError( - "Impossible de récupérer les devoirs : la source iCal a échoué" - ) from exc + return "ical", "pronotepy" if self._is_pronotepy_configured() else None if source is AgendaSource.PRONOTEPY: - try: - return self._fetch_homework_pronotepy() - except Exception as exc: - logger.error( - "Échec de la récupération pronotepy pour les devoirs : %s", - redact_exception(exc), - ) - raise PipelineCriticalError( - "Impossible de récupérer les devoirs : la source pronotepy a échoué" - ) from exc + return "pronotepy", "ical" if self._is_ical_configured() else None + if self._is_ical_configured(): + return "ical", "pronotepy" if self._is_pronotepy_configured() else None + if self._is_pronotepy_configured(): + return "pronotepy", None + raise PipelineCriticalError( + "Impossible de récupérer les devoirs : ni la source iCal ni pronotepy n'est configurée" + ) from None - # Mode AUTO : essayer iCal d'abord, puis replier sur pronotepy. - try: + def _fetch_homework_source(self, name: _SourceName, target_date: date) -> list[Homework]: + """Récupère les devoirs depuis la source nommée. + + :param name: Nom de la source (``"ical"`` ou ``"pronotepy"``). + :param target_date: Date cible pour laquelle collecter les devoirs. + :return: Liste des devoirs. + :rtype: list[Homework] + """ + if name == "ical": return self._fetch_homework_ical(target_date) - except Exception as exc: - logger.warning( - "Échec de la récupération iCal pour les devoirs : %s", - redact_exception(exc), - ) - logger.info("Repli sur pronotepy pour les devoirs.") + return self._fetch_homework_pronotepy(target_date) + + def fetch_homework(self, target_date: date) -> list[Homework]: + """Récupère les devoirs selon la source configurée. + + La source primaire est essayée en premier ; si elle échoue, la + source de repli unique (l'autre source, si configurée) est + essayée. Si la source primaire et le repli échouent — ou si + aucune source n'est configurée en mode ``AUTO`` — une erreur + critique est levée. + + :param target_date: Date cible pour laquelle collecter les devoirs. + :return: Liste des devoirs. + :rtype: list[Homework] + :raises PipelineCriticalError: Si toutes les sources tentées échouent. + """ + primary, fallback = self._homework_sources() try: - homeworks = self._fetch_homework_pronotepy() + return self._fetch_homework_source(primary, target_date) except Exception as exc: logger.error( - "Échec de la récupération pronotepy pour les devoirs : %s", + "Échec de la récupération %s pour les devoirs : %s", + primary, redact_exception(exc), ) - raise PipelineCriticalError( - "Impossible de récupérer les devoirs : les sources iCal et pronotepy ont échoué" - ) from exc - if not homeworks: - logger.warning( - "Le repli pronotepy pour les devoirs a retourné un résultat vide après l'échec " - "d'iCal : impossible de distinguer une absence de devoirs d'un échec silencieux." - ) - return homeworks + if fallback is None: + raise PipelineCriticalError( + f"Impossible de récupérer les devoirs : la source {primary} a échoué" + ) from None + logger.info("Repli sur %s pour les devoirs.", fallback) + try: + homeworks = self._fetch_homework_source(fallback, target_date) + except Exception as exc: + logger.error( + "Échec de la récupération %s pour les devoirs : %s", + fallback, + redact_exception(exc), + ) + raise PipelineCriticalError( + f"Impossible de récupérer les devoirs : les sources {primary}" + f" et {fallback} ont échoué" + ) from None + if not homeworks: + logger.warning( + "Le repli %s pour les devoirs a retourné un résultat vide après " + "l'échec de %s : impossible de distinguer une absence de devoirs " + "d'un échec silencieux.", + fallback, + primary, + ) + return homeworks def fetch_messages(self) -> list[Message]: """Récupère les messages des discussions Pronote (toujours via pronotepy). @@ -284,7 +380,14 @@ class PronoteFetcher: :return: Liste des messages. :rtype: list[Message] """ - return self._pronote_client.get_messages() + try: + return self._pronote_client.get_messages() + except Exception as exc: + logger.error( + "Échec de la récupération des messages : %s", + redact_exception(exc), + ) + raise def fetch_informations(self) -> list[Message]: """Récupère les informations et sondages Pronote (toujours via pronotepy). @@ -292,4 +395,11 @@ class PronoteFetcher: :return: Liste des informations et sondages. :rtype: list[Message] """ - return self._pronote_client.get_informations() + try: + return self._pronote_client.get_informations() + except Exception as exc: + logger.error( + "Échec de la récupération des informations : %s", + redact_exception(exc), + ) + raise diff --git a/pronote_sync/sources/pronote/ical.py b/pronote_sync/sources/pronote/ical.py index d3252f3..3a7e7d5 100644 --- a/pronote_sync/sources/pronote/ical.py +++ b/pronote_sync/sources/pronote/ical.py @@ -218,21 +218,23 @@ def _parse_french_date(value: str) -> date | None: return None -def parse_body(body: str) -> tuple[str | None, dict[date, str], dict[date, str]]: +def parse_body(body: str) -> tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]]: """Parse le corps HTML pour extraire contenu pédagogique et devoirs. Le contenu est extrait de la section ``Contenu pédagogique :``. Les devoirs à faire sont extraits des sections ``Pour le JJ/MM/AAAA :`` - (dict date → texte) et les devoirs donnés des sections - ``Donné le JJ/MM/AAAA :`` (dict date → texte). + (liste de tuples ``(date, texte)`` dans l'ordre du flux) et les devoirs donnés + des sections ``Donné le JJ/MM/AAAA :`` (liste de tuples + ``(date, texte)``). Les listes préservent tous les blocs, même lorsque plusieurs + sections partagent la même date. :param body: Corps HTML (à partir du premier ````). :return: Tuple ``(contenu pédagogique, devoirs dus, devoirs donnés)``. - :rtype: tuple[str | None, dict[date, str], dict[date, str]] + :rtype: tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]] """ content: str | None = None - due_blocks: dict[date, str] = {} - assigned_blocks: dict[date, str] = {} + due_blocks: list[tuple[date, str]] = [] + assigned_blocks: list[tuple[date, str]] = [] content_match = _CONTENT_PATTERN.search(body) if content_match is not None: @@ -241,34 +243,35 @@ def parse_body(body: str) -> tuple[str | None, dict[date, str], dict[date, str]] for match in _DUE_PATTERN.finditer(body): due_date = _parse_french_date(match.group(1)) if due_date is not None: - due_blocks[due_date] = _strip_html(match.group(2)) + due_blocks.append((due_date, _strip_html(match.group(2)))) for match in _ASSIGNED_PATTERN.finditer(body): assigned_date = _parse_french_date(match.group(1)) if assigned_date is not None: - assigned_blocks[assigned_date] = _strip_html(match.group(2)) + assigned_blocks.append((assigned_date, _strip_html(match.group(2)))) return content, due_blocks, assigned_blocks def parse_homework_blocks( - due_blocks: dict[date, str], - assigned_blocks: dict[date, str], + due_blocks: list[tuple[date, str]], + assigned_blocks: list[tuple[date, str]], ) -> tuple[HomeworkBlock, ...]: - """Construit les :class:`HomeworkBlock` depuis les dicts de devoirs. + """Construit les :class:`HomeworkBlock` depuis les listes de devoirs. Les blocs dus (``kind="due"``) précèdent les blocs donnés - (``kind="assigned"``), dans l'ordre d'insertion des dicts. + (``kind="assigned"``), dans l'ordre des listes. Tous les blocs + sont préservés, y compris lorsque plusieurs partagent la même date. - :param due_blocks: Dict date → texte des devoirs à faire. - :param assigned_blocks: Dict date → texte des devoirs donnés. + :param due_blocks: Liste de tuples ``(date, texte)`` des devoirs à faire. + :param assigned_blocks: Liste de tuples ``(date, texte)`` des devoirs donnés. :return: Tuple de blocs de devoirs pour le cours. :rtype: tuple[HomeworkBlock, ...] """ blocks: list[HomeworkBlock] = [] - for due_date, text in due_blocks.items(): + for due_date, text in due_blocks: blocks.append(HomeworkBlock(kind="due", date=due_date, text=text, html=text)) - for assigned_date, text in assigned_blocks.items(): + for assigned_date, text in assigned_blocks: blocks.append(HomeworkBlock(kind="assigned", date=assigned_date, text=text, html=text)) return tuple(blocks) @@ -360,9 +363,10 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School Les VEVENT de vacances/congés (tout le jour) deviennent des :class:`SchoolEvent` de type ``holiday``. Les VEVENT horodatés - deviennent des :class:`Lesson` dont le statut dérive de la - catégorie (``Cours - Cours annulé`` → ``CANCELLED``, - ``Cours - Cours déplacé`` → ``MOVED``). Les UID sont normalisés ; + deviennent des :class:`Lesson` dont le statut dérive de la propriété + ``STATUS`` (``CANCELLED`` → ``CANCELLED``) et de la catégorie + (``Cours - Cours annulé`` → ``CANCELLED``, ``Cours - Cours déplacé`` → + ``MOVED``). Les UID sont normalisés ; un événement sans UID reçoit un UID déterministe généré à partir de ses champs clés (début, fin, matière, enseignants, salles, groupe). @@ -416,11 +420,14 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School if not isinstance(start, datetime) or not isinstance(end, datetime): continue - status = LessonStatus.NORMAL - if "Cours - Cours annulé" in categories: + status_obj = component.get("status") + status_value = str(status_obj).strip().upper() if status_obj is not None else "" + if status_value == "CANCELLED" or "Cours - Cours annulé" in categories: status = LessonStatus.CANCELLED elif "Cours - Cours déplacé" in categories: status = LessonStatus.MOVED + else: + status = LessonStatus.NORMAL description = component.get("description") description_str = str(description) if description is not None else "" diff --git a/tests/unit/test_ical.py b/tests/unit/test_ical.py index 0f94ab3..0613ed5 100644 --- a/tests/unit/test_ical.py +++ b/tests/unit/test_ical.py @@ -26,6 +26,7 @@ from pronote_sync.sources.pronote.ical import ( generate_homework_id, get_calendar_name, normalize_homework_text, + parse_body, parse_ical, ) @@ -463,3 +464,137 @@ def test_generate_homework_id_deterministic() -> None: id1 = generate_homework_id(due_on, text) id2 = generate_homework_id(due_on, text) assert id1 == id2 + + +CANCELLED_STATUS_ICAL = """BEGIN:VCALENDAR +VERSION:2.0 +X-WR-CALNAME:Test +BEGIN:VEVENT +UID:Test-123-20260906T120000Z-Index-Education +DTSTART:20260907T080000Z +DTEND:20260907T090000Z +SUMMARY:Test Course +STATUS:CANCELLED +DESCRIPTION:
+END:VEVENT +END:VCALENDAR""" + + +NORMAL_STATUS_ICAL = """BEGIN:VCALENDAR +VERSION:2.0 +X-WR-CALNAME:Test +BEGIN:VEVENT +UID:Test-123-20260906T120000Z-Index-Education +DTSTART:20260907T080000Z +DTEND:20260907T090000Z +SUMMARY:Test Course +STATUS:CONFIRMED +DESCRIPTION:
+END:VEVENT +END:VCALENDAR""" + + +MOVED_BY_CATEGORY_ICAL = """BEGIN:VCALENDAR +VERSION:2.0 +X-WR-CALNAME:Test +BEGIN:VEVENT +UID:Test-123-20260906T120000Z-Index-Education +DTSTART:20260907T080000Z +DTEND:20260907T090000Z +SUMMARY:Test Course +CATEGORIES:Cours - Cours déplacé +DESCRIPTION:
+END:VEVENT +END:VCALENDAR""" + + +MULTIPLE_BLOCKS_SAME_DATE_ICAL = """BEGIN:VCALENDAR +VERSION:2.0 +X-WR-CALNAME:Test +BEGIN:VEVENT +UID:Test-123-20260906T120000Z-Index-Education +DTSTART:20260907T080000Z +DTEND:20260907T090000Z +SUMMARY:Test Course +CATEGORIES:Cours +DESCRIPTION:
+ Matière : Math + Professeur : M. Dupont + Salle : 204 + + Pour le 10/09/2026 : + Exercice 1 à 5 page 42. + Pour le 10/09/2026 : + Exercice 6 à 10 page 43. +
+END:VEVENT +END:VCALENDAR""" + + +def test_parse_ical_status_cancelled_only() -> None: + """Un cours avec STATUS:CANCELLED mais sans CATEGORIES contenant 'Cours annulé' a status == LessonStatus.CANCELLED. + + :return: None + """ + lessons, _, _ = parse_ical(CANCELLED_STATUS_ICAL) + assert len(lessons) == 1 + assert lessons[0].status == LessonStatus.CANCELLED + + +def test_parse_ical_status_normal_without_cancel() -> None: + """Un cours avec STATUS:CONFIRMED (ou sans STATUS) a status == LessonStatus.NORMAL. + + :return: None + """ + lessons, _, _ = parse_ical(NORMAL_STATUS_ICAL) + assert len(lessons) == 1 + assert lessons[0].status == LessonStatus.NORMAL + + +def test_parse_ical_moved_by_category_only() -> None: + """Un cours avec CATEGORIES:Cours - Cours déplacé et sans STATUS a status == LessonStatus.MOVED. + + :return: None + """ + lessons, _, _ = parse_ical(MOVED_BY_CATEGORY_ICAL) + assert len(lessons) == 1 + assert lessons[0].status == LessonStatus.MOVED + + +def test_parse_body_multiple_blocks_same_date() -> None: + """Un DESCRIPTION avec deux sections 'Pour le' à la même date conserve les deux blocs. + + :return: None + """ + body_html = ( + "
\n" + " Pour le 10/09/2026:\n" + " Exercice 1 à 5 page 42.\n" + " Pour le 10/09/2026:\n" + " Exercice 6 à 10 page 43.\n" + "
" + ) + content, due_blocks, assigned_blocks = parse_body(body_html) + assert len(due_blocks) == 2 + assert due_blocks[0][0] == date(2026, 9, 10) + assert due_blocks[0][1] == "Exercice 1 à 5 page 42." + assert due_blocks[1][0] == date(2026, 9, 10) + assert due_blocks[1][1] == "Exercice 6 à 10 page 43." + + +def test_collect_homeworks_from_fixture() -> None: + """Parse le fixture pronote-4e.ics, appelle collect_homeworks pour le 10/09/2026 et vérifie qu'au moins un devoir est retourné. + + :return: None + """ + fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics" + with open(fixture_path, encoding="utf-8") as f: + content = f.read() + + lessons, _, _ = parse_ical(content) + homeworks = collect_homeworks(lessons, date(2026, 9, 10)) + + assert len(homeworks) >= 1 + # Vérifie qu'au moins un devoir a le bon sujet et texte + assert any(hw.subject == "Mathématiques" for hw in homeworks) + assert any("Exercices 1 à 5 page 42" in hw.text for hw in homeworks)