fix(M4): pronote_url + account_type et parsing iCal (STATUS:CANCELLED + blocs multi-dates)

Configuration (D1+D3) :
- PronoteSettings : ajout pronote_url (str | None) et account_type
  (Literal student/parent, défaut parent)

Parsing iCal (Fix 4+7) :
- parse_ical : détection STATUS:CANCELLED en plus de CATEGORIES
- parse_body : list[tuple[date, str]] au lieu de dict[date, str]
  pour préserver les blocs multiples à la même date
- parse_homework_blocks : adapté aux listes

Co-authored-by: opencode/coder <coder@agents.invalid>
This commit is contained in:
2026-09-06 15:03:54 +02:00
parent e0cd073f9a
commit 1b550aa818
6 changed files with 578 additions and 198 deletions

View File

@@ -1,5 +1,7 @@
# --- Pronote --- # --- Pronote ---
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024 PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
PRONOTE_URL=https://college.ent/pronote/eleve.html
PRONOTE_ACCOUNT_TYPE=parent
PRONOTE_USERNAME=parent.dupont PRONOTE_USERNAME=parent.dupont
PRONOTE_PASSWORD=your_secure_password PRONOTE_PASSWORD=your_secure_password
PRONOTE_ENT=monbureaunumerique PRONOTE_ENT=monbureaunumerique

View File

@@ -27,6 +27,8 @@ class PronoteSettings(BaseSettings):
username: str | None = None username: str | None = None
password: SecretStr | None = None password: SecretStr | None = None
ent: str | None = None ent: str | None = None
pronote_url: str | None = None
account_type: Literal["student", "parent"] = "parent"
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto" agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
homework_source: Literal["auto", "ical", "pronotepy"] = "auto" homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
messages_source: Literal["pronotepy"] = "pronotepy" messages_source: Literal["pronotepy"] = "pronotepy"

View File

