148 lines
5.5 KiB
Python
148 lines
5.5 KiB
Python
"""É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,
|
|
)
|