diff --git a/pronote_sync/sources/pronote/fallback.py b/pronote_sync/sources/pronote/fallback.py new file mode 100644 index 0000000..46736f2 --- /dev/null +++ b/pronote_sync/sources/pronote/fallback.py @@ -0,0 +1,295 @@ +"""Logique de repli unifiant les sources iCal et pronotepy. + +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 +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`. +""" + +from __future__ import annotations + +import logging +from datetime import date, timedelta +from enum import StrEnum +from typing import Protocol + +from pronote_sync.config.settings import Settings +from pronote_sync.errors import PipelineCriticalError +from pronote_sync.models.agenda import Lesson, SchoolEvent +from pronote_sync.models.homework import Homework +from pronote_sync.models.message import Message +from pronote_sync.sources.pronote.client import PronoteClientProtocol +from pronote_sync.sources.pronote.ical import collect_homeworks, fetch_ical, parse_ical +from pronote_sync.utils.redaction import redact_exception + +logger = logging.getLogger(__name__) + + +class AgendaSource(StrEnum): + """Source configurée pour la récupération de l'agenda et des devoirs.""" + + AUTO = "auto" + ICAL = "ical" + PRONOTEPY = "pronotepy" + + +class PronoteFetcherProtocol(Protocol): + """Interface de récupération des données Pronote consommée par le pipeline.""" + + def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]: + """Récupère les cours et les événements scolaires. + + :return: Tuple ``(cours, événements scolaires)``. + :rtype: tuple[list[Lesson], list[SchoolEvent]] + :raises PipelineCriticalError: Si toutes les sources configurées échouent. + """ + ... + + def fetch_homework(self, target_date: date) -> list[Homework]: + """Récupère les devoirs pour la date cible. + + :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. + """ + ... + + def fetch_messages(self) -> list[Message]: + """Récupère les messages des discussions Pronote. + + :return: Liste des messages. + :rtype: list[Message] + """ + ... + + def fetch_informations(self) -> list[Message]: + """Récupère les informations et sondages Pronote. + + :return: Liste des informations et sondages. + :rtype: list[Message] + """ + ... + + +class PronoteFetcher: + """Récupère les données Pronote via iCal ou pronotepy avec repli. + + 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. + """ + + def __init__(self, settings: Settings, pronote_client: PronoteClientProtocol) -> None: + """Initialise le fetcher sans récupérer aucune donnée. + + :param settings: Configuration racine du pipeline (Pronote, fenêtre de synchronisation). + :param pronote_client: Client pronotepy utilisé pour les sources pronotepy. + """ + self._settings: Settings = settings + self._pronote_client: PronoteClientProtocol = pronote_client + + def _fetch_window(self) -> tuple[date, date]: + """Calcule la fenêtre de synchronisation autour de la date du jour. + + :return: Tuple ``(date de début, date de fin)`` de la fenêtre. + :rtype: tuple[date, date] + """ + today = date.today() + start = today - timedelta(days=self._settings.app.sync_past_days) + end = today + timedelta(days=self._settings.app.sync_future_days) + return start, end + + def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]: + """Récupère l'agenda depuis le flux iCal. + + :return: Tuple ``(cours, événements scolaires)``. + :rtype: tuple[list[Lesson], list[SchoolEvent]] + :raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide. + :raises OSError: Si le fichier iCal local est illisible. + :raises requests.RequestException: Si la récupération HTTP échoue. + """ + ical_url = self._settings.pronote.ical_url + if ical_url is None: + raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal") + raw_ical = fetch_ical(ical_url.get_secret_value()) + lessons, _, school_events = parse_ical(raw_ical) + return lessons, school_events + + def _fetch_agenda_pronotepy(self) -> tuple[list[Lesson], list[SchoolEvent]]: + """Récupère l'agenda depuis pronotepy. + + Les événements scolaires (vacances, jours fériés) ne sont pas + fournis par pronotepy : la liste retournée est vide. + + :return: Tuple ``(cours, événements scolaires)``. + :rtype: tuple[list[Lesson], list[SchoolEvent]] + """ + start, end = self._fetch_window() + lessons, _ = self._pronote_client.get_agenda_fallback(start, end) + return lessons, [] + + 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. + + :return: Tuple ``(cours, événements scolaires)``. + :rtype: tuple[list[Lesson], list[SchoolEvent]] + :raises PipelineCriticalError: Si toutes les sources configuré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. + 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() + 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 : 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 + + def _fetch_homework_ical(self, target_date: date) -> list[Homework]: + """Récupère les devoirs depuis le flux iCal pour la date cible. + + :param target_date: Date cible pour laquelle collecter les devoirs. + :return: Liste des devoirs. + :rtype: list[Homework] + :raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide. + :raises OSError: Si le fichier iCal local est illisible. + :raises requests.RequestException: Si la récupération HTTP échoue. + """ + 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. + + :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. + + :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. + """ + 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 + 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 + + # Mode AUTO : essayer iCal d'abord, puis replier sur pronotepy. + try: + 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.") + try: + homeworks = 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 : 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 + + def fetch_messages(self) -> list[Message]: + """Récupère les messages des discussions Pronote (toujours via pronotepy). + + :return: Liste des messages. + :rtype: list[Message] + """ + return self._pronote_client.get_messages() + + def fetch_informations(self) -> list[Message]: + """Récupère les informations et sondages Pronote (toujours via pronotepy). + + :return: Liste des informations et sondages. + :rtype: list[Message] + """ + return self._pronote_client.get_informations()