AntoineVe/college-infos
1
0
Fork 0
Code Issues 13 Pull Requests 1 Actions Packages Projects Releases 2 Wiki Activity
Files
feature/m7-caldav-sync
college-infos/GUIDE_DEV_PYTHON.md
Antoine Van Elstraete ebbe39f1f0 docs(M7): aligner guide, TODO et pre-commit pour la sync CalDAV 
Met à jour la documentation et la configuration pour le jalon M7 selon
les décisions d'architecture :

- GUIDE_DEV_PYTHON.md §7 : API réelle caldav>=1.3.0 (pas le pseudo-code),
  calendar_path (pas calendar_name), plan CalDAVSyncPlan explicite avant
  exécution, pas d'état local (scan distant), événements non marqués
  jamais modifiés
- TODO.md M7 : suppression de sync/state.py et BlogRSSState, ajout de
  l'exécution du plan et de la protection des événements non gérés
- .pre-commit-config.yaml : caldav>=1.3.0 ajouté aux additional_dependencies
  du hook mypy

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/tech-writer <tech-writer@agents.invalid>
2026-09-07 00:00:54 +02:00

6095 lines
268 KiB
Markdown
Raw Permalink Blame History

# Guide de Développement : Synchronisation Pronote → CalDAV + XMPP (Python)
> **Statut** : Guide de référence pour un futur projet Python inspiré de [`pronote-digest`](https://github.com/antoine-coulon/pronote-digest) (TypeScript).
> **Public cible** : Développeurs Python (≥ 3.13.5) familiers avec les concepts de CLI, synchronisation de calendriers et messagerie instantanée.
> **Objectif** : Fournir une base architecturale et technique pour un outil **synchronisant l'agenda Pronote vers CalDAV**, **comparant avec un agenda théorique**, **récupérant messages et informations**, et **envoyant une synthèse par XMPP**.
> **⚠️ À noter** : Ce guide est **volontairement détaillé** pour préserver les connaissances acquises sur les spécificités des flux Pronote (iCal) et les décisions architecturales du projet TypeScript. Certaines sections (ex: parsing iCal) contiennent des **observations précises** issues de l'analyse du code existant.
> **Mises à jour récentes** :
> - Ajout de la section **[5 bis. Sources externes : blog du collège (RSS)](#5-bis-sources-externes--blog-du-collège-rss)** pour le parsing du flux RSS du blog.
> - Mise à jour de la section **[10. Envoi XMPP](#10-envoi-xmpp)** avec la décision architecturale (compte bot dédié, messages directs, pas de PubSub).
> - Intégration des modèles `BlogArticle` et `ExternalInfo` dans la section **[6. Modèle de données Pydantic](#6-modèle-de-données-pydantic)**.
---
## 1. Vue d'ensemble du projet
### 1.1 Périmètre fonctionnel
Le projet doit implémenter les fonctionnalités suivantes, dans l'ordre logique du pipeline :
1. **Récupération des données Pronote** :
- Agenda (cours, annulations, déplacements) via **flux iCal officiel** (prioritaire) ou `pronotepy` (repli).
- Devoirs via **flux iCal** (si activé par l'établissement) ou `pronotepy`.
- Messages des professeurs, discussions, informations et sondages via **`pronotepy` uniquement** (non disponibles dans iCal).
2. **Synchronisation vers CalDAV** :
- Synchronisation **différentielle** (pas d'écrasement complet) des événements Pronote vers un calendrier CalDAV dédié.
- Gestion des **UID stables** (normalisation des UID Pronote ou hachage déterministe).
- Conservation des événements annulés avec `STATUS:CANCELLED`.
3. **Comparaison avec l'agenda théorique** :
- Détection des **changements** (ajouts, suppressions, modifications) entre l'agenda réel (Pronote) et un agenda théorique (fichier JSON local, avec parité des semaines et vacances scolaires).
- Génération d'une liste structurée des différences.
4. **Génération de la synthèse** :
- **Synthèse IA** (optionnelle) des changements d'agenda et des informations importantes (messages, annonces).
- **Liste brute des devoirs** (non modifiée par l'IA), préservée telle quelle.
5. **Envoi par XMPP** :
- Message structuré contenant :
- La synthèse IA (si disponible).
- La liste brute des devoirs.
### 1.2 Contraintes clés
- **Python ≥ 3.13.5** : Utilisation des dernières fonctionnalités (ex: `typing.Protocol`, `dataclasses`, `asyncio` pour XMPP).
- **Pas de secrets en clair** : Tokens, mots de passe et URLs sensibles **doivent** être masqués dans les logs, erreurs et fixtures.
- **Tests sans réseau** : Utilisation de **mocks** (ex: `pytest-mock`, `responses`, `aioresponses`) et **fixtures** anonymisées.
- **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).
- 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.
---
## 2. Architecture cible et pipeline
### 2.1 Diagramme textuel du pipeline
```
┌───────────────────────────────────────────────────────────────────────────────┐
│ CONFIGURATION │
│ (env vars + Pydantic Settings + .env.example) │
└───────────────────────────────────────────────────────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────────────────────────┐
│ RÉCUPÉRATION PRONOTE │
│ ┌─────────────────┐ ┌─────────────────┐ ┌───────────────────────────┐ │
│ │ Flux iCal │ │ pronotepy │ │ Règles de repli │ │
│ │ (agenda/devoirs)│ │ (messages/infos)│ │ (PRONOTE_AGENDA_SOURCE, │ │
│ │ │ │ │ │ PRONOTE_HOMEWORK_SOURCE)│ │
│ └────────┬────────┘ └────────┬────────┘ └──────────────┬────────────┘ │
│ │ │ │ │
│ └───────────────────────┼────────────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────────┐ │
│ │ NORMALISATION & PARSING │ │
│ │ - Parsing iCal (icalendar) → Modèles Pydantic (Lesson, Homework, ...) │ │
│ │ - Déduplication des devoirs (clé normalisée) │ │
│ │ - Normalisation des UID (suppression suffixes temporels) │ │
│ │ - Détection des statuts (annulé/déplacé via catégories) │ │
│ └─────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────────┐ │
│ │ RÉCUPÉRATION BLOG (RSS) │ │
│ │ - Fetch du flux RSS (feedparser) → Modèles BlogArticle │ │
│ │ - Déduplication par GUID (état local) │ │
│ │ - Cache HTTP (If-Modified-Since) │ │
│ └─────────────────────────────────────────────────────────────────────────┘
└───────────────────────────────────────────────────────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────────────────────────┐
│ COMPARAISON AVEC AGENDA THÉORIQUE │
│ ┌─────────────────────────────────────────────────────────────────────────┐ │
│ │ - Matching déterministe (date, créneau, matière normalisée) │ │
│ │ - Génération des différences (ajouts/suppressions/modifications) │ │
│ └─────────────────────────────────────────────────────────────────────────┘ │
└───────────────────────────────────────────────────────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────────────────────────┐
│ SYNCHRONISATION CALDAV │
│ ┌─────────────────┐ ┌─────────────────┐ ┌───────────────────────────┐ │
│ │ Plan de sync │ │ Sync │ │ État local │ │
│ │ (CalDavSyncPlan)│ │ différentielle │ │ (SQLite/JSON) │ │
│ └────────┬────────┘ └────────┬────────┘ └──────────────┬────────────┘ │
│ │ │ │ │
│ └───────────────────────┼────────────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────────┐ │
│ │ Résultat de sync (CalDavSyncResult) │ │
│ └─────────────────────────────────────────────────────────────────────────┘ │
└───────────────────────────────────────────────────────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────────────────────────┐
│ SYNTHÈSE IA (optionnelle) │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ Entrée │ │ Sortie │ │
│ │ (SynthesisInput)│ │ (SynthesisResult)│ │
│ └────────┬────────┘ └────────┬────────┘ │
│ │ │ │
│ └───────────────────────┼────────────────────────────┘ │
│ ▼ │
└───────────────────────────────────────────────────────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────────────────────────┐
│ CONSTRUCTION DU MESSAGE XMPP │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ Synthèse IA │ │ Liste brute │ │
│ │ (optionnelle) │ │ des devoirs │ │
│ └────────┬────────┘ └────────┬────────┘ │
│ │ │ │
│ └───────────────────────┼────────────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────────┐ │
│ │ Message XMPP (XmppMessage) │ │
│ │ - synthesis: Optional[str] │ │
│ │ - homeworks: List[Homework] │ │
│ │ - changes: List[AgendaChange] │ │
│ │ - messages: List[Message] │ │
│ └─────────────────────────────────────────────────────────────────────────┘ │
└───────────────────────────────────────────────────────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────────────────────────┐
│ ENVOI XMPP (slixmpp) │
│ ┌─────────────────────────────────────────────────────────────────────────┐ │
│ │ - Message structuré (synthèse + devoirs bruts) │ │
│ │ - Gestion des erreurs (reconnexion, timeout) │ │
│ └─────────────────────────────────────────────────────────────────────────┘ │
└───────────────────────────────────────────────────────────────────────────────┘
```
### 2.2 Équivalence avec `pronote-digest` (TypeScript)
| **Concept (TypeScript)** | **Équivalent Python** | **Bibliothèque/Outils** |
|---------------------------------|-----------------------------------------------|---------------------------------------|
| `fetch` (iCal) | `requests` + `icalendar` | `requests`, `icalendar` |
| `parse` (iCal → événements) | Parsing `icalendar` → Pydantic | `icalendar`, `pydantic` |
| `PronoteData` (données normalisées) | `PronoteData` (Pydantic) | `pydantic` |
| `AgendaDiff` (comparaison) | `AgendaDiff` (Pydantic) | `pydantic` |
| `CalDavSyncPlan`/`CalDavSyncResult` | `CalDavSyncPlan`/`CalDavSyncResult` (Pydantic) | `pydantic` |
| `SynthesisInput`/`SynthesisResult` | `SynthesisInput`/`SynthesisResult` (Pydantic) | `pydantic` |
| `XmppMessage` (message final) | `XmppMessage` (Pydantic) | `pydantic` |
| Pipeline injectable (`run.ts`) | Protocoles + composition root | `typing.Protocol`, `dataclasses` |
| Canaux (Email/File) | Adaptateurs CalDAV/XMPP | `caldav`, `slixmpp` |
| Tests (MSW) | Mocks HTTP (`responses`, `aioresponses`) | `pytest`, `pytest-mock` |
| Configuration (zod) | `pydantic-settings` | `pydantic-settings` |
| IA (Vercel AI SDK) | Adaptateur OpenAI/`litellm` (optionnel) | `openai`, `litellm` (optionnel) |
#### 2.3 Découpage en modules
```
pronote_sync/
├── __init__.py
├── config/ # Configuration (Pydantic Settings)
│ ├── __init__.py
│ ├── settings.py # Modèles de configuration
│ └── env.py # Chargement des variables d'environnement
├── models/ # Modèles de données (Pydantic)
│ ├── __init__.py
│ ├── agenda.py # Lesson, SchoolEvent, TheoreticalLesson
│ ├── homework.py # Homework
│ ├── message.py # Message (Pronote)
│ ├── diff.py # AgendaDiff (changements)
│ ├── pronote.py # PronoteData (données normalisées)
│ ├── sync.py # CalDavSyncPlan, CalDavSyncResult
│ ├── synthesis.py # SynthesisInput, SynthesisResult
│ └── xmpp.py # XmppMessage (synthèse + devoirs bruts)
├── sources/ # Sources de données
│ ├── __init__.py
│ ├── pronote/ # Pronote (iCal + pronotepy)
│ │ ├── __init__.py
│ │ ├── ical.py # Récupération/parsing iCal
│ │ ├── client.py # Client pronotepy (messages/infos)
│ │ └── fallback.py # Logique de repli
│ ├── blog/ # Blog (RSS)
│ │ ├── __init__.py
│ │ ├── rss.py # Client RSS (feedparser)
│ │ └── state.py # État local (déduplication, cache HTTP)
│ └── theoretical/ # Agenda théorique
│ ├── __init__.py
│ ├── file.py # Lecture fichier JSON (parité + vacances)
│ └── provider.py # Interface TheoreticalAgendaProvider
├── sync/ # Synchronisation CalDAV + Blog
│ ├── __init__.py
│ ├── caldav.py # Client CalDAV (caldav)
│ ├── state.py # État de sync CalDAV (SQLite/JSON)
│ ├── blog_state.py # État de sync Blog (déduplication, cache HTTP)
│ └── diff.py # Logique de comparaison
├── synthesis/ # Synthèse IA
│ ├── __init__.py
│ ├── provider.py # Protocole SynthesisProvider
│ ├── openai.py # Adaptateur OpenAI
│ └── litellm.py # Adaptateur litellm (optionnel)
├── channels/ # Canaux de sortie
│ ├── __init__.py
│ ├── xmpp.py # Envoi XMPP (slixmpp)
│ └── protocol.py # Protocole Channel
├── pipeline/ # Pipeline principal
│ ├── __init__.py
│ ├── run.py # Orchestration (composition root)
│ └── steps/ # Étapes du pipeline
│ ├── fetch.py
│ ├── normalize.py
│ ├── compare.py
│ ├── caldav_sync.py
│ ├── synthesis.py
│ └── send.py
├── utils/ # Utilitaires
│ ├── __init__.py
│ ├── logging.py # Configuration des logs (masquage secrets)
│ ├── redaction.py # Masquage des tokens/URLs
│ └── uid.py # Normalisation des UID
├── cli/ # Interface CLI
│ ├── __init__.py
│ └── main.py # Point d'entrée CLI
├── tests/ # Tests
│ ├── fixtures/ # Fixtures anonymisées
│ ├── conftest.py # Configuration pytest
│ └── ... # Tests par module
├── .env.example # Exemple de configuration
├── pyproject.toml # Dépendances et configuration projet
└── README.md # Documentation utilisateur
```
---
## 3. Configuration d'environnement
### 3.1 Variables d'environnement
Le projet utilise **`pydantic-settings`** pour valider et charger la configuration depuis les variables d'environnement ou un fichier `.env`.
#### 3.1.1 Variables obligatoires
| Variable | Description | Exemple (anonymisé) | Type |
|------------------------------|-----------------------------------------------------------------------------|---------------------------------------------|---------------|
| `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` | 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` |
| `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` |
> ⚠️ **Décision d'implémentation** :
> `XMPP_RECIPIENT` a été renommé en `XMPP_TO` dans l'implémentation (aligné avec §10.2.1).
> 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 |
|------------------------------|-----------------------------------------------------------------------------|-------------------------|---------------|
| `PRONOTE_AGENDA_SOURCE` | Source pour l'agenda (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` |
| `PRONOTE_HOMEWORK_SOURCE` | Source pour les devoirs (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` |
| `PRONOTE_MESSAGES_SOURCE` | Source pour les messages (`pronotepy` uniquement). | `pronotepy` | `Literal` |
| `SYNC_PAST_DAYS` | Nombre de jours dans le passé pour la sync CalDAV. | `7` | `int` |
| `SYNC_FUTURE_DAYS` | Nombre de jours dans le futur pour la sync CalDAV. | `30` | `int` |
> ⚠️ **Décision d'implémentation** :
> Ces variables sont désormais dans `AppSettings` (et non `CalDAVSettings`) car `CalDAVSettings` utilise `env_prefix="CALDAV_"`, ce qui nécessiterait `CALDAV_SYNC_PAST_DAYS`.
> Leur placement dans `AppSettings` (sans préfixe) garantit un mappage correct avec `SYNC_PAST_DAYS` / `SYNC_FUTURE_DAYS`.
| `THEORETICAL_AGENDA_PATH` | Chemin vers le fichier JSON de l'agenda théorique. | `None` | `str \| None`|
| `SCHOOL_HOLIDAYS_PATH` | Chemin vers le fichier JSON des vacances scolaires. | `None` | `str \| None`|
| `THEORETICAL_WEEK_ANCHOR_DATE` | Date de référence pour la parité des semaines (paire/impaire). | `None` | `date \| None`|
| `THEORETICAL_WEEK_ANCHOR_TYPE` | Parité de la semaine de référence (`even` ou `odd`). | `None` | `Literal["even", "odd"] \| None`|
| `AI_ENABLED` | Activer la synthèse IA. | `False` | `bool` |
| `AI_PROVIDER` | Fournisseur IA (`openai` ou `litellm`). | `openai` | `str` |
| `AI_BASE_URL` | URL de base pour l'API IA (ex: OpenAI compatible). | `None` | `str \| None`|
| `AI_API_KEY` | Clé API pour l'API IA. | `None` | `SecretStr` |
| `AI_MODEL` | Modèle IA à utiliser (exemple recommandé : `gpt-4o-mini`). | `None` | `str \| None`|
| `DRY_RUN` | Mode dry-run (pas de modifications CalDAV/XMPP). | `False` | `bool` |
| `LOG_LEVEL` | Niveau de log (`DEBUG`, `INFO`, `WARNING`, `ERROR`). | `INFO` | `str` |
#### 3.1.3 Exemple de fichier `.env.example`
```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
PRONOTE_ENT=monbureaunumerique
# Sources (auto = essayer iCal d'abord, puis pronotepy)
PRONOTE_AGENDA_SOURCE=auto
PRONOTE_HOMEWORK_SOURCE=auto
PRONOTE_MESSAGES_SOURCE=pronotepy
# --- CalDAV ---
CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
CALDAV_USERNAME=user@example.com
CALDAV_PASSWORD=your_caldav_password
CALDAV_CALENDAR_PATH=/pronote-sync/
# Fenêtre de synchronisation (jours)
SYNC_PAST_DAYS=7
SYNC_FUTURE_DAYS=30
# --- Agenda théorique (JSON) ---
THEORETICAL_AGENDA_PATH=./data/theoretical.json
SCHOOL_HOLIDAYS_PATH=./data/school_holidays.json
THEORETICAL_WEEK_ANCHOR_DATE=2026-09-01
THEORETICAL_WEEK_ANCHOR_TYPE=even
# --- XMPP ---
XMPP_JID=user@example.com
XMPP_PASSWORD=your_xmpp_password
XMPP_RECIPIENT=parent@example.com
# --- IA (optionnelle) ---
AI_ENABLED=true
AI_PROVIDER=openai
AI_BASE_URL=https://api.openai.com/v1
AI_API_KEY=your_ai_api_key
# AI_MODEL=gpt-4o-mini # exemple recommandé, non activé par défaut
# --- Divers ---
DRY_RUN=false
LOG_LEVEL=INFO
```
### 3.2 Modèle Pydantic pour la configuration
> ⚠️ **Décision d'implémentation** :
> L'implémentation utilise le style moderne de Pydantic v2 : `model_config = ConfigDict(frozen=True)` au lieu de `class Config`, pas de `json_encoders` (la sérialisation ISO est native en v2), `str | None` au lieu de `Optional[str]`, `list[str]` au lieu de `List[str]`.
> `AISettings.enabled` a pour valeur par défaut `False` (et non `True` comme indiqué dans le bloc de code).
> `XmppSettings` est entièrement défini en §10.2.3 avec tous les champs optionnels (valeurs par défaut) pour que `Settings()` fonctionne sans `.env`.
> `BlogSettings` a été ajouté (§5 bis.9.2) avec `enabled=False` et `rss_url` par défaut.
> `sync_past_days` et `sync_future_days` sont dans `AppSettings`, et non `CalDAVSettings`.
> `CalDAVSettings.calendar_path` a pour valeur par défaut `"/pronote-sync/"` (et non `"/pronote-digest/"`).
> `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"` (et non `"pronote-digest"`).
```python
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")
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"
class CalDAVSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="CALDAV_", env_file=".env", extra="ignore")
url: Optional[str] = None
username: Optional[str] = None
password: Optional[SecretStr] = None
calendar_path: str = "/pronote-digest/"
sync_past_days: int = 7
sync_future_days: int = 30
class AISettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="AI_", env_file=".env", extra="ignore")
enabled: bool = True
provider: Literal["openai", "litellm"] = "openai"
base_url: Optional[str] = None
api_key: Optional[SecretStr] = None
model: Optional[str] = None
class AppSettings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
dry_run: bool = False
log_level: str = "INFO"
theoretical_agenda_path: Optional[str] = None
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
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()
```
---
## 4. Gestion des secrets et *redaction*
### 4.1 Principes
- **Aucun secret** ne doit apparaître en clair dans :
- Le code source.
- Les logs (même en mode `DEBUG`).
- Les messages d'erreur.
- Les fixtures de test.
- **Masquage systématique** des :
- Tokens (`icalsecurise` dans les URLs iCal).
- Mots de passe (Pronote, CalDAV, XMPP, IA).
- URLs complètes contenant des tokens.
### 4.2 Implémentation
**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 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** :
> `redact_exception` est implémenté comme une **fonction au niveau du module** dans `utils/redaction.py`, et non comme une méthode de `RedactingFormatter` (contrairement à §4.2.2 où elle apparaît comme une méthode).
> `redact_url` utilise `urlsplit`/`urlunsplit`/`parse_qsl` au lieu de `urlparse`/`urlunparse`/`parse_qs`.
> La correspondance des clés sensibles est insensible à la casse.
```python
import re
from urllib.parse import urlparse, urlunparse, parse_qs, urlencode
def redact_url(url: str) -> str:
"""
Masque les paramètres sensibles dans une URL (ex: icalsecurise).
Inspiré de src/sources/pronote/fetch.ts dans pronote-digest.
"""
try:
parsed = urlparse(url)
query_params = parse_qs(parsed.query, keep_blank_values=True)
# Liste des paramètres à masquer
sensitive_keys = {"icalsecurise", "token", "key", "password", "secret"}
for key in sensitive_keys:
if key in query_params:
query_params[key] = ["REDACTED"]
# Reconstruire l'URL
new_query = urlencode(query_params, doseq=True)
redacted = urlunparse(parsed._replace(query=new_query))
return redacted
except Exception:
# En cas d'erreur, masquer toute l'URL
return "REDACTED_URL"
def redact_secrets(text: str) -> str:
"""
Masque les secrets dans un texte (URLs, tokens, mots de passe).
"""
# Masquer les URLs
text = re.sub(
r"(https?://[^\s]+)",
lambda m: redact_url(m.group(1)),
text,
)
# Masquer les tokens isolés (ex: icalsecurise=XXX)
text = re.sub(
r"(icalsecurise|token|password|secret)=[^\s&]+",
r"\1=REDACTED",
text,
flags=re.IGNORECASE,
)
return text
```
#### 4.2.2 Configuration des logs (`logging.py`)
> ⚠️ **Décision d'implémentation** :
> `redact_exception` est une fonction au niveau du module dans `redaction.py`, et non une méthode de `RedactingFormatter`.
> `setup_logging` utilise `logging.getLevelNamesMapping()` (Python 3.11+) au lieu de `getattr(logging, ...)`.
> `RedactingFormatter.format` gère à la fois les `record.args` de type tuple et dict.
```python
import logging
import sys
from typing import Any
from .redaction import redact_secrets
class RedactingFormatter(logging.Formatter):
"""Formatter qui masque les secrets dans les logs."""
def format(self, record: logging.LogRecord) -> str:
# Masquer les secrets dans le message
record.msg = redact_secrets(str(record.msg))
# Masquer les secrets dans les arguments
if record.args:
record.args = tuple(
redact_secrets(str(arg)) if isinstance(arg, str) else arg
for arg in record.args
)
return super().format(record)
def redact_exception(self, exc: Exception) -> str:
"""
Masque les secrets dans une exception avant journalisation ou réémission.
**Doit être appelé systématiquement** pour toutes les exceptions externes
(HTTP, CalDAV, XMPP, IA) avant de les logger ou de les réémettre.
"""
return redact_secrets(str(exc))
def setup_logging(level: str = "INFO") -> None:
"""Configure les logs avec masquage des secrets."""
log_level = getattr(logging, level.upper(), logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(RedactingFormatter(
fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
))
root_logger = logging.getLogger()
root_logger.setLevel(log_level)
root_logger.handlers.clear()
root_logger.addHandler(handler)
# Désactiver les logs des bibliothèques tierces (trop verbeuses)
logging.getLogger("urllib3").setLevel(logging.WARNING)
logging.getLogger("slixmpp").setLevel(logging.WARNING)
```
#### 4.2.3 Utilisation dans le code
```python
from .logging import setup_logging, redact_secrets
from .redaction import redact_url
# Initialisation des logs (au démarrage de l'application)
setup_logging(settings.log_level)
# Exemple d'utilisation dans une fonction
logger = logging.getLogger(__name__)
def fetch_ical(url: str) -> str:
try:
# ... logique de fetch ...
except Exception as e:
# Masquer l'URL dans l'erreur
safe_url = redact_url(url)
logger.error(f"Échec de la récupération de {safe_url}: {redact_secrets(str(e))}")
raise
```
---
## 5. Sources Pronote : iCal et `pronotepy`
---
## 5 bis. Sources externes : blog du collège (RSS)
### 5 bis.1 Description et objectifs
Le **blog du collège** ([https://blogpeda.ac-bordeaux.fr/cjeliote/](https://blogpeda.ac-bordeaux.fr/cjeliote/)) publie des **articles publics** (annonces, informations administratives, événements) qui doivent être intégrés dans les **"informations diverses"** du message XMPP.
**Objectifs** :
- Récupérer les nouveaux articles du blog via son **flux RSS 2.0** ou **Atom**.
- Les intégrer dans le message XMPP sous forme de **liste structurée** (titre, date, catégorie, extrait).
- Éviter les doublons grâce à une **déduplication par GUID**.
- Respecter les contraintes de **cache HTTP** pour limiter les requêtes inutiles.
---
### 5 bis.2 Flux RSS disponibles
Le blog utilise **WordPress Multisite** (plateforme `blogpeda.ac-bordeaux.fr`) avec les flux suivants :
| **Type** | **URL** | **Format** | **Contenu** |
|------------------------|-------------------------------------------------------------------------|------------|--------------------------------------|
| RSS 2.0 | `https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2` | RSS 2.0 | Articles complets (titre, lien, date, catégorie, contenu HTML) |
| Atom | `https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=atom` | Atom | Équivalent à RSS 2.0 (avec `<id>`, `<updated>`) |
| Commentaires RSS 2.0 | `https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=comments-rss2` | RSS 2.0 | Commentaires (non utilisé ici) |
**⚠️ Attention** :
- Le pattern `/feed/` **ne fonctionne pas** (retourne du HTML).
- Il faut **obligatoirement** utiliser `?feed=rss2` ou `?feed=atom`.
- La **REST API** (`?rest_route=/wp/v2/posts`) est **désactivée** (404).
---
### 5 bis.3 Structure des items RSS 2.0
Un item RSS 2.0 du blog contient les champs suivants :
```xml
<item>
<title>Titre de l'article</title>
<link>https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625</link>
<guid isPermaLink="false">https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625</guid>
<pubDate>Mon, 10 Aug 2026 09:00:11 +0000</pubDate>
<category>Administration</category>
<dc:creator>M. Dupont</dc:creator>
<description>Extrait en texte brut...</description>
<content:encoded><![CDATA[
<p>Contenu HTML complet de l'article...</p>
<a href="https://blogpeda.ac-bordeaux.fr/cjeliote/wp-content/uploads/2026/08/document.pdf">Lien vers un PDF</a>
<img src="..." alt="..." width="..." height="..." />
]]></content:encoded>
</item>
```
**Champs clés** :
| **Champ** | **Description** | **Format** | **Utilisation** |
|--------------------|-------------------------------------------------------------------------------|--------------------------|------------------------------------------|
| `<title>` | Titre de l'article. | Texte | Titre dans le message XMPP. |
| `<link>` | URL canonique de l'article. | URL | Lien cliquable dans le message. |
| `<guid>` | Identifiant unique (permalien ou non). | URL ou ID | **Clé de déduplication**. |
| `<pubDate>` | Date de publication (RFC 822). | `Mon, 10 Aug 2026 09:00:11 +0000` | Date de publication. |
| `<category>` | Catégorie de l'article (ex: `Administration`, `Pédagogie`). | Texte | Filtre ou affichage dans le message. |
| `<dc:creator>` | Auteur (souvent vide). | Texte | Optionnel (affichage si disponible). |
| `<description>` | Extrait en texte brut. | Texte | Texte court pour le message. |
| `<content:encoded>`| **Contenu HTML complet** de l'article. | HTML | **Source principale** pour le texte. |
**Flux Atom** :
- `<id>` : Identifiant unique (équivalent au GUID).
- `<published>` : Date de publication (ISO 8601).
- `<updated>` : Date de dernière modification (pour détecter les mises à jour).
- `<content type="html">` : Contenu HTML complet.
---
### 5 bis.4 Bibliothèque recommandée : `feedparser`
**Pourquoi `feedparser` ?**
- **Standard de facto** pour le parsing de flux RSS/Atom en Python.
- Gère **RSS 0.9x/1.0/2.0** et **Atom** de manière unifiée.
- **Normalise** les champs (ex: `published_parsed` pour les dates).
- Extrait automatiquement `<content:encoded>` dans `entry.content[0].value`.
- Compatible **Python 3.13+**.
**Installation** :
```bash
pip install feedparser
```
---
### 5 bis.5 Modèle de données : `BlogArticle`
Un article du blog est représenté par le modèle Pydantic suivant :
```python
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field
class BlogArticle(BaseModel):
"""
Représente un article du blog du collège.
Utilisé pour l'intégration dans les "informations diverses" du message XMPP.
"""
id: str = Field(..., description="GUID de l'article (clé de déduplication)")
title: str = Field(..., description="Titre de l'article")
url: str = Field(..., description="URL canonique de l'article")
published_at: datetime = Field(..., description="Date de publication (UTC)")
updated_at: Optional[datetime] = Field(
None, description="Date de dernière mise à jour (si disponible)"
)
category: Optional[str] = Field(None, description="Catégorie de l'article")
author: Optional[str] = Field(None, description="Auteur (si disponible)")
content_html: str = Field(..., description="Contenu HTML complet")
content_text: str = Field(..., description="Contenu en texte brut (pour XMPP)")
class Config:
frozen = True # Immuable
json_encoders = {
datetime: lambda v: v.isoformat(),
}
```
**Exemple d'utilisation** :
```python
article = BlogArticle(
id="https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625",
title="Réunion de rentrée",
url="https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625",
published_at=datetime(2026, 8, 10, 9, 0, 11, tzinfo=timezone.utc),
updated_at=None,
category="Administration",
author="M. Dupont",
content_html="<p>La réunion de rentrée aura lieu le 1er septembre.</p>",
content_text="La réunion de rentrée aura lieu le 1er septembre.",
)
```
---
### 5 bis.6 Intégration dans le modèle `ExternalInfo`
Les articles du blog sont agrégés avec d'autres sources externes (ex: messages Pronote) dans un modèle `ExternalInfo` :
```python
from typing import List
from datetime import datetime
from pydantic import BaseModel, Field
class ExternalInfo(BaseModel):
"""
Agrège les informations externes (blog, messages Pronote, etc.)
pour les intégrer dans le message XMPP.
"""
blog_articles: List[BlogArticle] = Field(
default_factory=list, description="Liste des nouveaux articles du blog"
)
pronote_messages: List[Message] = Field(
default_factory=list, description="Liste des messages Pronote"
)
other_info: List[str] = Field(
default_factory=list, description="Autres informations (extensible)"
)
class Config:
json_encoders = {
datetime: lambda v: v.isoformat(),
}
```
**Intégration dans `PronoteData`** :
```python
class PronoteData(BaseModel):
# ... champs existants ...
# Note: external_info est géré uniquement dans XmppMessage.
```
---
### 5 bis.7 Récupération et parsing du flux RSS
#### 5 bis.7.1 Client RSS (`sources/blog/rss.py`)
Le résultat d'une récupération est un modèle Pydantic figé, `BlogRSSFetchResult`
(module `sources/blog/result.py`) :
```python
from pydantic import BaseModel, ConfigDict, Field
from ..models.blog import BlogArticle
class BlogRSSFetchResult(BaseModel):
"""
Résultat d'une récupération du flux RSS du blog du collège.
Modèle figé (``frozen``) : les instances sont immuables après création.
"""
model_config = ConfigDict(frozen=True)
articles: tuple[BlogArticle, ...] = Field(
default=(),
description=(
"Nouveaux articles absents de known_guids, triés par date de "
"publication décroissante puis par identifiant croissant"
),
)
etag: str | None = Field(
default=None,
description="Valeur de l'en-tête ETag de la réponse RSS, si disponible",
)
last_modified: str | None = Field(
default=None,
description="Valeur de l'en-tête Last-Modified de la réponse RSS, si disponible",
)
not_modified: bool = Field(
default=False,
description="Vaut True si le serveur a répondu 304 Not Modified",
)
```
**Attributs** :
- `articles` : nouveaux articles absents de `known_guids`, triés par date de
publication décroissante puis par identifiant croissant. Tuple vide si aucun
nouvel article (ou en cas de réponse `304 Not Modified`).
- `etag` : valeur de l'en-tête `ETag` de la réponse RSS, ou `None` si
indisponible.
- `last_modified` : valeur de l'en-tête `Last-Modified` de la réponse RSS, ou
`None` si indisponible.
- `not_modified` : vaut `True` si le serveur a répondu `304 Not Modified`,
`False` sinon.
Client de récupération et de parsing du flux (`sources/blog/rss.py`) :
```python
import logging
import re
from datetime import UTC, datetime
from html import unescape
import feedparser
import requests
from bs4 import BeautifulSoup
from ..models.blog import BlogArticle
from ..sources.blog.result import BlogRSSFetchResult
from ..utils.redaction import redact_url
logger = logging.getLogger(__name__)
class BlogRSSClient:
"""
Client pour récupérer et parser le flux RSS du blog du collège.
"""
def __init__(self, rss_url: str, timeout: int = 20):
self.rss_url = rss_url
self.timeout = timeout
def fetch_and_parse(
self,
*,
known_guids: frozenset[str] | None = None,
etag: str | None = None,
last_modified: str | None = None,
) -> BlogRSSFetchResult:
"""
Récupère le flux RSS et parse les nouveaux articles.
Args:
known_guids: Ensemble des GUID d'articles déjà traités (pour la déduplication).
Si None, retourne tous les articles.
etag: Valeur de l'en-tête ``ETag`` mémorisée pour la requête conditionnelle, ou None.
last_modified: Valeur de l'en-tête ``Last-Modified`` mémorisée pour la requête
conditionnelle, ou None.
Returns:
Résultat de la récupération : nouveaux articles (triés par date de publication
décroissante puis par identifiant croissant), en-têtes de cache reçus et
indicateur ``304 Not Modified``.
"""
try:
# Transport HTTP séparé du parsing
headers: dict[str, str] = {"user-agent": "pronote-sync"}
if etag is not None:
headers["If-None-Match"] = etag
if last_modified is not None:
headers["If-Modified-Since"] = last_modified
response = requests.get(self.rss_url, headers=headers, timeout=self.timeout)
# 304 Not Modified : pas de nouveaux articles
if response.status_code == 304:
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=True,
)
# Rejeter les statuts d'erreur (4xx/5xx)
response.raise_for_status()
# Extraire les en-têtes de cache de la réponse (ETag / Last-Modified)
response_etag: str | None = response.headers.get("ETag")
response_last_modified: str | None = response.headers.get("Last-Modified")
# Parsing du contenu reçu (pas de l'URL)
feed = feedparser.parse(response.content)
# Flux invalide (XML malformé, etc.) : résultat vide, sans erreur ;
# les en-têtes de cache d'entrée sont conservés tels quels
if getattr(feed, "bozo", None):
logger.warning(
"Flux RSS du blog invalide, ignoré : %s",
redact_url(self.rss_url),
)
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=False,
)
articles: list[BlogArticle] = []
seen_guids: set[str] = set(known_guids) if known_guids is not None else set()
for entry in getattr(feed, "entries", []):
# Extraire le GUID (utiliser link si le GUID est vide)
guid = str(entry.get("id") or entry.get("link") or "")
# Ignorer les articles déjà connus ou en double dans le flux
if not guid or guid in seen_guids:
continue
# Parser la date de publication (RFC 822 ou ISO 8601)
published_at = self._parse_date(
entry.get("published_parsed") or entry.get("pubdate_parsed")
)
if published_at is None:
continue
# Parser la date de mise à jour (si disponible)
updated_at = self._parse_date(entry.get("updated_parsed"))
# Extraire le contenu HTML (content:encoded ou description)
raw_content = entry.get("content")
if raw_content:
content_html = str(raw_content[0].get("value") or "")
else:
content_html = str(entry.get("description") or "")
# Extraire la catégorie (tags ou champ category)
tags = entry.get("tags")
category_value = tags[0].get("term") if tags else None
if not category_value:
category_value = entry.get("category")
category = str(category_value) if category_value else None
# Créer l'article
articles.append(
BlogArticle(
id=guid,
title=str(entry.get("title") or guid),
url=str(entry.get("link") or guid),
published_at=published_at,
updated_at=updated_at,
category=category,
author=str(entry.get("author")) if entry.get("author") else None,
content_html=content_html,
content_text=self._html_to_text(content_html),
)
)
seen_guids.add(guid)
# Tri stable : d'abord par date de publication décroissante, puis par identifiant croissant
articles.sort(key=lambda article: article.id)
articles.sort(key=lambda article: article.published_at, reverse=True)
return BlogRSSFetchResult(
articles=tuple(articles),
etag=response_etag,
last_modified=response_last_modified,
not_modified=False,
)
except Exception as e:
safe_url = redact_url(self.rss_url)
logger.error(f"Échec de la récupération du flux RSS {safe_url}: {e}")
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=False,
)
@staticmethod
def _parse_date(date_tuple: tuple[int, ...] | None) -> datetime | None:
"""
Convertit un tuple de date (RFC 822 ou ISO 8601) en datetime UTC.
Args:
date_tuple: Tuple retourné par feedparser (ex: (2026, 8, 10, 9, 0, 11, 0, 1, -1)),
ou None si absent.
Returns:
datetime en UTC, ou None si le tuple est absent, vide ou invalide.
"""
if not date_tuple:
return None
# feedparser retourne un tuple struct_time (année, mois, jour, heure, minute, seconde, jour_semaine, jour_année, DST)
try:
return datetime(
date_tuple[0], # année
date_tuple[1], # mois
date_tuple[2], # jour
date_tuple[3], # heure
date_tuple[4], # minute
date_tuple[5], # seconde
tzinfo=UTC,
)
except (ValueError, IndexError):
return None
@staticmethod
def _html_to_text(html: str) -> str:
"""
Convertit du HTML en texte brut (supprime les balises, décode les entités).
Args:
html: Contenu HTML.
Returns:
Texte brut.
"""
if not html:
return ""
# Utiliser BeautifulSoup pour extraire le texte
soup = BeautifulSoup(html, "html.parser")
text = soup.get_text(separator=" ", strip=True)
# Décoder les entités HTML
text = unescape(text)
# Nettoyer les espaces multiples
text = re.sub(r"\s+", " ", text).strip()
return text
```
#### 5 bis.7.2 Déduplication et état local
La déduplication des articles du blog repose sur leur **GUID** (ou leur URL si le GUID est vide).
**Stratégie** :
1. Stocker un **fichier d'état local** (ex: `.blog_rss_state.json`) contenant la version du
format, l'**ensemble des GUID déjà traités** et les en-têtes de cache HTTP (`ETag` /
`Last-Modified`) de la dernière réponse.
2. À chaque récupération, ignorer les articles dont le GUID est **déjà présent** dans
l'ensemble des GUID connus.
3. Utiliser le **cache HTTP** (`If-Modified-Since` / `If-None-Match`) via `feedparser` pour
éviter les requêtes inutiles.
**Exemple de fichier d'état** :
```json
{
"version": 1,
"known_guids": [
"https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625",
"https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626"
],
"etag": "abc123",
"last_modified": "Wed, 01 Sep 2026 00:00:00 GMT"
}
```
Les GUID sont triés alphabétiquement pour une sortie JSON déterministe.
**Gestionnaire d'état** (`sources/blog/state.py`) :
```python
import json
import logging
from collections.abc import Iterable
from pathlib import Path
logger = logging.getLogger(__name__)
_STATE_VERSION = 1
class BlogRSSState:
"""
Gère l'état local pour la déduplication des articles du blog et le cache HTTP.
"""
def __init__(self, state_file: str = ".blog_rss_state.json"):
self._state_file = Path(state_file)
self._known_guids: set[str] = set()
self._etag: str | None = None
self._last_modified: str | None = None
self._load()
def _load(self) -> None:
"""Charge l'état depuis le fichier."""
if not self._state_file.exists():
return
try:
data = json.loads(self._state_file.read_text(encoding="utf-8"))
if not isinstance(data, dict) or data.get("version") != _STATE_VERSION:
logger.warning(
"Fichier d'état blog RSS : version absente ou non supportée, "
"démarrage avec un état vide."
)
return
guids_data = data.get("known_guids", [])
if isinstance(guids_data, list):
self._known_guids = {guid for guid in guids_data if isinstance(guid, str)}
etag_data = data.get("etag")
if isinstance(etag_data, str):
self._etag = etag_data
last_modified_data = data.get("last_modified")
if isinstance(last_modified_data, str):
self._last_modified = last_modified_data
except Exception as e:
logger.warning(f"Échec du chargement de l'état du blog: {e}")
self._known_guids = set()
self._etag = None
self._last_modified = None
def _save(self) -> None:
"""Sauvegarde l'état dans le fichier."""
payload = {
"version": _STATE_VERSION,
"known_guids": sorted(self._known_guids),
"etag": self._etag,
"last_modified": self._last_modified,
}
try:
with self._state_file.open("w", encoding="utf-8") as f:
json.dump(payload, f, indent=2)
except Exception as e:
logger.error(f"Échec de la sauvegarde de l'état du blog: {e}")
def get_known_guids(self) -> frozenset[str]:
"""Retourne une copie immuable des GUID connus."""
return frozenset(self._known_guids)
def add_guids(self, guids: Iterable[str]) -> None:
"""Ajoute des GUID à l'ensemble des GUID connus et sauvegarde."""
new_guids = set(guids)
if not new_guids:
return
self._known_guids.update(new_guids)
self._save()
def get_cache_headers(self) -> tuple[str | None, str | None]:
"""Retourne les en-têtes de cache HTTP mémorisés (etag, last_modified)."""
return self._etag, self._last_modified
def update_cache_headers(self, etag: str | None, last_modified: str | None) -> None:
"""Met à jour les en-têtes de cache HTTP et sauvegarde."""
self._etag = etag
self._last_modified = last_modified
self._save()
def clear(self) -> None:
"""Efface l'état (GUID et en-têtes de cache) et sauvegarde."""
self._known_guids = set()
self._etag = None
self._last_modified = None
self._save()
```
**Utilisation dans le pipeline** :
```python
# Initialisation
rss_client = BlogRSSClient(rss_url=settings.blog.rss_url)
blog_state = BlogRSSState()
# Récupération des nouveaux articles
known_guids = blog_state.get_known_guids()
etag, last_modified = blog_state.get_cache_headers()
result = rss_client.fetch_and_parse(
known_guids=known_guids,
etag=etag,
last_modified=last_modified,
)
# Mise à jour de l'état avec les nouveaux GUID et les en-têtes de cache
blog_state.add_guids(article.id for article in result.articles)
blog_state.update_cache_headers(result.etag, result.last_modified)
```
---
### 5 bis.8 Intégration dans le pipeline
#### 5 bis.8.1 Étape de récupération du blog (`pipeline/steps/fetch_blog.py`)
```python
from typing import List
from ..models.blog import BlogArticle
from ..sources.blog.rss import BlogRSSClient
from ..sources.blog.state import BlogRSSState
def fetch_blog_step(
rss_client: BlogRSSClient,
blog_state: BlogRSSState,
enabled: bool = True,
) -> List[BlogArticle]:
"""
Étape de récupération des articles du blog.
Args:
rss_client: Client RSS configuré.
blog_state: État local pour la déduplication et le cache HTTP.
enabled: Si False, retourne une liste vide.
Returns:
Liste des nouveaux articles.
"""
if not enabled:
return []
known_guids = blog_state.get_known_guids()
etag, last_modified = blog_state.get_cache_headers()
result = rss_client.fetch_and_parse(
known_guids=known_guids,
etag=etag,
last_modified=last_modified,
)
# Mettre à jour l'état avec les nouveaux GUID et les en-têtes de cache
blog_state.add_guids(article.id for article in result.articles)
blog_state.update_cache_headers(result.etag, result.last_modified)
return list(result.articles)
```
#### 5 bis.8.2 Intégration dans le pipeline principal
L'étape de récupération du blog est insérée **après la récupération Pronote** et **avant la synthèse IA** :
```python
# Dans pipeline/run.py
def run(self) -> Tuple[Optional[PronoteData], List[PipelineError]]:
# ... étapes existantes (fetch Pronote, normalize, compare) ...
# Étape 5 bis: Récupération du blog
try:
blog_articles = fetch_blog_step(
self.blog_rss_client,
self.blog_state,
enabled=self.settings.blog.enabled,
)
except PipelineError as e:
self._warnings.append(PipelineWarning(
message=f"Récupération du blog échouée: {e.message}",
step="fetch_blog",
))
blog_articles = []
# Intégration dans PronoteData
pronote_data.external_info.blog_articles = blog_articles
# ... suite du pipeline (synthèse IA, envoi XMPP) ...
```
---
### 5 bis.9 Configuration
#### 5 bis.9.1 Variables d'environnement
Ajouter les variables suivantes dans la configuration :
| **Variable** | **Description** | **Valeur par défaut** | **Type** |
|----------------------------|-------------------------------------------------------------------------------|-----------------------|-------------------|
| `BLOG_ENABLED` | Activer la récupération du blog. | `False` | `bool` |
| `BLOG_RSS_URL` | URL du flux RSS du blog. | `https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2` | `str` |
#### 5 bis.9.2 Modèle Pydantic pour la configuration du blog
```python
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class BlogSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="BLOG_", env_file=".env", extra="ignore")
enabled: bool = Field(False, description="Activer la récupération du blog")
rss_url: str = Field(
"https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2",
description="URL du flux RSS du blog",
)
```
**Intégration dans `Settings`** :
```python
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()
blog: BlogSettings = BlogSettings() # Nouveau
```
#### 5 bis.9.3 Exemple de configuration dans `.env`
```ini
# --- Blog du collège ---
BLOG_ENABLED=true
BLOG_RSS_URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2
```
---
### 5 bis.10 Tests
#### 5 bis.10.1 Fixtures RSS
Créer un fichier de test anonymisé dans `tests/fixtures/blog_rss.xml` :
```xml
<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0" xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:content="http://purl.org/rss/1.0/modules/content/">
<channel>
<title>Blog du Collège Jéliote</title>
<link>https://blogpeda.ac-bordeaux.fr/cjeliote/</link>
<description>Blog du collège</description>
<item>
<title>Réunion de rentrée</title>
<link>https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625</link>
<guid isPermaLink="false">https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625</guid>
<pubDate>Mon, 10 Aug 2026 09:00:11 +0000</pubDate>
<category>Administration</category>
<dc:creator>M. Dupont</dc:creator>
<description>La réunion de rentrée aura lieu le 1er septembre.</description>
<content:encoded><![CDATA[
<p>La réunion de rentrée aura lieu le <strong>1er septembre</strong> à 18h en salle 204.</p>
<p><a href="https://blogpeda.ac-bordeaux.fr/cjeliote/wp-content/uploads/2026/08/ordre-du-jour.pdf">Ordre du jour</a></p>
]]></content:encoded>
</item>
<item>
<title>Sortie pédagogique</title>
<link>https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626</link>
<guid isPermaLink="false">https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626</guid>
<pubDate>Tue, 11 Aug 2026 14:30:00 +0000</pubDate>
<category>Pédagogie</category>
<dc:creator>Mme Martin</dc:creator>
<description>Sortie prévue au musée le 15 septembre.</description>
<content:encoded><![CDATA[
<p>Une sortie pédagogique au <a href="https://musee.example.com">musée</a> est prévue le 15 septembre.</p>
]]></content:encoded>
</item>
</channel>
</rss>
```
#### 5 bis.10.2 Tests unitaires
**Test du parsing RSS** :
```python
import pytest
from datetime import datetime, timezone
from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.models.blog import BlogArticle
@pytest.fixture
def mock_blog_rss_client():
"""Retourne un client RSS mocké pour les tests."""
client = BlogRSSClient(rss_url="file://tests/fixtures/blog_rss.xml")
return client
@pytest.mark.unittest
def test_parse_blog_rss(mock_blog_rss_client):
"""Test le parsing d'un flux RSS du blog."""
result = mock_blog_rss_client.fetch_and_parse()
assert len(result.articles) == 2
# Vérifier le premier article
article1 = result.articles[0]
assert article1.title == "Sortie pédagogique"
assert article1.url == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626"
assert article1.category == "Pédagogie"
assert article1.author == "Mme Martin"
assert "musée" in article1.content_text
assert article1.published_at == datetime(2026, 8, 11, 14, 30, 0, tzinfo=timezone.utc)
# Vérifier le deuxième article
article2 = result.articles[1]
assert article2.title == "Réunion de rentrée"
assert article2.url == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625"
assert article2.category == "Administration"
assert "1er septembre" in article2.content_text
```
**Test de la déduplication** :
```python
@pytest.mark.unittest
def test_blog_deduplication(tmp_path):
"""Test la déduplication des articles du blog."""
from pronote_sync.sources.blog.state import BlogRSSState
# Créer un fichier d'état temporaire
state_file = tmp_path / "blog_state.json"
state = BlogRSSState(state_file=str(state_file))
# Initialement, aucun article connu
assert state.get_known_guids() == frozenset()
# Simuler la récupération de 2 articles
state.add_guids(["https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625"])
assert "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625" in state.get_known_guids()
# Simuler une nouvelle récupération : seul le nouvel article doit être retourné
client = BlogRSSClient(rss_url="file://tests/fixtures/blog_rss.xml")
result = client.fetch_and_parse(known_guids=state.get_known_guids())
# Seul l'article avec p=1626 doit être retourné (car p=1625 est déjà connu)
assert len(result.articles) == 1
assert result.articles[0].id == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626"
```
---
### 5 bis.11 Intégration dans le message XMPP
Les articles du blog sont intégrés dans la section **"Informations diverses"** du message XMPP :
```python
# Dans channels/xmpp.py, méthode _format_message
def _format_message(self, message: XmppMessage) -> str:
lines = []
# ... sections existantes (synthèse, changements, devoirs) ...
# Informations diverses (blog + messages Pronote)
if message.external_info.blog_articles or message.external_info.pronote_messages:
lines.append("📢 Informations diverses :")
# Articles du blog
for article in message.external_info.blog_articles:
lines.append(f" - [{article.category or 'Info'}] {article.title} ({article.published_at.strftime('%d/%m')})")
lines.append(f" {article.content_text[:100]}...") # Extrait court
lines.append(f" 🔗 {article.url}")
# Messages Pronote
for msg in message.external_info.pronote_messages:
lines.append(f" - [Message] {msg.title} (de {msg.author})")
lines.append("")
return "\n".join(lines)
```
---
### 5 bis.12 Points clés
| **Aspect** | **Décision** | **Justification** |
|--------------------------|-------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------|
| **Flux RSS** | Utiliser `?feed=rss2` (pas `/feed/`). | Seule URL fonctionnelle sur WordPress Multisite. |
| **Bibliothèque** | `feedparser` | Standard, gère RSS/Atom, compatible Python 3.13+. |
| **Déduplication** | Par GUID (ou URL si GUID vide). | Éviter les doublons entre les exécutions. |
| **Cache HTTP** | Utiliser `If-Modified-Since` / `If-None-Match` (géré par `feedparser`). | Limiter les requêtes inutiles. |
| **Intégration XMPP** | Dans "Informations diverses" (avec les messages Pronote). | Cohérence avec l'objectif de synthèse globale. |
| **Contenu** | Utiliser `<content:encoded>` (HTML complet) + conversion en texte brut. | Le HTML contient toutes les informations (liens, images). |
| **Sécurité** | Masquer les URLs dans les logs/erreurs. | Éviter les fuites de données (même si le blog est public). |
| **Tests** | Fixtures XML anonymisées + mocks. | Pas de dépendance réseau, données reproductibles. |
---
### 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` :
1. **Format des événements** :
- Les cours sont des événements **chronométrés** (`DTSTART` et `DTEND` avec heure, ex: `DTSTART:20260905T080000Z`).
- Les jours fériés/vacances sont des événements **tout le jour** (`DTSTART;VALUE=DATE`, ex: `DTSTART;VALUE=DATE:20260920`).
2. **Catégories et statuts** :
- `CATEGORIES: Cours - Cours annulé` → Statut **annulé** (`STATUS:CANCELLED` dans iCal).
- `CATEGORIES: Cours - Cours déplacé` → Statut **déplacé** (à traiter comme une modification).
- `CATEGORIES: Congés` ou `Vacances` → Événement de type **vacances** (ex: `SUMMARY:Vacances de Noël`).
3. **Description HTML** :
La `DESCRIPTION` contient des balises HTML avec des **labels en français** :
```html
<div>
Matière : Mathématiques
Professeur : M. Dupont
Salle : 204
Groupe : Classe entière
<strong>Contenu pédagogique :
</strong>
Résoudre des équations du second degré.
<strong>Pour le 10/09/2026 :
</strong>
Exercices 1 à 5 page 42.
<strong>Donné le 05/09/2026 :
</strong>
Exercices 1 à 5 page 42.
</div>
```
- **Structure réelle** :
- La `DESCRIPTION` contient d'abord un **en-tête texte** (avant le premier `<strong>`) avec des lignes au format `Label : Valeur`.
- Les labels d'en-tête sont : `Matière :`, `Professeur :` ou `Professeurs :`, `Salle :` ou `Salles :`, `Groupe :`.
- Le **corps HTML** commence à partir du premier `<strong>`.
- Les sections sont identifiées par les balises exactes :
- `<strong>Contenu pédagogique : \n</strong>` → contenu pédagogique (texte brut).
- `<strong>Pour le JJ/MM/AAAA : \n</strong>` → devoir à faire (date d'échéance).
- `<strong>Donné le JJ/MM/AAAA : \n</strong>` → devoir donné (date d'attribution).
- **Devoirs en double** : Les devoirs apparaissent **deux fois** :
- Une fois sous `Pour le JJ/MM/AAAA` (date d'échéance).
- Une fois sous `Donné le JJ/MM/AAAA` (date de distribution).
- **Déduplication nécessaire** (voir [Section 5.1.4](#514-déduplication-des-devoirs)).
4. **UID des événements** :
- Format réel observé : `Cours-16027-1-20260904T120218Z-Index-Education` (le préfixe peut varier : `Cours-...`, `Edt_...`, etc.).
- **Suffixes temporels** : Le suffixe `-YYYYMMDDTHHMMSSZ-Index-Education` (ou `-Index-Education` si pas de timestamp) change à chaque export.
- **Normalisation requise** : Supprimer le suffixe final `-YYYYMMDDTHHMMSSZ-Index-Education` ou `-Index-Education` pour obtenir un UID stable. Si aucun UID exploitable n'existe, produire un UID déterministe par hachage de champs clés (date de début, date de fin, matière, professeur, salle, groupe).
5. **Nom du calendrier** :
- Lu depuis `X-WR-CALNAME` (ex: `X-WR-CALNAME:Edt Jean DUPONT 4ème A`).
- La bibliothèque `icalendar` ignore les propriétés `X-WR-*` avec paramètres → **lecture manuelle** du texte source (voir [Section 5.1.2](#512-récupération-du-nom-du-calendrier)).
6. **Validation du flux** :
- Pronote renvoie du **HTML** (ex: page "Session expirée") si le token `icalsecurise` est invalide ou expiré.
- **Validation minimale** : Vérifier que le flux contient `BEGIN:VCALENDAR`.
#### 5.1.2 Règles du jour cible
La règle du jour cible est implémentée dans `src/core/calendar.ts` (fonction `resolveTarget`).
Elle détermine la date pour laquelle générer le digest (planning ou devoirs).
**3 cas exacts** :
1. Si **J+1** a au moins un cours → `school-day` à J+1.
2. Sinon, si **J** a des cours **ET** qu'il existe un prochain jour avec cours → `school-day` au prochain jour avec cours (ex: vendredi → lundi).
3. Sinon → `no-school` à J+1, avec libellé de vacances si applicable, et date de reprise si connue.
**Pseudocode** :
```
Si cours_existent(J+1) :
retourner J+1 (school-day)
Sinon si cours_existent(J) ET prochain_jour_avec_cours_existe :
retourner prochain_jour_avec_cours (school-day)
Sinon :
retourner J+1 (no-school, avec libellé de vacances si applicable)
```
**Exemple Python** :
```python
from typing import Optional, Tuple, List
from datetime import date, timedelta
from ..models.agenda import Lesson, SchoolEvent
def resolve_target_day(
today: date,
lessons: List[Lesson],
school_events: List[SchoolEvent],
) -> Tuple[date, str, Optional[date], Optional[str]]:
"""
Détermine le jour cible pour le digest.
Args:
today: Date du jour.
lessons: Liste des cours.
school_events: Liste des événements scolaires (vacances).
Returns:
Tuple (date_cible, type, date_de_reprise, holiday_label).
- type : "school-day" ou "no-school".
- date_de_reprise : Date de reprise si en vacances, sinon None.
- holiday_label : Libellé des vacances si applicable, sinon None.
"""
tomorrow = today + timedelta(days=1)
# Fonction helper pour vérifier si un jour a des cours
def has_lessons(day: date) -> bool:
return any(lesson.start.date() == day for lesson in lessons)
# Cas 1 : J+1 a des cours
if has_lessons(tomorrow):
return (tomorrow, "school-day", None, None)
# Cas 2 : J a des cours ET il existe un prochain jour avec cours
if has_lessons(today):
# Trouver le prochain jour avec cours après J (recherche illimitée)
next_day = today + timedelta(days=1)
while True:
if has_lessons(next_day):
return (next_day, "school-day", None, None)
next_day += timedelta(days=1)
# Cas 3 : Aucun cours à J+1 ou J → no-school
# Vérifier si J+1 est en vacances
holiday_label = None
resume_date = None
for event in school_events:
if event.kind == "holiday" and event.from_date <= tomorrow < event.to_date:
holiday_label = event.label
resume_date = event.to_date
break
return (tomorrow, "no-school", resume_date, holiday_label)
```
**Cas de test à couvrir** :
| Scénario | Entrée (J) | Sortie attendue (date cible) | Type | Date de reprise | Libellé vacances |
|-----------------------------------|------------------|-------------------------------|---------------|-----------------|------------------|
| J+1 a des cours | Lundi | Mardi | school-day | None | None |
| J a des cours, J+1 sans cours | Vendredi | Lundi | school-day | None | None |
| J+1 en vacances | Veille de vacances | J+1 | no-school | Fin des vacances| "Vacances de Noël" |
| J en vacances | Dimanche | Lundi | no-school | Fin des vacances| "Vacances de Noël" |
| Aucune activité (week-end normal) | Samedi | Dimanche | no-school | None | None |
#### 5.1.3 Récupération du flux iCal (`sources/pronote/ical.py`)
```python
import requests
from typing import Optional
from urllib.parse import urlparse
from .redaction import redact_url, redact_secrets
from ..models.agenda import RawCalendarData
def fetch_ical(url: str, timeout: int = 20) -> str:
"""
Récupère le flux iCal depuis une URL Pronote.
Inspiré de src/sources/pronote/fetch.ts.
Args:
url: URL du flux iCal (peut être file:// pour les tests).
timeout: Timeout en secondes (défaut: 20s).
Returns:
Contenu brut du flux iCal.
Raises:
ValueError: Si le flux est invalide (pas de BEGIN:VCALENDAR).
requests.exceptions.RequestException: En cas d'erreur HTTP.
"""
headers = {
"accept": "text/calendar, */*;q=0.5",
"user-agent": "pronote-sync",
}
# Gestion des URLs file:// pour les tests
if url.startswith("file://"):
import pathlib
file_path = pathlib.Path(url.replace("file://", ""))
content = file_path.read_text(encoding="utf-8")
if "BEGIN:VCALENDAR" not in content:
raise ValueError(f"Fichier iCal invalide: {redact_url(url)}")
return content
# Récupération HTTP
try:
response = requests.get(
url,
headers=headers,
timeout=timeout,
)
response.raise_for_status()
content = response.text
except requests.exceptions.RequestException as e:
# Masquer l'URL et les détails de l'erreur
safe_url = redact_url(url)
safe_error = redact_secrets(str(e))
raise requests.exceptions.RequestException(
f"Échec de la récupération de {safe_url}: {safe_error}"
) from None
# Validation du flux
if "BEGIN:VCALENDAR" not in content:
raise ValueError(
f"Flux iCal invalide (pas de BEGIN:VCALENDAR) pour {redact_url(url)}"
)
return content
def get_calendar_name(raw_ical: str) -> Optional[str]:
"""
Extrait le nom du calendrier depuis X-WR-CALNAME.
Args:
raw_ical: Contenu brut du flux iCal.
Returns:
Nom du calendrier ou None.
"""
import re
# Recherche de X-WR-CALNAME (peut être sur une ligne ou plié)
match = re.search(r"X-WR-CALNAME:(.+?)(?:\r?\n|$)", raw_ical)
if match:
return match.group(1).strip()
return None
```
#### 5.1.4 Déduplication des devoirs
Les devoirs apparaissent **deux fois** dans le flux iCal Pronote :
- Une fois sous `Pour le JJ/MM/AAAA` (date d'échéance).
- Une fois sous `Donné le JJ/MM/AAAA` (date de distribution).
**Stratégie** (inspirée de `src/core/homework.ts`) :
1. **Collecter tous les blocs homework** de tous les VEVENT d'abord.
2. **Passe 1** : Parcourir **globalement** les blocs `Pour le` (devoirs à faire) dont la date correspond à la date d'échéance cible.
3. **Passe 2** : Parcourir **globalement** les blocs `Donné le` sur les cours du jour cible (pour les flux sans `Pour le`).
4. **Clé de déduplication** : Texte normalisé (`text.replace(/\s+/g, ' ').trim().toLowerCase()`).
5. **ID stable** : `sha1([due_on, key]).slice(0, 12)`.
6. **Tri** : Par matière puis texte (locale française).
**Implémentation** :
```python
import re
import hashlib
from datetime import date
from typing import Optional, List
from ..models.agenda import Lesson
from ..models.homework import Homework as HomeworkModel
def normalize_homework_text(text: str) -> str:
"""
Normalise le texte d'un devoir pour la déduplication.
Inspiré de src/core/homework.ts.
Args:
text: Texte brut du devoir.
Returns:
Texte normalisé (espaces unifiés, minuscules, sans balises HTML).
"""
# Remplacer les espaces multiples par un seul
text = re.sub(r"\s+", " ", text)
# Supprimer les balises HTML
text = re.sub(r"<[^>]+>", "", text)
# Trim et minuscules
return text.strip().lower()
def generate_homework_id(due_on: date, normalized_text: str) -> str:
"""
Génère un ID stable pour un devoir.
Inspiré de src/core/homework.ts.
Args:
due_on: Date d'échéance (requise).
normalized_text: Texte normalisé du devoir.
Returns:
ID stable (12 premiers caractères du hash SHA-1).
"""
due_on_str = due_on.isoformat()
key = f"{due_on_str}|{normalized_text}"
return hashlib.sha1(key.encode("utf-8")).hexdigest()[:12]
def collect_homeworks(lessons: List[Lesson], target_date: date) -> List[HomeworkModel]:
"""
Collecte et déduplique les devoirs en deux passes globales.
Inspiré de src/core/homework.ts.
Args:
lessons: Liste de tous les cours (VEVENT) parsés.
target_date: Date cible pour laquelle collecter les devoirs.
Returns:
Liste unique de devoirs, triée par matière puis texte.
"""
by_text: dict[str, HomeworkModel] = {}
# Passe 1: blocs "Pour le" (due) — date d'échéance
for lesson in lessons:
for block in lesson.homework_blocks:
if block.kind == "due" and block.date == target_date:
key = normalize_homework_text(block.text)
if key not in by_text:
by_text[key] = HomeworkModel(
id=generate_homework_id(target_date, key),
subject=lesson.subject,
teachers=lesson.teachers,
assigned_on=lesson.start.date(),
due_on=block.date,
text=block.text,
html=block.html,
)
# Passe 2: blocs "Donné le" (assigned) — cours du jour cible
for lesson in lessons:
if not lesson.start.date() == target_date:
continue
for block in lesson.homework_blocks:
if block.kind == "assigned":
key = normalize_homework_text(block.text)
if key not in by_text:
by_text[key] = HomeworkModel(
id=generate_homework_id(target_date, key),
subject=lesson.subject,
teachers=lesson.teachers,
assigned_on=block.date,
due_on=target_date,
text=block.text,
html=block.html,
)
return sorted(by_text.values(), key=lambda h: (h.subject.lower(), h.text.lower()))
```
**Algorithme de déduplication** :
- Les deux passes partagent la même `Map` (clé normalisée → devoir).
- Un devoir présent dans `Pour le` **ET** `Donné le` est compté **une seule fois** (first-in-wins).
- **Résultat** : Liste unique de devoirs, triée par matière puis texte.
- **Note** : `Homework.due_on` est toujours requis (pour les blocs `Donné le`, on déduit `due_on = target_date`).
**Intégration dans le pipeline** :
La fonction `collect_homeworks(lessons, target_date)` est appelée **après le parsing** de tous les VEVENT,
une fois que `target_date` est connu (via `resolve_target_day`).
Cela permet de :
1. Parcourir tous les cours pour extraire les blocs `Pour le` et `Donné le`.
2. Dédupliquer globalement les devoirs en utilisant la clé normalisée.
3. Retourner une liste unique de devoirs, triée par matière puis texte.
**Exemple d'utilisation dans le pipeline** :
```python
# Après parsing de tous les VEVENT et résolution de target_date
target_date = resolve_target_day(today, lessons, school_events)[0]
homeworks = collect_homeworks(lessons, target_date)
```
#### 5.1.5 Normalisation des UID
> ⚠️ **Décision d'implémentation** :
> La fonction est nommée `normalize_pronote_uid` (et non `normalize_uid` comme référencé dans TODO.md).
> `generate_deterministic_uid` utilise `hashlib.sha1(payload, usedforsecurity=False)` pour satisfaire la règle bandit B324 (le hachage n'est pas utilisé pour la sécurité).
Les UID Pronote contiennent des **suffixes temporels** qui changent à chaque export. Exemple réel :
```
UID:Cours-16027-1-20260904T120218Z-Index-Education
```
**Solution** :
1. Supprimer le suffixe final `-YYYYMMDDTHHMMSSZ-Index-Education` (ou `-Index-Education` si pas de timestamp).
2. Si aucun UID exploitable n'existe, générer un UID déterministe par hachage des champs clés (date de début, date de fin, matière, professeur, salle, groupe).
```python
import re
import hashlib
from datetime import datetime
from typing import Optional
def normalize_pronote_uid(uid: str) -> str:
"""
Normalise un UID Pronote en supprimant le suffixe temporel.
Inspiré de src/sources/pronote/parse.ts (lignes 161-164).
Args:
uid: UID brut de l'événement Pronote.
Returns:
UID stable sans suffixe temporel.
"""
# Supprimer le suffixe -YYYYMMDDTHHMMSSZ-Index-Education ou -Index-Education
normalized = re.sub(r"-\d{8}T\d{6}Z-Index-Education$", "", uid)
normalized = re.sub(r"-Index-Education$", "", normalized)
return normalized
def generate_deterministic_uid(
start: datetime,
end: datetime,
subject: str,
teachers: list[str],
rooms: list[str],
group: Optional[str] = None,
) -> str:
"""
Génère un UID déterministe si aucun UID exploitable n'existe.
Args:
start: Date/heure de début du cours.
end: Date/heure de fin du cours.
subject: Matière.
teachers: Liste des professeurs.
rooms: Liste des salles.
group: Groupe (optionnel).
Returns:
UID déterministe (hash SHA-1 des champs clés).
"""
key_parts = [
start.isoformat(),
end.isoformat(),
subject,
",".join(sorted(teachers)),
",".join(sorted(rooms)),
group or "",
]
key = "|".join(key_parts)
return hashlib.sha1(key.encode("utf-8")).hexdigest()[:12]
#### 5.1.6 Parsing complet du flux iCal (`sources/pronote/ical.py`)
```python
from typing import List, Optional, Tuple
from datetime import datetime, date
from icalendar import Calendar, Event
from ..models.agenda import Lesson, Homework, SchoolEvent, LessonStatus
from ..models.homework import Homework as HomeworkModel
def split_header_and_body(description: str) -> Tuple[str, str]:
"""
Sépare l'en-tête texte du corps HTML dans la DESCRIPTION.
L'en-tête est avant le premier `<strong>`, le corps HTML commence à partir du premier `<strong>`.
Args:
description: Contenu brut de la DESCRIPTION.
Returns:
Tuple (en-tête texte, corps HTML).
"""
# Trouver la position du premier `<strong>`
strong_start = description.find("<strong>")
if strong_start == -1:
return description, ""
header = description[:strong_start].strip()
body = description[strong_start:]
return header, body
def parse_header(header: str) -> dict:
"""
Parse l'en-tête texte pour extraire les métadonnées du cours.
Les labels sont : `Matière :`, `Professeur :`/`Professeurs :`, `Salle :`/`Salles :`, `Groupe :`.
Args:
header: En-tête texte (avant le premier `<strong>`).
Returns:
Dictionnaire avec les champs : subject, teachers, rooms, group.
"""
import re
from html import unescape
result = {
"subject": "",
"teachers": [],
"rooms": [],
"group": None,
}
# Parser chaque ligne de l'en-tête (format : `Label : Valeur`)
for line in header.split("\n"):
line = line.strip()
if not line:
continue
# Extraire le label et la valeur
match = re.match(r"^([^:]+) :\s*(.+)$", line)
if not match:
continue
label = match.group(1).strip()
value = unescape(match.group(2).strip())
if label.lower() == "matière":
result["subject"] = value
elif label.lower() in ("professeur", "professeurs"):
# Split sur les virgules pour les professeurs multiples
result["teachers"] = [t.strip() for t in value.split(",") if t.strip()]
elif label.lower() in ("salle", "salles"):
# Split sur les virgules pour les salles multiples
result["rooms"] = [r.strip() for r in value.split(",") if r.strip()]
elif label.lower() == "groupe":
result["group"] = value
return result
def parse_body(body: str) -> Tuple[Optional[str], List[dict]]:
"""
Parse le corps HTML pour extraire le contenu pédagogique et les devoirs.
Args:
body: Corps HTML (à partir du premier `<strong>`).
Returns:
Tuple (contenu pédagogique, liste des devoirs).
"""
import re
from html import unescape
content = None
homeworks = []
# Extraire le contenu pédagogique
content_match = re.search(
r"<strong>Contenu pédagogique : \n</strong>(.+?)(?:<strong>|$)",
body,
re.DOTALL,
)
if content_match:
content_html = content_match.group(1).strip()
# Nettoyer les balises HTML pour le texte brut
content = re.sub(r"<[^>]+>", "", content_html)
content = unescape(content).strip()
# Extraire les devoirs (blocs "Pour le" et "Donné le")
# Passe 1 : blocs "Pour le JJ/MM/AAAA" (date d'échéance)
pour_le_matches = re.finditer(
r"<strong>Pour le (\d{2}/\d{2}/\d{4}) : \n</strong>(.+?)(?:<strong>|$)",
body,
re.DOTALL,
)
for match in pour_le_matches:
due_date_str = match.group(1)
text_html = match.group(2).strip()
# Nettoyer le texte pour la clé de déduplication
text_clean = re.sub(r"<[^>]+>", "", text_html)
text_clean = unescape(text_clean).strip()
# Parser la date (format JJ/MM/AAAA → AAAA-MM-JJ)
try:
due_date = datetime.strptime(due_date_str, "%d/%m/%Y").date()
except ValueError:
continue
homeworks.append({
"type": "due",
"date": due_date,
"text": text_clean,
"html": text_html,
})
# Passe 2 : blocs "Donné le JJ/MM/AAAA" (date d'attribution)
donne_le_matches = re.finditer(
r"<strong>Donné le (\d{2}/\d{2}/\d{4}) : \n</strong>(.+?)(?:<strong>|$)",
body,
re.DOTALL,
)
for match in donne_le_matches:
assigned_date_str = match.group(1)
text_html = match.group(2).strip()
# Nettoyer le texte
text_clean = re.sub(r"<[^>]+>", "", text_html)
text_clean = unescape(text_clean).strip()
# Parser la date
try:
assigned_date = datetime.strptime(assigned_date_str, "%d/%m/%Y").date()
except ValueError:
continue
homeworks.append({
"type": "assigned",
"date": assigned_date,
"text": text_clean,
"html": text_html,
})
return content, homeworks
def parse_homework_blocks(body: str) -> List[dict]:
"""
Parse le corps HTML pour extraire les blocs de devoirs (Pour le / Donné le).
Args:
body: Corps HTML (à partir du premier `<strong>`).
Returns:
Liste des blocs de devoirs avec type, date, texte et HTML.
"""
homeworks = []
# Passe 1 : blocs "Pour le JJ/MM/AAAA" (date d'échéance)
pour_le_matches = re.finditer(
r"<strong>Pour le (\d{2}/\d{2}/\d{4}) : \n</strong>(.+?)(?:<strong>|$)",
body,
re.DOTALL,
)
for match in pour_le_matches:
due_date_str = match.group(1)
text_html = match.group(2).strip()
# Nettoyer le texte pour la clé de déduplication
text_clean = re.sub(r"<[^>]+>", "", text_html)
text_clean = unescape(text_clean).strip()
# Parser la date (format JJ/MM/AAAA → AAAA-MM-JJ)
try:
due_date = datetime.strptime(due_date_str, "%d/%m/%Y").date()
except ValueError:
continue
homeworks.append({
"type": "due",
"date": due_date,
"text": text_clean,
"html": text_html,
})
# Passe 2 : blocs "Donné le JJ/MM/AAAA" (date d'attribution)
donne_le_matches = re.finditer(
r"<strong>Donné le (\d{2}/\d{2}/\d{4}) : \n</strong>(.+?)(?:<strong>|$)",
body,
re.DOTALL,
)
for match in donne_le_matches:
assigned_date_str = match.group(1)
text_html = match.group(2).strip()
# Nettoyer le texte
text_clean = re.sub(r"<[^>]+>", "", text_html)
text_clean = unescape(text_clean).strip()
# Parser la date
try:
assigned_date = datetime.strptime(assigned_date_str, "%d/%m/%Y").date()
except ValueError:
continue
homeworks.append({
"type": "assigned",
"date": assigned_date,
"text": text_clean,
"html": text_html,
})
return homeworks
def parse_ical(raw_ical: str) -> tuple[List[Lesson], List[HomeworkModel], List[SchoolEvent]]:
"""
Parse un flux iCal Pronote en événements typés.
Args:
raw_ical: Contenu brut du flux iCal.
Returns:
Tuple (lessons, homeworks, school_events).
- lessons : Liste des cours avec leurs blocs de devoirs bruts (homework_blocks).
- homeworks : **Toujours vide** (la collecte/déduplication se fait plus tard dans le pipeline via `collect_homeworks(lessons, target_date)`).
- school_events : Liste des événements scolaires (vacances).
**Note importante** :
La déduplication globale des devoirs est effectuée **après le parsing** de tous les VEVENT,
une fois que `target_date` est connu (via `resolve_target_day`).
Voir la section [5.1.4 Déduplication des devoirs](#514-déduplication-des-devoirs) pour plus de détails.
"""
cal = Calendar.from_ical(raw_ical)
lessons: List[Lesson] = []
homeworks: List[HomeworkModel] = [] # Toujours vide : la collecte se fait via collect_homeworks(lessons, target_date)
school_events: List[SchoolEvent] = []
for component in cal.walk():
if not isinstance(component, Event):
continue
# Déterminer le type d'événement
categories = getattr(component, "categories", None)
if categories:
categories = [c.to_unicode() for c in categories.cats]
else:
categories = []
# Événements de type "vacances"
if any(cat in ["Congés", "Vacances"] for cat in categories):
school_events.append(SchoolEvent(
kind="holiday",
label=str(component.get("summary")),
from_date=component.get("dtstart").dt,
to_date=component.get("dtend").dt,
))
continue
# Cours annulés ou déplacés
status = LessonStatus.NORMAL
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
# Parsing de la description
description = str(component.get("description", ""))
header, body = split_header_and_body(description)
# Parser l'en-tête pour les métadonnées du cours
lesson_data = parse_header(header)
# Parser le corps pour le contenu et les devoirs
content, raw_homeworks = parse_body(body)
# Créer le cours avec les blocs de devoirs bruts (pour déduplication globale)
start = component.get("dtstart").dt
end = component.get("dtend").dt
uid = normalize_pronote_uid(str(component.get("uid")))
lesson = Lesson(
id=uid,
start=start,
end=end,
subject=lesson_data.get("subject", ""),
teachers=lesson_data.get("teachers", []),
rooms=lesson_data.get("rooms", []),
group=lesson_data.get("group"),
status=status,
content=content,
homework_blocks=raw_homeworks, # Stockage temporaire pour déduplication globale
)
lessons.append(lesson)
return lessons, homeworks, school_events
```
#### 5.1.7 Client `pronotepy`
`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
import pronotepy.ent as pronote_ent
from pronotepy import ParentClient
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")
client = ParentClient(
settings.url,
settings.username,
settings.password.get_secret_value(),
ent=ent_function,
)
```
Le client applicatif expose des méthodes distinctes :
- `get_lessons(start, end) -> list[Lesson]` ;
- `get_homeworks(start, end) -> list[Homework]` ;
- `get_messages() -> list[Message]` ;
- `get_informations() -> list[Message]`.
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`)
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é. |
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
| **Aspect** | **iCal** | **pronotepy** | **Recommandation** |
|--------------------------|-----------------------------------|-----------------------------------|--------------------------------------------|
| **Agenda** | ✅ Disponible | ✅ Disponible | Préférer iCal (officiel, stable). |
| **Devoirs** | ✅ Si activé par l'établissement | ✅ Toujours disponible | Préférer iCal si disponible. |
| **Messages** | ❌ Non disponible | ✅ Disponible | Utiliser pronotepy. |
| **Informations** | ❌ Non disponible | ✅ Disponible | Utiliser pronotepy. |
| **Stabilité** | ✅ Très stable | ⚠️ Peut casser (reverse-engineering) | iCal en priorité. |
| **Authentification** | ❌ Pas besoin (token dans URL) | ✅ Nécessaire (login/mot de passe) | Masquer les secrets. |
| **Performances** | ✅ Rapide (1 requête HTTP) | ⚠️ Plus lent (plusieurs requêtes) | iCal en priorité. |
---
## 6. Modèle de données Pydantic
> ⚠️ **Décision d'implémentation (M3)** :
> Tous les modèles utilisent `model_config = ConfigDict(frozen=True)` (Pydantic v2), et non `class Config: frozen = True`.
> Pas de `json_encoders` : la sérialisation ISO pour `datetime`/`date` est native dans Pydantic v2.
> Les énumérations utilisent `enum.StrEnum` (Python 3.11+) au lieu de `(str, Enum)` (règle ruff UP042).
> Dans `HomeworkBlock`, le champ `date` utilise un alias d'import `_date` (`from datetime import date as _date`) pour éviter un conflit de nom/champ dans Pydantic v2. Même chose pour `time` → `_time` dans `TheoreticalLesson`.
> `XmppMessage.external_info` est de type `ExternalInfo | None` avec une valeur par défaut `None` (et non `default_factory=ExternalInfo`) : le pipeline passe `None` lorsqu'il n'y a pas d'informations externes.
> Les modèles sont répartis en 10 modules domaines (agenda, homework, message, blog, diff, pronote, sync, synthesis, xmpp) avec `__init__.py` réexportant les 22 noms via `__all__`.
### 6.1 Principes
- **Modèles distincts par domaine** : Ne pas créer un unique modèle fourre-tout. Chaque étape du pipeline utilise des modèles dédiés (décision architecturale).
- **Validation stricte** : Utiliser Pydantic pour valider les données dès leur création.
- **Immuabilité** : Les modèles de **contrat** (ex: `Lesson`, `Homework`, `Message`) doivent être immuables (`frozen=True`). Les modèles de **travail** (ex: `CalDAVSyncResult`, `PronoteData`) peuvent être mutables pour permettre les mises à jour progressives.
- **Sérialisation** : Tous les modèles doivent supporter la sérialisation JSON (pour l'archivage et les tests).
### 6.2 Modèles de base (`models/__init__.py`)
```python
from datetime import datetime, date, time
from typing import List, Optional, Literal
from enum import Enum
from pydantic import BaseModel, Field, validator
# --- Types de base ---
class Status(str, Enum):
"""Statut générique pour les événements."""
NORMAL = "normal"
CANCELLED = "cancelled"
MOVED = "moved"
# --- Modèles d'agenda ---
class LessonStatus(str, Enum):
"""Statut d'un cours."""
NORMAL = "normal"
CANCELLED = "cancelled"
MOVED = "moved"
class HomeworkBlock(BaseModel):
"""
Représente un bloc de devoir extrait de la description d'un cours.
Utilisé pour la déduplication globale des devoirs.
"""
kind: Literal["due", "assigned"] = Field(..., description="Type de bloc (échéance ou attribution)")
date: date = Field(..., description="Date associée au bloc")
text: str = Field(..., description="Texte du devoir (brut)")
html: str = Field(default="", description="Texte du devoir (HTML)")
class Lesson(BaseModel):
"""
Représente un cours dans l'agenda Pronote.
Équivalent de `Lesson` dans src/core/model.ts.
"""
id: str = Field(..., description="UID stable du cours (normalisé)")
start: datetime = Field(..., description="Date/heure de début")
end: datetime = Field(..., description="Date/heure de fin")
subject: str = Field(..., description="Matière (ex: Mathématiques)")
teachers: List[str] = Field(default_factory=list, description="Liste des professeurs")
rooms: List[str] = Field(default_factory=list, description="Liste des salles")
group: Optional[str] = Field(None, description="Groupe (ex: Classe entière)")
status: LessonStatus = Field(LessonStatus.NORMAL, description="Statut du cours")
content: Optional[str] = Field(None, description="Contenu pédagogique")
homework_blocks: List[HomeworkBlock] = Field(
default_factory=list, description="Blocs de devoirs extraits de la description"
)
class Config:
frozen = True # Immuable
json_encoders = {
datetime: lambda v: v.isoformat(),
}
class SchoolEventKind(str, Enum):
"""Type d'événement scolaire."""
HOLIDAY = "holiday"
PUBLIC_HOLIDAY = "public_holiday"
class SchoolEvent(BaseModel):
"""
Représente un événement scolaire (vacances, jours fériés).
Équivalent de `SchoolEvent` dans src/core/model.ts.
"""
kind: SchoolEventKind = Field(..., description="Type d'événement")
label: str = Field(..., description="Libellé (ex: Vacances de Noël)")
from_date: date = Field(..., description="Date de début (inclusive)")
to_date: date = Field(..., description="Date de fin (exclusive)")
class Config:
frozen = True
class TheoreticalLesson(BaseModel):
"""
Représente un cours dans l'agenda théorique.
Utilisé pour la comparaison avec l'agenda réel.
"""
id: str = Field(..., description="Identifiant unique")
day_of_week: int = Field(..., description="Jour de la semaine (0=lundi, 6=dimanche)")
start_time: time = Field(..., description="Heure de début")
end_time: time = Field(..., description="Heure de fin")
subject: str = Field(..., description="Matière")
teachers: List[str] = Field(default_factory=list, description="Liste des professeurs")
rooms: List[str] = Field(default_factory=list, description="Liste des salles")
class Config:
frozen = True
# --- Modèles de devoirs ---
class Homework(BaseModel):
"""
Représente un devoir.
Équivalent de `Homework` dans src/core/model.ts.
"""
id: str = Field(..., description="ID stable (hachage)")
subject: str = Field(..., description="Matière")
teachers: List[str] = Field(default_factory=list, description="Liste des professeurs")
assigned_on: Optional[date] = Field(None, description="Date de distribution")
due_on: date = Field(..., description="Date d'échéance")
text: str = Field(..., description="Texte du devoir (brut)")
html: str = Field(default="", description="Texte du devoir (HTML)")
class Config:
frozen = True
# --- Modèles de messages ---
class MessageType(str, Enum):
"""Type de message Pronote."""
DISCUSSION = "discussion"
INFORMATION = "information"
SURVEY = "survey"
class Message(BaseModel):
"""
Représente un message ou une information Pronote.
"""
id: str = Field(..., description="Identifiant unique")
type: MessageType = Field(..., description="Type de message")
title: str = Field(..., description="Titre")
content: str = Field(..., description="Contenu")
author: str = Field(..., description="Auteur")
date: datetime = Field(..., description="Date de création")
read: bool = Field(False, description="Lu ou non")
class Config:
frozen = True
# --- Modèles de comparaison ---
class AgendaChangeType(str, Enum):
"""Type de changement dans l'agenda."""
ADDED = "added"
REMOVED = "removed"
MODIFIED = "modified"
class AgendaChange(BaseModel):
"""
Représente un changement entre l'agenda réel et l'agenda théorique.
"""
type: AgendaChangeType = Field(..., description="Type de changement")
lesson: Optional[Lesson] = Field(None, description="Cours concerné (pour ADDED/MODIFIED)")
theoretical_lesson: Optional[TheoreticalLesson] = Field(
None, description="Cours théorique concerné (pour REMOVED/MODIFIED)"
)
details: str = Field(default="", description="Détails du changement")
class Config:
frozen = True
class AgendaDiff(BaseModel):
"""
Représente les différences entre l'agenda réel et l'agenda théorique.
"""
target_date: date = Field(..., description="Date cible de la comparaison")
changes: List[AgendaChange] = Field(default_factory=list, description="Liste des changements")
class Config:
frozen = True
# --- Modèle XMPP ---
class XmppMessage(BaseModel):
"""
Représente le message final à envoyer par XMPP.
**Sépare clairement la synthèse IA et la liste brute des devoirs** (décision [5](#5-synthèse-ia---protocole-pas-de-sdk-imposé)).
"""
target_date: date = Field(..., description="Date cible")
synthesis: Optional[str] = Field(
None,
description="Synthèse IA (optionnelle). 3-5 phrases, ton chaleureux et sobre."
)
homeworks: List[Homework] = Field(
default_factory=list,
description="Liste **brute** des devoirs (non modifiée par l'IA)"
)
changes: List[AgendaChange] = Field(
default_factory=list,
description="Liste des changements d'agenda"
)
messages: List[Message] = Field(
default_factory=list,
description="Liste des messages/informations importants"
)
external_info: ExternalInfo = Field(
default_factory=ExternalInfo,
description="Informations externes (blog, messages Pronote)"
)
class Config:
frozen = True
# --- Modèle de sync CalDAV ---
class CalDAVSyncStatus(str, Enum):
"""Statut de synchronisation CalDAV."""
SUCCESS = "success"
FAILED = "failed"
SKIPPED = "skipped" # Déjà synchronisé
# --- Modèles de données normalisées (Pronote) ---
class PronoteData(BaseModel):
"""
Données normalisées récupérées depuis Pronote.
Résultat de l'étape de récupération et normalisation.
"""
lessons: List[Lesson] = Field(default_factory=list, description="Liste des cours")
homeworks: List[Homework] = Field(default_factory=list, description="Liste des devoirs")
school_events: List[SchoolEvent] = Field(
default_factory=list,
description="Liste des événements scolaires"
)
messages: List[Message] = Field(
default_factory=list,
description="Liste des messages/informations"
)
target_date: date = Field(..., description="Date cible (J+1 ou prochain jour scolaire)")
generated_at: datetime = Field(..., description="Date/heure de génération")
class Config:
json_encoders = {
datetime: lambda v: v.isoformat(),
date: lambda v: v.isoformat(),
}
# --- Modèles de synchronisation CalDAV ---
class CalDAVSyncPlan(BaseModel):
"""
Plan de synchronisation CalDAV.
Contient les événements à ajouter, modifier ou supprimer.
"""
lessons_to_add: List[Lesson] = Field(default_factory=list)
lessons_to_update: List[Lesson] = Field(default_factory=list)
lessons_to_remove: List[str] = Field(default_factory=list) # Liste d'UID
homeworks_to_add: List[Homework] = Field(default_factory=list)
homeworks_to_update: List[Homework] = Field(default_factory=list)
homeworks_to_remove: List[str] = Field(default_factory=list) # Liste d'UID
school_events_to_add: List[SchoolEvent] = Field(default_factory=list)
school_events_to_update: List[SchoolEvent] = Field(default_factory=list)
school_events_to_remove: List[str] = Field(default_factory=list) # Liste d'UID
class CalDAVSyncResult(BaseModel):
"""
Résultat d'une synchronisation CalDAV.
**Mutable** : les compteurs sont incrémentés pendant la synchronisation.
"""
status: CalDAVSyncStatus = Field(..., description="Statut global")
added: int = Field(0, description="Nombre d'événements ajoutés")
updated: int = Field(0, description="Nombre d'événements mis à jour")
removed: int = Field(0, description="Nombre d'événements supprimés")
errors: List[str] = Field(default_factory=list, description="Liste des erreurs")
# --- Modèles de synthèse IA ---
class SynthesisInput(BaseModel):
"""
Entrée pour la synthèse IA.
Contient les données nécessaires à la génération de la synthèse.
"""
agenda_diff: Optional[AgendaDiff] = Field(None, description="Différences d'agenda")
messages: List[Message] = Field(default_factory=list, description="Messages importants")
school_events: List[SchoolEvent] = Field(default_factory=list, description="Événements scolaires")
target_date: date = Field(..., description="Date cible")
class SynthesisResult(BaseModel):
"""
Résultat de la synthèse IA.
"""
text: Optional[str] = Field(None, description="Texte de la synthèse IA")
### 6.3 Points clés
- **Immuabilité** : Les modèles de **contrat** (ex: `Lesson`, `Homework`, `Message`, `XmppMessage`) utilisent `frozen=True`. Les modèles de **travail** (ex: `CalDAVSyncResult`, `PronoteData`) sont mutables pour permettre les mises à jour progressives.
- **Sérialisation** : Les champs `datetime` et `date` sont sérialisés en ISO format pour JSON.
- **Validation** : Pydantic valide automatiquement les types et les contraintes (ex: `date` doit être un objet `date` valide).
- **Séparation des préoccupations** :
- `PronoteData` : Données normalisées de Pronote.
- `AgendaDiff` : Résultat de la comparaison avec l'agenda théorique.
- `CalDAVSyncPlan`/`CalDAVSyncResult` : Plan et résultat de la synchronisation CalDAV.
- `SynthesisInput`/`SynthesisResult` : Entrée et sortie de la synthèse IA.
- `XmppMessage` : Message prêt à être envoyé.
---
## 7. Synchronisation CalDAV
### 7.1 Principes
- **Synchronisation différentielle** : Ne pas écraser le calendrier distant, mais **mettre à jour uniquement les événements modifiés** (décision [3](#3-synchronisation-caldav---différentielle-avec-uid-stables)).
- **UID stables** : Utiliser des UID normalisés pour éviter les doublons.
- **Canonicalisation à la frontière des sources** : les UID sont normalisés une seule fois à la frontière des sources via `utils.uid.normalize_pronote_uid`, afin qu'un même cours provenant d'iCal ou de `pronotepy` produise le **même identifiant canonique** (aucun doublon ni suppression/ajout artificiel lors d'un changement de source).
- **Fenêtre de synchronisation** : Configurable via `SYNC_PAST_DAYS` et `SYNC_FUTURE_DAYS`.
- **Mode dry-run** : Obligatoire pour tester sans modifier le calendrier distant.
- **Idempotence** : Deux exécutions identiques **doivent** produire le même état CalDAV.
- **Scan du calendrier distant** : La synchronisation repose sur le **scan du calendrier CalDAV distant** (les événements gérés sont relus à chaque exécution) — **aucun état local** n'est conservé (voir §7.3).
- **Plan explicite** : Le `CalDAVSyncPlan` (ajouts / mises à jour / suppressions) est **calculé explicitement avant l'exécution** de la synchronisation (voir §7.2).
- **Événements non gérés** : Les événements **non marqués** `X-PRONOTE-SYNC-MANAGED: v1` **ne sont jamais modifiés ni supprimés** : ils appartiennent à d'autres outils ou à l'utilisateur.
### 7.2 Client CalDAV (`sync/caldav.py`)
Utilisation de la bibliothèque [`caldav`](https://pypi.org/project/caldav/) (Python 3.8+, maintenue).
> **⚠️ Code illustratif** : le bloc de code ci-dessous est **illustratif** : il montre les
> règles métier de la synchronisation (marqueur, comparaison, cours annulés, dry-run).
> L'**implémentation réelle** doit s'adapter à la version de la bibliothèque `caldav`
> installée (`caldav>=1.3.0`) : l'**API réelle** documentée ci-dessous **prévaut** sur les
> anciens appels encore présents dans l'exemple (ex: `calendar(name=...)`,
> `calendar.add_event(...)`, `event.properties`, `vobject_instance`).
#### API réelle (`caldav>=1.3.0`)
- **Connexion** : `caldav.DAVClient(url, username, password)` — les paramètres proviennent
de `CalDAVSettings` (`CALDAV_URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`).
- **Résolution du calendrier** : `DAVClient.principal()` puis `principal.calendars()` ;
sélectionner le calendrier dont l'URL correspond à **`CalDAVSettings.calendar_path`**
(ex: `/pronote-sync/`). La résolution ne se fait **pas** par nom de calendrier :
`calendar_name` est abandonné au profit de `calendar_path`.
- **Lecture des événements** : `calendar.objects()` liste les objets du calendrier
(`calendar.date_search(start=..., end=...)` reste utilisable pour une fenêtre selon la
version installée).
- **Contenu iCalendar** : chaque objet expose `event.icalendar_component` (un
`icalendar.Event`) donnant accès aux propriétés (`uid`, `summary`, `dtstart`, `dtend`,
`status`, `categories`, `X-PRONOTE-SYNC-MANAGED`).
- **Ajout** : `calendar.save_event(ical_text)` crée un événement (UID normalisé et
marqueur inclus).
- **Mise à jour** : modifier les propriétés de l'`icalendar_component` puis
`event.save()` (si la version installée le supporte), sinon supprimer puis recréer sur
le même UID via `save_event()`.
- **Suppression** : `event.delete()` — **uniquement** pour les événements marqués.
**Règles métier conservées** (indépendantes de la version de `caldav`) :
- **Marqueur** : chaque événement géré porte `X-PRONOTE-SYNC-MANAGED: v1`.
- **Comparaison `_events_equal`** : ne compare que les **champs gérés** (UID, DTSTART,
DTEND, SUMMARY, DESCRIPTION, STATUS, CATEGORIES, marqueur) et ignore les propriétés
**volatiles** (`DTSTAMP`, `CREATED`, `LAST-MODIFIED`) qui changent à chaque écriture
côté serveur.
- **Cours annulés** : conservés avec `STATUS:CANCELLED` (ne pas supprimer).
- **Mode dry-run** : logue le plan sans écrire sur le calendrier distant.
#### Approche en trois phases
1. **Collecte / scan** : lister les événements du calendrier distant (fenêtre
`SYNC_PAST_DAYS` / `SYNC_FUTURE_DAYS`) et ne retenir que les événements **gérés**
(`X-PRONOTE-SYNC-MANAGED: v1`), indexés par UID normalisé (`normalize_pronote_uid`).
2. **Calcul du plan** : comparer (via `_events_equal`) les événements distants gérés avec
les données Pronote (cours, devoirs, événements scolaires) et produire un
**`CalDAVSyncPlan` explicite** (`lessons_to_add`, `lessons_to_update`,
`lessons_to_remove`, etc.) — **aucune écriture** à ce stade.
3. **Exécution** : appliquer le plan (ajouts via `calendar.save_event()`, mises à jour si
les événements diffèrent, suppressions si l'UID est absent des données Pronote) ; en
mode `dry_run`, loguer le plan **sans rien écrire**.
Exemple minimal (API réelle) :
```python
import caldav
# CalDAVSettings : url, username, password (SecretStr), calendar_path = "/pronote-sync/"
settings = None # instance de CalDAVSettings (pydantic-settings)
client = caldav.DAVClient(
url=settings.url,
username=settings.username,
password=settings.password.get_secret_value(),
)
principal = client.principal()
calendar = next(
c for c in principal.calendars()
if str(c.url).rstrip("/").endswith(settings.calendar_path.rstrip("/"))
)
for obj in calendar.objects():
vevent = obj.icalendar_component # icalendar.Event
if vevent.get("X-PRONOTE-SYNC-MANAGED") == "v1":
print(vevent.get("uid"))
```
**Exemple illustratif** (règles métier complètes — API partiellement ancienne) :
```python
from typing import List, Optional, Dict, Any
from datetime import datetime, timedelta
import caldav
from caldav.elements import DAVCalendar, DAVEvent
from ..models.agenda import Lesson, Homework, SchoolEvent
from ..models.sync import CalDAVSyncResult, CalDAVSyncStatus
from ..utils.uid import normalize_pronote_uid
import logging
logger = logging.getLogger(__name__)
class CalDAVClient:
"""
Client pour interagir avec un serveur CalDAV.
Gère la synchronisation différentielle des événements Pronote.
"""
# Marqueur pour identifier les événements gérés par l'outil
MANAGED_PROPERTY = "X-PRONOTE-SYNC-MANAGED"
MANAGED_VALUE = "v1"
def __init__(
self,
url: str,
username: str,
password: str,
calendar_path: str = "/pronote-sync/",
dry_run: bool = False,
):
self.url = url
self.username = username
self.password = password
self.calendar_path = calendar_path
self.dry_run = dry_run
self._client: Optional[caldav.DAVClient] = None
self._calendar: Optional[DAVCalendar] = None
def connect(self) -> None:
"""Établit la connexion au serveur CalDAV."""
self._client = caldav.DAVClient(
url=self.url,
username=self.username,
password=self.password,
)
# Résoudre le calendrier via calendar_path (cf. CalDAVSettings) :
# API réelle (caldav>=1.3.0) : principal.calendars() puis correspondance
# sur l'URL du calendrier.
principal = self._client.principal()
matches = [
c for c in principal.calendars()
if str(c.url).rstrip("/").endswith(self.calendar_path.rstrip("/"))
]
if matches:
self._calendar = matches[0]
else:
# Pas de création automatique : la résolution se fait par chemin uniquement.
logger.warning(
f"Calendrier {self.calendar_path} introuvable"
f"{' et dry_run activé. Aucune modification ne sera effectuée.' if self.dry_run else ' : vérifier CalDAVSettings.calendar_path.'}"
)
self._calendar = None
def _is_managed_event(self, event: DAVEvent) -> bool:
"""Vérifie si un événement est géré par l'outil."""
# API réelle : event.icalendar_component (icalendar.Event)
vevent = event.icalendar_component
managed = vevent.get(self.MANAGED_PROPERTY)
return managed is not None and str(managed) == self.MANAGED_VALUE
def _get_event_uid(self, event: DAVEvent) -> str:
"""Récupère l'UID normalisé d'un événement."""
uid = str(event.icalendar_component.get("uid"))
return normalize_pronote_uid(uid)
def _build_event(
self,
lesson: Lesson,
) -> DAVEvent:
"""Construit un événement CalDAV à partir d'un cours Pronote."""
from icalendar import Event, vDatetime, vDate, vText, vUri
event = Event()
event.add("uid", vUri(lesson.id))
event.add("summary", vText(lesson.subject))
event.add("dtstart", vDatetime(lesson.start))
event.add("dtend", vDatetime(lesson.end))
# Ajouter les professeurs et salles dans la description
teachers = ", ".join(lesson.teachers) if lesson.teachers else ""
rooms = ", ".join(lesson.rooms) if lesson.rooms else ""
description = f"Matière : {lesson.subject}\n"
if teachers:
description += f"Professeur(s) : {teachers}\n"
if rooms:
description += f"Salle(s) : {rooms}\n"
if lesson.content:
description += f"\nContenu : {lesson.content}"
event.add("description", vText(description))
# Statut
if lesson.status == LessonStatus.CANCELLED:
event.add("status", "CANCELLED")
elif lesson.status == LessonStatus.MOVED:
event.add("status", "CONFIRMED") # ou un statut personnalisé
else:
event.add("status", "CONFIRMED")
# Marqueur pour identifier les événements gérés
event.add(self.MANAGED_PROPERTY, self.MANAGED_VALUE)
# Catégories
categories = ["Pronote"]
if lesson.status == LessonStatus.CANCELLED:
categories.append("Annulé")
elif lesson.status == LessonStatus.MOVED:
categories.append("Déplacé")
event.add("categories", categories)
return DAVEvent(event)
def _build_homework_event(self, homework: Homework) -> DAVEvent:
"""Construit un événement CalDAV à partir d'un devoir."""
from icalendar import Event, vDatetime, vDate, vText, vUri
# Utiliser la date d'échéance comme date de début/fin
due_date = homework.due_on
start = datetime(due_date.year, due_date.month, due_date.day, 8, 0, 0)
end = datetime(due_date.year, due_date.month, due_date.day, 18, 0, 0)
event = Event()
event.add("uid", vUri(f"homework-{homework.id}"))
event.add("summary", vText(f"Devoir : {homework.subject}"))
event.add("dtstart", vDatetime(start))
event.add("dtend", vDatetime(end))
# Description
description = f"Matière : {homework.subject}\n"
description += f"À faire pour le : {due_date.strftime('%d/%m/%Y')}\n"
description += f"\n{homework.text}"
event.add("description", vText(description))
# Statut : Tâche (TODO)
event.add("status", "NEEDS-ACTION")
# Marqueur
event.add(self.MANAGED_PROPERTY, self.MANAGED_VALUE)
event.add("categories", ["Pronote", "Devoir"])
return DAVEvent(event)
def _build_school_event_event(self, school_event: SchoolEvent) -> DAVEvent:
"""Construit un événement CalDAV à partir d'un événement scolaire."""
from icalendar import Event, vDate, vText, vUri
event = Event()
event.add("uid", vUri(f"school-event-{school_event.label}-{school_event.from_date.isoformat()}"))
event.add("summary", vText(school_event.label))
event.add("dtstart", vDate(school_event.from_date))
event.add("dtend", vDate(school_event.to_date))
# Statut
event.add("status", "CONFIRMED")
# Marqueur
event.add(self.MANAGED_PROPERTY, self.MANAGED_VALUE)
event.add("categories", ["Pronote", school_event.kind.value])
return DAVEvent(event)
def _events_equal(self, event1: DAVEvent, event2: DAVEvent) -> bool:
"""
Compare les champs gérés pour déterminer si une mise à jour est nécessaire.
Seuls les champs explicitement gérés par l'outil sont comparés :
UID, DTSTART, DTEND, SUMMARY, DESCRIPTION, STATUS, CATEGORIES, et
X-PRONOTE-SYNC-MANAGED.
Les propriétés volatiles (DTSTAMP, CREATED, LAST-MODIFIED) sont
volontairement **exclues** de la comparaison : elles sont modifiées par le
serveur à chaque écriture et ne reflètent aucun changement Pronote.
Args:
event1: Événement existant dans CalDAV.
event2: Nouvel événement à synchroniser.
Returns:
True si les événements sont identiques pour les champs gérés, False sinon.
"""
# Comparaison des UID normalisés
if self._get_event_uid(event1) != self._get_event_uid(event2):
return False
# Comparaison des champs gérés (API réelle : icalendar_component)
vobj1 = event1.icalendar_component
vobj2 = event2.icalendar_component
# DTSTART et DTEND
if vobj1.get("dtstart").dt != vobj2.get("dtstart").dt:
return False
if vobj1.get("dtend").dt != vobj2.get("dtend").dt:
return False
# SUMMARY
if str(vobj1.get("summary")) != str(vobj2.get("summary")):
return False
# DESCRIPTION
if str(vobj1.get("description")) != str(vobj2.get("description")):
return False
# STATUS
if str(vobj1.get("status")) != str(vobj2.get("status")):
return False
# CATEGORIES (comparaison des listes)
cats1 = [str(c) for c in vobj1.get("categories", [])]
cats2 = [str(c) for c in vobj2.get("categories", [])]
if sorted(cats1) != sorted(cats2):
return False
# X-PRONOTE-SYNC-MANAGED (doit toujours être présent et égal)
if str(vobj1.get(self.MANAGED_PROPERTY)) != str(vobj2.get(self.MANAGED_PROPERTY)):
return False
return True
def sync(
self,
lessons: List[Lesson],
homeworks: List[Homework],
school_events: List[SchoolEvent],
past_days: int = 7,
future_days: int = 30,
) -> CalDAVSyncResult:
"""
Synchronise les événements Pronote vers CalDAV.
**Idempotent** : Deux exécutions identiques sans changement externe ne modifient pas le calendrier.
Args:
lessons: Liste des cours à synchroniser.
homeworks: Liste des devoirs à synchroniser.
school_events: Liste des événements scolaires à synchroniser.
past_days: Nombre de jours dans le passé à synchroniser.
future_days: Nombre de jours dans le futur à synchroniser.
Returns:
Résultat de la synchronisation.
"""
if self._calendar is None:
return CalDAVSyncResult(
status=CalDAVSyncStatus.SKIPPED,
errors=["Aucun calendrier disponible (dry_run ou erreur de connexion)"],
)
result = CalDAVSyncResult(status=CalDAVSyncStatus.SUCCESS)
# Calculer la plage de dates
today = datetime.now().date()
start_date = today - timedelta(days=past_days)
end_date = today + timedelta(days=future_days)
# Récupérer les événements existants dans la plage
try:
existing_events = list(
self._calendar.date_search(
start=start_date,
end=end_date,
expand=True,
)
)
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de la récupération des événements CalDAV: {safe_error}")
return CalDAVSyncResult(
status=CalDAVSyncStatus.FAILED,
errors=[f"Échec de la récupération des événements: {safe_error}"],
)
# Indexer les événements existants par UID normalisé
existing_by_uid: Dict[str, DAVEvent] = {}
for event in existing_events:
if self._is_managed_event(event):
uid = self._get_event_uid(event)
existing_by_uid[uid] = event
# **Tests d'idempotence** : Deux exécutions consécutives avec les mêmes données
# ne doivent effectuer **aucune écriture** (result.added = 0, result.updated = 0, result.removed = 0).
# Voir les tests dans `tests/integration/test_caldav.py` (ex: `test_sync_idempotent`).
# Synchroniser les cours
for lesson in lessons:
if not (start_date <= lesson.start.date() <= end_date):
continue
uid = lesson.id
if uid in existing_by_uid:
# Comparer l'événement existant avec le nouvel événement
existing_event = existing_by_uid[uid]
new_event = self._build_event(lesson)
# Ne mettre à jour que si les événements diffèrent
if not self._events_equal(existing_event, new_event):
if not self.dry_run:
try:
existing_event.vobject_instance = new_event.vobject_instance
existing_event.save()
result.updated += 1
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de la mise à jour de {uid}: {safe_error}")
result.errors.append(f"Mise à jour {uid}: {safe_error}")
else:
result.updated += 1
logger.info(f"[DRY-RUN] Mise à jour de {uid}")
# Sinon, aucun changement : ne pas compter comme mise à jour
else:
# Ajouter un nouvel événement
new_event = self._build_event(lesson)
if not self.dry_run:
try:
self._calendar.add_event(new_event)
result.added += 1
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de l'ajout de {uid}: {safe_error}")
result.errors.append(f"Ajout {uid}: {safe_error}")
else:
result.added += 1
logger.info(f"[DRY-RUN] Ajout de {uid}")
# Synchroniser les devoirs
for homework in homeworks:
if not (start_date <= homework.due_on <= end_date):
continue
uid = f"homework-{homework.id}"
if uid in existing_by_uid:
# Comparer l'événement existant avec le nouvel événement
existing_event = existing_by_uid[uid]
new_event = self._build_homework_event(homework)
# Ne mettre à jour que si les événements diffèrent
if not self._events_equal(existing_event, new_event):
if not self.dry_run:
try:
existing_event.vobject_instance = new_event.vobject_instance
existing_event.save()
result.updated += 1
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de la mise à jour du devoir {uid}: {safe_error}")
result.errors.append(f"Mise à jour devoir {uid}: {safe_error}")
else:
result.updated += 1
logger.info(f"[DRY-RUN] Mise à jour du devoir {uid}")
else:
# Ajouter
new_event = self._build_homework_event(homework)
if not self.dry_run:
try:
self._calendar.add_event(new_event)
result.added += 1
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de l'ajout du devoir {uid}: {safe_error}")
result.errors.append(f"Ajout devoir {uid}: {safe_error}")
else:
result.added += 1
logger.info(f"[DRY-RUN] Ajout du devoir {uid}")
# Synchroniser les événements scolaires
for school_event in school_events:
if not (school_event.from_date >= start_date and school_event.to_date <= end_date):
continue
uid = f"school-event-{school_event.label}-{school_event.from_date.isoformat()}"
if uid in existing_by_uid:
# Comparer l'événement existant avec le nouvel événement
existing_event = existing_by_uid[uid]
new_event = self._build_school_event_event(school_event)
# Ne mettre à jour que si les événements diffèrent
if not self._events_equal(existing_event, new_event):
if not self.dry_run:
try:
existing_event.vobject_instance = new_event.vobject_instance
existing_event.save()
result.updated += 1
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de la mise à jour de l'événement {uid}: {safe_error}")
result.errors.append(f"Mise à jour événement {uid}: {safe_error}")
else:
result.updated += 1
logger.info(f"[DRY-RUN] Mise à jour de l'événement {uid}")
else:
# Ajouter
new_event = self._build_school_event_event(school_event)
if not self.dry_run:
try:
self._calendar.add_event(new_event)
result.added += 1
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de l'ajout de l'événement {uid}: {safe_error}")
result.errors.append(f"Ajout événement {uid}: {safe_error}")
else:
result.added += 1
logger.info(f"[DRY-RUN] Ajout de l'événement {uid}")
# Supprimer les événements gérés qui n'existent plus
# **À implémenter avec prudence** :
# - Ne supprimer que les événements marqués comme gérés.
# - Vérifier qu'ils ne sont plus dans les listes lessons/homeworks/school_events.
# Exemple :
current_uids = {
lesson.id for lesson in lessons
} | {
f"homework-{hw.id}" for hw in homeworks
} | {
f"school-event-{se.label}-{se.from_date.isoformat()}" for se in school_events
}
for uid, event in existing_by_uid.items():
if uid not in current_uids:
if not self.dry_run:
try:
event.delete()
result.removed += 1
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de la suppression de {uid}: {safe_error}")
result.errors.append(f"Suppression {uid}: {safe_error}")
else:
result.removed += 1
logger.info(f"[DRY-RUN] Suppression de {uid}")
# Définir le statut final
if result.errors:
result.status = CalDAVSyncStatus.FAILED
elif result.added == 0 and result.updated == 0 and result.removed == 0:
result.status = CalDAVSyncStatus.SKIPPED
return result
def close(self) -> None:
"""Fermeture de la connexion."""
self._client = None
self._calendar = None
```
### 7.3 État de synchronisation
**M7 ne stocke aucun état local** : il n'existe ni fichier d'état ni module
`sync/state.py`.
- **Scan du calendrier distant** : à chaque exécution, la synchronisation **scanne le
calendrier CalDAV distant** pour retrouver les événements gérés (marqueur
`X-PRONOTE-SYNC-MANAGED: v1`), indexés par UID normalisé.
- **Le calendrier distant est la source de vérité** : la comparaison entre événements
distants gérés et données Pronote se fait directement sur le calendrier, sans fichier
intermédiaire. Cela garantit une **idempotence naturelle** (deux exécutions identiques
produisent le même état) et supprime les risques liés à un fichier local (corruption,
perte, fuite de données, permissions `chmod 600`, exclusion `.gitignore` ou des
sauvegardes).
> **Évolution future** : si les performances l'exigent (calendrier très chargé, scans
> trop coûteux), un **état local** (fichier JSON, SQLite ou sync-token CalDAV) pourra
> être ajouté dans un jalon ultérieur, sans changer le contrat de la synchronisation
> (§7.1, §7.2 et §7.4 restent valables).
### 7.4 Points clés
- **Différentielle** : La synchronisation compare les UID existants avec ceux à synchroniser.
- **Idempotence** : Deux exécutions identiques ne modifient pas le calendrier.
- **Dry-run** : Mode obligatoire pour tester sans effet de bord.
- **Marquage** : Les événements gérés sont marqués avec `X-PRONOTE-SYNC-MANAGED: v1` pour éviter les conflits.
- **Cours annulés** : Conservés avec `STATUS:CANCELLED` (ne pas supprimer).
- **Plan explicite** : Le `CalDAVSyncPlan` est calculé avant l'exécution.
- **Pas d'état local** : La comparaison se fait avec le calendrier distant (scan).
- **Événements non gérés** : Les événements non marqués ne sont jamais modifiés ni supprimés.
---
## 8. Comparaison avec l'agenda théorique
### 8.1 Principes
- **Agenda théorique** : Représente l'emploi du temps **attendu** (ex: emploi du temps officiel de l'établissement).
- **Agenda réel** : Représente l'emploi du temps **réel** (récupéré depuis Pronote).
- **Objectif** : Détecter les **changements** (ajouts, suppressions, modifications) entre les deux.
- **Format JSON** : L'agenda théorique est désormais décrit par un **fichier JSON** (et non plus iCal/CSV), avec gestion de la **parité des semaines** (paire/impaire) et des **vacances scolaires**.
- **Parité des semaines** : Chaque leçon peut s'appliquer à **toutes** les semaines (`all`), uniquement aux semaines **paires** (`even`) ou **impaires** (`odd`). La parité d'une date est calculée par rapport à une **date de référence** configurée.
- **Vacances scolaires** : Un **fichier JSON séparé** liste les périodes de vacances (ex: zone A) ; aucune leçon théorique n'est produite pendant ces périodes.
- **Encapsulation** : Le provider encapsule **en interne** le calcul de la parité et le filtrage des vacances ; l'appelant (ex: `AgendaComparator`) reçoit simplement les cours théoriques ou une liste vide.
- **Matching déterministe** : Utiliser des règles claires pour associer un cours réel à un cours théorique (décision [4](#4-agenda-théorique---interface-abstraite--implémentation-fichier)).
### 8.2 Interface `TheoreticalAgendaProvider` (`sources/theoretical/provider.py`)
```python
from typing import Protocol, List, Optional
from datetime import date, time
from ..models.agenda import TheoreticalLesson
class TheoreticalAgendaProvider(Protocol):
"""
Protocole pour les fournisseurs d'agenda théorique.
Permet de changer facilement la source (fichier, API, etc.).
"""
def get_lessons(self, date: date) -> List[TheoreticalLesson]:
"""
Récupère les cours théoriques pour une date donnée.
Args:
date: Date pour laquelle récupérer les cours.
Returns:
Liste des cours théoriques.
"""
...
def get_lessons_for_range(
self,
start_date: date,
end_date: date,
) -> List[TheoreticalLesson]:
"""
Récupère les cours théoriques pour une plage de dates.
Args:
start_date: Date de début (inclusive).
end_date: Date de fin (inclusive).
Returns:
Liste des cours théoriques.
"""
...
```
> **Note sur le périmètre du provider** : Le fournisseur gère **en interne** la **parité des semaines** (paire/impaire) et les **vacances scolaires**. L'appelant ne connaît ni la date de référence de parité, ni les périodes de vacances : en période de vacances (ou pour une semaine dont la parité ne correspond à aucune leçon), il reçoit simplement une **liste vide**. Le contrat `TheoreticalAgendaProvider` reste donc volontairement minimal et stable.
### 8.3 Implémentation par fichier (`sources/theoretical/file.py`)
#### 8.3.1 Format du fichier JSON de l'agenda théorique
L'agenda théorique est fourni sous forme de **fichier JSON** (ex: `./data/theoretical.json`) :
```json
{
"version": 1,
"lessons": [
{
"week": "all",
"day_of_week": 0,
"start_time": "08:00",
"end_time": "09:00",
"subject": "Mathématiques",
"teachers": ["M. Dupont"],
"rooms": ["101"]
},
{
"week": "even",
"day_of_week": 1,
"start_time": "10:00",
"end_time": "11:00",
"subject": "Anglais",
"teachers": [],
"rooms": []
},
{
"week": "odd",
"day_of_week": 1,
"start_time": "10:00",
"end_time": "11:00",
"subject": "Espagnol",
"teachers": [],
"rooms": []
}
]
}
```
- `week` : `"all"` (toutes les semaines), `"even"` (semaines paires) ou `"odd"` (semaines impaires).
- `day_of_week` : entier de **0 (lundi)** à **6 (dimanche)**.
- `start_time` / `end_time` : chaînes au format `"HH:MM"`.
- `teachers` / `rooms` : listes de chaînes, **optionnelles** (défaut : liste vide).
- `id` : **optionnel** ; s'il est absent, le provider génère un identifiant **déterministe incluant le type de semaine**, afin que deux leçons de parité différente sur le même créneau aient des identifiants distincts.
#### 8.3.2 Format du fichier JSON des vacances scolaires (fichier séparé)
Les vacances scolaires sont décrites dans un **fichier JSON séparé** (ex: `./data/school_holidays.json`) :
```json
{
"zone": "A",
"school_year": "2026-2027",
"periods": [
{
"start_date": "2026-10-17",
"end_date": "2026-11-02",
"label": "Toussaint"
}
]
}
```
- `start_date` et `end_date` sont des dates ISO (`YYYY-MM-DD`) **inclusives**.
- `is_holiday(date)` retourne `True` si la date tombe dans **l'une** des périodes (`start_date <= date <= end_date`).
#### 8.3.3 Service de parité de semaine
La parité des semaines est configurée via deux paramètres :
- `THEORETICAL_WEEK_ANCHOR_DATE` : date de référence (ex: `2026-09-01`).
- `THEORETICAL_WEEK_ANCHOR_TYPE` : `"even"` ou `"odd"` (parité de la semaine de référence).
**Algorithme** : on calcule le nombre de semaines entre le **lundi de la semaine cible** et le **lundi de la semaine de référence**. Si ce décalage est **pair**, la semaine cible a la **même parité** que l'ancre ; s'il est **impair**, la parité est **opposée**.
```python
from datetime import date, timedelta
from typing import Literal
def week_parity(
target: date,
anchor_date: date,
anchor_type: Literal["even", "odd"],
) -> Literal["even", "odd"]:
"""Détermine la parité (paire/impaire) de la semaine d'une date cible.
:param target: Date dont on veut connaître la parité de semaine.
:param anchor_date: Date de référence (semaine de parité ``anchor_type``).
:param anchor_type: Parité de la semaine de référence (``"even"`` ou ``"odd"``).
:return: ``"even"`` ou ``"odd"`` selon la parité calculée.
:rtype: Literal["even", "odd"]
"""
target_monday = target - timedelta(days=target.weekday())
anchor_monday = anchor_date - timedelta(days=anchor_date.weekday())
offset_weeks = (target_monday - anchor_monday).days // 7
if offset_weeks % 2 == 0:
return anchor_type
return "odd" if anchor_type == "even" else "even"
```
#### 8.3.4 Comportement du provider JSON
- `get_lessons(date)` : si la date tombe pendant les **vacances scolaires**, retourner `[]`. Sinon, déterminer la **parité de la semaine**, filtrer les leçons selon le champ `week` (`all` correspond à toutes les semaines, `even`/`odd` à leur parité respective), construire les objets `TheoreticalLesson` et retourner la liste **triée par `id`**.
- `get_lessons_for_range(start_date, end_date)` : itérer sur **chaque date** de la plage, ignorer les **vacances scolaires**, appliquer le **filtrage de parité** à chaque jour et retourner la liste cumulée (éventuellement **dédupliquée par `id`**).
#### 8.3.5 Configuration
- `THEORETICAL_AGENDA_PATH` : chemin vers le fichier JSON de l'agenda théorique (ex: `./data/theoretical.json`).
- `SCHOOL_HOLIDAYS_PATH` : chemin vers le fichier JSON des vacances scolaires.
- `THEORETICAL_WEEK_ANCHOR_DATE` : date de référence pour la parité (ex: `2026-09-01`).
- `THEORETICAL_WEEK_ANCHOR_TYPE` : `"even"` ou `"odd"`.
- Si `THEORETICAL_AGENDA_PATH` est `None`, le provider est **désactivé** (M8 retourne un diff vide, non bloquant).
- Si le fichier d'agenda contient des leçons `even`/`odd` mais **aucune ancre n'est configurée**, lever une **erreur de configuration** explicite.
### 8.4 Politique de départage pour les collisions
**Règle déterministe** pour les collisions entre cours théoriques et réels :
1. **Tri par identifiant stable** : Les cours sont triés par ID ou clé de matching (ex: `theoretical-{day_of_week}-{start_time}-{subject}`).
2. **Comparaison exacte** : Les créneaux horaires et la matière normalisée doivent correspondre.
3. **Choix de la première correspondance** : En cas de multiples correspondances admissibles, choisir la **première** après tri déterministe.
**Exemple de tri** :
```python
# Tri des cours théoriques par ID stable (pour un matching déterministe)
theoretical_lessons_sorted = sorted(
theoretical_lessons,
key=lambda lesson: (
lesson.day_of_week,
lesson.start_time,
lesson.end_time,
lesson.subject.lower(),
),
)
```
**Exemple de matching avec départage déterministe** :
```python
def match_theoretical_lesson(
real_lesson: Lesson,
theoretical_events: list[TheoreticalLesson],
tolerance_minutes: int = 15,
) -> TheoreticalLesson | None:
"""Trouve la leçon théorique correspondant à une leçon réelle.
:param real_lesson: Leçon réelle depuis Pronote.
:param theoretical_events: Liste des leçons théoriques candidates.
:param tolerance_minutes: Tolérance en minutes pour le créneau horaire.
:return: La leçon théorique correspondante, ou None.
:rtype: TheoreticalLesson | None
"""
real_start = real_lesson.start
real_day = real_start.weekday()
def to_minutes(t: time) -> int:
return t.hour * 60 + t.minute
# ``normalize_subject`` sera défini dans ``sync/diff.py`` (M8) ou dans le
# module théorique ; il normalise les matières pour un matching déterministe.
start_minutes = real_start.hour * 60 + real_start.minute
end_minutes = real_lesson.end.hour * 60 + real_lesson.end.minute
candidates = [
t
for t in theoretical_events
if t.day_of_week == real_day
and abs(to_minutes(t.start_time) - start_minutes) <= tolerance_minutes
and abs(to_minutes(t.end_time) - end_minutes) <= tolerance_minutes
and normalize_subject(t.subject) == normalize_subject(real_lesson.subject)
]
if not candidates:
return None
candidates.sort(key=lambda t: (t.id, t.start_time))
return candidates[0]
```
---
### 8.5 Logique de comparaison (`sync/diff.py`)
```python
from typing import List, Tuple, Optional
from datetime import date, time, timedelta
from ..models.agenda import Lesson, TheoreticalLesson
from ..models.diff import AgendaDiff, AgendaChange, AgendaChangeType
import logging
logger = logging.getLogger(__name__)
class AgendaComparator:
"""
Compare l'agenda réel (Pronote) avec l'agenda théorique.
"""
# Tolérance pour le matching des heures (en minutes)
TIME_TOLERANCE = 5
def __init__(self, theoretical_provider: TheoreticalAgendaProvider):
self.theoretical_provider = theoretical_provider
def _normalize_subject(self, subject: str) -> str:
"""Normalise le nom d'une matière pour le matching."""
import re
# Supprimer les accents, passer en minuscules, supprimer les espaces multiples
subject = re.sub(r"[^\w\s]", "", subject) # Supprimer la ponctuation
subject = re.sub(r"\s+", " ", subject).strip().lower()
return subject
def _normalize_time(self, t: time) -> time:
"""Normalise une heure (arrondir à 5 minutes près)."""
minute = (t.minute // 5) * 5
return time(t.hour, minute)
def _match_lesson(
self,
real_lesson: Lesson,
theoretical_lessons: List[TheoreticalLesson],
) -> Optional[TheoreticalLesson]:
"""
Trouve le cours théorique correspondant à un cours réel.
Args:
real_lesson: Cours réel (Pronote).
theoretical_lessons: Liste des cours théoriques pour le même jour.
Returns:
Cours théorique correspondant ou None.
**Politique de départage** :
Si plusieurs cours théoriques correspondent, on trie par UID stable (pour un matching déterministe)
et on retourne le premier.
"""
real_day = real_lesson.start.weekday()
real_start = self._normalize_time(real_lesson.start.time())
real_end = self._normalize_time(real_lesson.end.time())
real_subject = self._normalize_subject(real_lesson.subject)
# Collecter tous les candidats correspondants
candidates = []
for theoretical in theoretical_lessons:
if theoretical.day_of_week != real_day:
continue
theo_start = self._normalize_time(theoretical.start_time)
theo_end = self._normalize_time(theoretical.end_time)
theo_subject = self._normalize_subject(theoretical.subject)
# Matching sur :
# 1. Créneau horaire (avec tolérance)
# 2. Matière normalisée
if (
theo_start == real_start
and theo_end == real_end
and theo_subject == real_subject
):
candidates.append(theoretical)
# Trier les candidats par UID stable pour un matching déterministe
candidates.sort(key=lambda t: t.id)
return candidates[0] if candidates else None
def compare_for_date(self, date: date, real_lessons: List[Lesson]) -> AgendaDiff:
"""
Compare l'agenda réel et théorique pour une date donnée.
Args:
date: Date à comparer.
real_lessons: Liste des cours réels pour cette date.
Returns:
Différences entre les deux agendas.
"""
theoretical_lessons = self.theoretical_provider.get_lessons(date)
changes: List[AgendaChange] = []
# Indexer les cours réels par ID pour éviter les doublons
real_by_id = {lesson.id: lesson for lesson in real_lessons}
# 1. Trouver les cours ajoutés ou modifiés
for real_lesson in real_lessons:
matched = self._match_lesson(real_lesson, theoretical_lessons)
if matched is None:
# Cours ajouté (pas dans l'agenda théorique)
changes.append(AgendaChange(
type=AgendaChangeType.ADDED,
lesson=real_lesson,
theoretical_lesson=None,
details="Cours ajouté par rapport à l'agenda théorique",
))
else:
# Vérifier si le cours a été modifié
if (
real_lesson.subject != matched.subject
or real_lesson.teachers != matched.teachers
or real_lesson.rooms != matched.rooms
or real_lesson.status != LessonStatus.NORMAL
):
changes.append(AgendaChange(
type=AgendaChangeType.MODIFIED,
lesson=real_lesson,
theoretical_lesson=matched,
details=self._describe_changes(real_lesson, matched),
))
# 2. Trouver les cours supprimés
for theoretical in theoretical_lessons:
# Vérifier si ce cours théorique a un correspondant réel
has_match = any(
self._match_lesson(real, [theoretical]) is not None
for real in real_lessons
)
if not has_match:
changes.append(AgendaChange(
type=AgendaChangeType.REMOVED,
lesson=None,
theoretical_lesson=theoretical,
details="Cours supprimé par rapport à l'agenda théorique",
))
return AgendaDiff(target_date=date, changes=changes)
def _describe_changes(
self,
real: Lesson,
theoretical: TheoreticalLesson,
) -> str:
"""Décrit les différences entre un cours réel et un cours théorique."""
differences = []
if real.subject != theoretical.subject:
differences.append(f"matière: {theoretical.subject} → {real.subject}")
if set(real.teachers) != set(theoretical.teachers):
differences.append(
f"professeurs: {theoretical.teachers} → {real.teachers}"
)
if set(real.rooms) != set(theoretical.rooms):
differences.append(f"salles: {theoretical.rooms} → {real.rooms}")
if real.status != LessonStatus.NORMAL:
differences.append(f"statut: {real.status.value}")
return "; ".join(differences)
def compare_for_range(
self,
start_date: date,
end_date: date,
real_lessons_by_date: dict[date, List[Lesson]],
) -> List[AgendaDiff]:
"""
Compare les agendas pour une plage de dates.
Args:
start_date: Date de début.
end_date: Date de fin.
real_lessons_by_date: Dictionnaire {date: liste des cours réels}.
Returns:
Liste des différences par date.
"""
diffs = []
current_date = start_date
while current_date <= end_date:
real_lessons = real_lessons_by_date.get(current_date, [])
diff = self.compare_for_date(current_date, real_lessons)
if diff.changes:
diffs.append(diff)
current_date += timedelta(days=1)
return diffs
```
### 8.7 Points clés
- **Format JSON** : L'agenda théorique est décrit par un **fichier JSON** (leçons `all`/`even`/`odd`) ; les vacances scolaires sont décrites par un **fichier JSON séparé**.
- **Parité des semaines** : Déterminée par `THEORETICAL_WEEK_ANCHOR_DATE` et `THEORETICAL_WEEK_ANCHOR_TYPE` (décalage en semaines entre le lundi de référence et le lundi cible).
- **Vacances scolaires** : Les jours de vacances retournent une **liste vide** (aucune leçon théorique).
- **Identifiants déterministes** : Générés par le provider (type de semaine inclus) pour garantir des IDs distincts et stables.
- **Matching déterministe** : Basé sur le jour, le créneau horaire (avec tolérance) et la matière normalisée.
- **Normalisation** : Les matières et heures sont normalisées pour éviter les faux négatifs.
- **Types de changements** : Ajout, suppression, modification.
- **Politique de départage** : Tri par identifiant stable (ID), puis comparaison exacte des créneaux et matière normalisée. En cas de multiples correspondances, choix de la première après tri déterministe.
---
## 9. Synthèse IA
### 9.1 Principes
- **Optionnelle** : La synthèse IA ne doit **jamais bloquer** le pipeline (décision [5](#5-synthèse-ia---protocole-pas-de-sdk-imposé)).
- **Périmètre limité** :
- **Inclus** : Changements d'agenda, messages importants, informations.
- **Exclus** : La liste brute des devoirs (doit rester **intacte**).
- **Contraintes** :
- 3-5 phrases maximum.
- Ton **chaleureux et sobre**.
- **Pas d'emoji dans le texte généré par l'IA**, pas de titre, pas de liste.
*Note* : Les emojis sont autorisés dans le **formatage du message XMPP** (ex: 📌, 📅, 📚, 💬) pour améliorer la lisibilité.
- **Ne pas inventer** d'informations.
- **Rejeter** les horaires non connus (ex: "à 14h" si l'heure n'est pas dans les données).
- **Timeout** : 30 secondes maximum.
- **Longueur maximale** : 800 caractères.
### 9.2 Protocole `SynthesisProvider` (`synthesis/provider.py`)
```python
from typing import Protocol, Optional
from ..models.synthesis import SynthesisInput, SynthesisResult
class SynthesisProvider(Protocol):
"""
Protocole pour les fournisseurs de synthèse IA.
Permet de changer facilement de fournisseur (OpenAI, Mistral, etc.).
"""
def generate(self, input_data: SynthesisInput) -> Optional[SynthesisResult]:
"""
Génère une synthèse IA à partir des données Pronote.
Args:
input_data: Données Pronote (`PronoteData`) à synthétiser.
Returns:
Synthèse IA (string) ou None en cas d'échec.
**Ne doit jamais lever d'exception** (retourner None à la place).
"""
...
```
### 9.3 Adaptateur OpenAI (`synthesis/openai.py`)
```python
from typing import Optional
import httpx
from ..models.synthesis import SynthesisInput, SynthesisResult
from .provider import SynthesisProvider
import logging
logger = logging.getLogger(__name__)
class OpenAISynthesisProvider:
"""
Fournisseur de synthèse IA utilisant l'API OpenAI.
Compatible avec les API OpenAI-compatibles (ex: Mistral, Google via litellm).
"""
# Prompt système en français (inspiré de src/intro/prompt.ts)
SYSTEM_PROMPT = """
Tu es un assistant bienveillant qui résume les informations importantes pour un parent.
Rédige une synthèse en **3 à 5 phrases maximum**, dans un **ton chaleureux et sobre**.
Règles strictes :
- N'utilise **aucun emoji**, aucun titre, aucune liste.
- Ne mentionne **aucun horaire** (ex: "à 14h") sauf si l'heure est explicitement dans les données.
- **N'invente rien** : ne mentionne que ce qui est présent dans les données.
- Sois concis et direct.
- Si aucune information importante n'est disponible, retourne une chaîne vide.
Exemple de format attendu :
"Le cours de mathématiques de Jean a été annulé demain. Un devoir de français est à rendre pour vendredi. Le professeur a envoyé un message concernant la sortie pédagogique."
"""
# Longueur maximale autorisée
MAX_LENGTH = 800
# Timeout en secondes
TIMEOUT = 30
def __init__(
self,
base_url: Optional[str] = None,
api_key: Optional[str] = None,
model: str = "gpt-4o-mini",
):
self.base_url = base_url.rstrip("/") if base_url else "https://api.openai.com/v1"
self.api_key = api_key or ""
self.model = model
def _build_prompt(self, input_data: SynthesisInput) -> str:
"""Construit le prompt utilisateur à partir des données d'entrée."""
parts = []
# Changements d'agenda
if input_data.agenda_diff and input_data.agenda_diff.changes:
changes = []
for change in input_data.agenda_diff.changes:
if change.type == "added":
changes.append(f"Cours ajouté : {change.lesson.subject} le {input_data.target_date.strftime('%d/%m/%Y')}")
elif change.type == "removed":
changes.append(f"Cours supprimé : {change.theoretical_lesson.subject}")
elif change.type == "modified":
changes.append(f"Cours modifié : {change.lesson.subject} ({change.details})")
if changes:
parts.append("Changements d'agenda : " + "; ".join(changes))
# Messages importants
if input_data.messages:
messages = []
for msg in input_data.messages:
if not msg.read: # Seuls les messages non lus sont importants
messages.append(f"Message de {msg.author} : {msg.title}")
if messages:
parts.append("Messages : " + "; ".join(messages))
# Événements scolaires
if input_data.school_events:
events = []
for event in input_data.school_events:
events.append(f"{event.label} du {event.from_date.strftime('%d/%m')}")
if events:
parts.append("Événements : " + "; ".join(events))
if not parts:
return "Aucune information importante à signaler."
return "\n".join(parts)
def generate(self, input_data: SynthesisInput) -> Optional[SynthesisResult]:
"""Génère une synthèse IA."""
if not self.api_key:
logger.warning("Clé API non configurée. Synthèse IA désactivée.")
return None
try:
user_prompt = self._build_prompt(input_data)
# Appel à l'API OpenAI
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
"max_tokens": self.MAX_LENGTH,
"temperature": 0.3, # Ton sobre et déterministe
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
with httpx.Client(timeout=self.TIMEOUT) as client:
response = client.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
)
response.raise_for_status()
result = response.json()
synthesis_text = result["choices"][0]["message"]["content"].strip()
# Vérifier la longueur
if len(synthesis_text) > self.MAX_LENGTH:
synthesis_text = synthesis_text[:self.MAX_LENGTH]
# Nettoyer les éventuels artefacts
synthesis_text = synthesis_text.replace("\n", " ").strip()
return SynthesisResult(text=synthesis_text) if synthesis_text else None
except Exception as e:
safe_error = redact_secrets(str(e))
logger.warning(f"Échec de la génération de la synthèse IA: {safe_error}")
return None
```
### 9.4 Adaptateur `litellm` (optionnel) (`synthesis/litellm.py`)
`litellm` permet d'utiliser **plusieurs fournisseurs IA** (OpenAI, Mistral, Google, etc.) avec une seule API.
**Installation** : Pour activer le support `litellm`, installer le package optionnel :
```bash
pip install .[ai-litellm]
```
```python
from typing import Optional
import litellm
from ..models.synthesis import SynthesisInput, SynthesisResult
from .provider import SynthesisProvider
import logging
logger = logging.getLogger(__name__)
class LiteLLMSynthesisProvider:
"""
Fournisseur de synthèse IA utilisant litellm.
Permet de basculer facilement entre plusieurs modèles.
"""
SYSTEM_PROMPT = OpenAISynthesisProvider.SYSTEM_PROMPT
MAX_LENGTH = 800
TIMEOUT = 30
def __init__(
self,
model: str = "gpt-4o-mini",
api_key: Optional[str] = None,
base_url: Optional[str] = None,
):
self.model = model
self.api_key = api_key
self.base_url = base_url
# Configuration de litellm (si base_url fourni)
if self.base_url:
litellm.api_base = self.base_url
if self.api_key:
litellm.api_key = self.api_key
def _build_prompt(self, input_data: SynthesisInput) -> str:
"""Construit le prompt utilisateur."""
# Réutiliser la logique de OpenAISynthesisProvider
provider = OpenAISynthesisProvider()
return provider._build_prompt(input_data)
def generate(self, input_data: SynthesisInput) -> Optional[SynthesisResult]:
"""Génère une synthèse IA via litellm."""
try:
user_prompt = self._build_prompt(input_data)
response = litellm.completion(
model=self.model,
messages=[
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
max_tokens=self.MAX_LENGTH,
temperature=0.3,
)
synthesis_text = response.choices[0].message.content.strip()
if len(synthesis_text) > self.MAX_LENGTH:
synthesis_text = synthesis_text[:self.MAX_LENGTH]
synthesis_text = synthesis_text.replace("\n", " ").strip()
return SynthesisResult(text=synthesis_text) if synthesis_text else None
except Exception as e:
safe_error = redact_secrets(str(e))
logger.warning(f"Échec de la génération de la synthèse IA (litellm): {safe_error}")
return None
```
### 9.5 Factory pour les fournisseurs IA (`synthesis/__init__.py`)
```python
from typing import Optional
from .provider import SynthesisProvider
from .openai import OpenAISynthesisProvider
from .litellm import LiteLLMSynthesisProvider
from ..config.settings import AISettings
def get_synthesis_provider(settings: AISettings, provider: Optional[str] = None) -> Optional[SynthesisProvider]:
"""
Fabrique un fournisseur de synthèse IA selon la configuration.
Args:
settings: Configuration IA.
provider: Fournisseur explicite à utiliser (ex: "litellm" ou "openai").
Si non spécifié, utilise OpenAI-compatible par défaut.
Returns:
Fournisseur de synthèse IA ou None si désactivé.
"""
if not settings.enabled:
return None
if not settings.api_key:
return None
# Utiliser litellm uniquement si explicitement demandé via AI_PROVIDER=litellm
if provider == "litellm" or (provider is None and settings.base_url and "litellm" in settings.base_url.lower()):
return LiteLLMSynthesisProvider(
model=settings.model,
api_key=settings.api_key.get_secret_value(),
base_url=settings.base_url,
)
# Par défaut : adaptateur OpenAI-compatible (fonctionne avec OpenAI, Mistral, etc.)
return OpenAISynthesisProvider(
base_url=settings.base_url or "https://api.openai.com/v1",
api_key=settings.api_key.get_secret_value(),
model=settings.model,
)
```
### 9.6 Points clés
- **Protocole** : `SynthesisProvider` permet de changer facilement de fournisseur.
- **Mode dégradé** : Si la synthèse échoue, retourner `None` (le pipeline continue).
- **Prompt système** : En français, avec des contraintes strictes (pas d'emoji, pas d'invention).
- **Longueur limitée** : 800 caractères maximum.
- **Timeout** : 30 secondes pour éviter les blocages.
- **Pas de secrets** : La clé API est masquée dans les logs.
---
## 10. Envoi XMPP
### 10.1 Décision architecturale : Compte XMPP dédié avec message direct
**Option retenue** : **Compte bot dédié** (`pronote-bot@exemple.org`) envoyant un **message direct** au parent.
**Pas de PubSub (XEP-0060)**.
#### 10.1.1 Rationale
| **Critère** | **Option A : Compte dédié + message direct** | **Option B : PubSub (XEP-0060)** | **Décision** |
|--------------------------|-----------------------------------------------|----------------------------------|--------------|
| **Complexité** | ⭐ Simple (1 compte, 1 destinataire) | ⭐⭐⭐ Complexe (nœud, ACL, abonnements) | **Option A** |
| **Robustesse** | ⭐⭐⭐ Compatible avec tous les serveurs XMPP | ⭐ Dépend du support PubSub côté serveur | **Option A** |
| **Historique** | ⭐⭐ Géré par MAM (XEP-0313) côté serveur ou client | ⭐⭐⭐ Historique natif via PubSub | **Option A** |
| **Évolutivité** | ⭐⭐ Possible (liste de JIDs → MUC → PubSub) | ⭐⭐⭐ Évolutif par conception | **Option A** |
| **Maintenance** | ⭐⭐⭐ Faible (pas de configuration serveur) | ⭐ Maintenance serveur requise | **Option A** |
**Justification** :
- Pour un **destinataire unique connu** (ex: `parent@exemple.org`), PubSub ajoute une **complexité inutile** (création de nœud, gestion des ACL, abonnements) sans bénéfice suffisant.
- Un compte bot dédié est **simple, robuste et compatible** avec tous les serveurs XMPP (Prosody, ejabberd, etc.).
- L'historique peut être assuré par :
- **MAM (XEP-0313)** côté serveur.
- Le client XMPP (ex: Gajim, Conversations).
- L'**archivage local** du digest (déjà implémenté dans le projet TypeScript).
#### 10.1.2 Évolution future
Si le besoin évolue (ex: **plusieurs destinataires**), les étapes suivantes sont envisagées :
1. **Liste de JIDs** :
- Le compte bot envoie le message à **plusieurs destinataires** (ex: `parent1@exemple.org`, `parent2@exemple.org`).
- **Condition** : Nombre de destinataires ≤ 10.
- **Avantage** : Simple à implémenter (boucle sur les JIDs).
- **Inconvénient** : Pas de partage de l'historique entre destinataires.
2. **MUC (Multi-User Chat)** :
- Créer une **salle privée** (ex: `pronote-digest@muc.exemple.org`) et y inviter les destinataires.
- **Condition** : Nombre de destinataires > 10 ou besoin de partage d'historique.
- **Avantage** : Historique partagé, gestion centralisée.
- **Inconvénient** : Configuration serveur requise (création de salle, gestion des membres).
3. **PubSub (XEP-0060)** :
- Créer un **nœud PubSub** (ex: `pronote-digest@pubsub.exemple.org`) et publier les messages.
- **Condition** : Besoin de **diffusion large** (ex: toute une classe) ou intégration avec d'autres outils.
- **Avantage** : Découplage total entre producteur et consommateurs.
- **Inconvénient** : Complexité accrue (création de nœud, ACL, abonnements).
**Règle** : **Ne pas introduire PubSub** tant que le besoin ne dépasse pas les capacités d'un compte dédié + message direct.
---
### 10.2 Configuration XMPP
#### 10.2.1 Variables d'environnement
| **Variable** | **Description** | **Valeur par défaut** | **Type** | **Obligatoire** |
|----------------------------|-------------------------------------------------------------------------------|-----------------------|-------------------|-----------------|
| `XMPP_ENABLED` | Activer l'envoi XMPP. | `False` | `bool` | ❌ Non |
| `XMPP_JID` | Identifiant du compte bot (ex: `pronote-bot@exemple.org`). | `None` | `str` | ✅ Oui |
| `XMPP_PASSWORD` | Mot de passe du compte bot. | `None` | `SecretStr` | ✅ Oui |
| `XMPP_HOST` | Hôte XMPP **explicite** (ex: `exemple.org`). | `None` | `str` | ✅ Oui |
| `XMPP_PORT` | Port XMPP (5222 pour TLS, 5223 pour SSL). | `5222` | `int` | ❌ Non |
| `XMPP_TO` | Destinataire unique (ex: `parent@exemple.org`). | `None` | `str` | ✅ Oui |
| `XMPP_RESOURCE` | Ressource XMPP (ex: `pronote-digest`). | `pronote-digest` | `str` | ❌ Non |
| `XMPP_USE_TLS` | Utiliser TLS pour la connexion. | `True` | `bool` | ❌ Non |
| `XMPP_TIMEOUT` | Timeout de connexion (secondes). | `30` | `int` | ❌ Non |
**⚠️ Notes** :
- **`XMPP_HOST` doit être explicite** : Éviter les ambiguïtés DNS/SRV (ex: `exemple.org` au lieu de `xmpp.exemple.org` si le SRV pointe vers `exemple.org`).
- **Pas de variables PubSub** : `XMPP_PUBSUB_NODE`, `XMPP_ROOM`, `XMPP_SUBSCRIBERS` **ne doivent pas être introduites** pour l'instant.
- **Sécurité** : `XMPP_JID`, `XMPP_PASSWORD` et `XMPP_TO` **ne doivent jamais apparaître** dans les logs, erreurs ou fixtures.
- **Standardisation** : `XMPP_TO` est mappé sur le champ `to` dans le modèle Pydantic.
#### 10.2.2 Exemple de configuration dans `.env`
```ini
# --- XMPP ---
XMPP_ENABLED=true
XMPP_JID=pronote-bot@exemple.org
XMPP_PASSWORD=secret_password # Masqué via SecretStr
XMPP_HOST=exemple.org
XMPP_PORT=5222
XMPP_TO=parent@exemple.org
XMPP_RESOURCE=pronote-digest
XMPP_USE_TLS=true
XMPP_TIMEOUT=30
```
#### 10.2.3 Modèle Pydantic pour la configuration XMPP
```python
from pydantic import SecretStr, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class XmppSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="XMPP_", env_file=".env", extra="ignore")
enabled: bool = Field(False, description="Activer l'envoi XMPP")
jid: str = Field(..., description="Identifiant du compte bot (ex: pronote-bot@exemple.org)")
password: SecretStr = Field(..., description="Mot de passe du compte bot")
host: str = Field(..., description="Hôte XMPP explicite (ex: exemple.org)")
port: int = Field(5222, description="Port XMPP (5222 pour TLS)")
to: str = Field(..., description="Destinataire unique (ex: parent@exemple.org)")
resource: str = Field("pronote-digest", description="Ressource XMPP")
use_tls: bool = Field(True, description="Utiliser TLS pour la connexion")
timeout: int = Field(30, description="Timeout de connexion (secondes)")
```
---
### 10.3 Protocole `Channel` (`channels/protocol.py`)
```python
from typing import Protocol
from ..models.xmpp import XmppMessage
class Channel(Protocol):
"""
Protocole pour les canaux de sortie (XMPP, fichier, etc.).
**Synchrone** : Le pipeline appelle `send()` sans await.
Inspiré de l'interface `Channel` dans src/channels/ du projet TypeScript.
"""
name: str
def send(self, message: XmppMessage) -> bool:
"""
Envoie un message de manière **synchrone**.
Args:
message: Message à envoyer.
Returns:
True si l'envoi a réussi, False sinon.
"""
...
```
---
### 10.3 Client XMPP (`channels/xmpp.py`)
```python
import asyncio
from typing import Optional, Awaitable
import slixmpp
from slixmpp.exceptions import IqError, IqTimeout
from ..models.xmpp import XmppMessage
from .protocol import Channel
import logging
logger = logging.getLogger(__name__)
class XmppChannel(Channel):
"""
Canal XMPP pour l'envoi des messages.
Utilise slixmpp en mode asynchrone.
"""
name = "xmpp"
def __init__(
self,
jid: str,
password: str,
recipient: str,
dry_run: bool = False,
):
self.jid = jid
self.password = password
self.recipient = recipient
self.dry_run = dry_run
self._client: Optional[slixmpp.ClientXMPP] = None
self._connected = False
self._message_sent = False
async def connect(self) -> bool:
"""Établit la connexion XMPP."""
if self._connected:
return True
try:
# Créer le client
self._client = slixmpp.ClientXMPP(self.jid, self.password)
# Configurer les handlers
self._client.add_event_handler("session_start", self._on_session_start)
self._client.add_event_handler("failed_auth", self._on_failed_auth)
self._client.add_event_handler("disconnected", self._on_disconnected)
# Se connecter (async)
self._client.connect()
self._client.process(block=False)
# Attendre la connexion (timeout: 30s)
await asyncio.wait_for(
self._wait_for_connection(),
timeout=30.0,
)
return self._connected
except Exception as e:
logger.error(f"Échec de la connexion XMPP: {redact_secrets(str(e))}")
return False
def _on_session_start(self, event: slixmpp.Event) -> None:
"""Handler appelé quand la session XMPP est établie."""
self._connected = True
logger.info("Connexion XMPP établie")
def _on_failed_auth(self, event: slixmpp.Event) -> None:
"""Handler appelé en cas d'échec d'authentification."""
logger.error("Échec de l'authentification XMPP")
self._connected = False
def _on_disconnected(self, event: slixmpp.Event) -> None:
"""Handler appelé en cas de déconnexion."""
logger.warning("Déconnexion XMPP")
self._connected = False
async def _wait_for_connection(self) -> None:
"""Attend que la connexion soit établie."""
while not self._connected:
await asyncio.sleep(0.1)
def _format_message(self, message: XmppMessage) -> str:
"""Formate le message XMPP en texte brut."""
lines = []
# Titre (date cible)
lines.append(f"=== Pronote - {message.target_date.strftime('%A %d %B %Y')} ===")
lines.append("")
# Synthèse IA (si disponible)
if message.synthesis:
lines.append("📌 Synthèse :")
lines.append(message.synthesis)
lines.append("")
# Changements d'agenda
if message.changes:
lines.append("📅 Changements d'agenda :")
for change in message.changes:
if change.type == "added":
lines.append(f" + {change.lesson.subject} ({change.lesson.start.strftime('%H:%M')})")
elif change.type == "removed":
lines.append(f" - {change.theoretical_lesson.subject}")
elif change.type == "modified":
lines.append(f" ~ {change.lesson.subject} ({change.details})")
lines.append("")
# Liste brute des devoirs
if message.homeworks:
lines.append("📚 Devoirs :")
for hw in message.homeworks:
due_date = hw.due_on.strftime("%d/%m/%Y")
lines.append(f" - {hw.subject} (pour le {due_date}) : {hw.text}")
lines.append("")
# Messages
if message.messages:
lines.append("💬 Messages :")
for msg in message.messages:
lines.append(f" - {msg.author} : {msg.title}")
return "\n".join(lines)
async def send(self, message: XmppMessage) -> bool:
"""Envoie un message XMPP."""
if not self._connected:
# Se connecter si ce n'est pas déjà fait
if not await self.connect():
return False
if self.dry_run:
logger.info(f"[DRY-RUN] Envoi XMPP à {self.recipient}")
logger.info(f"Contenu:\n{self._format_message(message)}")
return True
try:
# Formater le message
body = self._format_message(message)
# Envoyer le message
self._client.send_message(
mto=self.recipient,
mbody=body,
mtype="chat",
)
logger.info(f"Message XMPP envoyé à {self.recipient}")
return True
except Exception as e:
safe_error = redact_secrets(str(e))
logger.error(f"Échec de l'envoi XMPP: {safe_error}")
return False
async def disconnect(self) -> None:
"""Déconnecte le client XMPP."""
if self._client:
self._client.disconnect()
self._connected = False
class SyncXmppChannel:
"""
Adaptateur synchrone pour XMPP.
Encapsule asyncio avec une stratégie robuste pour éviter les conflits de boucle d'événements.
**Important** : Si le pipeline est appelé depuis un contexte asynchrone, l'envoi XMPP doit être isolé
dans un thread séparé pour éviter les conflits de boucle.
"""
def __init__(
self,
jid: str,
password: str,
recipient: str,
dry_run: bool = False,
):
self.jid = jid
self.password = password
self.recipient = recipient
self.dry_run = dry_run
self._xmpp_channel = XmppChannel(jid, password, recipient, dry_run)
def send(self, message: XmppMessage) -> bool:
"""Envoie un message XMPP de manière synchrone."""
import asyncio
# Créer une nouvelle boucle d'événements pour éviter les conflits
loop = asyncio.new_event_loop()
try:
asyncio.set_event_loop(loop)
return loop.run_until_complete(self._xmpp_channel.send(message))
finally:
loop.close()
asyncio.set_event_loop(None)
```
### 10.4 Factory pour les canaux (`channels/__init__.py`)
```python
from typing import List, Dict, Type
from .protocol import Channel
from .xmpp import SyncXmppChannel
from ..config.settings import Settings
# Registre des factories de canaux
_CHANNEL_FACTORIES: Dict[str, Type[Channel]] = {
"xmpp": SyncXmppChannel,
}
def get_channel(settings: Settings, channel_name: str = "xmpp") -> Channel:
"""
Fabrique un canal selon la configuration.
Args:
settings: Configuration globale.
channel_name: Nom du canal (défaut: "xmpp").
Returns:
Canal configuré.
"""
factory = _CHANNEL_FACTORIES.get(channel_name)
if factory is None:
raise ValueError(f"Canal inconnu: {channel_name}")
if channel_name == "xmpp":
if not settings.xmpp.enabled:
raise ValueError("XMPP est désactivé (XMPP_ENABLED=False)")
return factory(
jid=settings.xmpp.jid,
password=settings.xmpp.password.get_secret_value(),
recipient=settings.xmpp.to,
dry_run=settings.app.dry_run,
)
raise ValueError(f"Canal {channel_name} non implémenté")
```
### 10.5 Points clés
- **slixmpp** : Bibliothèque recommandée pour XMPP (asyncio, maintenue).
- **Format du message** : Structuré avec sections claires (synthèse, changements, devoirs, messages).
- **Mode dégradé** : Si XMPP échoue, le pipeline peut continuer (mais le message ne sera pas envoyé).
- **Dry-run** : Mode obligatoire pour tester sans envoyer de message.
- **Reconnexion** : Gestion des erreurs de connexion.
---
## 11. Gestion des erreurs et modes dégradés
### 11.1 Principes
- **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).
- **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** :
- **Aucune source disponible** (iCal + pronotepy échouent) → **échec explicite** avec message clair.
- **Configuration invalide** (ex: `PRONOTE_ICAL_URL` manquant) → **échec explicite**.
### 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
class ErrorSeverity(Enum):
"""Niveau de gravité d'une erreur."""
DEBUG = auto() # Erreur mineure (ex: warning de parsing)
WARNING = auto() # Erreur non bloquante (ex: synthèse IA échouée)
ERROR = auto() # Erreur bloquante pour une étape (ex: récupération Pronote échouée)
CRITICAL = auto() # Erreur bloquante pour le pipeline (ex: aucune source disponible)
class PipelineError(Exception):
"""Erreur dans le pipeline."""
def __init__(
self,
message: str,
severity: ErrorSeverity = ErrorSeverity.ERROR,
step: Optional[str] = None,
recoverable: bool = False,
):
super().__init__(message)
self.message = message
self.severity = severity
self.step = step
self.recoverable = recoverable
class PipelineWarning(PipelineError):
"""Avertissement dans le pipeline (non bloquant)."""
def __init__(self, message: str, step: Optional[str] = None):
super().__init__(message, ErrorSeverity.WARNING, step, recoverable=True)
class PipelineCriticalError(PipelineError):
"""Erreur critique dans le pipeline (bloquante)."""
def __init__(self, message: str, step: Optional[str] = None):
super().__init__(message, ErrorSeverity.CRITICAL, step, recoverable=False)
```
### 11.3 Gestion des erreurs dans le pipeline (`pipeline/run.py`)
```python
from typing import List, Optional, Tuple
from ..models.agenda import Lesson, Homework, SchoolEvent
from ..models.xmpp import XmppMessage
from ..models.pronote import PronoteData
from ..models.sync import CalDAVSyncResult
from ..models.synthesis import SynthesisInput, SynthesisResult
from ..sources.pronote.fallback import PronoteFetcher
from ..sync.caldav import CalDAVClient
from ..sync.diff import AgendaComparator
from ..synthesis.provider import SynthesisProvider
from ..channels.protocol import Channel
from .steps import (
fetch_step,
normalize_step,
compare_step,
caldav_sync_step,
synthesis_step,
send_step,
fetch_blog_step,
)
import logging
logger = logging.getLogger(__name__)
class PipelineRunner:
"""
Orchestre l'exécution du pipeline avec gestion des erreurs.
Ordre des étapes :
1. Récupération Pronote
2. Normalisation
2 bis. Récupération du blog (RSS)
3. Comparaison avec l'agenda théorique
4. Synchronisation CalDAV
5. Synthèse IA (optionnelle)
6. Construction du message XMPP
7. Envoi XMPP
"""
def __init__(
self,
pronote_fetcher: PronoteFetcher,
caldav_client: CalDAVClient,
agenda_comparator: AgendaComparator,
synthesis_provider: Optional[SynthesisProvider],
channel: Channel,
blog_rss_client: Optional["BlogRSSClient"] = None,
blog_state: Optional["BlogRSSState"] = None,
dry_run: bool = False,
):
self.pronote_fetcher = pronote_fetcher
self.caldav_client = caldav_client
self.agenda_comparator = agenda_comparator
self.synthesis_provider = synthesis_provider
self.channel = channel
self.blog_rss_client = blog_rss_client
self.blog_state = blog_state
self.dry_run = dry_run
self._errors: List[PipelineError] = []
self._warnings: List[PipelineWarning] = []
def run(self) -> Tuple[Optional[PronoteData], List[PipelineError]]:
"""
Exécute le pipeline complet.
Returns:
Tuple (PronoteData final, liste des erreurs).
"""
pronote_data: Optional[PronoteData] = None
agenda_diff = None
sync_result: Optional[CalDAVSyncResult] = None
synthesis_result: Optional[SynthesisResult] = None
blog_articles: List["BlogArticle"] = []
try:
# Étape 1: Récupération Pronote
try:
lessons, homeworks, school_events, messages = fetch_step(
self.pronote_fetcher
)
except PipelineError as e:
if e.severity == ErrorSeverity.CRITICAL:
raise
self._errors.append(e)
logger.warning(f"Étape 'fetch' échouée (non critique): {e.message}")
return None, self._errors + self._warnings
# Étape 2: Normalisation
try:
pronote_data = normalize_step(lessons, homeworks, school_events, messages)
except PipelineError as e:
self._errors.append(e)
logger.warning(f"Étape 'normalize' échouée: {e.message}")
return None, self._errors + self._warnings
# Étape 2 bis: Récupération du blog (RSS)
if self.blog_rss_client and self.blog_state:
try:
blog_articles = fetch_blog_step(
self.blog_rss_client,
self.blog_state,
enabled=True,
)
except PipelineError as e:
self._warnings.append(PipelineWarning(
message=f"Récupération du blog échouée: {e.message}",
step="fetch_blog",
))
logger.warning(f"Étape 'fetch_blog' échouée (non bloquante): {e.message}")
blog_articles = []
# Étape 3: Comparaison avec l'agenda théorique
try:
agenda_diff = compare_step(
self.agenda_comparator,
pronote_data.lessons,
pronote_data.target_date,
)
except PipelineError as e:
self._warnings.append(PipelineWarning(
message=f"Comparaison échouée: {e.message}",
step="compare",
))
logger.warning(f"Étape 'compare' échouée (non bloquante): {e.message}")
# Étape 4: Synchronisation CalDAV
try:
sync_result = caldav_sync_step(
self.caldav_client,
pronote_data.lessons,
pronote_data.homeworks,
pronote_data.school_events,
)
if sync_result and sync_result.status.value == "failed":
self._warnings.append(PipelineWarning(
message=f"Synchronisation CalDAV échouée: {sync_result.errors}",
step="sync",
))
logger.warning("Synchronisation CalDAV échouée (non bloquante)")
except PipelineError as e:
self._warnings.append(PipelineWarning(
message=f"Synchronisation CalDAV échouée: {e.message}",
step="sync",
))
logger.warning(f"Étape 'sync' échouée (non bloquante): {e.message}")
# Étape 5: Synthèse IA (optionnelle)
if self.synthesis_provider and agenda_diff:
try:
synthesis_input = SynthesisInput(
agenda_diff=agenda_diff,
messages=pronote_data.messages,
school_events=pronote_data.school_events,
target_date=pronote_data.target_date,
)
synthesis_result = synthesis_step(self.synthesis_provider, synthesis_input)
except PipelineError as e:
self._warnings.append(PipelineWarning(
message=f"Synthèse IA échouée: {e.message}",
step="synthesis",
))
logger.warning(f"Étape 'synthesis' échouée (non bloquante): {e.message}")
# Étape 6: Construction du message XMPP
xmpp_message = XmppMessage(
target_date=pronote_data.target_date,
synthesis=synthesis_result.text if synthesis_result else None,
homeworks=pronote_data.homeworks,
changes=agenda_diff.changes if agenda_diff else [],
messages=pronote_data.messages,
external_info=ExternalInfo(
blog_articles=blog_articles,
pronote_messages=pronote_data.messages,
) if blog_articles or pronote_data.messages else None,
)
# Étape 7: Envoi XMPP
try:
send_step(self.channel, xmpp_message)
except PipelineError as e:
self._warnings.append(PipelineWarning(
message=f"Envoi XMPP échoué: {e.message}",
step="send",
))
logger.warning(f"Étape 'send' échouée (non bloquante): {e.message}")
return pronote_data, self._errors + self._warnings
except PipelineCriticalError as e:
logger.error(f"Erreur critique dans le pipeline: {e.message}")
return None, [e]
except Exception as e:
from ..utils.redaction import redact_secrets
safe_error = redact_secrets(str(e))
logger.error(f"Erreur inattendue dans le pipeline: {safe_error}")
return None, [PipelineCriticalError(
message=safe_error,
step="unknown",
)]
def get_errors(self) -> List[PipelineError]:
"""Récupère la liste des erreurs."""
return self._errors
def get_warnings(self) -> List[PipelineWarning]:
"""Récupère la liste des avertissements."""
return self._warnings
```
### 11.4 Étapes du pipeline (`pipeline/steps/`)
Chaque étape du pipeline est **isolée** et peut lever des `PipelineError` ou `PipelineWarning`.
#### 11.4.1 `fetch.py`
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`.
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`
```python
from pronote_sync.models.blog import BlogArticle
from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.sources.blog.state import BlogRSSState
from .errors import PipelineError, ErrorSeverity
def fetch_blog_step(
rss_client: BlogRSSClient,
blog_state: BlogRSSState,
enabled: bool = True,
) -> list[BlogArticle]:
"""
Étape de récupération des articles du blog du collège.
Lit les GUID déjà connus et les en-têtes de cache HTTP depuis l'état,
puis appelle le client RSS avec ces valeurs pour une requête
conditionnelle. Si la réponse n'est pas ``304 Not Modified`` et que de
nouveaux articles sont présents, les GUID et les en-têtes de cache sont
enregistrés dans l'état. Retourne les nouveaux articles sous forme de
liste.
:param rss_client: Client RSS configuré.
:param blog_state: État local pour la déduplication et le cache HTTP.
:param enabled: Si False, retourne une liste vide.
:return: Liste des nouveaux articles.
:rtype: list[BlogArticle]
:raises PipelineError: Si la récupération échoue (non bloquante pour le
pipeline).
"""
if not enabled:
return []
try:
known_guids = blog_state.get_known_guids()
etag, last_modified = blog_state.get_cache_headers()
result = rss_client.fetch_and_parse(
known_guids=known_guids,
etag=etag,
last_modified=last_modified,
)
# Mettre à jour l'état si de nouveaux articles sont trouvés
if not result.not_modified and result.articles:
blog_state.add_guids(article.id for article in result.articles)
blog_state.update_cache_headers(result.etag, result.last_modified)
return list(result.articles)
except Exception as e:
raise PipelineError(
message=f"Échec de la récupération du blog: {e}",
severity=ErrorSeverity.WARNING,
step="fetch_blog",
recoverable=True,
) from e
```
#### 11.4.2 Autres étapes
Les autres étapes (`normalize_step`, `compare_step`, etc.) suivent le même principe :
- **Lever `PipelineCriticalError`** pour les erreurs bloquantes.
- **Lever `PipelineError`** pour les erreurs non bloquantes.
- **Retourner un résultat partiel** si possible.
### 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** :
- 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.
---
## 12. Tests, fixtures et mocks
### 12.1 Principes
- **Pas de réseau en tests** : Utiliser des **mocks** pour toutes les requêtes HTTP (Pronote, CalDAV) et XMPP.
- **Fixtures anonymisées** : Utiliser des **données réelles anonymisées** (pas de noms d'élèves, professeurs, établissements réels).
- **Couverture élevée** : Viser **≥ 90%** de couverture (comme dans `pronote-digest`).
- **Tests déterministes** : Les tests doivent être **reproductibles** (pas de dépendance à l'heure ou à des données externes).
- **Vitesse** : Les tests doivent s'exécuter **rapidement** (éviter les sleeps inutiles).
### 12.2 Structure des tests
```
tests/
├── __init__.py
├── conftest.py # Fixtures pytest partagées
├── fixtures/ # Fichiers de fixtures (iCal, JSON, XML, etc.)
│ ├── pronote-4e.ics # Flux iCal Pronote anonymisé (4ème)
│ ├── pronote-6e.ics # Flux iCal Pronote anonymisé (6ème)
│ ├── theoretical.json # Agenda théorique JSON
│ ├── school_holidays.json # Vacances scolaires JSON
│ └── blog_rss.xml # Flux RSS du blog anonymisé
├── unit/ # Tests unitaires
│ ├── test_models.py # Tests des modèles Pydantic
│ ├── test_parsing.py # Tests du parsing iCal
│ ├── test_uid.py # Tests de normalisation des UID
│ └── ...
├── integration/ # Tests d'intégration
│ ├── test_pipeline.py # Tests du pipeline complet
│ ├── test_caldav.py # Tests de la sync CalDAV (mockée)
│ └── ...
└── e2e/ # Tests end-to-end
└── test_cli.py # Tests de l'interface CLI
```
### 12.3 Fixtures anonymisées
#### 12.3.1 Exemple de flux iCal Pronote anonymisé (`tests/fixtures/pronote-4e.ics`)
```icalendar
BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Index Education//Pronote//FR
X-WR-CALNAME:Edt Jean DUPONT 4ème A
BEGIN:VEVENT
UID:Edt_12345@index-education.net-20260905T120000Z-Index-Education
DTSTAMP:20260905T120000Z
DTSTART:20260905T080000Z
DTEND:20260905T090000Z
SUMMARY:Mathématiques
CATEGORIES:Cours
DESCRIPTION:EDUCATION MUSICALE
Professeur : DURAND G.
Salle : S002 Education musicale
<strong>Contenu pédagogique :
</strong>Pratique vocale
<strong>Pour le 10/09/2026 :
</strong>Réviser les chansons apprises
<strong>Donné le 03/09/2026 :
</strong>Apporter le cahier de chants
END:VEVENT
BEGIN:VEVENT
UID:Edt_67890@index-education.net-20260905T120000Z-Index-Education
DTSTAMP:20260905T120000Z
DTSTART:20260905T090000Z
DTEND:20260905T100000Z
SUMMARY:Français
CATEGORIES:Cours - Cours annulé
DESCRIPTION:Français
Professeur : Mme Martin
Salle : 205
Groupe : Classe entière
<strong>Contenu pédagogique :
</strong>Étude d'un texte littéraire.
END:VEVENT
STATUS:CANCELLED
BEGIN:VEVENT
UID:Edt_11111@index-education.net-20260905T120000Z-Index-Education
DTSTAMP:20260905T120000Z
DTSTART:20260920
DTEND:20260921
SUMMARY:Vacances de la Toussaint
CATEGORIES:Congés
END:VEVENT
END:VCALENDAR
```
**Points clés** :
- **Anonymisation** : Noms d'élèves (`Jean DUPONT`), professeurs (`M. Dupont`, `Mme Martin`), salles (`204`, `205`) et établissements sont **fictifs**.
- **Tokens supprimés** : Les URLs ne contiennent **pas** de `icalsecurise`.
- **Données réalistes** : Structure identique aux flux réels Pronote.
#### 12.3.2 Exemple de fichier JSON théorique (`tests/fixtures/theoretical.json`)
```json
{
"version": 1,
"lessons": [
{
"id": "theoretical-maths-monday-1",
"week": "all",
"day_of_week": 0,
"start_time": "08:00",
"end_time": "09:00",
"subject": "Mathématiques",
"teachers": ["Mme Martin"],
"rooms": ["101"]
},
{
"week": "all",
"day_of_week": 0,
"start_time": "09:00",
"end_time": "10:00",
"subject": "Français",
"teachers": ["M. Dupont"],
"rooms": ["102"]
},
{
"week": "even",
"day_of_week": 1,
"start_time": "10:00",
"end_time": "11:00",
"subject": "Anglais",
"teachers": ["Mme Bernard"],
"rooms": ["201"]
}
]
}
```
**Champs clés** :
- `week` : `all` (toutes les semaines), `even` (semaines paires) ou `odd` (semaines impaires).
- `day_of_week` : jour de la semaine (0 = lundi, 4 = vendredi).
- `start_time` / `end_time` : créneau horaire au format `HH:MM`.
### 12.4 Configuration pytest (`tests/conftest.py`)
```python
import pytest
from datetime import datetime, date
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.synthesis import SynthesisInput, SynthesisResult
# --- Fixtures pour les modèles ---
@pytest.fixture
def sample_lesson():
"""Retourne un cours Pronote de test."""
return Lesson(
id="Edt_12345@index-education.net",
start=datetime(2026, 9, 5, 8, 0, 0),
end=datetime(2026, 9, 5, 9, 0, 0),
subject="Mathématiques",
teachers=["M. Dupont"],
rooms=["204"],
status=LessonStatus.NORMAL,
content="Résoudre des équations du second degré.",
)
@pytest.fixture
def sample_cancelled_lesson():
"""Retourne un cours annulé."""
return Lesson(
id="Edt_67890@index-education.net",
start=datetime(2026, 9, 5, 9, 0, 0),
end=datetime(2026, 9, 5, 10, 0, 0),
subject="Français",
teachers=["Mme Martin"],
rooms=["205"],
status=LessonStatus.CANCELLED,
content="Étude d'un texte littéraire.",
)
@pytest.fixture
def sample_homework():
"""Retourne un devoir de test."""
return Homework(
id="abc123def456",
subject="Mathématiques",
teachers=["M. Dupont"],
assigned_on=date(2026, 9, 5),
due_on=date(2026, 9, 10),
text="Exercices 1 à 5 page 42.",
html="<p>Exercices 1 à 5 page 42.</p>",
)
@pytest.fixture
def sample_school_event():
"""Retourne un événement scolaire de test."""
return SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Vacances de la Toussaint",
from_date=date(2026, 10, 18),
to_date=date(2026, 11, 3),
)
@pytest.fixture
def sample_message():
"""Retourne un message de test."""
return Message(
id="msg_123",
type=MessageType.INFORMATION,
title="Sortie pédagogique",
content="Une sortie est prévue le 15 octobre.",
author="M. Dupont",
date=datetime(2026, 9, 1, 10, 0, 0),
read=False,
)
@pytest.fixture
def sample_pronote_data(sample_lesson, sample_homework, sample_message):
"""Retourne un jeu de données Pronote de test."""
return PronoteData(
lessons=[sample_lesson],
homeworks=[sample_homework],
messages=[sample_message],
)
# --- Fixtures pour les mocks ---
@pytest.fixture
def mock_ical_content():
"""Retourne un contenu iCal de test."""
return """BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Index Education//Pronote//FR
X-WR-CALNAME:Edt Test
BEGIN:VEVENT
UID:Edt_12345@index-education.net-20260905T120000Z-Index-Education
DTSTAMP:20260905T120000Z
DTSTART:20260905T080000Z
DTEND:20260905T090000Z
SUMMARY:Mathématiques
CATEGORIES:Cours
DESCRIPTION:Matière : Mathématiques
Professeur : M. Dupont
Salle : 204
<strong>Contenu pédagogique :
</strong>Résoudre des équations du second degré.
<strong>Pour le 10/09/2026 :
</strong>Exercices 1 à 5 page 42.
END:VEVENT
END:VCALENDAR"""
@pytest.fixture
def mock_pronotepy_lessons():
"""Retourne une liste de cours mockés (simule pronotepy)."""
class MockLesson:
def __init__(self, id, start, end, subject, teachers, rooms, content):
self.id = id
self.start = start
self.end = end
self.subject = subject
self.teachers = teachers
self.rooms = rooms
self.content = content
return [
MockLesson(
id=12345,
start=datetime(2026, 9, 5, 8, 0, 0),
end=datetime(2026, 9, 5, 9, 0, 0),
subject="Mathématiques",
teachers=[type("Teacher", (), {"name": "M. Dupont"})()],
rooms=[type("Room", (), {"name": "204"})()],
content="Résoudre des équations.",
),
]
# --- Fixtures pour les mocks HTTP ---
@pytest.fixture
def mock_requests_get():
"""Mock requests.get pour les tests iCal."""
import requests_mock
with requests_mock.Mocker() as m:
m.get(
"https://test.ent/pronote/ical/test.ics",
text=mock_ical_content(),
status_code=200,
)
yield m
# --- Fixtures pour les tests de synthèse IA ---
@pytest.fixture
def mock_ai_provider():
"""Mock un fournisseur de synthèse IA."""
from unittest.mock import MagicMock
from pronote_sync.synthesis.provider import SynthesisProvider
provider = MagicMock(spec=SynthesisProvider)
provider.generate.return_value = "Synthèse de test."
return provider
@pytest.fixture
def mock_failing_ai_provider():
"""Mock un fournisseur de synthèse IA qui échoue."""
from unittest.mock import MagicMock
from pronote_sync.synthesis.provider import SynthesisProvider
provider = MagicMock(spec=SynthesisProvider)
provider.generate.return_value = None
return provider
# --- Fixtures pour les tests XMPP ---
@pytest.fixture
def mock_xmpp_channel():
"""Mock un canal XMPP."""
from unittest.mock import MagicMock
from pronote_sync.channels.protocol import Channel
channel = MagicMock(spec=Channel)
channel.name = "xmpp"
channel.send.return_value = True
return channel
# --- Fixtures pour les tests de configuration ---
@pytest.fixture
def sample_settings():
"""Retourne une configuration de test."""
from pydantic import SecretStr
from pronote_sync.config.settings import Settings, PronoteSettings, CalDAVSettings, XmppSettings, AISettings, AppSettings
return Settings(
pronote=PronoteSettings(
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="monbureaunumerique",
agenda_source="auto",
homework_source="auto",
messages_source="pronotepy",
),
caldav=CalDAVSettings(
url="https://caldav.test.com/calendars/test/",
username="test_user",
password=SecretStr("test_password"),
sync_past_days=7,
sync_future_days=30,
),
xmpp=XmppSettings(
jid="test@example.com",
password=SecretStr("test_password"),
recipient="recipient@example.com",
),
ai=AISettings(
enabled=True,
base_url="https://api.test.com/v1",
api_key=SecretStr("test_api_key"),
model="gpt-4o-mini",
),
app=AppSettings(
dry_run=True,
log_level="DEBUG",
theoretical_agenda_path="./tests/fixtures/theoretical.json",
),
)
# --- Exemple de test unitaire ---
@pytest.mark.unittest
def test_parse_ical_lesson(parsed_lessons):
"""Test le parsing d'un cours depuis iCal."""
lessons, homeworks, school_events = parsed_lessons
assert len(lessons) == 1
lesson = lessons[0]
assert lesson.subject == "Mathématiques"
assert lesson.teachers == ["M. Dupont"]
assert lesson.rooms == ["204"]
assert lesson.start == datetime(2026, 9, 5, 8, 0, 0)
assert lesson.end == datetime(2026, 9, 5, 9, 0, 0)
assert lesson.status == LessonStatus.NORMAL
@pytest.mark.unittest
def test_parse_ical_homework(parsed_lessons):
"""Test le parsing des devoirs depuis iCal."""
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]
assert homework.subject == "Mathématiques"
assert homework.due_on == date(2026, 9, 10)
assert "Exercices 1 à 5 page 42" in homework.text
# --- Exemple de test d'intégration ---
@pytest.mark.integration
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.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 JsonTheoreticalAgendaProvider
# Configurer le fetcher Pronote
pronote_client = PronoteClient(sample_settings.pronote)
fetcher = PronoteFetcher(sample_settings, pronote_client)
# Configurer le client CalDAV
caldav_client = CalDAVClient(
url=sample_settings.caldav.url,
username=sample_settings.caldav.username,
password=sample_settings.caldav.password.get_secret_value(),
dry_run=True,
)
# Configurer le comparateur d'agenda
theoretical_provider = JsonTheoreticalAgendaProvider(
file_path=sample_settings.app.theoretical_agenda_path
)
comparator = AgendaComparator(theoretical_provider)
# Configurer le pipeline
runner = PipelineRunner(
pronote_fetcher=fetcher,
caldav_client=caldav_client,
agenda_comparator=comparator,
synthesis_provider=mock_ai_provider,
channel=mock_xmpp_channel,
dry_run=True,
)
# Exécuter le pipeline
pronote_data, errors = runner.run()
# Vérifications
assert pronote_data is not None
assert len(pronote_data.lessons) >= 0
assert len(pronote_data.homeworks) >= 0
assert len(errors) == 0 # Aucun erreur critique
# --- Exemple de test de parsing des UID ---
@pytest.mark.unittest
def test_normalize_pronote_uid():
"""Test la normalisation des UID Pronote."""
from pronote_sync.utils.uid import normalize_pronote_uid
# UID avec suffixe temporel
uid_with_suffix = "Edt_12345@index-education.net-20260905T120000Z-Index-Education"
normalized = normalize_pronote_uid(uid_with_suffix)
assert normalized == "Edt_12345@index-education.net"
# UID déjà normalisé
uid_normalized = "Edt_12345@index-education.net"
assert normalize_pronote_uid(uid_normalized) == uid_normalized
# --- Exemple de test de déduplication des devoirs ---
@pytest.mark.unittest
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.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)
# Cours 1 : contient un bloc "Pour le" et un bloc "Donné le" pour le même devoir
lesson1 = Lesson(
id="lesson-1",
start=datetime(2026, 9, 5, 8, 0, 0),
end=datetime(2026, 9, 5, 9, 0, 0),
subject="Mathématiques",
teachers=["M. Dupont"],
rooms=["204"],
status=LessonStatus.NORMAL,
content="Résoudre des équations.",
homework_blocks=[
HomeworkBlock(
kind="due",
date=date(2026, 9, 10),
text="Exercices 1 à 5 page 42.",
html="<p>Exercices 1 à 5 page 42.</p>",
),
HomeworkBlock(
kind="assigned",
date=date(2026, 9, 5),
text="Exercices 1 à 5 page 42.", # Même texte que le bloc "Pour le"
html="<p>Exercices 1 à 5 page 42.</p>",
),
],
)
# Cours 2 : contient un bloc "Donné le" pour un autre devoir
lesson2 = Lesson(
id="lesson-2",
start=datetime(2026, 9, 5, 9, 0, 0),
end=datetime(2026, 9, 5, 10, 0, 0),
subject="Français",
teachers=["Mme Martin"],
rooms=["205"],
status=LessonStatus.NORMAL,
content="Étude d'un texte.",
homework_blocks=[
HomeworkBlock(
kind="assigned",
date=date(2026, 9, 5),
text="Lire les pages 10 à 15.",
html="<p>Lire les pages 10 à 15.</p>",
),
],
)
# Date cible : 10 septembre 2026
target_date = date(2026, 9, 10)
# Collecter et dédupliquer les devoirs
homeworks = collect_homeworks([lesson1, lesson2], target_date)
# Vérifications :
# - Le devoir "Exercices 1 à 5 page 42" doit apparaître une seule fois (dédupliqué)
# - Le devoir "Lire les pages 10 à 15" ne doit pas apparaître (car sa date d'échéance n'est pas le 10/09)
assert len(homeworks) == 1
assert homeworks[0].text == "Exercices 1 à 5 page 42."
assert homeworks[0].subject == "Mathématiques"
assert homeworks[0].due_on == date(2026, 9, 10)
### 12.5 Exécution des tests
#### 12.5.1 Commandes pytest
| Commande | Description |
|-----------------------------------|--------------------------------------------------|
| `pytest` | Exécute tous les tests. |
| `pytest tests/unit/` | Exécute uniquement les tests unitaires. |
| `pytest tests/integration/` | Exécute uniquement les tests d'intégration. |
| `pytest -v` | Mode verbeux (affiche les noms des tests). |
| `pytest -x` | Arrête au premier échec. |
| `pytest --tb=short` | Affiche une traceback courte. |
| `pytest --cov=pronote_sync` | Mesure la couverture de code. |
| `pytest --cov=pronote_sync --cov-report=html` | Génère un rapport HTML de couverture. |
#### 12.5.2 Configuration de la couverture (`pyproject.toml`)
```toml
[tool.pytest.ini_options]
minversion = "7.0"
testpaths = ["tests"]
python_files = ["test_*.py"]
python_functions = ["test_*"]
addopts = "-v --tb=short"
[tool.coverage.run]
source = ["pronote_sync"]
branch = true
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise NotImplementedError",
"if TYPE_CHECKING:",
]
fail_under = 90 # Échec si couverture < 90%
[tool.coverage.html]
directory = "coverage_html"
```
#### 12.5.3 Exemple de rapport de couverture
```
Name Stmts Miss Cover Missing
--------------------------------------------------------------
pronote_sync/config/settings.py 50 0 100%
pronote_sync/models/agenda.py 100 0 100%
pronote_sync/sources/pronote/ical.py 200 5 97% 120-125
pronote_sync/pipeline/run.py 150 2 99% 200
--------------------------------------------------------------
TOTAL 1000 10 99%
```
### 12.6 Points clés
- **Mocks** : Utiliser `requests-mock` pour les requêtes HTTP, `unittest.mock` pour les dépendances.
- **Fixtures** : Stocker les données de test dans `tests/fixtures/`.
- **Anonymisation** : **Jamais** de données réelles dans les fixtures.
- **Couverture** : Viser **≥ 90%** (comme dans `pronote-digest`).
- **Vitesse** : Les tests doivent s'exécuter en **quelques secondes** (pas de sleeps inutiles).
- **Déterminisme** : Les tests doivent être **reproductibles** (pas de dépendance à l'heure ou à des données externes).
---
## 13. Checklist de sécurité
### 13.1 Secrets et données sensibles
| **Risque** | **Mesure de mitigation** | **Vérification** | **Statut** |
|-------------------------------------|----------------------------------------------------------------------------------------|-------------------------------------------|------------|
| Tokens dans le code | Utiliser `pydantic-settings` + `SecretStr` pour les variables d'environnement. | `grep -r "icalsecurise\|password\|api_key" src/` | ❌ Interdit |
| Tokens dans les logs | Masquage systématique via `RedactingFormatter` (voir [Section 4.2](#42-implémentation)). | Tests avec `PRONOTE_ICAL_URL` contenant un token. | ✅ Obligatoire |
| Tokens dans les erreurs | Masquage dans les messages d'erreur (voir `redact_url` et `redact_secrets`). | Tests avec URLs contenant des tokens. | ✅ Obligatoire |
| Tokens dans les fixtures | **Anonymiser** toutes les fixtures (pas de tokens réels). | Vérification manuelle des fixtures. | ✅ Obligatoire |
| Tokens dans les commits Git | Utiliser `.gitignore` pour `.env` et `pre-commit` pour bloquer les secrets. | `git grep "icalsecurise\|password" -- .` (contenu suivi courant) | ❌ Interdit |
| Clés API dans le code | Toujours charger depuis les variables d'environnement. | `grep -r "api_key\s*=" src/` | ❌ Interdit |
| Mots de passe en clair | Toujours utiliser `SecretStr` ou `getpass`. | `grep -r "password\s*=" src/` | ❌ Interdit |
| Fichiers d'état non protégés | Appliquer `chmod 600` et exclure du Git (`.gitignore`). | `ls -la .pronote_sync_state.json` (doit être `-rw-------`) | ✅ Obligatoire |
### 13.2 Validation des entrées
| **Risque** | **Mesure de mitigation** | **Vérification** | **Statut** |
|-------------------------------------|----------------------------------------------------------------------------------------|-------------------------------------------|------------|
| Injection SQL (si SQLite) | Utiliser des requêtes paramétrées (pas de string formatting). | Revue du code utilisant SQLite. | ✅ Obligatoire |
| Injection XMPP | Échapper les messages XMPP (slixmpp le fait automatiquement). | Tests avec des messages contenant `<`, `>`, `&`. | ✅ Obligatoire |
| Parsing iCal malveillant | Valider que le flux contient `BEGIN:VCALENDAR` avant parsing. | Tests avec des flux invalides. | ✅ Obligatoire |
| URLs malveillantes | Valider les URLs avec `urllib.parse` avant utilisation. | Tests avec des URLs malformées. | ✅ Obligatoire |
| Taille des requêtes IA | Limiter la taille du prompt (`MAX_LENGTH = 800`). | Tests avec de grands prompts. | ✅ Obligatoire |
### 13.3 Authentification et autorisation
| **Risque** | **Mesure de mitigation** | **Vérification** | **Statut** |
|-------------------------------------|----------------------------------------------------------------------------------------|-------------------------------------------|------------|
| Accès non autorisé à Pronote | Utiliser les identifiants fournis par l'utilisateur (pas de hardcoding). | Revue du code d'authentification. | ✅ Obligatoire |
| Accès non autorisé à CalDAV | Utiliser les identifiants fournis par l'utilisateur. | Revue du code CalDAV. | ✅ Obligatoire |
| Accès non autorisé à XMPP | Utiliser les identifiants fournis par l'utilisateur. | Revue du code XMPP. | ✅ Obligatoire |
| Accès non autorisé à l'API IA | Utiliser les clés API fournies par l'utilisateur. | Revue du code IA. | ✅ Obligatoire |
| Stockage des secrets | **Ne jamais stocker** les secrets en base de données ou dans des fichiers non sécurisés. | Revue de l'architecture. | ✅ Obligatoire |
### 13.4 Chiffrement et réseau
| **Risque** | **Mesure de mitigation** | **Vérification** | **Statut** |
|-------------------------------------|----------------------------------------------------------------------------------------|-------------------------------------------|------------|
| Requêtes HTTP non chiffrées | Toujours utiliser `https://` pour Pronote, CalDAV, IA. | Vérification des URLs dans le code. | ✅ Obligatoire |
| Certificats SSL invalides | Utiliser `verify=True` par défaut dans `requests` (désactiver uniquement pour les tests). | `grep -r "verify=False" src/` | ⚠️ À éviter |
| Timeout des requêtes | Configurer des timeouts (20s pour iCal, 30s pour IA). | Revue des appels HTTP. | ✅ Obligatoire |
| Fuites de mémoire (secrets) | **Ne jamais stocker** les secrets en mémoire plus longtemps que nécessaire. | Revue du code de gestion des secrets. | ✅ Obligatoire |
### 13.5 Audit et logging
| **Risque** | **Mesure de mitigation** | **Vérification** | **Statut** |
|-------------------------------------|----------------------------------------------------------------------------------------|-------------------------------------------|------------|
| Logs contenant des secrets | Toujours utiliser `RedactingFormatter` pour les logs. | Tests avec des secrets dans les logs. | ✅ Obligatoire |
| Logs trop verbeux | Limiter le niveau de log à `INFO` par défaut. | Revue de la configuration des logs. | ✅ Obligatoire |
| Logs des erreurs sensibles | Masquer les détails sensibles dans les erreurs (ex: URLs avec tokens). | Tests avec des erreurs contenant des secrets. | ✅ Obligatoire |
| Audit des accès | **Ne pas implémenter** de logging des accès (hors scope). | Revue de l'architecture. | ⚠️ Hors scope |
### 13.6 Exemple de script de vérification de sécurité
```python
#!/usr/bin/env python3
"""
Script de vérification de sécurité pour le projet.
À exécuter avant chaque commit ou release.
"""
import subprocess
import sys
from pathlib import Path
def run_command(cmd: list, description: str) -> bool:
"""Exécute une commande et affiche le résultat.
Args:
cmd: Liste d'arguments pour subprocess.run (pas de shell=True pour éviter les injections).
description: Description de la vérification.
"""
print(f"🔍 {description}...")
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
print(f"❌ ÉCHEC: {cmd}")
print(result.stdout)
print(result.stderr)
return False
if result.stdout.strip():
print(f"⚠️ TROUVÉ:")
print(result.stdout)
return False
print(f"✅ OK")
return True
def check_secrets_in_code():
"""Vérifie qu'il n'y a pas de secrets dans le code."""
checks = [
(["grep", "-r", "icalsecurise=", "src/", "tests/", "--include=*.py"], "Tokens iCal dans le code"),
(["grep", "-r", "password\s*=", "src/", "tests/", "--include=*.py"], "Mots de passe en clair"),
(["grep", "-r", "api_key\s*=", "src/", "tests/", "--include=*.py"], "Clés API en clair"),
(["grep", "-r", "PRONOTE_ICAL_URL.*=", "src/", "tests/", "--include=*.py"], "URLs iCal en clair"),
]
all_ok = True
for cmd, desc in checks:
if not run_command(cmd, desc):
all_ok = False
return all_ok
def check_secrets_in_git():
"""Vérifie qu'il n'y a pas de secrets dans le contenu suivi courant.
Note : `git grep` recherche dans le contenu **suivi courant** (working tree + index),
pas dans l'historique Git. Pour rechercher dans l'historique, utiliser :
- `git log -p` (pour voir les diffs complets)
- `git log -S 'icalsecurise='` (pour trouver les commits contenant une chaîne)
- Un outil dédié comme `trufflehog` ou `git-secrets --scan-history`.
"""
checks = [
["git", "grep", "-l", "icalsecurise=", "--", "."],
["git", "grep", "-l", "password=", "--", "."],
["git", "grep", "-l", "api_key=", "--", "."],
]
all_ok = True
for cmd, desc in checks:
if not run_command(cmd, desc):
all_ok = False
return all_ok
def check_fixtures():
"""Vérifie que les fixtures sont anonymisées."""
fixtures_dir = Path("tests/fixtures")
if not fixtures_dir.exists():
print("⚠️ Dossier tests/fixtures/ introuvable")
return True
# Vérifier qu'il n'y a pas de tokens dans les fixtures
for fixture_file in fixtures_dir.glob("*"):
if fixture_file.suffix == ".ics":
content = fixture_file.read_text()
if "icalsecurise=" in content:
print(f"❌ Token trouvé dans {fixture_file}")
return False
print("✅ Fixtures OK")
return True
def main():
"""Exécute toutes les vérifications."""
print("🔒 Vérification de sécurité\n")
all_ok = True
# Vérifier les secrets dans le code
if not check_secrets_in_code():
all_ok = False
print()
# Vérifier les secrets dans Git
if not check_secrets_in_git():
all_ok = False
print()
# Vérifier les fixtures
if not check_fixtures():
all_ok = False
print()
if all_ok:
print("✅ Toutes les vérifications de sécurité ont réussi!")
return 0
else:
print("❌ Certaines vérifications de sécurité ont échoué!")
return 1
if __name__ == "__main__":
sys.exit(main())
```
### 13.7 Outils recommandés
| **Outil** | **Usage** | **Installation** |
|-------------------------|---------------------------------------------------------------------------|--------------------------------|
| `grep` | Recherche de secrets dans le code. | Natif (Linux/macOS) |
| `git-secrets` | Détection de secrets dans Git (historique et contenu suivi). | `git clone https://github.com/awslabs/git-secrets.git && cd git-secrets && sudo ./install.sh` |
| `trufflehog` | Détection de secrets dans les dépôts Git. | `pip install trufflehog` |
| `bandit` | Analyse de sécurité Python. | `pip install bandit` |
| `safety` | Vérification des dépendances vulnérables. | `pip install safety` |
| `pre-commit` | Exécution de hooks avant commit (ex: `detect-secrets`). | `pip install pre-commit` |
### 13.8 Exemple de configuration `pre-commit` (`.pre-commit-config.yaml`)
```yaml
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.4.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: check-added-large-files
- repo: https://github.com/Yelp/detect-secrets
rev: v1.4.0
hooks:
- id: detect-secrets
args: [--baseline, .secrets.baseline]
- repo: https://github.com/PyCQA/bandit
rev: 1.7.5
hooks:
- id: bandit
args: [-ll, -ii]
- repo: https://github.com/awslabs/git-secrets
rev: master
hooks:
- id: git-secrets
args: [--scan, --no-index]
- repo: local
hooks:
- id: security-check
name: Vérification de sécurité
entry: python scripts/security_check.py
language: system
pass_filenames: true
stages: [commit]
```
### 13.9 Points clés
- **Zéro secret en clair** : Aucun token, mot de passe ou clé API ne doit apparaître dans le code ou les logs.
- **Masquage systématique** : Toujours masquer les secrets dans les logs et les erreurs.
- **Anonymisation** : Toutes les fixtures doivent être anonymisées.
- **Validation des entrées** : Toujours valider les URLs, les flux iCal et les requêtes IA.
- **Chiffrement** : Toujours utiliser HTTPS pour les requêtes réseau.
- **Audit régulier** : Exécuter des vérifications de sécurité avant chaque commit/release.
---
## 14. Checklist d'exploitation
### 14.1 Déploiement
| **Tâche** | **Description** | **Obligatoire** | **Statut** |
|----------------------------------------|-----------------------------------------------------------------------------------------------------|-----------------|------------|
| Configuration des variables d'environnement | Vérifier que toutes les variables obligatoires sont définies (voir [Section 3.1](#31-variables-denvironnement)). | ✅ Oui | |
| Vérification des secrets | Exécuter le script de vérification de sécurité (voir [Section 13.6](#136-exemple-de-script-de-vérification-de-sécurité)). | ✅ Oui | |
| Test en mode dry-run | Exécuter le pipeline avec `DRY_RUN=true` pour vérifier que tout fonctionne sans effet de bord. | ✅ Oui | |
| Configuration des logs | Vérifier que les logs sont configurés avec masquage des secrets (voir [Section 4.2](#42-implémentation)). | ✅ Oui | |
| Vérification des dépendances | Exécuter `pip check` pour vérifier que toutes les dépendances sont installées. | ✅ Oui | |
| Configuration du cron (si planifié) | Configurer une tâche cron pour exécuter le script régulièrement (ex: tous les jours à 18h). | ⚠️ Non | |
### 14.2 Configuration du cron (optionnel)
Si le projet est exécuté **régulièrement** (ex: tous les jours), on peut configurer une tâche cron :
```bash
# Éditer le crontab
crontab -e
```
Exemple de ligne cron (exécution tous les jours à 18h) :
```
0 18 * * * /usr/bin/python3 /chemin/vers/pronote_sync/cli/main.py >> /var/log/pronote_sync.log 2>&1
```
**Bonnes pratiques** :
- **Rediriger les logs** vers un fichier pour le débogage.
- **Utiliser un environnement virtuel** pour isoler les dépendances.
- **Vérifier les permissions** : Le script doit être exécutable (`chmod +x`).
- **Tester la commande** manuellement avant de l'ajouter au cron.
### 14.3 Supervision
| **Tâche** | **Description** | **Obligatoire** | **Statut** |
|----------------------------------------|-----------------------------------------------------------------------------------------------------|-----------------|------------|
| Vérification des logs | Surveiller les logs pour détecter les erreurs (ex: `tail -f /var/log/pronote_sync.log`). | ✅ Oui | |
| Alertes en cas d'échec | Configurer des alertes (ex: email, notification XMPP) si le pipeline échoue. | ⚠️ Non | |
| Rotation des logs | Configurer une rotation des logs (ex: `logrotate`) pour éviter les fichiers trop volumineux. | ⚠️ Non | |
| Sauvegarde des données | Sauvegarder régulièrement les données synchronisées (ex: CalDAV, état local). | ⚠️ Non | |
### 14.4 Exemple de configuration `logrotate` (`/etc/logrotate.d/pronote_sync`)
```
/var/log/pronote_sync.log {
daily
missingok
rotate 7
compress
delaycompress
notifempty
create 0640 user user
}
```
### 14.5 Maintenance
| **Tâche** | **Description** | **Fréquence** | **Statut** |
|----------------------------------------|-----------------------------------------------------------------------------------------------------|---------------|------------|
| Mise à jour des dépendances | Exécuter `pip list --outdated` et mettre à jour les dépendances. | Mensuelle | |
| Vérification des secrets | Exécuter le script de vérification de sécurité. | Avant chaque mise à jour | |
| Test du pipeline | Exécuter le pipeline en mode dry-run pour vérifier que tout fonctionne. | Avant chaque mise à jour | |
| Sauvegarde de la configuration | Sauvegarder le fichier `.env` et les fichiers de configuration. | Avant chaque mise à jour | |
| Revue des logs | Vérifier les logs pour détecter des erreurs récurrentes. | Hebdomadaire | |
### 14.6 Dépannage
#### 14.6.1 Problèmes courants
| **Problème** | **Cause possible** | **Solution** |
|---------------------------------------|------------------------------------------------------------------------------------|------------------------------------------------------------------------------|
| Échec de la récupération iCal | Token `icalsecurise` expiré ou invalide. | Régénérer le token depuis Pronote. |
| Échec de la connexion Pronote (`pronotepy`) | Identifiants incorrects ou ENT non supporté. | Vérifier `PRONOTE_USERNAME`, `PRONOTE_PASSWORD`, `PRONOTE_ENT`. |
| Échec de la connexion CalDAV | URL, identifiant ou mot de passe CalDAV incorrect. | Vérifier `CALDAV_URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`. |
| Échec de la connexion XMPP | Identifiant ou mot de passe XMPP incorrect. | Vérifier `XMPP_JID`, `XMPP_PASSWORD`. |
| Échec de la synthèse IA | Clé API IA invalide ou modèle non disponible. | Vérifier `AI_API_KEY`, `AI_BASE_URL`, `AI_MODEL`. |
| Aucun cours récupéré | Flux iCal vide ou `pronotepy` non configuré. | Vérifier `PRONOTE_ICAL_URL` ou les identifiants `pronotepy`. |
| Doublons dans les devoirs | Problème de déduplication. | Vérifier la logique de déduplication (voir [Section 5.1.4](#514-déduplication-des-devoirs)). |
| Synchronisation CalDAV lente | Trop d'événements à synchroniser. | Réduire `SYNC_PAST_DAYS` ou `SYNC_FUTURE_DAYS`. |
#### 14.6.2 Commandes de débogage
| **Commande** | **Description** |
|---------------------------------------|-----------------------------------------------------------------------------------------------------|
| `python -m pronote_sync.cli.main --dry-run --log-level DEBUG` | Exécute le pipeline en mode dry-run avec des logs détaillés. |
| `python -c "from pronote_sync.sources.pronote.ical import fetch_ical; print(fetch_ical('file://tests/fixtures/pronote-4e.ics'))"` | Teste le parsing d'un fichier iCal local. |
| `python -c "from pronote_sync.config.settings import settings; print(settings)"` | Affiche la configuration chargée. |
| `python -c "import caldav; print(caldav.__version__)"` | Vérifie la version de la bibliothèque CalDAV. |
| `python -c "import slixmpp; print(slixmpp.__version__)"` | Vérifie la version de la bibliothèque XMPP. |
### 14.7 Points clés
- **Test en dry-run** : Toujours tester avec `DRY_RUN=true` avant de passer en production.
- **Vérification des secrets** : Exécuter le script de vérification de sécurité avant chaque déploiement.
- **Supervision** : Surveiller les logs pour détecter les erreurs.
- **Maintenance** : Mettre à jour régulièrement les dépendances.
- **Dépannage** : Utiliser les commandes de débogage pour diagnostiquer les problèmes.
---
## 15. Limites connues et risques
### 15.1 Limites techniques
| **Limite** | **Description** | **Impact** | **Solution proposée** |
|-----------------------------------------|-----------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------|---------------------------------------------------------------------------------------|
| **Pas d'API officielle Pronote** | Pronote ne fournit pas d'API publique pour les élèves/parents. | Dépendance au flux iCal ou à `pronotepy` (reverse-engineering). | Utiliser le flux iCal officiel en priorité. |
| **Flux iCal incomplet** | Certains établissements désactivent l'export des devoirs dans iCal. | Impossible de récupérer les devoirs via iCal. | Basculer sur `pronotepy` pour les devoirs. |
| **`pronotepy` en maintenance** | `pronotepy` est en mode maintenance (bugfixes uniquement). | Risque de cassure si Pronote met à jour son protocole. | Surveiller les issues GitHub de `pronotepy`. |
| **Messages non disponibles dans iCal** | Les messages, discussions et informations ne sont **pas** dans le flux iCal. | Impossible de récupérer ces données sans `pronotepy`. | Utiliser `pronotepy` pour les messages. |
| **CalDAV : support variable** | Certains serveurs CalDAV ont des limitations (ex: pas de sync-token). | Synchronisation moins efficace. | Utiliser un état local (SQLite/JSON) pour compenser. |
| **XMPP : serveurs variés** | Les serveurs XMPP ont des configurations différentes (ex: authentification, TLS). | Problèmes de compatibilité possibles. | Tester avec le serveur XMPP cible avant déploiement. |
| **IA : coûts et latence** | Les API IA peuvent être coûteuses et lentes. | Synthèse IA peut être désactivée ou lente. | Limiter la taille du prompt et utiliser un timeout. |
| **Python 3.13.5+** | Le projet nécessite Python ≥ 3.13.5. | Incompatibilité avec les anciennes versions de Python. | Documenter clairement la version requise. |
### 15.2 Risques de sécurité
| **Risque** | **Description** | **Impact** | **Mitigation** |
|-----------------------------------------|-----------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------|---------------------------------------------------------------------------------------|
| **Fuites de tokens** | Un token `icalsecurise` ou une clé API pourrait fuir dans les logs ou le code. | Accès non autorisé à Pronote ou à l'API IA. | Masquage systématique des secrets (voir [Section 4](#4-gestion-des-secrets-et-redaction)). |
| **Reverse-engineering de Pronote** | `pronotepy` utilise du reverse-engineering, ce qui peut violer les CGU de Pronote. | Risque juridique ou blocage par Index Éducation. | Utiliser le flux iCal officiel en priorité. |
| **Injections XMPP** | Un message XMPP malveillant pourrait être envoyé. | Exécution de code arbitraire (si le client XMPP est vulnérable). | Utiliser `slixmpp` (maintenu) et échapper les messages. |
| **Attaques par force brute** | Un attaquant pourrait essayer de deviner les identifiants Pronote/CalDAV/XMPP. | Accès non autorisé aux données. | Limiter les tentatives de connexion et utiliser des mots de passe robustes. |
| **Fuites de données** | Les données Pronote (cours, devoirs, messages) sont sensibles. | Violation de la vie privée. | **Ne jamais stocker** les données en clair (sauf si nécessaire). Anonymiser les fixtures. |
### 15.3 Risques opérationnels
| **Risque** | **Description** | **Impact** | **Mitigation** |
|-----------------------------------------|-----------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------|---------------------------------------------------------------------------------------|
| **Changements dans Pronote** | Pronote pourrait modifier son format iCal ou son protocole interne. | Le projet pourrait cesser de fonctionner. | Surveiller les mises à jour de Pronote et adapter le code. |
| **Changements dans CalDAV** | Le serveur CalDAV pourrait changer son API ou ses limitations. | La synchronisation pourrait échouer. | Tester régulièrement avec le serveur CalDAV cible. |
| **Changements dans XMPP** | Le serveur XMPP pourrait changer sa configuration. | L'envoi des messages pourrait échouer. | Tester régulièrement avec le serveur XMPP cible. |
| **Changements dans les API IA** | Les fournisseurs IA pourraient modifier leurs API ou leurs modèles. | La synthèse IA pourrait échouer. | Utiliser un adaptateur générique (ex: `litellm`) et gérer les erreurs. |
| **Problèmes réseau** | Le projet dépend de connexions réseau (Pronote, CalDAV, XMPP, IA). | Le pipeline pourrait échouer. | Implémenter des timeouts et des modes dégradés. |
| **Problèmes de performance** | Le projet pourrait être lent avec de nombreux événements ou devoirs. | Expérience utilisateur dégradée. | Optimiser le code et limiter la fenêtre de synchronisation. |
### 15.4 Risques juridiques
| **Risque** | **Description** | **Impact** | **Mitigation** |
|-----------------------------------------|-----------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------|---------------------------------------------------------------------------------------|
| **Violation des CGU de Pronote** | L'utilisation de `pronotepy` pourrait violer les CGU de Pronote/Index Éducation. | Risque juridique (poursuites, blocage). | Utiliser le flux iCal officiel en priorité. Préférer une solution officielle si disponible. |
| **Violation du RGPD** | Le projet manipule des données personnelles (noms, devoirs, messages). | Risque juridique (amendes). | **Anonymiser** toutes les données stockées ou loggées. Obtenir le consentement des utilisateurs. |
| **Utilisation non autorisée des API IA** | Certaines API IA ont des restrictions d'usage (ex: interdiction d'usage commercial). | Risque juridique (violation des contrats). | Vérifier les CGU des fournisseurs IA et respecter leurs limitations. |
### 15.5 Recommandations générales
1. **Privilégier le flux iCal officiel** :
- Moins risqué que `pronotepy` (pas de reverse-engineering).
- Plus stable (format standardisé).
2. **Limiter l'usage de `pronotepy`** :
- Utiliser uniquement pour les données **non disponibles dans iCal** (messages, informations).
- Surveiller les mises à jour de `pronotepy` pour détecter les cassures.
3. **Gérer les erreurs avec grâce** :
- **Ne jamais bloquer** le pipeline pour des erreurs non critiques.
- Toujours fournir un **mode dégradé** (ex: envoyer le message sans synthèse IA).
4. **Protéger les secrets** :
- **Masquer** systématiquement les tokens, mots de passe et clés API.
- **Ne jamais stocker** les secrets en clair dans le code ou les logs.
5. **Respecter la vie privée** :
- **Anonymiser** toutes les données stockées ou loggées.
- **Ne pas collecter** de données inutiles.
6. **Documenter clairement** :
- **Informer les utilisateurs** des risques (ex: `pronotepy` pourrait casser).
- **Documenter les limitations** (ex: certains établissements désactivent l'export des devoirs dans iCal).
7. **Tester régulièrement** :
- **Tester avec les serveurs cibles** (CalDAV, XMPP) avant déploiement.
- **Mettre à jour les dépendances** régulièrement.
---
## Annexes
### Glossaire
| **Terme** | **Définition** |
|-------------------------|-----------------------------------------------------------------------------------------------------|
| **Pronote** | Logiciel de gestion de vie scolaire (collège/lycée) développé par Index Éducation. |
| **Pronote Campus** | Version de Pronote pour l'enseignement supérieur (non ciblé par ce projet). |
| **iCal** | Format standard pour les calendriers (RFC 5545). |
| **CalDAV** | Protocole pour synchroniser des calendriers via HTTP. |
| **XMPP** | Protocole de messagerie instantanée (anciennement Jabber). |
| **UID** | Identifiant unique pour un événement iCal/CalDAV. |
| **Dry-run** | Mode de test où aucune modification n'est appliquée (lecture seule). |
| **Idempotence** | Propriété d'une opération qui produit le même résultat si elle est exécutée plusieurs fois. |
| **Reverse-engineering** | Technique consistant à analyser un logiciel pour en comprendre le fonctionnement interne. |
### Ressources utiles
| **Ressource** | **Lien** | **Description** |
|----------------------------------------|---------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------|
| Pronote (officiel) | [https://www.index-education.com](https://www.index-education.com) | Site officiel de Pronote. |
| `pronotepy` | [https://github.com/bain3/pronotepy](https://github.com/bain3/pronotepy) | Bibliothèque Python pour interagir avec Pronote. |
| `caldav` (PyPI) | [https://pypi.org/project/caldav/](https://pypi.org/project/caldav/) | Client CalDAV pour Python. |
| `slixmpp` | [https://github.com/poezio/slixmpp](https://github.com/poezio/slixmpp) | Bibliothèque XMPP pour Python (asyncio). |
| `icalendar` | [https://pypi.org/project/icalendar/](https://pypi.org/project/icalendar/) | Bibliothèque pour parser/générer des fichiers iCal. |
| `pydantic` | [https://pydantic.dev/](https://pydantic.dev/) | Bibliothèque pour la validation des données. |
| `pydantic-settings` | [https://pydantic.dev/latest/usage/pydantic_settings/](https://pydantic.dev/latest/usage/pydantic_settings/) | Extension de Pydantic pour gérer les variables d'environnement. |
| 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. |
### C. Exemple de fichier `pyproject.toml`
```toml
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "pronote-sync"
version = "0.1.0"
description = "Synchronisation Pronote → CalDAV + XMPP"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.13.5"
authors = [
{name = "Votre Nom", email = "votre@email.com"}
]
keywords = ["pronote", "caldav", "xmpp", "sync", "school"]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: End Users/Desktop",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Topic :: Office/Business :: Scheduling",
"Topic :: Communications :: Chat",
"Topic :: Utilities",
]
dependencies = [
"requests>=2.31.0",
"icalendar>=5.0.0",
"caldav>=1.3.0",
"slixmpp>=1.8.0",
"pydantic>=2.0.0",
"pydantic-settings>=2.0.0",
"pronotepy>=2.15.0",
"openai>=1.0.0",
"httpx>=0.25.0",
]
[project.optional-dependencies]
ai-litellm = ["litellm>=1.0"]
dev = [
"pytest>=7.0.0",
"pytest-cov>=4.0.0",
"pytest-mock>=3.0.0",
"requests-mock>=1.11.0",
"aioresponses>=0.7.0",
"bandit>=1.7.0",
"safety>=2.0.0",
"pre-commit>=3.0.0",
"detect-secrets>=1.4.0",
]
[project.scripts]
pronote-sync = "pronote_sync.cli.main:main"
[project.urls]
Homepage = "https://github.com/votre-utilisateur/pronote-sync"
Documentation = "https://github.com/votre-utilisateur/pronote-sync#readme"
Repository = "https://github.com/votre-utilisateur/pronote-sync"
Issues = "https://github.com/votre-utilisateur/pronote-sync/issues"
[tool.setuptools.packages.find]
where = ["."]
include = ["pronote_sync*"]
[tool.pytest.ini_options]
minversion = "7.0"
testpaths = ["tests"]
python_files = ["test_*.py"]
python_functions = ["test_*"]
addopts = "-v --tb=short"
[tool.coverage.run]
source = ["pronote_sync"]
branch = true
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise NotImplementedError",
"if TYPE_CHECKING:",
]
fail_under = 90
[tool.bandit]
exclude_dirs = ["tests", "venv"]
skips = ["B101"] # Ignorer les assertions (utilisées dans les tests)
[tool.ruff]
line-length = 100
target-version = "py313"
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # Pyflakes
"I", # isort
"B", # flake8-bugbear
"C4", # flake8-comprehensions
"UP", # pyupgrade
]
ignore = [
"E501", # line too long (géré par line-length)
]
[tool.mypy]
python_version = "3.13"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
strict = true
---
## Conclusion
Ce guide fournit une **base architecturale et technique solide** pour développer un outil Python de synchronisation Pronote → CalDAV + XMPP, inspiré du projet TypeScript `pronote-digest`. Il couvre :
- **L'architecture** : Pipeline modulaire, injectable et testable.
- **Les spécificités Pronote** : Parsing des flux iCal, déduplication des devoirs, normalisation des UID.
- **Les intégrations** : CalDAV (synchronisation différentielle), XMPP (envoi de messages structurés), IA (synthèse optionnelle).
- **La sécurité** : Gestion des secrets, masquage des logs, validation des entrées.
- **Les tests** : Mocks, fixtures anonymisées, couverture élevée.
- **L'exploitation** : Déploiement, supervision, maintenance.
### Prochaines étapes
1. **Créer le dépôt** : Initialiser un nouveau dépôt Python avec la structure proposée.
2. **Implémenter le cœur** : Commencer par les modules `models/`, `sources/pronote/ical.py` et `utils/`.
3. **Ajouter les tests** : Écrire des tests unitaires pour chaque module dès le début.
4. **Configurer CI/CD** : Mettre en place GitHub Actions pour exécuter les tests et vérifier la sécurité.
5. **Tester en conditions réelles** : Utiliser des flux iCal Pronote anonymisés pour valider le parsing.
> **⚠️ Rappel** : Ce guide est **volontairement détaillé** pour préserver les connaissances acquises sur les spécificités de Pronote. Certaines sections (ex: parsing iCal) contiennent des **observations précises** issues de l'analyse du code TypeScript existant. **Ne pas sous-estimer l'importance de ces détails** : ils sont critiques pour un fonctionnement fiable du projet.
---
## Annexes
| **Terme** | **Définition** |
|-------------------------|-----------------------------------------------------------------------------------------------------|
| **Pronote** | Logiciel de gestion de vie scolaire (collège/lycée) développé par Index Éducation. |
| **Pronote Campus** | Version de Pronote pour l'enseignement supérieur (non ciblé par ce projet). |
| **iCal** | Format standard pour les calendriers (RFC 5545). |
| **CalDAV** | Protocole pour synchroniser des calendriers via HTTP. |
| **XMPP** | Protocole de messagerie instantanée (anciennement Jabber). |
| **UID** | Identifiant unique pour un événement iCal/CalDAV. |
| **Dry-run** | Mode de test où aucune modification n'est appliquée (lecture seule). |
| **Idempotence** | Propriété d'une opération qui produit le même résultat si elle est exécutée plusieurs fois. |
| **Reverse-engineering** | Technique consistant à analyser un logiciel pour en comprendre le fonctionnement interne. |
### B. Ressources utiles
| **Ressource** | **Lien** | **Description** |
|----------------------------------------|---------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------|
| Pronote (officiel) | [https://www.index-education.com](https://www.index-education.com) | Site officiel de Pronote. |
| `pronotepy` | [https://github.com/bain3/pronotepy](https://github.com/bain3/pronotepy) | Bibliothèque Python pour interagir avec Pronote. |
| `caldav` (PyPI) | [https://pypi.org/project/caldav/](https://pypi.org/project/caldav/) | Client CalDAV pour Python. |
| `slixmpp` | [https://github.com/poezio/slixmpp](https://github.com/poezio/slixmpp) | Bibliothèque XMPP pour Python (asyncio). |
| `icalendar` | [https://pypi.org/project/icalendar/](https://pypi.org/project/icalendar/) | Bibliothèque pour parser/générer des fichiers iCal. |
| `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.
Reference in New Issue View Git Blame Copy Permalink