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``).
Contrat strict : les modes explicites n'utilisent que la source
configurée, sans aucun repli ; seul le mode ``auto`` applique un repli
unique iCal → pronotepy, en cas d'exception uniquement. 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 collections.abc import Iterator
from contextlib import contextmanager
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, repli réservé au mode ``auto``.

    Unifie les sources iCal et pronotepy selon la source configurée
    (``agenda_source`` / ``homework_source``) : les modes explicites
    n'utilisent que la source configurée, sans aucun repli ; seul le mode
    ``auto`` essaie une source primaire puis, si elle échoue, une seule
    source de repli 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
        self._run_ical_agenda: tuple[list[Lesson], list[SchoolEvent]] | None = None
        self._cache_ical_for_run = False

    @contextmanager
    def run_context(self) -> Iterator[None]:
        """Active un cache iCal éphémère pour une exécution du pipeline.

        Le cache couvre à la fois le téléchargement et le parsing du flux.
        Il est toujours supprimé à la sortie du contexte, y compris si une
        étape échoue : il ne peut donc pas devenir un cache global ou
        persistant entre deux exécutions.

        :yield: Aucun objet.
        :rtype: Iterator[None]
        """
        previous_cache = self._run_ical_agenda
        previous_enabled = self._cache_ical_for_run
        self._run_ical_agenda = None
        self._cache_ical_for_run = True
        try:
            yield
        finally:
            self._run_ical_agenda = previous_cache
            self._cache_ical_for_run = previous_enabled

    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.
        """
        if self._cache_ical_for_run and self._run_ical_agenda is not None:
            return self._run_ical_agenda
        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)
        result = (lessons, school_events)
        if self._cache_ical_for_run:
            self._run_ical_agenda = result
        return result

    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.

        Les modes explicites ``ICAL`` et ``PRONOTEPY`` désignent la seule
        source utilisée, sans aucun repli. En mode ``AUTO``, iCal est
        primaire si ``ical_url`` est configuré (repli pronotepy si la
        configuration pronotepy est complète), sinon pronotepy sans repli.

        :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", None
        if source is AgendaSource.PRONOTEPY:
            return "pronotepy", 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 explicite (``ical`` ou ``pronotepy``), la source désignée
        est la seule tentée : si elle échoue, une erreur critique est levée
        sans repli. En mode ``auto``, la source primaire est essayée en
        premier puis, si elle échoue, la source de repli unique (l'autre
        source, si configurée) l'est à son tour ; si la source primaire et
        le repli échouent — ou si aucune source n'est configurée — 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.

        Les modes explicites ``ICAL`` et ``PRONOTEPY`` désignent la seule
        source utilisée, sans aucun repli. En mode ``AUTO``, iCal est
        primaire si ``ical_url`` est configuré (repli pronotepy si la
        configuration pronotepy est complète), sinon pronotepy sans repli.

        :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", None
        if source is AgendaSource.PRONOTEPY:
            return "pronotepy", 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.

        En mode explicite (``ical`` ou ``pronotepy``), la source désignée
        est la seule tentée : si elle échoue, une erreur critique est levée
        sans repli. En mode ``auto``, la source primaire est essayée en
        premier puis, si elle échoue, la source de repli unique (l'autre
        source, si configurée) l'est à son tour ; si la source primaire et
        le repli échouent — ou si aucune source n'est configurée — 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