AntoineVe/college-infos
1
0
Fork 0
Code Issues 13 Pull Requests 1 Actions Packages Projects Releases 2 Wiki Activity

docs: aligner les jalons sur le contrat M4

Browse Source 
Co-authored-by: codex/gpt-5.6-sol <codex-gpt-5.6-sol@agents.invalid>
This commit is contained in:
Antoine Van Elstraete
2026-09-06 14:59:06 +02:00
parent 292aef4f5b
commit d0e1f92d7f
2 changed files with 190 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files

564
GUIDE_DEV_PYTHON.md
View File

@@ -47,8 +47,10 @@ Le projet doit implémenter les fonctionnalités suivantes, dans l'ordre logique
- **Idempotence** : Deux exécutions identiques sans changement externe **doivent** produire le même résultat (aucune modification en base ou CalDAV).
- **Mode dégradé** :
- Si la synthèse IA échoue → envoyer le message **sans synthèse** (mais avec la liste brute des devoirs).
- Si `pronotepy` échoue → basculer sur iCal (si disponible) pour l'agenda/devoirs.
- Si iCal et `pronotepy` échouent → **échec explicite** avec message clair.
- En mode source `auto`, si iCal échoue → basculer sur `pronotepy` pour
l'agenda/devoirs.
- En mode source explicite (`ical` ou `pronotepy`), ne pas basculer silencieusement.
- En mode `auto`, si iCal et `pronotepy` échouent → **échec explicite** avec message clair.
---
@@ -260,15 +262,15 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
| Variable | Description | Exemple (anonymisé) | Type |
|------------------------------|-----------------------------------------------------------------------------|---------------------------------------------|---------------|
| `PRONOTE_ICAL_URL` | URL du flux iCal Pronote (contient `icalsecurise`). | `https://college.ent/pronote/ical/...` | `str` |
| `PRONOTE_URL` | URL de la page Pronote utilisée par `pronotepy` (page parent). | `https://college.ent/pronote/parent.html` | `str` |
| `PRONOTE_ICAL_URL` | URL du flux iCal Pronote (contient `icalsecurise`). | `https://college.ent/pronote/ical/...` | `SecretStr` |
| `PRONOTE_USERNAME` | Identifiant Pronote (si `pronotepy` utilisé). | `parent.dupont` | `str` |
| `PRONOTE_PASSWORD` | Mot de passe Pronote (si `pronotepy` utilisé). | `SecretStr` (masqué) | `SecretStr` |
| `PRONOTE_ENT` | ENT Pronote (ex: `monbureaunumerique`, `atrium`). | `monbureaunumerique` | `str` |
| `PRONOTE_ENT` | Slug ENT supporté, résolu vers une fonction de `pronotepy.ent`. | `monbureaunumerique` | `str` |
| `CALDAV_URL` | URL du serveur CalDAV. | `https://caldav.example.com/calendars/...` | `str` |
| `CALDAV_USERNAME` | Identifiant CalDAV. | `user@example.com` | `str` |
| `CALDAV_PASSWORD` | Mot de passe CalDAV. | `SecretStr` (masqué) | `SecretStr` |
| `CALDAV_CALENDAR_PATH` | Chemin du calendrier CalDAV de destination. | `/pronote-sync/` | `str` |
| `CALDAV_CALENDAR_PATH` | Chemin du calendrier CalDAV de destination. | `/pronote-sync/` | `str` |
| `XMPP_JID` | Identifiant XMPP (ex: `user@example.com`). | `user@example.com` | `str` |
| `XMPP_PASSWORD` | Mot de passe XMPP. | `SecretStr` (masqué) | `SecretStr` |
| `XMPP_RECIPIENT` | Destinataire XMPP (ex: `parent@example.com`). | `parent@example.com` | `str` |
@@ -278,6 +280,19 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
> Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`.
> Une section `BLOG_ENABLED` et `BLOG_RSS_URL` a été ajoutée dans `.env.example`.
Les variables Pronote sont obligatoires selon les sources activées :
- la source iCal exige `PRONOTE_ICAL_URL` ;
- la source `pronotepy` exige `PRONOTE_URL`, `PRONOTE_USERNAME` et
`PRONOTE_PASSWORD` ;
- `PRONOTE_ENT` reste optionnel pour une connexion directe, mais, s'il est fourni, son slug doit
appartenir à une liste fermée et être résolu vers la fonction correspondante de `pronotepy.ent`.
`PRONOTE_URL` et `PRONOTE_ICAL_URL` sont deux contrats distincts : l'un ne doit jamais être déduit
de l'autre. Le cas d'usage actuel est un compte parent ; le client à construire est donc
`pronotepy.ParentClient`. Une généralisation à plusieurs profils ne sera ajoutée qu'en présence
d'un besoin réel et testé.
#### 3.1.2 Variables optionnelles
| Variable | Description | Valeur par défaut | Type |
@@ -305,6 +320,7 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
```ini
# --- Pronote ---
PRONOTE_URL=https://college.ent/pronote/parent.html
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
PRONOTE_USERNAME=parent.dupont
PRONOTE_PASSWORD=your_secure_password
@@ -358,17 +374,18 @@ LOG_LEVEL=INFO
> `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"` (et non `"pronote-digest"`).
```python
from typing import Literal, Optional
from typing import Literal
from pydantic import SecretStr, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class PronoteSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="PRONOTE_", env_file=".env", extra="ignore")
ical_url: Optional[str] = None
username: Optional[str] = None
password: Optional[SecretStr] = None
ent: Optional[str] = None
url: str | None = None
ical_url: SecretStr | None = None
username: str | None = None
password: SecretStr | None = None
ent: str | None = None
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
messages_source: Literal["pronotepy"] = "pronotepy"
@@ -402,11 +419,11 @@ class AppSettings(BaseSettings):
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
pronote: PronoteSettings = PronoteSettings()
caldav: CalDAVSettings = CalDAVSettings()
xmpp: XmppSettings = XmppSettings()
ai: AISettings = AISettings()
app: AppSettings = AppSettings()
pronote: PronoteSettings = Field(default_factory=PronoteSettings)
caldav: CalDAVSettings = Field(default_factory=CalDAVSettings)
xmpp: XmppSettings = Field(default_factory=XmppSettings)
ai: AISettings = Field(default_factory=AISettings)
app: AppSettings = Field(default_factory=AppSettings)
settings = Settings()
@@ -431,47 +448,20 @@ settings = Settings()
**Note importante** : Tous les messages d'erreur externes (HTTP, Pronote, CalDAV, XMPP, IA) **doivent** être systématiquement expurgés des secrets avant journalisation ou réémission. Utiliser `redact_secrets(str(e))` ou `redact_exception()` pour les logs.
**Tests négatifs recommandés** :
- Vérifier que les messages d'erreur ne contiennent ni `icalsecurise`, ni `SECRET`, ni mots de passe, ni clés API.
- Exemple de test :
```python
def test_error_messages_do_not_leak_secrets():
"""Vérifie que les messages d'erreur ne fuient pas de secrets."""
from pronote_sync.utils.redaction import redact_secrets
from pronote_sync.sources.pronote.ical import fetch_ical
# Simuler une URL avec token
bad_url = "https://example.com/ical?icalsecurise=SECRET_TOKEN"
try:
fetch_ical(bad_url)
except Exception as e:
error_msg = str(e)
assert "SECRET_TOKEN" not in error_msg
assert "icalsecurise" not in error_msg.lower()
```
**Tests négatifs recommandés** :
- Vérifier que les tokens (`icalsecurise`), clés API et URLs ne apparaissent **jamais** dans les logs ou les messages d'erreur, même en cas d'exception.
- Exemple de test :
```python
def test_error_messages_do_not_leak_secrets():
"""Vérifie que les messages d'erreur ne fuient pas de secrets."""
from pronote_sync.utils.redaction import redact_secrets
from pronote_sync.sources.pronote.ical import fetch_ical
# Simuler une URL avec token
bad_url = "https://example.com/ical?icalsecurise=SECRET_TOKEN"
try:
fetch_ical(bad_url)
except Exception as e:
error_msg = str(e)
assert "SECRET_TOKEN" not in error_msg
assert "icalsecurise" not in error_msg.lower()
```
**Tests négatifs obligatoires** : injecter des sentinelles distinctes dans l'URL, les identifiants,
le mot de passe et la clé API, provoquer une erreur externe, puis vérifier leur absence dans le
message, les logs, `__cause__`, `__context__` et le traceback formaté.
**Application systématique** :
Toutes les exceptions externes (HTTP, Pronote, CalDAV, XMPP, IA) **doivent** être traitées avec `redact_secrets()` ou `redact_exception()` avant toute journalisation ou réémission.
Le masquage du message extérieur ne suffit pas si l'exception brute reste chaînée dans
`__cause__` ou `__context__` : un traceback complet pourrait alors révéler l'URL ou les
identifiants d'origine. À la frontière avec une bibliothèque externe, journaliser uniquement la
version expurgée puis lever l'exception applicative avec `raise ... from None`, ou chaîner une
cause elle-même expurgée. Les tests de non-fuite doivent inspecter `str(exc)`, les logs, la cause,
le contexte et le traceback complet.
#### 4.2.1 Masquage des URLs (`redaction.py`)
> ⚠️ **Décision d'implémentation** :
@@ -1349,6 +1339,40 @@ def _format_message(self, message: XmppMessage) -> str:
### 5.1 Flux iCal Pronote
#### Contrat normatif des sources Pronote (M4 et jalons suivants)
Les règles ci-dessous priment sur les exemples historiques de cette section :
1. `parse_ical()` retourne les cours et événements scolaires. Sa liste de `Homework` reste vide :
les blocs bruts conservés dans chaque `Lesson` sont transformés ensuite par
`collect_homeworks(lessons, target_date)`.
2. Les blocs de devoirs sont conservés dans une séquence. Une structure `date -> texte` est
interdite, car plusieurs devoirs peuvent partager la même date. La déduplication ne s'effectue
qu'au moment de `collect_homeworks`.
3. Un cours est annulé si `STATUS:CANCELLED` **ou** la catégorie Pronote correspondante est
présente. Le statut déplacé est détecté par sa catégorie.
4. Le client `pronotepy` reçoit l'URL Pronote en premier argument, puis les identifiants, avec une
fonction ENT résolue depuis une liste fermée. Pour le compte parent actuellement visé, utiliser
`pronotepy.ParentClient(pronote_url, username, password, ent=ent_function)`.
5. Le client expose séparément la récupération des cours et celle des devoirs. Les devoirs
`pronotepy` sont filtrés strictement sur `due_on == target_date` avant d'être retournés au
pipeline.
6. Une liste vide est un résultat valide ; une exception signale un échec de source. Les méthodes
critiques d'agenda et de devoirs propagent donc une erreur expurgée au `PronoteFetcher`. Les
messages et informations, non critiques, peuvent se dégrader en listes vides accompagnées d'un
warning.
7. En mode `auto`, iCal est essayé en premier puis `pronotepy` sert de repli. Les modes explicites
`ical` et `pronotepy` sont stricts et ne changent pas silencieusement de source. En mode `auto`,
l'échec des deux sources lève `PipelineCriticalError`.
8. Pendant une exécution du pipeline, un flux iCal déjà téléchargé et parsé est réutilisé pour
l'agenda et les devoirs. Ce partage reste limité à l'exécution courante : aucun cache global ou
persistant n'est nécessaire.
Avant M7, une fixture anonymisée doit confirmer que deux événements équivalents provenant d'iCal
et de `pronotepy` aboutissent au même identifiant canonique. Si ce n'est pas le cas, la
normalisation doit être corrigée à la frontière des sources avant toute synchronisation CalDAV ;
ne pas introduire de moteur de rapprochement complexe sans données qui le justifient.
#### 5.1.1 Observations sur les flux réels
Les flux iCal générés par Pronote ont des **spécificités importantes** à prendre en compte, basées sur l'analyse du projet TypeScript `pronote-digest` :
@@ -1551,7 +1575,7 @@ def fetch_ical(url: str, timeout: int = 20) -> str:
safe_error = redact_secrets(str(e))
raise requests.exceptions.RequestException(
f"Échec de la récupération de {safe_url}: {safe_error}"
) from e
) from None
# Validation du flux
if "BEGIN:VCALENDAR" not in content:
@@ -2073,7 +2097,8 @@ def parse_ical(raw_ical: str) -> tuple[List[Lesson], List[HomeworkModel], List[S
# Cours annulés ou déplacés
status = LessonStatus.NORMAL
if "Cours - Cours annulé" in categories:
ical_status = str(component.get("status", "")).upper()
if ical_status == "CANCELLED" or "Cours - Cours annulé" in categories:
status = LessonStatus.CANCELLED
elif "Cours - Cours déplacé" in categories:
status = LessonStatus.MOVED
@@ -2111,274 +2136,72 @@ def parse_ical(raw_ical: str) -> tuple[List[Lesson], List[HomeworkModel], List[S
```
#### 5.1.7 Client `pronotepy` pour messages et informations
#### 5.1.7 Client `pronotepy`
`pronotepy` est utilisé **uniquement** pour :
- Les **messages** des professeurs (`client.get_discussions()`).
- Les **informations et sondages** (`client.get_information_and_surveys()`).
`pronotepy` fournit les cours et devoirs de repli, ainsi que les messages, informations et
sondages. La signature réelle de la bibliothèque doit être respectée ; l'URL Pronote est le
premier argument et l'ENT est une fonction, pas une chaîne :
```python
from typing import List, Optional
from pronotepy import Client, PronoteAPIError
from ..models.message import Message, MessageType
from ..redaction import redact_url
import logging
import pronotepy.ent as pronote_ent
from pronotepy import ParentClient
logger = logging.getLogger(__name__)
ENT_RESOLVERS = {
"monbureaunumerique": pronote_ent.monbureaunumerique,
# Ajouter uniquement les ENT effectivement pris en charge et testés.
}
ent_function = ENT_RESOLVERS.get(settings.ent) if settings.ent else None
if settings.ent and ent_function is None:
raise ValueError("PRONOTE_ENT n'est pas pris en charge")
if settings.url is None or settings.username is None or settings.password is None:
raise ValueError("Configuration pronotepy incomplète")
class PronoteClient:
"""Client pour interagir avec Pronote via pronotepy."""
client = ParentClient(
settings.url,
settings.username,
settings.password.get_secret_value(),
ent=ent_function,
)
```
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
ent: Optional[str] = None,
ical_url: Optional[str] = None,
):
self.username = username
self.password = password
self.ent = ent
self.ical_url = ical_url
self._client: Optional[Client] = None
Le client applicatif expose des méthodes distinctes :
def _get_client(self) -> Client:
"""Initialise et retourne le client pronotepy."""
if self._client is None:
if not all([self.username, self.password, self.ent]):
raise ValueError("Username, password et ENT sont requis pour pronotepy")
- `get_lessons(start, end) -> list[Lesson]` ;
- `get_homeworks(start, end) -> list[Homework]` ;
- `get_messages() -> list[Message]` ;
- `get_informations() -> list[Message]`.
self._client = Client(
self.username,
self.password,
self.ent,
)
return self._client
def get_messages(self) -> List[Message]:
"""Récupère les messages des professeurs."""
try:
client = self._get_client()
discussions = client.get_discussions()
messages = []
for discussion in discussions:
for message in discussion.messages:
messages.append(Message(
id=str(message.id),
type=MessageType.DISCUSSION,
title=message.title,
content=message.content,
author=message.author,
date=message.date,
read=message.is_read,
))
return messages
except PronoteAPIError as e:
logger.error(f"Échec de la récupération des messages Pronote: {redact_secrets(str(e))}")
return []
def get_informations(self) -> List[Message]:
"""Récupère les informations et sondages."""
try:
client = self._get_client()
informations = client.get_information_and_surveys()
messages = []
for info in informations:
messages.append(Message(
id=str(info.id),
type=MessageType.INFORMATION,
title=info.title,
content=info.content,
author=info.author,
date=info.date,
read=False, # Par défaut non lu
))
return messages
except PronoteAPIError as e:
logger.error(f"Échec de la récupération des informations Pronote: {redact_secrets(str(e))}")
return []
def get_agenda_fallback(self) -> tuple[List[Lesson], List[HomeworkModel]]:
"""
Récupère l'agenda et les devoirs via pronotepy (repli si iCal échoue).
**À utiliser uniquement si PRONOTE_AGENDA_SOURCE=pronotepy ou PRONOTE_HOMEWORK_SOURCE=pronotepy**.
"""
try:
client = self._get_client()
lessons = []
for lesson in client.get_lessons():
lessons.append(Lesson(
id=str(lesson.id),
start=lesson.start,
end=lesson.end,
subject=lesson.subject,
teachers=[t.name for t in lesson.teachers],
rooms=[r.name for r in lesson.rooms],
status=LessonStatus.NORMAL, # À adapter selon les données
content=lesson.content,
))
homeworks = []
for hw in client.get_homework():
homeworks.append(HomeworkModel(
id=str(hw.id),
subject=hw.subject,
teachers=[t.name for t in hw.teachers],
assigned_on=hw.given_date,
due_on=hw.due_date,
text=hw.description,
html=hw.description, # pronotepy ne fournit pas de HTML
))
return lessons, homeworks
except PronoteAPIError as e:
logger.error(f"Échec de la récupération de l'agenda via pronotepy: {redact_secrets(str(e))}")
return [], []
def close(self) -> None:
"""Fermeture du client."""
if self._client:
self._client.close()
self._client = None
Cette séparation évite qu'un appel agenda récupère inutilement les devoirs, et inversement. Les
méthodes agenda/devoirs ne transforment jamais une erreur en liste vide : elles journalisent une
version expurgée puis lèvent une erreur expurgée avec `from None`. Les méthodes de messages et
d'informations sont non critiques et peuvent retourner une liste vide avec un warning.
Les objets renvoyés par `client.homework(start, end)` couvrent une fenêtre. Le résultat destiné à
un jour cible est donc filtré explicitement sur `homework.date == target_date`.
#### 5.1.8 Logique de repli (`sources/pronote/fallback.py`)
```python
from typing import Literal, Optional
from enum import Enum
from .ical import fetch_ical, parse_ical
from .client import PronoteClient
from ..models.agenda import Lesson, Homework
Le `PronoteFetcher` dépend de `Settings` et d'un protocole de client injecté ; il ne construit pas
de singleton et ne contient pas d'identifiants dupliqués.
| Mode | Comportement agenda/devoirs |
|------|------------------------------|
| `ical` | iCal uniquement ; toute erreur devient critique. |
| `pronotepy` | `pronotepy` uniquement ; toute erreur devient critique. |
| `auto` | iCal d'abord, puis `pronotepy` uniquement si iCal lève une erreur. |
| `auto`, deux échecs | Lever `PipelineCriticalError` avec un message expurgé. |
class AgendaSource(Enum):
AUTO = "auto"
ICAL = "ical"
PRONOTEPY = "pronotepy"
class PronoteFetcher:
"""Gère la récupération des données Pronote avec repli."""
def __init__(
self,
ical_url: Optional[str] = None,
username: Optional[str] = None,
password: Optional[str] = None,
ent: Optional[str] = None,
agenda_source: str = "auto",
homework_source: str = "auto",
):
self.ical_url = ical_url
self.username = username
self.password = password
self.ent = ent
self.agenda_source = AgendaSource(agenda_source)
self.homework_source = AgendaSource(homework_source)
self._pronote_client: Optional[PronoteClient] = None
def _get_pronote_client(self) -> PronoteClient:
if self._pronote_client is None:
self._pronote_client = PronoteClient(
username=self.username,
password=self.password,
ent=self.ent,
ical_url=self.ical_url,
)
return self._pronote_client
def fetch_agenda(self) -> tuple[List[Lesson], List[Homework]]:
"""Récupère l'agenda selon la source configurée (`agenda_source`)."""
if self.agenda_source == AgendaSource.ICAL:
return self._fetch_agenda_ical()
elif self.agenda_source == AgendaSource.PRONOTEPY:
return self._fetch_agenda_pronotepy()
else: # AUTO
# Essayer iCal d'abord
try:
lessons, homeworks = self._fetch_agenda_ical()
if lessons or homeworks:
return lessons, homeworks
except Exception as e:
logger.warning(f"Échec de la récupération iCal pour l'agenda: {redact_secrets(str(e))}")
# Repli sur pronotepy
logger.info("Repli sur pronotepy pour l'agenda.")
return self._fetch_agenda_pronotepy()
def fetch_homework(self) -> List[Homework]:
"""Récupère les devoirs selon la source configurée (`homework_source`)."""
if self.homework_source == AgendaSource.ICAL:
# Récupérer uniquement les devoirs depuis iCal
try:
_, homeworks = self._fetch_agenda_ical()
return homeworks
except Exception as e:
logger.warning(f"Échec de la récupération iCal pour les devoirs: {redact_secrets(str(e))}")
return []
elif self.homework_source == AgendaSource.PRONOTEPY:
# Récupérer uniquement les devoirs depuis pronotepy
try:
_, homeworks = self._fetch_agenda_pronotepy()
return homeworks
except Exception as e:
logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}")
return []
else: # AUTO
# Essayer iCal d'abord
try:
_, homeworks = self._fetch_agenda_ical()
if homeworks:
return homeworks
except Exception as e:
logger.warning(f"Échec de la récupération iCal pour les devoirs: {redact_secrets(str(e))}")
# Repli sur pronotepy
logger.info("Repli sur pronotepy pour les devoirs.")
try:
_, homeworks = self._fetch_agenda_pronotepy()
return homeworks
except Exception as e:
logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}")
return []
def _fetch_agenda_ical(self) -> tuple[List[Lesson], List[Homework]]:
"""Récupère l'agenda depuis iCal."""
if not self.ical_url:
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
raw_ical = fetch_ical(self.ical_url)
lessons, homeworks, _ = parse_ical(raw_ical)
return lessons, homeworks
def _fetch_agenda_pronotepy(self) -> tuple[List[Lesson], List[Homework]]:
"""Récupère l'agenda depuis pronotepy."""
client = self._get_pronote_client()
lessons, homeworks = client.get_agenda_fallback()
return lessons, homeworks
def fetch_messages(self) -> List[Message]:
"""Récupère les messages (toujours via pronotepy)."""
client = self._get_pronote_client()
return client.get_messages()
def fetch_informations(self) -> List[Message]:
"""Récupère les informations (toujours via pronotepy)."""
client = self._get_pronote_client()
return client.get_informations()
def close(self) -> None:
"""Fermeture des ressources."""
if self._pronote_client:
self._pronote_client.close()
self._pronote_client = None
```
Une réponse vide est un succès et ne déclenche pas de repli : une journée peut réellement ne
contenir aucun cours ou devoir. Inversement, une exception ne doit jamais être convertie en
`([], [])`, car le `PronoteFetcher` perdrait alors l'information nécessaire pour distinguer un
échec d'un résultat vide.
`fetch_homework(target_date)` applique le même contrat aux deux sources. Pour iCal, il appelle
`collect_homeworks(lessons, target_date)`. Pour `pronotepy`, il filtre les devoirs récupérés sur la
même date cible. En M11, la composition root fournit un contexte d'exécution permettant de
réutiliser le même téléchargement/parsing iCal pour l'agenda et les devoirs lorsque les deux
sélections le permettent.
### 5.2 Résumé des points clés
@@ -4546,8 +4369,10 @@ def get_channel(settings: Settings, channel_name: str = "xmpp") -> Channel:
- **Ne jamais bloquer le pipeline** : Une erreur dans une étape ne doit pas empêcher les autres étapes de s'exécuter (sauf si critique).
- **Modes dégradés** :
- **Synthèse IA** : Si elle échoue → envoyer le message **sans synthèse** (mais avec la liste brute des devoirs).
- **pronotepy** : Si la récupération échoue → basculer sur **iCal** (si disponible).
- **iCal** : Si la récupération échoue → basculer sur **pronotepy** (si configuré).
- **Sources Pronote en mode `auto`** : essayer iCal, puis basculer sur `pronotepy`
uniquement si iCal lève une erreur.
- **Source Pronote explicite** : `ical` et `pronotepy` sont des modes stricts, sans repli
implicite.
- **CalDAV** : Si la synchronisation échoue → **logger l'erreur** mais continuer le pipeline.
- **XMPP** : Si l'envoi échoue → **logger l'erreur** mais continuer le pipeline.
- **Erreurs critiques** :
@@ -4556,6 +4381,9 @@ def get_channel(settings: Settings, channel_name: str = "xmpp") -> Channel:
### 11.2 Hiérarchie des erreurs
La hiérarchie canonique réside dans `pronote_sync/errors.py`. Les jalons suivants la complètent si
nécessaire mais ne créent pas une seconde hiérarchie dans `pipeline/steps/errors.py`.
```python
from enum import Enum, auto
from typing import Optional
@@ -4610,7 +4438,7 @@ from ..models.xmpp import XmppMessage
from ..models.pronote import PronoteData
from ..models.sync import CalDAVSyncResult
from ..models.synthesis import SynthesisInput, SynthesisResult
from ..sources.pronote.fetcher import PronoteFetcher
from ..sources.pronote.fallback import PronoteFetcher
from ..sync.caldav import CalDAVClient
from ..sync.diff import AgendaComparator
from ..synthesis.provider import SynthesisProvider
@@ -4818,78 +4646,26 @@ class PipelineRunner:
Chaque étape du pipeline est **isolée** et peut lever des `PipelineError` ou `PipelineWarning`.
#### 11.4.1 `fetch_step.py`
#### 11.4.1 `fetch.py`
```python
from typing import Tuple, List
from ..models.agenda import Lesson, Homework, SchoolEvent
from ..models.message import Message
from ..sources.pronote.fetcher import PronoteFetcher
from ..utils.redaction import redact_secrets
from .errors import PipelineError, ErrorSeverity, PipelineCriticalError
L'étape de récupération orchestre le contrat de `PronoteFetcher` sans réimplémenter la sélection
des sources :
1. récupérer l'agenda et les événements scolaires ;
2. résoudre le jour cible ;
3. récupérer les devoirs pour cette date cible ;
4. récupérer les messages et informations non critiques ;
5. assembler `PronoteData`.
def fetch_step(fetcher: PronoteFetcher) -> Tuple[List[Lesson], List[Homework], List[SchoolEvent], List[Message]]:
"""
Étape de récupération des données Pronote.
**Logique de precedence** :
- Si `agenda_source=ical` et `homework_source=ical`, un seul fetch iCal suffit (les devoirs sont extraits du même flux).
- Si `agenda_source=ical` et `homework_source=pronotepy`, deux sources distinctes sont utilisées.
- La déduplication globale est effectuée après fusion des résultats.
Args:
fetcher: Fetcher Pronote configuré.
Returns:
Tuple (lessons, homeworks, school_events, messages).
Raises:
PipelineCriticalError: Si aucune source n'est disponible.
PipelineError: Si une source échoue mais qu'une autre est disponible.
"""
try:
# Récupérer l'agenda (cours + événements scolaires)
lessons, agenda_homeworks = fetcher.fetch_agenda()
school_events = [] # À récupérer depuis iCal ou autre source
# Récupérer les devoirs selon la source configurée
# Si la source est iCal et que l'agenda a déjà été récupéré depuis iCal,
# les devoirs sont déjà inclus dans agenda_homeworks (via parsing iCal).
# Sinon, récupérer les devoirs depuis la source dédiée (ex: pronotepy).
if fetcher.homework_source.value == "pronotepy" or (
fetcher.homework_source.value == "auto" and fetcher.agenda_source.value != "ical"
):
# Récupérer les devoirs depuis pronotepy
homework_list = fetcher.fetch_homework()
# Fusionner les devoirs (agenda_homeworks peut être vide si agenda_source != ical)
homeworks = agenda_homeworks + homework_list
else:
# Utiliser les devoirs déjà extraits de l'agenda iCal
homeworks = agenda_homeworks
# Récupérer les messages et informations (toujours via pronotepy)
messages = fetcher.fetch_messages()
informations = fetcher.fetch_informations()
messages.extend(informations)
if not lessons and not homeworks:
raise PipelineCriticalError(
message="Aucun cours ou devoir récupéré depuis Pronote",
step="fetch",
)
return lessons, homeworks, school_events, messages
except Exception as e:
raise PipelineError(
message=f"Échec de la récupération Pronote: {redact_secrets(str(e))}",
severity=ErrorSeverity.ERROR,
step="fetch",
recoverable=False,
) from e
```
Quand l'agenda et les devoirs utilisent iCal pendant la même exécution, le téléchargement et le
parsing sont partagés dans un contexte local au run. Une simple valeur mémorisée dans l'instance du
fetcher ou dans le contexte d'exécution suffit ; aucun cache global, persistant ou système
d'invalidation n'est requis.
Une liste vide est une donnée valide et ne doit pas provoquer d'erreur critique. La criticité dépend
des exceptions remontées par les sources. L'étape importe les erreurs depuis
`pronote_sync.errors`, journalise uniquement des contenus expurgés et ne chaîne jamais une
exception externe brute susceptible de contenir un secret.
#### 11.4.1 bis `fetch_blog_step.py`
@@ -4953,9 +4729,9 @@ Les autres étapes (`normalize_step`, `compare_step`, etc.) suivent le même pri
### 11.5 Points clés
- **Ne jamais bloquer** : Les erreurs non critiques (ex: synthèse IA) ne bloquent pas le pipeline.
- **Modes dégradés** :
- Si iCal échoue → basculer sur `pronotepy`.
- Si `pronotepy` échoue → basculer sur iCal.
- Si les deux échouent → **échec critique**.
- En mode `auto`, si iCal échoue → basculer sur `pronotepy`.
- En mode explicite, ne pas changer de source.
- En mode `auto`, si les deux sources échouent → **échec critique**.
- **Logs clairs** : Chaque erreur est loggée avec son niveau de gravité.
- **Retour d'erreur** : Le pipeline retourne toujours une liste des erreurs/warnings rencontrés.
@@ -5276,10 +5052,11 @@ def sample_settings():
return Settings(
pronote=PronoteSettings(
ical_url="https://test.ent/pronote/ical/test.ics",
url="https://test.ent/pronote/parent.html",
ical_url=SecretStr("https://test.ent/pronote/ical/test.ics"),
username="test_user",
password=SecretStr("test_password"),
ent="test_ent",
ent="monbureaunumerique",
agenda_source="auto",
homework_source="auto",
messages_source="pronotepy",
@@ -5331,7 +5108,12 @@ def test_parse_ical_lesson(parsed_lessons):
@pytest.mark.unittest
def test_parse_ical_homework(parsed_lessons):
"""Test le parsing des devoirs depuis iCal."""
lessons, homeworks, school_events = parsed_lessons
from pronote_sync.sources.pronote.ical import collect_homeworks
lessons, parsed_homeworks, school_events = parsed_lessons
assert parsed_homeworks == []
homeworks = collect_homeworks(lessons, date(2026, 9, 10))
assert len(homeworks) == 1
homework = homeworks[0]
@@ -5347,20 +5129,15 @@ def test_parse_ical_homework(parsed_lessons):
def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, mock_xmpp_channel, sample_settings):
"""Test le pipeline complet avec des mocks."""
from pronote_sync.pipeline.run import PipelineRunner
from pronote_sync.sources.pronote.fetcher import PronoteFetcher
from pronote_sync.sources.pronote.client import PronoteClient
from pronote_sync.sources.pronote.fallback import PronoteFetcher
from pronote_sync.sync.caldav import CalDAVClient
from pronote_sync.sync.diff import AgendaComparator
from pronote_sync.sources.theoretical.file import CSVTheoreticalAgendaProvider
# Configurer le fetcher Pronote
fetcher = PronoteFetcher(
ical_url=sample_settings.pronote.ical_url,
username=sample_settings.pronote.username,
password=sample_settings.pronote.password.get_secret_value(),
ent=sample_settings.pronote.ent,
agenda_source=sample_settings.pronote.agenda_source,
homework_source=sample_settings.pronote.homework_source,
)
pronote_client = PronoteClient(sample_settings.pronote)
fetcher = PronoteFetcher(sample_settings, pronote_client)
# Configurer le client CalDAV
caldav_client = CalDAVClient(
@@ -5421,7 +5198,7 @@ def test_collect_homeworks():
"""Test la collecte et déduplication des devoirs depuis des blocs de plusieurs VEVENT."""
from datetime import date, datetime
from pronote_sync.models.agenda import Lesson, LessonStatus
from pronote_sync.models.homework import HomeworkBlock
from pronote_sync.models.agenda import HomeworkBlock
from pronote_sync.sources.pronote.ical import collect_homeworks
# Créer des cours avec des blocs de devoirs (simulant des VEVENT parsés)
@@ -6174,4 +5951,3 @@ Ce guide fournit une **base architecturale et technique solide** pour développe
| `pydantic` | [https://pydantic.dev/](https://pydantic.dev/) | Bibliothèque pour la validation des données. |
| RFC 5545 (iCal) | [https://datatracker.ietf.org/doc/html/rfc5545](https://datatracker.ietf.org/doc/html/rfc5545) | Spécification officielle du format iCal. |
| RFC 4791 (CalDAV) | [https://datatracker.ietf.org/doc/html/rfc4791](https://datatracker.ietf.org/doc/html/rfc4791) | Spécification officielle de CalDAV.