@@ -3,16 +3,20 @@
Ce module fournit l'encapsulation du client ``pronotepy`` pour la source Ce module fournit l'encapsulation du client ``pronotepy`` pour la source
Pronote : récupération des messages des professeurs, des informations et Pronote : récupération des messages des professeurs, des informations et
sondages, ainsi que des cours et devoirs en mode repli lorsque le flux sondages, ainsi que des cours et devoirs en mode repli lorsque le flux
iCal échoue. Toutes les erreurs sont journalisées avec des secrets masqués. iCal échoue. Les erreurs des méthodes dégradées (messages, informations)
sont journalisées avec des secrets masqués ; les erreurs de récupération
des cours et des devoirs se propagent pour déclencher le repli iCal.
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
from datetime import date from datetime import date
from typing import Protocol from typing import Any, Protocol
import pronotepy import pronotepy
import pronotepy.ent as pronotepy_ent
import requests
from pronote_sync.config.settings import PronoteSettings from pronote_sync.config.settings import PronoteSettings
from pronote_sync.models.agenda import Lesson, LessonStatus from pronote_sync.models.agenda import Lesson, LessonStatus
@@ -23,6 +27,69 @@ from pronote_sync.utils.redaction import redact_exception
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _get_ent_callable(name: str) -> Any:
"""Retourne le callable ``pronotepy`` associé à un nom d'ENT.
L'accès par :func:`getattr` évite les erreurs ``attr-defined`` de mypy
sur les attributs non exportés explicitement par ``pronotepy.ent``.
:param name: Nom de l'attribut dans ``pronotepy.ent``.
:return: Callable ``pronotepy`` associé.
:rtype: Any
"""
return getattr(pronotepy_ent, name)
_ENT_NAMES: list[str] = [
"monbureaunumerique",
"ent_elyco",
"bordeaux",
"ent_creuse",
"occitanie_montpellier",
"paris_classe_numerique",
"ile_de_france",
"ent_hdf",
"ac_orleans_tours",
"ac_poitiers",
"ac_rennes",
"laclasse_educonnect",
"ent77",
"ent_ecollege78",
"ent_essonne",
"val_doise",
"val_de_marne",
"ent_var",
"atrium_sud",
"laclasse_lyon",
"eclat_bfc",
"cas_arsene76",
"cas_ent27",
"cas_kosmos",
"ent_creuse_educonnect",
"ent_mayotte",
"ent_somme",
"ent_94",
"extranet_colleges_somme",
"ac_reunion",
]
_ENT_RESOLVERS: dict[str, Any] = {name: _get_ent_callable(name) for name in _ENT_NAMES}
def _resolve_ent(ent_name: str) -> Any:
"""Résout un nom d'ENT en callable ``pronotepy``.
:param ent_name: Nom de l'ENT tel que configuré (ex. ``"bordeaux"``).
:return: Callable ``pronotepy`` associé à l'ENT.
:raises ValueError: Si le nom d'ENT n'est pas reconnu.
"""
resolver = _ENT_RESOLVERS.get(ent_name)
if resolver is None:
supported = ", ".join(sorted(_ENT_RESOLVERS.keys()))
raise ValueError(f"ENT inconnu : {ent_name!r}. ENT supportés : {supported}")
return resolver
class PronoteClientProtocol(Protocol): class PronoteClientProtocol(Protocol):
"""Interface du client Pronote consommée par la logique de repli.""" """Interface du client Pronote consommée par la logique de repli."""
@@ -42,13 +109,23 @@ class PronoteClientProtocol(Protocol):
""" """
... ...
def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]: def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours et les devoirs via ``pronotepy`` (repli iCal). """Récupère les cours via ``pronotepy`` (repli iCal).
:param start: Date de début de la fenêtre (incluse). :param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse).
:return: Tuple ``(cours, devoirs)``. :return: Liste des cours.
:rtype: tuple[list[Lesson], list[Homework]] :rtype: list[Lesson]
"""
...
def get_homeworks(self, start: date, end: date) -> list[Homework]:
"""Récupère les devoirs via ``pronotepy``.
:param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse).
:return: Liste des devoirs.
:rtype: list[Homework]
""" """
... ...
@@ -56,11 +133,13 @@ class PronoteClientProtocol(Protocol):
class PronoteClient: class PronoteClient:
"""Client d'accès à Pronote via ``pronotepy``. """Client d'accès à Pronote via ``pronotepy``.
Encapsule ``pronotepy.Client`` avec une connexion paresseuse : la Encapsule ``pronotepy.Client`` ou ``pronotepy.ParentClient`` selon le
connexion n'est établie qu'à la première méthode de récupération type de compte, avec une connexion paresseuse : la connexion n'est
appelée. Les erreurs ``pronotepy.PronoteAPIError`` sont journalisées établie qu'à la première méthode de récupération appelée. Les erreurs
avec des secrets masqués et les méthodes de récupération retournent des méthodes dégradées (``get_messages()``, ``get_informations()``)
alors une valeur vide au lieu de propager l'exception. sont journalisées avec des secrets masqués et retournent une valeur
vide ; ``get_lessons()`` et ``get_homeworks()`` laissent les
exceptions se propager pour déclencher le repli iCal.
""" """
def __init__(self, settings: PronoteSettings) -> None: def __init__(self, settings: PronoteSettings) -> None:
@@ -75,24 +154,39 @@ class PronoteClient:
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse). """Crée et connecte le client ``pronotepy`` (connexion paresseuse).
Le client est créé une seule fois puis réutilisé pour les appels Le client est créé une seule fois puis réutilisé pour les appels
suivants. L'erreur de connexion est relancée sans journalisation, suivants. Le nom d'ENT est résolu via :func:`_resolve_ent` et le
la méthode publique appelante étant responsable de la journaliser. type de compte (``student`` ou ``parent``) détermine la classe de
client utilisée. L'erreur de connexion est relancée sans
journalisation, la méthode publique appelante étant responsable
de la journaliser.
:return: Le client ``pronotepy`` connecté. :return: Le client ``pronotepy`` connecté.
:rtype: pronotepy.Client :rtype: pronotepy.Client
:raises ValueError: Si ``username``, ``password`` ou ``ent`` est manquant. :raises ValueError: Si ``pronote_url``, ``username``, ``password``
ou ``ent`` est manquant, ou si l'ENT est inconnu.
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue. :raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
""" """
if self._client is None: if self._client is None:
pronote_url = self._settings.pronote_url
username = self._settings.username username = self._settings.username
password = self._settings.password password = self._settings.password
ent = self._settings.ent ent = self._settings.ent
if username is None or password is None or ent is None: if pronote_url is None or username is None or password is None or ent is None:
raise ValueError("username, password et ent sont requis pour pronotepy") raise ValueError(
try: "pronote_url, username, password et ent sont requis pour pronotepy"
self._client = pronotepy.Client(username, password.get_secret_value(), ent) )
except pronotepy.PronoteAPIError: resolver = _resolve_ent(ent)
raise client_class: type[pronotepy.Client] = (
pronotepy.ParentClient
if self._settings.account_type == "parent"
else pronotepy.Client
)
self._client = client_class(
pronote_url=pronote_url,
username=username,
password=password.get_secret_value(),
ent=resolver,
)
return self._client return self._client
def get_messages(self) -> list[Message]: def get_messages(self) -> list[Message]:
@@ -122,7 +216,13 @@ class PronoteClient:
) )
) )
return messages return messages
except (pronotepy.PronoteAPIError, ValueError) as exc: except (
pronotepy.PronoteAPIError,
ValueError,
requests.RequestException,
ConnectionError,
TimeoutError,
) as exc:
logger.error( logger.error(
"Échec de la récupération des messages Pronote : %s", "Échec de la récupération des messages Pronote : %s",
redact_exception(exc), redact_exception(exc),
@@ -154,62 +254,86 @@ class PronoteClient:
) )
) )
return messages return messages
except (pronotepy.PronoteAPIError, ValueError) as exc: except (
pronotepy.PronoteAPIError,
ValueError,
requests.RequestException,
ConnectionError,
TimeoutError,
) as exc:
logger.error( logger.error(
"Échec de la récupération des informations Pronote : %s", "Échec de la récupération des informations Pronote : %s",
redact_exception(exc), redact_exception(exc),
) )
return [] return []
def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]: def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours et les devoirs via ``pronotepy``. """Récupère les cours via ``pronotepy`` (repli iCal).
À utiliser uniquement si les sources iCal sont indisponibles ou en Les exceptions ne sont pas attrapées : elles se propagent afin que
repli automatique. Les cours annulés sont mappés sur le statut l'appelant puisse détecter l'échec et déclencher le repli (ou une
``CANCELLED`` ; **pronotepy** ne fournissant ni la date de erreur explicite).
:param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse).
:return: Liste des cours.
:rtype: list[Lesson]
: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.
"""
client = self._connect()
lessons: list[Lesson] = []
for lesson in client.lessons(start, end):
content = lesson.content
lessons.append(
Lesson(
id=lesson.id,
start=lesson.start,
end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "",
teachers=tuple(lesson.teacher_names or ()),
rooms=tuple(lesson.classrooms or ()),
group=lesson.group_name,
status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL),
content=content.description if content is not None else None,
)
)
return lessons
def get_homeworks(self, start: date, end: date) -> list[Homework]:
"""Récupère les devoirs via ``pronotepy``.
Les exceptions ne sont pas attrapées : elles se propagent afin que
l'appelant puisse détecter l'échec et déclencher le repli (ou une
erreur explicite). **pronotepy** ne fournissant ni la date de
distribution ni les professeurs des devoirs, ces champs restent distribution ni les professeurs des devoirs, ces champs restent
vides. vides.
:param start: Date de début de la fenêtre (incluse). :param start: Date de début de la fenêtre (incluse).
:param end: Date de fin de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse).
:return: Tuple ``(cours, devoirs)`` ; vide en cas d'erreur. :return: Liste des devoirs.
:rtype: tuple[list[Lesson], list[Homework]] :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.
""" """
try: client = self._connect()
client = self._connect() homeworks: list[Homework] = []
lessons: list[Lesson] = [] for hw in client.homework(start, end):
for lesson in client.lessons(start, end): homeworks.append(
content = lesson.content Homework(
lessons.append( id=hw.id,
Lesson( subject=hw.subject.name,
id=lesson.id, teachers=(),
start=lesson.start, assigned_on=None,
end=lesson.end, due_on=hw.date,
subject=lesson.subject.name if lesson.subject is not None else "", text=hw.description,
teachers=tuple(lesson.teacher_names or ()), html=hw.description,
rooms=tuple(lesson.classrooms or ()),
group=lesson.group_name,
status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL),
content=content.description if content is not None else None,
)
) )
homeworks: list[Homework] = []
for hw in client.homework(start, end):
homeworks.append(
Homework(
id=hw.id,
subject=hw.subject.name,
teachers=(),
assigned_on=None,
due_on=hw.date,
text=hw.description,
html=hw.description,
)
)
return lessons, homeworks
except (pronotepy.PronoteAPIError, ValueError) as exc:
logger.error(
"Échec de la récupération de l'agenda via pronotepy : %s",
redact_exception(exc),
) )
return [], [] return homeworks

