college-infos/pronote_sync/sources/pronote/fallback.py

"""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()