# 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 iCal/CSV local). - 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 iCal/CSV │ └── 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 iCal/CSV de l'agenda théorique. | `None` | `str \| 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 --- THEORETICAL_AGENDA_PATH=./data/theoretical.ics # --- 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 ``, ``) | | 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 Titre de l'article https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625 https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625 Mon, 10 Aug 2026 09:00:11 +0000 Administration M. Dupont Extrait en texte brut... Contenu HTML complet de l'article...

Lien vers un PDF ... ]]> ``` **Champs clés** : | **Champ** | **Description** | **Format** | **Utilisation** | |--------------------|-------------------------------------------------------------------------------|--------------------------|------------------------------------------| | `` | 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 https://blogpeda.ac-bordeaux.fr/cjeliote/ Blog du collège Réunion de rentrée https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625 https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625 Mon, 10 Aug 2026 09:00:11 +0000 Administration M. Dupont La réunion de rentrée aura lieu le 1er septembre. La réunion de rentrée aura lieu le 1er septembre à 18h en salle 204.

Ordre du jour

]]> Sortie pédagogique https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626 https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626 Tue, 11 Aug 2026 14:30:00 +0000 Pédagogie Mme Martin Sortie prévue au musée le 15 septembre. Une sortie pédagogique au musée est prévue le 15 septembre.

]]> ``` #### 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 `` (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
Matière : Mathématiques Professeur : M. Dupont Salle : 204 Groupe : Classe entière Contenu pédagogique : Résoudre des équations du second degré. Pour le 10/09/2026 : Exercices 1 à 5 page 42. Donné le 05/09/2026 : Exercices 1 à 5 page 42.
``` - **Structure réelle** : - La `DESCRIPTION` contient d'abord un **en-tête texte** (avant le premier ``) 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 ``. - Les sections sont identifiées par les balises exactes : - `Contenu pédagogique : \n` → contenu pédagogique (texte brut). - `Pour le JJ/MM/AAAA : \n` → devoir à faire (date d'échéance). - `Donné le JJ/MM/AAAA : \n` → 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 ``, le corps HTML commence à partir du premier ``. Args: description: Contenu brut de la DESCRIPTION. Returns: Tuple (en-tête texte, corps HTML). """ # Trouver la position du premier `` strong_start = description.find("") 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 ``). 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 ``). 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"Contenu pédagogique : \n(.+?)(?:|$)", 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"Pour le (\d{2}/\d{2}/\d{4}) : \n(.+?)(?:|$)", 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"Donné le (\d{2}/\d{2}/\d{4}) : \n(.+?)(?:|$)", 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 ``). 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"Pour le (\d{2}/\d{2}/\d{4}) : \n(.+?)(?:|$)", 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"Donné le (\d{2}/\d{2}/\d{4}) : \n(.+?)(?:|$)", 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. - **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. ### 7.2 Client CalDAV (`sync/caldav.py`) Utilisation de la bibliothèque [`caldav`](https://pypi.org/project/caldav/) (Python 3.8+, maintenue). ```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_name: str = "Pronote", dry_run: bool = False, ): self.url = url self.username = username self.password = password self.calendar_name = calendar_name 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écupérer ou créer le calendrier try: self._calendar = self._client.calendar(name=self.calendar_name) except caldav.lib.error.NotFoundError: # Créer le calendrier s'il n'existe pas if not self.dry_run: self._calendar = self._client.make_calendar( name=self.calendar_name, supported_calendar_components=["VEVENT"], ) else: logger.warning( f"Calendrier {self.calendar_name} introuvable et dry_run activé. " "Aucune modification ne sera effectuée." ) # Créer un calendrier fictif pour les tests self._calendar = None def _is_managed_event(self, event: DAVEvent) -> bool: """Vérifie si un événement est géré par l'outil.""" # Vérifier la présence du marqueur X-PRONOTE-SYNC-MANAGED props = event.properties managed = props.get(self.MANAGED_PROPERTY, None) return managed and managed.value == self.MANAGED_VALUE def _get_event_uid(self, event: DAVEvent) -> str: """Récupère l'UID normalisé d'un événement.""" uid = event.vobject_instance.uid.value return normalize_pronote_uid(uid) def _build_event( self, lesson: Lesson, calendar_name: Optional[str] = None, ) -> 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) # Nom du calendrier (si disponible) if calendar_name: event.add("x-wr-calname", calendar_name) 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. 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 uid1 = self._get_event_uid(event1) uid2 = self._get_event_uid(event2) if uid1 != uid2: return False # Comparaison des champs gérés vobj1 = event1.vobject_instance vobj2 = event2.vobject_instance # DTSTART et DTEND if vobj1.get("dtstart").value != vobj2.get("dtstart").value: return False if vobj1.get("dtend").value != vobj2.get("dtend").value: 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", []).cats] if hasattr(vobj1.get("categories", None), "cats") else [] cats2 = [str(c) for c in vobj2.get("categories", []).cats] if hasattr(vobj2.get("categories", None), "cats") else [] 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 (`sync/state.py`) Pour éviter de synchroniser à chaque exécution tous les événements depuis le début des temps, on peut stocker un **état local** de la synchronisation. **Protéger les fichiers d'état** : - **Permissions** : Appliquer `chmod 600` sur les fichiers d'état (ex: `.pronote_sync_state.json`) pour limiter l'accès au propriétaire. - **Exclusion Git** : Ajouter les fichiers d'état au `.gitignore` pour éviter de les commiter. - **Exclusion des sauvegardes** : Exclure les fichiers d'état des sauvegardes automatiques (ex: Time Machine, rsync). - **Emplacement** : Stocker les fichiers d'état dans un répertoire dédié (ex: `~/.config/pronote-sync/`) hors de l'arborescence Git. #### 7.3.1 Options pour l'état local | **Option** | **Avantages** | **Inconvénients** | **Recommandation** | |------------------|----------------------------------------|---------------------------------------|-----------------------------| | Fichier JSON | Simple, portable, pas de dépendance | Moins performant pour les gros volumes | ✅ Pour un usage simple | | SQLite | Performant, requêtes complexes | Dépendance supplémentaire | ✅ Pour un usage avancé | | Sync-token CalDAV| Natif, optimisé | Pas toujours supporté par les serveurs | ⚠️ Si disponible | #### 7.3.2 Implémentation avec JSON (`sync/state.py`) ```python import json from datetime import datetime from pathlib import Path from typing import Dict, Optional, Any from ..models.agenda import Lesson, Homework import logging logger = logging.getLogger(__name__) class SyncState: """ Gère l'état de synchronisation local (fichier JSON). Stocke les UID et les timestamps des dernières synchronisations. """ def __init__(self, state_file: str = ".pronote_sync_state.json"): self.state_file = Path(state_file) self._state: Dict[str, Any] = { "last_sync": None, "synced_uids": { "lessons": set(), "homeworks": set(), "school_events": set(), }, "sync_history": [], } self._load() def _load(self) -> None: """Charge l'état depuis le fichier.""" if self.state_file.exists(): try: with open(self.state_file, "r", encoding="utf-8") as f: self._state = json.load(f) # Convertir les sets en sets (JSON les stocke comme des listes) self._state["synced_uids"] = { k: set(v) for k, v in self._state.get("synced_uids", {}).items() } except Exception as e: logger.warning(f"Échec du chargement de l'état: {e}") self._state = { "last_sync": None, "synced_uids": { "lessons": set(), "homeworks": set(), "school_events": set(), }, "sync_history": [], } def _save(self) -> None: """Sauvegarde l'état dans le fichier.""" # Convertir les sets en listes pour JSON state_to_save = { **self._state, "synced_uids": { k: list(v) for k, v in self._state["synced_uids"].items() }, } try: with open(self.state_file, "w", encoding="utf-8") as f: json.dump(state_to_save, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"Échec de la sauvegarde de l'état: {e}") def mark_synced( self, lessons: list[Lesson], homeworks: list[Homework], school_events: list, ) -> None: """Marque les UID comme synchronisés.""" self._state["synced_uids"]["lessons"].update(lesson.id for lesson in lessons) self._state["synced_uids"]["homeworks"].update( f"homework-{hw.id}" for hw in homeworks ) self._state["synced_uids"]["school_events"].update( f"school-event-{se.label}-{se.from_date.isoformat()}" for se in school_events ) self._state["last_sync"] = datetime.now().isoformat() self._state["sync_history"].append({ "timestamp": datetime.now().isoformat(), "lessons": len(lessons), "homeworks": len(homeworks), "school_events": len(school_events), }) self._save() def is_synced(self, uid: str, kind: str = "lessons") -> bool: """Vérifie si un UID a déjà été synchronisé.""" return uid in self._state["synced_uids"].get(kind, set()) def get_last_sync(self) -> Optional[datetime]: """Récupère la date de la dernière synchronisation.""" if self._state["last_sync"]: return datetime.fromisoformat(self._state["last_sync"]) return None def clear(self) -> None: """Efface l'état.""" self._state = { "last_sync": None, "synced_uids": { "lessons": set(), "homeworks": set(), "school_events": set(), }, "sync_history": [], } self._save() ``` ### 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). --- ## 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. - **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. """ ... ``` ### 8.3 Implémentation par fichier (`sources/theoretical/file.py`) #### 8.3.1 Fichier iCal Si l'agenda théorique est fourni sous forme de **fichier iCal** (ex: export depuis un autre outil), on peut le parser de la même manière que le flux Pronote. ```python from typing import List from datetime import date, time from pathlib import Path from icalendar import Calendar, Event from ..models.agenda import TheoreticalLesson from .provider import TheoreticalAgendaProvider class ICalTheoreticalAgendaProvider: """Fournisseur d'agenda théorique depuis un fichier iCal.""" def __init__(self, file_path: str): self.file_path = Path(file_path) self._lessons: List[TheoreticalLesson] = [] self._load() def _load(self) -> None: """Charge le fichier iCal et parse les cours.""" if not self.file_path.exists(): raise FileNotFoundError(f"Fichier iCal introuvable: {self.file_path}") with open(self.file_path, "rb") as f: cal = Calendar.from_ical(f.read()) for component in cal.walk(): if not isinstance(component, Event): continue # Ignorer les événements tout le jour (vacances, etc.) if hasattr(component.get("dtstart"), "dt") and not hasattr(component.get("dtstart").dt, "hour"): continue start = component.get("dtstart").dt end = component.get("dtend").dt # Générer un ID stable (basé sur le jour, l'heure et la matière) summary = str(component.get("summary", "")) uid = f"theoretical-{start.strftime('%Y%m%d')}-{start.hour}{start.minute}-{summary}" lesson = TheoreticalLesson( id=uid, day_of_week=start.weekday(), start_time=time(start.hour, start.minute), end_time=time(end.hour, end.minute), subject=summary, teachers=[], # À extraire de la description si disponible rooms=[], ) self._lessons.append(lesson) def get_lessons(self, date: date) -> List[TheoreticalLesson]: """Récupère les cours pour une date donnée.""" day_of_week = date.weekday() return [ lesson for lesson in self._lessons if lesson.day_of_week == day_of_week ] def get_lessons_for_range( self, start_date: date, end_date: date, ) -> List[TheoreticalLesson]: """Récupère les cours pour une plage de dates.""" from datetime import timedelta result = [] current_date = start_date while current_date <= end_date: result.extend(self.get_lessons(current_date)) current_date += timedelta(days=1) return result #### 8.3.2 Fichier CSV Si l'agenda théorique est fourni sous forme de **fichier CSV**, on peut le parser ainsi : ```csv jour,semaine,heure_debut,heure_fin,matiere,professeur,salle lundi,1,08:00,09:00,Mathématiques,M. Dupont,204 lundi,1,09:00,10:00,Français,Mme Martin,205 ... ``` ```python import csv from typing import List from datetime import date, time from pathlib import Path from ..models.agenda import TheoreticalLesson from .provider import TheoreticalAgendaProvider class CSVTheoreticalAgendaProvider: """Fournisseur d'agenda théorique depuis un fichier CSV.""" def __init__(self, file_path: str): self.file_path = Path(file_path) self._lessons: List[TheoreticalLesson] = [] self._load() def _load(self) -> None: """Charge le fichier CSV et parse les cours.""" if not self.file_path.exists(): raise FileNotFoundError(f"Fichier CSV introuvable: {self.file_path}") with open(self.file_path, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: day_of_week = self._parse_day(row["jour"]) start_time = self._parse_time(row["heure_debut"]) end_time = self._parse_time(row["heure_fin"]) # Générer un ID stable uid = f"theoretical-{day_of_week}-{start_time.isoformat()}-{row['matiere']}" lesson = TheoreticalLesson( id=uid, day_of_week=day_of_week, start_time=start_time, end_time=end_time, subject=row["matiere"], teachers=[row["professeur"]] if row.get("professeur") else [], rooms=[row["salle"]] if row.get("salle") else [], ) self._lessons.append(lesson) def _parse_day(self, day: str) -> int: """Convertit un nom de jour en index (0=lundi, 6=dimanche).""" days = { "lundi": 0, "mardi": 1, "mercredi": 2, "jeudi": 3, "vendredi": 4, "samedi": 5, "dimanche": 6, } return days.get(day.lower(), 0) def _parse_time(self, time_str: str) -> time: """Parse une chaîne de temps (ex: 08:00).""" hour, minute = map(int, time_str.split(":")) return time(hour, minute) def get_lessons(self, date: date) -> List[TheoreticalLesson]: """Récupère les cours pour une date donnée.""" day_of_week = date.weekday() return [ lesson for lesson in self._lessons if lesson.day_of_week == day_of_week ] def get_lessons_for_range( self, start_date: date, end_date: date, ) -> List[TheoreticalLesson]: """Récupère les cours pour une plage de dates.""" from datetime import timedelta result = [] current_date = start_date while current_date <= end_date: result.extend(self.get_lessons(current_date)) current_date += timedelta(days=1) return result ``` ### 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 UID 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_event( real_lesson: PronoteLesson, theoretical_events: list[TheoreticalEvent], tolerance_minutes: int = 15, ) -> TheoreticalEvent | None: """Trouve l'événement théorique correspondant, avec départage déterministe.""" candidates = [ t for t in theoretical_events if abs((t.start - real_lesson.start).total_seconds()) <= tolerance_minutes * 60 and normalize_subject(t.subject) == normalize_subject(real_lesson.subject) and t.start.date() == real_lesson.start.date() ] if not candidates: return None # Tri déterministe par UID stable, puis par créneau candidates.sort(key=lambda t: (t.uid or "", t.start)) 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 - **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. - **Agenda théorique** : Peut être fourni via fichier iCal ou CSV (extensible à d'autres sources). - **Politique de départage** : Tri par identifiant stable (UID), 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, CSV, XML, etc.) │ ├── pronote-4e.ics # Flux iCal Pronote anonymisé (4ème) │ ├── pronote-6e.ics # Flux iCal Pronote anonymisé (6ème) │ ├── theoretical.ics # Agenda théorique iCal │ ├── theoretical.csv # Agenda théorique CSV │ └── 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 Contenu pédagogique : Pratique vocale Pour le 10/09/2026 : Réviser les chansons apprises Donné le 03/09/2026 : 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 Contenu pédagogique : É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 CSV théorique (`tests/fixtures/theoretical.csv`) ```csv jour,semaine,heure_debut,heure_fin,matiere,professeur,salle lundi,1,08:00,09:00,Mathématiques,M. Dupont,204 lundi,1,09:00,10:00,Français,Mme Martin,205 lundi,1,10:00,11:00,Histoire-Géographie,M. Bernard,206 mardi,1,08:00,09:00,Physique-Chimie,Mme Durand,301 ... ``` ### 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="

Exercices 1 à 5 page 42.

", ) @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 Contenu pédagogique : Résoudre des équations du second degré. Pour le 10/09/2026 : 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.ics", ), ) # --- 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 CSVTheoreticalAgendaProvider # 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 = CSVTheoreticalAgendaProvider( 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="

Exercices 1 à 5 page 42.

", ), HomeworkBlock( kind="assigned", date=date(2026, 9, 5), text="Exercices 1 à 5 page 42.", # Même texte que le bloc "Pour le" html="

Exercices 1 à 5 page 42.

", ), ], ) # 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="

Lire les pages 10 à 15.

", ), ], ) # 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.