View File

@@ -3,11 +3,14 @@
Ce module fournit l'enum :class:`AgendaSource`, le protocole Ce module fournit l'enum :class:`AgendaSource`, le protocole
:class:`PronoteFetcherProtocol` consommé par le pipeline ainsi que la :class:`PronoteFetcherProtocol` consommé par le pipeline ainsi que la
classe :class:`PronoteFetcher` qui sélectionne la source selon la classe :class:`PronoteFetcher` qui sélectionne la source selon la
configuration (``PRONOTE_AGENDA_SOURCE`` / ``PRONOTE_HOMEWORK_SOURCE``) configuration (``PRONOTE_AGENDA_SOURCE`` / ``PRONOTE_HOMEWORK_SOURCE``).
avec repli automatique iCal → pronotepy en mode ``auto``. Les messages La logique applique un repli **unique** : une source primaire est
essayée en premier et, en cas d'échec, une seule source de repli (jamais
réciproque ni itératif) est essayée si elle est configurée. Les messages
et informations proviennent toujours de pronotepy. Toutes les erreurs et informations proviennent toujours de pronotepy. Toutes les erreurs
sont journalisées avec des secrets masqués via sont journalisées avec des secrets masqués via
:func:`~pronote_sync.utils.redaction.redact_exception`. :func:`~pronote_sync.utils.redaction.redact_exception` ; les exceptions
d'origine ne sont jamais chaînées (``from None``).
""" """
from __future__ import annotations from __future__ import annotations
@@ -15,7 +18,7 @@ from __future__ import annotations
import logging import logging
from datetime import date, timedelta from datetime import date, timedelta
from enum import StrEnum from enum import StrEnum
from typing import Protocol from typing import Literal, Protocol
from pronote_sync.config.settings import Settings from pronote_sync.config.settings import Settings
from pronote_sync.errors import PipelineCriticalError from pronote_sync.errors import PipelineCriticalError
@@ -28,6 +31,8 @@ from pronote_sync.utils.redaction import redact_exception
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_SourceName = Literal["ical", "pronotepy"]
class AgendaSource(StrEnum): class AgendaSource(StrEnum):
"""Source configurée pour la récupération de l'agenda et des devoirs.""" """Source configurée pour la récupération de l'agenda et des devoirs."""
@@ -77,12 +82,13 @@ class PronoteFetcherProtocol(Protocol):
class PronoteFetcher: class PronoteFetcher:
"""Récupère les données Pronote via iCal ou pronotepy avec repli. """Récupère les données Pronote via iCal ou pronotepy avec repli unique.
Unifie les sources iCal et pronotepy selon la source configurée Unifie les sources iCal et pronotepy selon la source configurée
(``agenda_source`` / ``homework_source``) : en mode ``AUTO``, le flux (``agenda_source`` / ``homework_source``) : la source primaire est
iCal est essayé en premier et pronotepy sert de repli. Les messages et essayée en premier et, si elle échoue, une seule source de repli est
informations proviennent toujours de pronotepy. essayée lorsqu'elle est configurée. Les messages et informations
proviennent toujours de pronotepy.
""" """
def __init__(self, settings: Settings, pronote_client: PronoteClientProtocol) -> None: def __init__(self, settings: Settings, pronote_client: PronoteClientProtocol) -> None:
@@ -105,6 +111,29 @@ class PronoteFetcher:
end = today + timedelta(days=self._settings.app.sync_future_days) end = today + timedelta(days=self._settings.app.sync_future_days)
return start, end 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]]: def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère l'agenda depuis le flux iCal. """Récupère l'agenda depuis le flux iCal.
@@ -129,70 +158,101 @@ class PronoteFetcher:
:return: Tuple ``(cours, événements scolaires)``. :return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]] :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() start, end = self._fetch_window()
lessons, _ = self._pronote_client.get_agenda_fallback(start, end) lessons = self._pronote_client.get_lessons(start, end)
return lessons, [] return lessons, []
def _agenda_sources(self) -> tuple[_SourceName, _SourceName | None]:
"""Sélectionne la source primaire et le repli unique pour l'agenda.
En mode ``AUTO``, iCal est primaire si ``ical_url`` est configuré
(repli pronotepy si la configuration pronotepy est complète),
sinon pronotepy sans repli. Les modes explicites ``ICAL`` et
``PRONOTEPY`` désignent la source primaire et tentent l'autre
source en repli unique si elle est configurée.
: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", "pronotepy" if self._is_pronotepy_configured() else None
if source is AgendaSource.PRONOTEPY:
return "pronotepy", "ical" if self._is_ical_configured() else 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]]: def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère les cours et les événements scolaires selon la source configurée. """Récupère les cours et les événements scolaires selon la source configurée.
En mode ``AUTO``, iCal est essayé en premier et pronotepy sert de La source primaire est essayée en premier ; si elle échoue, la
repli ; si les deux sources échouent, une erreur critique est levée. source de repli unique (l'autre source, si configurée) est
essayée. Si la source primaire et le repli échouent — ou si
aucune source n'est configurée en mode ``AUTO`` — une erreur
critique est levée.
:return: Tuple ``(cours, événements scolaires)``. :return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]] :rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises PipelineCriticalError: Si toutes les sources configurées échouent. :raises PipelineCriticalError: Si toutes les sources tentées échouent.
""" """
source = AgendaSource(self._settings.pronote.agenda_source) primary, fallback = self._agenda_sources()
if source is AgendaSource.ICAL:
try:
return self._fetch_agenda_ical()
except Exception as exc:
logger.error(
"Échec de la récupération iCal pour l'agenda : %s",
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer l'agenda : la source iCal a échoué"
) from exc
if source is AgendaSource.PRONOTEPY:
try:
return self._fetch_agenda_pronotepy()
except Exception as exc:
logger.error(
"Échec de la récupération pronotepy pour l'agenda : %s",
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer l'agenda : la source pronotepy a échoué"
) from exc
# Mode AUTO : essayer iCal d'abord, puis replier sur pronotepy.
try: try:
return self._fetch_agenda_ical() return self._fetch_agenda_source(primary)
except Exception as exc:
logger.warning(
"Échec de la récupération iCal pour l'agenda : %s",
redact_exception(exc),
)
logger.info("Repli sur pronotepy pour l'agenda.")
try:
lessons, school_events = self._fetch_agenda_pronotepy()
except Exception as exc: except Exception as exc:
logger.error( logger.error(
"Échec de la récupération pronotepy pour l'agenda : %s", "Échec de la récupération %s pour l'agenda : %s",
primary,
redact_exception(exc), redact_exception(exc),
) )
raise PipelineCriticalError( if fallback is None:
"Impossible de récupérer l'agenda : les sources iCal et pronotepy ont échoué" raise PipelineCriticalError(
) from exc f"Impossible de récupérer l'agenda : la source {primary} a échoué"
if not lessons: ) from None
logger.warning( logger.info("Repli sur %s pour l'agenda.", fallback)
"Le repli pronotepy pour l'agenda a retourné un résultat vide après l'échec " try:
"d'iCal : impossible de distinguer une absence de cours d'un échec silencieux." lessons, school_events = self._fetch_agenda_source(fallback)
) except Exception as exc:
return lessons, school_events 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]: def _fetch_homework_ical(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis le flux iCal pour la date cible. """Récupère les devoirs depuis le flux iCal pour la date cible.
@@ -207,76 +267,112 @@ class PronoteFetcher:
lessons, _ = self._fetch_agenda_ical() lessons, _ = self._fetch_agenda_ical()
return collect_homeworks(lessons, target_date) return collect_homeworks(lessons, target_date)
def _fetch_homework_pronotepy(self) -> list[Homework]: def _fetch_homework_pronotepy(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis pronotepy. """Récupère les devoirs depuis pronotepy pour la date cible.
:return: Liste des devoirs. Les devoirs sont filtrés sur la date d'échéance : seuls ceux dont
:rtype: list[Homework] ``due_on`` correspond à ``target_date`` sont conservés.
"""
start, end = self._fetch_window()
_, homeworks = self._pronote_client.get_agenda_fallback(start, end)
return homeworks
def fetch_homework(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs selon la source configurée.
En mode ``AUTO``, iCal est essayé en premier et pronotepy sert de
repli ; si les deux sources échouent, une erreur critique est levée.
:param target_date: Date cible pour laquelle collecter les devoirs. :param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs. :return: Liste des devoirs.
:rtype: list[Homework] :rtype: list[Homework]
:raises PipelineCriticalError: Si toutes les sources configurées échouent. :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.
En mode ``AUTO``, iCal est primaire si ``ical_url`` est configuré
(repli pronotepy si la configuration pronotepy est complète),
sinon pronotepy sans repli. Les modes explicites ``ICAL`` et
``PRONOTEPY`` désignent la source primaire et tentent l'autre
source en repli unique si elle est configurée.
: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) source = AgendaSource(self._settings.pronote.homework_source)
if source is AgendaSource.ICAL: if source is AgendaSource.ICAL:
try: return "ical", "pronotepy" if self._is_pronotepy_configured() else None
return self._fetch_homework_ical(target_date)
except Exception as exc:
logger.error(
"Échec de la récupération iCal pour les devoirs : %s",
redact_exception(exc),
)
raise PipelineCriticalError(
"Impossible de récupérer les devoirs : la source iCal a échoué"
) from exc
if source is AgendaSource.PRONOTEPY: if source is AgendaSource.PRONOTEPY:
try: return "pronotepy", "ical" if self._is_ical_configured() else None
return self._fetch_homework_pronotepy() if self._is_ical_configured():
except Exception as exc: return "ical", "pronotepy" if self._is_pronotepy_configured() else None
logger.error( if self._is_pronotepy_configured():
"Échec de la récupération pronotepy pour les devoirs : %s", return "pronotepy", None
redact_exception(exc), raise PipelineCriticalError(
) "Impossible de récupérer les devoirs : ni la source iCal ni pronotepy n'est configurée"
raise PipelineCriticalError( ) from None
"Impossible de récupérer les devoirs : la source pronotepy a échoué"
) from exc
# Mode AUTO : essayer iCal d'abord, puis replier sur pronotepy. def _fetch_homework_source(self, name: _SourceName, target_date: date) -> list[Homework]:
try: """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_ical(target_date)
except Exception as exc: return self._fetch_homework_pronotepy(target_date)
logger.warning(
"Échec de la récupération iCal pour les devoirs : %s", def fetch_homework(self, target_date: date) -> list[Homework]:
redact_exception(exc), """Récupère les devoirs selon la source configurée.
)
logger.info("Repli sur pronotepy pour les devoirs.") La source primaire est essayée en premier ; si elle échoue, la
source de repli unique (l'autre source, si configurée) est
essayée. Si la source primaire et le repli échouent — ou si
aucune source n'est configurée en mode ``AUTO`` — 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: try:
homeworks = self._fetch_homework_pronotepy() return self._fetch_homework_source(primary, target_date)
except Exception as exc: except Exception as exc:
logger.error( logger.error(
"Échec de la récupération pronotepy pour les devoirs : %s", "Échec de la récupération %s pour les devoirs : %s",
primary,
redact_exception(exc), redact_exception(exc),
) )
raise PipelineCriticalError( if fallback is None:
"Impossible de récupérer les devoirs : les sources iCal et pronotepy ont échoué" raise PipelineCriticalError(
) from exc f"Impossible de récupérer les devoirs : la source {primary} a échoué"
if not homeworks: ) from None
logger.warning( logger.info("Repli sur %s pour les devoirs.", fallback)
"Le repli pronotepy pour les devoirs a retourné un résultat vide après l'échec " try:
"d'iCal : impossible de distinguer une absence de devoirs d'un échec silencieux." homeworks = self._fetch_homework_source(fallback, target_date)
) except Exception as exc:
return homeworks 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]: def fetch_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote (toujours via pronotepy). """Récupère les messages des discussions Pronote (toujours via pronotepy).
@@ -284,7 +380,14 @@ class PronoteFetcher:
:return: Liste des messages. :return: Liste des messages.
:rtype: list[Message] :rtype: list[Message]
""" """
return self._pronote_client.get_messages() 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]: def fetch_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote (toujours via pronotepy). """Récupère les informations et sondages Pronote (toujours via pronotepy).
@@ -292,4 +395,11 @@ class PronoteFetcher:
:return: Liste des informations et sondages. :return: Liste des informations et sondages.
:rtype: list[Message] :rtype: list[Message]
""" """
return self._pronote_client.get_informations() 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

View File

@@ -218,21 +218,23 @@ def _parse_french_date(value: str) -> date | None:
return None return None
def parse_body(body: str) -> tuple[str | None, dict[date, str], dict[date, str]]: def parse_body(body: str) -> tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]]:
"""Parse le corps HTML pour extraire contenu pédagogique et devoirs. """Parse le corps HTML pour extraire contenu pédagogique et devoirs.
Le contenu est extrait de la section ``<strong>Contenu pédagogique :</strong>``. Le contenu est extrait de la section ``<strong>Contenu pédagogique :</strong>``.
Les devoirs à faire sont extraits des sections ``<strong>Pour le JJ/MM/AAAA :</strong>`` Les devoirs à faire sont extraits des sections ``<strong>Pour le JJ/MM/AAAA :</strong>``
(dict date texte) et les devoirs donnés des sections (liste de tuples ``(date, texte)`` dans l'ordre du flux) et les devoirs donnés
``<strong>Donné le JJ/MM/AAAA :</strong>`` (dict date → texte). des sections ``<strong>Donné le JJ/MM/AAAA :</strong>`` (liste de tuples
``(date, texte)``). Les listes préservent tous les blocs, même lorsque plusieurs
sections partagent la même date.
:param body: Corps HTML (à partir du premier ``<strong>``). :param body: Corps HTML (à partir du premier ``<strong>``).
:return: Tuple ``(contenu pédagogique, devoirs dus, devoirs donnés)``. :return: Tuple ``(contenu pédagogique, devoirs dus, devoirs donnés)``.
:rtype: tuple[str | None, dict[date, str], dict[date, str]] :rtype: tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]]
""" """
content: str | None = None content: str | None = None
due_blocks: dict[date, str] = {} due_blocks: list[tuple[date, str]] = []
assigned_blocks: dict[date, str] = {} assigned_blocks: list[tuple[date, str]] = []
content_match = _CONTENT_PATTERN.search(body) content_match = _CONTENT_PATTERN.search(body)
if content_match is not None: if content_match is not None:
@@ -241,34 +243,35 @@ def parse_body(body: str) -> tuple[str | None, dict[date, str], dict[date, str]]
for match in _DUE_PATTERN.finditer(body): for match in _DUE_PATTERN.finditer(body):
due_date = _parse_french_date(match.group(1)) due_date = _parse_french_date(match.group(1))
if due_date is not None: if due_date is not None:
due_blocks[due_date] = _strip_html(match.group(2)) due_blocks.append((due_date, _strip_html(match.group(2))))
for match in _ASSIGNED_PATTERN.finditer(body): for match in _ASSIGNED_PATTERN.finditer(body):
assigned_date = _parse_french_date(match.group(1)) assigned_date = _parse_french_date(match.group(1))
if assigned_date is not None: if assigned_date is not None:
assigned_blocks[assigned_date] = _strip_html(match.group(2)) assigned_blocks.append((assigned_date, _strip_html(match.group(2))))
return content, due_blocks, assigned_blocks return content, due_blocks, assigned_blocks
def parse_homework_blocks( def parse_homework_blocks(
due_blocks: dict[date, str], due_blocks: list[tuple[date, str]],
assigned_blocks: dict[date, str], assigned_blocks: list[tuple[date, str]],
) -> tuple[HomeworkBlock, ...]: ) -> tuple[HomeworkBlock, ...]:
"""Construit les :class:`HomeworkBlock` depuis les dicts de devoirs. """Construit les :class:`HomeworkBlock` depuis les listes de devoirs.
Les blocs dus (``kind="due"``) précèdent les blocs donnés Les blocs dus (``kind="due"``) précèdent les blocs donnés
(``kind="assigned"``), dans l'ordre d'insertion des dicts. (``kind="assigned"``), dans l'ordre des listes. Tous les blocs
sont préservés, y compris lorsque plusieurs partagent la même date.
:param due_blocks: Dict date → texte des devoirs à faire. :param due_blocks: Liste de tuples ``(date, texte)`` des devoirs à faire.
:param assigned_blocks: Dict date → texte des devoirs donnés. :param assigned_blocks: Liste de tuples ``(date, texte)`` des devoirs donnés.
:return: Tuple de blocs de devoirs pour le cours. :return: Tuple de blocs de devoirs pour le cours.
:rtype: tuple[HomeworkBlock, ...] :rtype: tuple[HomeworkBlock, ...]
""" """
blocks: list[HomeworkBlock] = [] blocks: list[HomeworkBlock] = []
for due_date, text in due_blocks.items(): for due_date, text in due_blocks:
blocks.append(HomeworkBlock(kind="due", date=due_date, text=text, html=text)) blocks.append(HomeworkBlock(kind="due", date=due_date, text=text, html=text))
for assigned_date, text in assigned_blocks.items(): for assigned_date, text in assigned_blocks:
blocks.append(HomeworkBlock(kind="assigned", date=assigned_date, text=text, html=text)) blocks.append(HomeworkBlock(kind="assigned", date=assigned_date, text=text, html=text))
return tuple(blocks) return tuple(blocks)
@@ -360,9 +363,10 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School
Les VEVENT de vacances/congés (tout le jour) deviennent des Les VEVENT de vacances/congés (tout le jour) deviennent des
:class:`SchoolEvent` de type ``holiday``. Les VEVENT horodatés :class:`SchoolEvent` de type ``holiday``. Les VEVENT horodatés
deviennent des :class:`Lesson` dont le statut dérive de la deviennent des :class:`Lesson` dont le statut dérive de la propriété
catégorie (``Cours - Cours annulé`` → ``CANCELLED``, ``STATUS`` (``CANCELLED`` → ``CANCELLED``) et de la catégorie
``Cours - Cours déplacé`` → ``MOVED``). Les UID sont normalisés ; (``Cours - Cours annulé`` → ``CANCELLED``, ``Cours - Cours déplacé`` →
``MOVED``). Les UID sont normalisés ;
un événement sans UID reçoit un UID déterministe généré à partir un événement sans UID reçoit un UID déterministe généré à partir
de ses champs clés (début, fin, matière, enseignants, salles, groupe). de ses champs clés (début, fin, matière, enseignants, salles, groupe).
@@ -416,11 +420,14 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School
if not isinstance(start, datetime) or not isinstance(end, datetime): if not isinstance(start, datetime) or not isinstance(end, datetime):
continue continue
status = LessonStatus.NORMAL status_obj = component.get("status")
if "Cours - Cours annulé" in categories: status_value = str(status_obj).strip().upper() if status_obj is not None else ""
if status_value == "CANCELLED" or "Cours - Cours annulé" in categories:
status = LessonStatus.CANCELLED status = LessonStatus.CANCELLED
elif "Cours - Cours déplacé" in categories: elif "Cours - Cours déplacé" in categories:
status = LessonStatus.MOVED status = LessonStatus.MOVED
else:
status = LessonStatus.NORMAL
description = component.get("description") description = component.get("description")
description_str = str(description) if description is not None else "" description_str = str(description) if description is not None else ""

View File

@@ -26,6 +26,7 @@ from pronote_sync.sources.pronote.ical import (
generate_homework_id, generate_homework_id,
get_calendar_name, get_calendar_name,
normalize_homework_text, normalize_homework_text,
parse_body,
parse_ical, parse_ical,
) )
@@ -463,3 +464,137 @@ def test_generate_homework_id_deterministic() -> None:
id1 = generate_homework_id(due_on, text) id1 = generate_homework_id(due_on, text)
id2 = generate_homework_id(due_on, text) id2 = generate_homework_id(due_on, text)
assert id1 == id2 assert id1 == id2
CANCELLED_STATUS_ICAL = """BEGIN:VCALENDAR
VERSION:2.0
X-WR-CALNAME:Test
BEGIN:VEVENT
UID:Test-123-20260906T120000Z-Index-Education
DTSTART:20260907T080000Z
DTEND:20260907T090000Z
SUMMARY:Test Course
STATUS:CANCELLED
DESCRIPTION:<div></div>
END:VEVENT
END:VCALENDAR"""
NORMAL_STATUS_ICAL = """BEGIN:VCALENDAR
VERSION:2.0
X-WR-CALNAME:Test
BEGIN:VEVENT
UID:Test-123-20260906T120000Z-Index-Education
DTSTART:20260907T080000Z
DTEND:20260907T090000Z
SUMMARY:Test Course
STATUS:CONFIRMED
DESCRIPTION:<div></div>
END:VEVENT
END:VCALENDAR"""
MOVED_BY_CATEGORY_ICAL = """BEGIN:VCALENDAR
VERSION:2.0
X-WR-CALNAME:Test
BEGIN:VEVENT
UID:Test-123-20260906T120000Z-Index-Education
DTSTART:20260907T080000Z
DTEND:20260907T090000Z
SUMMARY:Test Course
CATEGORIES:Cours - Cours déplacé
DESCRIPTION:<div></div>
END:VEVENT
END:VCALENDAR"""
MULTIPLE_BLOCKS_SAME_DATE_ICAL = """BEGIN:VCALENDAR
VERSION:2.0
X-WR-CALNAME:Test
BEGIN:VEVENT
UID:Test-123-20260906T120000Z-Index-Education
DTSTART:20260907T080000Z
DTEND:20260907T090000Z
SUMMARY:Test Course
CATEGORIES:Cours
DESCRIPTION:<div>
Matière : Math
Professeur : M. Dupont
Salle : 204
<strong>Pour le 10/09/2026 :</strong>
Exercice 1 à 5 page 42.
<strong>Pour le 10/09/2026 :</strong>
Exercice 6 à 10 page 43.
</div>
END:VEVENT
END:VCALENDAR"""
def test_parse_ical_status_cancelled_only() -> None:
"""Un cours avec STATUS:CANCELLED mais sans CATEGORIES contenant 'Cours annulé' a status == LessonStatus.CANCELLED.
:return: None
"""
lessons, _, _ = parse_ical(CANCELLED_STATUS_ICAL)
assert len(lessons) == 1
assert lessons[0].status == LessonStatus.CANCELLED
def test_parse_ical_status_normal_without_cancel() -> None:
"""Un cours avec STATUS:CONFIRMED (ou sans STATUS) a status == LessonStatus.NORMAL.
:return: None
"""
lessons, _, _ = parse_ical(NORMAL_STATUS_ICAL)
assert len(lessons) == 1
assert lessons[0].status == LessonStatus.NORMAL
def test_parse_ical_moved_by_category_only() -> None:
"""Un cours avec CATEGORIES:Cours - Cours déplacé et sans STATUS a status == LessonStatus.MOVED.
:return: None
"""
lessons, _, _ = parse_ical(MOVED_BY_CATEGORY_ICAL)
assert len(lessons) == 1
assert lessons[0].status == LessonStatus.MOVED
def test_parse_body_multiple_blocks_same_date() -> None:
"""Un DESCRIPTION avec deux sections 'Pour le' à la même date conserve les deux blocs.
:return: None
"""
body_html = (
"<div>\n"
" <strong>Pour le 10/09/2026:</strong>\n"
" Exercice 1 à 5 page 42.\n"
" <strong>Pour le 10/09/2026:</strong>\n"
" Exercice 6 à 10 page 43.\n"
"</div>"
)
content, due_blocks, assigned_blocks = parse_body(body_html)
assert len(due_blocks) == 2
assert due_blocks[0][0] == date(2026, 9, 10)
assert due_blocks[0][1] == "Exercice 1 à 5 page 42."
assert due_blocks[1][0] == date(2026, 9, 10)
assert due_blocks[1][1] == "Exercice 6 à 10 page 43."
def test_collect_homeworks_from_fixture() -> None:
"""Parse le fixture pronote-4e.ics, appelle collect_homeworks pour le 10/09/2026 et vérifie qu'au moins un devoir est retourné.
:return: None
"""
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
with open(fixture_path, encoding="utf-8") as f:
content = f.read()
lessons, _, _ = parse_ical(content)
homeworks = collect_homeworks(lessons, date(2026, 9, 10))
assert len(homeworks) >= 1
# Vérifie qu'au moins un devoir a le bon sujet et texte
assert any(hw.subject == "Mathématiques" for hw in homeworks)
assert any("Exercices 1 à 5 page 42" in hw.text for hw in homeworks)