college-infos/pronote_sync/pipeline/steps/fetch.py

"""Étape de récupération des données Pronote pour une exécution du pipeline."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import date

from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError
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.fallback import PronoteFetcherProtocol
from pronote_sync.utils.redaction import redact_exception


@dataclass(frozen=True)
class FetchedPronoteData:
    """Représente les données brutes récupérées pendant une exécution.

    :ivar lessons: Cours récupérés depuis la source sélectionnée.
    :ivar homeworks: Devoirs destinés à la date cible.
    :ivar school_events: Événements scolaires récupérés avec l'agenda.
    :ivar messages: Messages et informations Pronote disponibles.
    :ivar target_date: Date cible du digest.
    """

    lessons: list[Lesson]
    homeworks: list[Homework]
    school_events: list[SchoolEvent]
    messages: list[Message]
    target_date: date


def resolve_target_date(
    today: date, lessons: list[Lesson], school_events: list[SchoolEvent]
) -> date:
    """Détermine la date cible du digest à partir de l'agenda disponible.

    La règle privilégie J+1 lorsqu'il contient des cours. Si la journée en
    cours contient des cours mais pas J+1, le prochain cours connu est choisi.
    Sans cours correspondant, J+1 est conservé, y compris pendant les vacances.

    :param today: Date de référence de l'exécution.
    :param lessons: Cours récupérés pour la fenêtre de synchronisation.
    :param school_events: Événements scolaires récupérés (réservés aux évolutions
        du libellé de jour sans cours).
    :return: Date cible du digest.
    :rtype: date
    """
    del school_events
    tomorrow = date.fromordinal(today.toordinal() + 1)
    lesson_dates = {lesson.start.date() for lesson in lessons}
    if tomorrow in lesson_dates:
        return tomorrow
    if today in lesson_dates:
        future_dates = sorted(day for day in lesson_dates if day > today)
        if future_dates:
            return future_dates[0]
    return tomorrow


def _fetch_optional_messages(
    fetcher: PronoteFetcherProtocol,
) -> tuple[list[Message], list[PipelineWarning]]:
    """Récupère les messages et informations sans bloquer le pipeline.

    :param fetcher: Fetcher Pronote configuré.
    :return: Messages disponibles et avertissements éventuels.
    :rtype: tuple[list[Message], list[PipelineWarning]]
    """
    messages: list[Message] = []
    warnings: list[PipelineWarning] = []
    for step, method in (
        ("fetch_messages", fetcher.fetch_messages),
        ("fetch_informations", fetcher.fetch_informations),
    ):
        try:
            messages.extend(method())
        except Exception as exc:
            warnings.append(
                PipelineWarning(
                    f"Récupération non critique échouée : {redact_exception(exc)}",
                    step=step,
                )
            )
    return messages, warnings


def fetch_step(
    fetcher: PronoteFetcherProtocol, *, today: date | None = None
) -> tuple[FetchedPronoteData, list[PipelineWarning]]:
    """Récupère les données Pronote critiques et les compléments dégradables.

    L'agenda et les devoirs sont critiques : leur échec empêche de produire un
    digest fiable et est donc propagé comme :class:`PipelineCriticalError`.
    Les messages et informations sont facultatifs ; leur échec produit un
    avertissement et une liste partielle reste valide.

    :param fetcher: Fetcher Pronote configuré.
    :param today: Date de référence, injectée par les tests ; J courant par défaut.
    :return: Données récupérées et avertissements non critiques.
    :rtype: tuple[FetchedPronoteData, list[PipelineWarning]]
    :raises PipelineCriticalError: Si l'agenda ou les devoirs ne sont pas disponibles.
    :raises PronoteAuthRotationError: Si une rotation du token d'authentification
        pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline.
    """
    try:
        lessons, school_events = fetcher.fetch_agenda()
        target_date = resolve_target_date(today or date.today(), lessons, school_events)
        homeworks = fetcher.fetch_homework(target_date)
    except PipelineCriticalError:
        raise
    except PronoteAuthRotationError:
        raise
    except Exception as exc:
        raise PipelineCriticalError(
            f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
        ) from None

    messages, warnings = _fetch_optional_messages(fetcher)
    return (
        FetchedPronoteData(
            lessons=lessons,
            homeworks=homeworks,
            school_events=school_events,
            messages=messages,
            target_date=target_date,
        ),
        warnings,
    )