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``).
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` ; les exceptions
d'origine ne sont jamais chaînées (``from None``).
"""

from __future__ import annotations

import logging
from datetime import date, timedelta
from enum import StrEnum
from typing import Literal, 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__)

_SourceName = Literal["ical", "pronotepy"]


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 unique.

    Unifie les sources iCal et pronotepy selon la source configurée
    (``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:
        """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 _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.

        :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]]
        :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_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.

        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 tentées échouent.
        """
        primary, fallback = self._agenda_sources()
        try:
            return self._fetch_agenda_source(primary)
        except Exception as exc:
            logger.error(
                "Échec de la récupération %s pour l'agenda : %s",
                primary,
                redact_exception(exc),
            )
            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.

        :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, target_date: date) -> list[Homework]:
        """Récupère les devoirs depuis pronotepy pour la date cible.

        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 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:
            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 les devoirs : ni la source iCal ni pronotepy n'est configurée"
        ) from None

    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)
        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:
            return self._fetch_homework_source(primary, target_date)
        except Exception as exc:
            logger.error(
                "Échec de la récupération %s pour les devoirs : %s",
                primary,
                redact_exception(exc),
            )
            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).

        :return: Liste des messages.
        :rtype: list[Message]
        """
        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).

        :return: Liste des informations et sondages.
        :rtype: list[Message]
        """
        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