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

import logging
from dataclasses import dataclass
from datetime import date

from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError
from pronote_sync.models.agenda import Lesson, LessonStatus, 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

logger = logging.getLogger(__name__)


@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 un cours non annulé. Sinon,
    elle choisit le prochain jour possédant un cours effectif dans la fenêtre.
    Si aucune reprise n'est visible, la fin exclusive d'un événement scolaire
    couvrant J+1 est utilisée ; à défaut, J+1 reste le meilleur choix connu.

    :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, avec borne de fin exclusive.
    :return: Date cible du digest.
    :rtype: date
    """
    tomorrow = date.fromordinal(today.toordinal() + 1)
    effective_dates = {
        lesson.start.date()
        for lesson in lessons
        if lesson.status != LessonStatus.CANCELLED and lesson.start.date() > today
    }
    if tomorrow in effective_dates:
        reason = "cours effectif demain"
        target = tomorrow
    elif effective_dates:
        reason = "prochain cours effectif connu"
        target = min(effective_dates)
    else:
        event_ends = [
            event.to_date for event in school_events if event.from_date <= tomorrow < event.to_date
        ]
        if event_ends:
            reason = "fin exclusive de l'événement couvrant demain"
            target = max(event_ends)
        else:
            reason = "aucune reprise connue"
            target = tomorrow
    logger.info("Date cible %s : %s", target.isoformat(), reason)
    return target


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,
    )