Antoine Van Elstraete
d7d31e14ff
Co-authored-by: Codex/gpt-5.6-terra <codex-gpt-5-6-terra@agents.invalid>
127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
"""É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
|
|
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.
|
|
"""
|
|
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 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,
|
|
)
|