From aaca78c55d088ced63f8f9a96511c14647dc6275 Mon Sep 17 00:00:00 2001 From: Antoine Van Elstraete Date: Sat, 5 Sep 2026 23:52:02 +0200 Subject: [PATCH] =?UTF-8?q?docs+fix:=20renforcement=20de=20l'architecture?= =?UTF-8?q?=20agentique=20et=20corrections=20de=20s=C3=A9curit=C3=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AGENTS.md : - Rôles d'agents renforcés : @coder ne valide pas, @debugger ne code pas, @verifier ne modifie pas, etc. - Table « Séparation des rôles » : tâche → agent responsable → ne pas confier à - Workflow étape 5 : délégation explicite à @verifier pour la validation GUIDE_DEV_PYTHON.md : - 7 notes « Décision d'implémentation » ajoutées aux sections concernées : 3.1.1 (XMPP_RECIPIENT→XMPP_TO, vars ajoutées), 3.1.2 (SYNC_PAST_DAYS→AppSettings), 3.2 (Pydantic v2 style, defaults corrigés), 4.2.1 (redact_exception module function), 4.2.2 (getLevelNamesMapping), 5.1.5 (normalize_pronote_uid, usedforsecurity=False), 6 (ConfigDict, StrEnum, alias _date, external_info Optional) Sécurité (audit @security-auditor, corrections @coder, validation @verifier) : - RedactingFormatter : redaction APRÈS formatage (corrige TypeError %s + fuite traceback) - redact_url : masquage des credentials dans userinfo URL (HTTP Basic Auth) - redact_secrets : patterns étendus (api_key, access_token, authorization, auth) - redact_secrets : support JSON-style « key: value » avec guillemets Validations (@verifier) : - ruff check : PASS | mypy strict : PASS | bandit : PASS (0 issue) - %s formatting : OK (password=REDACTED, pas de TypeError) - Traceback redaction : OK (icalsecurise=REDACTED) - URL userinfo : OK (user:REDACTED@host) - JSON-style redaction : OK ({"token": "REDACTED"}) - Régression red-to-green : OK (historical HEAD reproduction) Co-authored-by: OpenCode/orchestrator --- .secrets.baseline | 8 +- AGENTS.md | 44 +- GUIDE_DEV_PYTHON.md | 911 +++++++++++++++++--------------- pronote_sync/utils/logging.py | 19 +- pronote_sync/utils/redaction.py | 47 +- 5 files changed, 554 insertions(+), 475 deletions(-) diff --git a/.secrets.baseline b/.secrets.baseline index 81af743..4d590f9 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -90,6 +90,10 @@ { "path": "detect_secrets.filters.allowlist.is_line_allowlisted" }, + { + "path": "detect_secrets.filters.common.is_baseline_file", + "filename": ".secrets.baseline" + }, { "path": "detect_secrets.filters.common.is_ignored_due_to_verification_policies", "min_level": 2 @@ -136,9 +140,9 @@ "filename": "GUIDE_DEV_PYTHON.md", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "is_verified": false, - "line_number": 5073 + "line_number": 5112 } ] }, - "generated_at": "2026-09-05T17:53:24Z" + "generated_at": "2026-09-05T21:51:55Z" } diff --git a/AGENTS.md b/AGENTS.md index 1bcad9c..0c72669 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -189,22 +189,38 @@ Cette section s'applique uniquement lorsque le travail est exécuté avec le sys Les rôles d'agents disponibles pour ce projet sont les suivants : -- `@architect` : Arbitrages d'architecture et choix techniques structurants pour le pipeline `pronote-sync`. -- `@coder` : Opérations de développement et changements de code dans le projet. -- `@debugger` : Reproduction d'un symptôme et établissement de sa cause profonde (ex. : échec de synchronisation, repli iCal/pronotepy). -- `@explorer` : Exploration du dépôt en lecture seule et fourniture de contexte factuel. -- `@orchestrator` : Compréhension globale du projet, définition des jalons, coordination et garantie du résultat. -- `@planner` : Transformation d'une demande complexe en unités exécutables avec frontières et dépendances claires. -- `@reviewer` : Revues indépendantes de correction, régression, contrats et maintenabilité. -- `@security-auditor` : Audit indépendant d'une surface de sécurité désignée (ex. : gestion des secrets, masquage des données). -- `@tech-writer` : Rédaction et maintenance de documentation technique exacte et vérifiable. -- `@test-engineer` : Conception, écriture et exécution de tests ciblés (unitaires, intégration, mocks). -- `@ui-designer` : conception et implémentation d'interfaces Web et terminal. -- `@verifier` : Vérification indépendante du comportement livré, des régressions et du respect des conventions (idempotence, mode dégradé). -- `@web-explorer` : Recherche et extraction de sources Web vérifiables (ex. : documentation Pronote, CalDAV, XMPP). +- `@architect` : Arbitrages d'architecture et choix techniques structurants. **Ne produit pas de code.** +- `@coder` : Écrit et modifie du code, de la configuration et des scripts. **Ne valide pas** (ruff, mypy, pytest) — c'est le rôle de `@verifier`. **Ne diagnostique pas** — c'est le rôle de `@debugger`. +- `@debugger` : Reproduit un symptôme et établit sa cause profonde. **Ne modifie pas le code.** +- `@explorer` : Explore le dépôt en lecture seule. **Ne modifie rien, n'exécute pas de commandes.** +- `@orchestrator` : Compréhension globale, définition des jalons, coordination et garantie du résultat. **N'écrit pas de code.** +- `@planner` : Transforme une demande complexe en unités exécutables. **Ne dirige aucun technicien.** +- `@reviewer` : Revues indépendantes de correction, régression, contrats et maintenabilité. **Ne modifie pas le code.** +- `@security-auditor` : Audit indépendant d'une surface de sécurité. **Ne modifie pas le code.** +- `@tech-writer` : Rédige et maintient la documentation. **N'écrit pas de code applicatif.** +- `@test-engineer` : Conçoit, écrit et exécute des tests ciblés. **N'écrit pas de code de production.** +- `@ui-designer` : Conçoit et implémente les interfaces Web et terminal. +- `@verifier` : Vérifie indépendamment le comportement livré, les régressions et le respect des conventions (ruff, mypy, pytest, bandit, idempotence, mode dégradé). **Ne modifie pas le code.** +- `@web-explorer` : Recherche et extrait des sources Web vérifiables. **Ne modifie pas le dépôt.** > **Note** : Ne pas utiliser `@coder` pour les tâches de documentation (`@tech-writer`) ni pour les tests (`@test-engineer`). +### Séparation des rôles + +| Type de tâche | Agent responsable | Ne pas confier à | +|---|---|---| +| Écrire/modifier du code | `@coder` | `@verifier`, `@explorer` | +| Valider (ruff, mypy, pytest, bandit) | `@verifier` | `@coder` | +| Diagnostiquer un bug | `@debugger` | `@coder` | +| Écrire un test | `@test-engineer` | `@coder` | +| Rédiger de la documentation | `@tech-writer` | `@coder` | +| Explorer le dépôt (lecture) | `@explorer` | `@coder`, `@verifier` | +| Arbitrage technique structurant | `@architect` | `@coder`, `@planner` | +| Revue de code | `@reviewer` | `@coder`, `@verifier` | +| Audit de sécurité | `@security-auditor` | `@coder`, `@verifier` | +| Recherche web | `@web-explorer` | `@explorer` | +| Découpage de travail complexe | `@planner` | `@coder` | + --- ## 10. Workflow de modification @@ -213,7 +229,7 @@ Les rôles d'agents disponibles pour ce projet sont les suivants : 2. Préserver les changements existants de l'utilisateur. 3. Pour une correction, reproduire d'abord le défaut avec un test automatisé lorsque c'est raisonnable. 4. Faire une modification étroite et cohérente, en respectant les conventions du projet (idempotence, mode dégradé, repli iCal/pronotepy). -5. Vérifier le comportement nominal et les cas d'erreur, notamment : +5. Faire vérifier le comportement par `@verifier` (ruff, mypy, pytest, bandit) et les cas d'erreur, notamment : - Succès de la synchronisation Pronote → CalDAV/XMPP. - Repli vers iCal en cas d'échec de `pronotepy`. - Gestion des erreurs explicites. diff --git a/GUIDE_DEV_PYTHON.md b/GUIDE_DEV_PYTHON.md index 662d1f8..9230121 100644 --- a/GUIDE_DEV_PYTHON.md +++ b/GUIDE_DEV_PYTHON.md @@ -271,6 +271,10 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati | `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`. #### 3.1.2 Variables optionnelles @@ -281,6 +285,10 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati | `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_BASE_URL` | URL de base pour l'API IA (ex: OpenAI compatible). | `None` | `str \| None`| @@ -335,6 +343,15 @@ 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, Optional from pydantic import SecretStr, Field @@ -417,7 +434,7 @@ settings = Settings() """Vérifie que les messages d'erreur ne fuient pas de secrets.""" from pronote_sync.utils.redaction import redact_secrets from pronote_sync.sources.pronote.ical import fetch_ical - + # Simuler une URL avec token bad_url = "https://example.com/ical?icalsecurise=SECRET_TOKEN" try: @@ -436,7 +453,7 @@ settings = Settings() """Vérifie que les messages d'erreur ne fuient pas de secrets.""" from pronote_sync.utils.redaction import redact_secrets from pronote_sync.sources.pronote.ical import fetch_ical - + # Simuler une URL avec token bad_url = "https://example.com/ical?icalsecurise=SECRET_TOKEN" try: @@ -447,11 +464,16 @@ settings = Settings() assert "icalsecurise" not in error_msg.lower() ``` -**Application systématique** : +**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. #### 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 @@ -465,14 +487,14 @@ def redact_url(url: str) -> str: 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)) @@ -492,7 +514,7 @@ def redact_secrets(text: str) -> str: 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&]+", @@ -500,12 +522,17 @@ def redact_secrets(text: str) -> str: 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 @@ -515,18 +542,18 @@ 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: @@ -541,18 +568,18 @@ class RedactingFormatter(logging.Formatter): 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) @@ -798,10 +825,10 @@ class BlogRSSClient: def fetch_and_parse(self, known_guids: Optional[set] = None) -> List[BlogArticle]: """ Récupère le flux RSS et parse les nouveaux articles. - + Args: known_guids: Ensemble des GUID déjà connus (pour la déduplication). Si None, retourne tous les articles. - + Returns: Liste des nouveaux articles (triés par date de publication décroissante). """ @@ -822,7 +849,7 @@ class BlogRSSClient: for entry in feed.entries: # Extraire le GUID (utiliser link si guid est vide) guid = getattr(entry, "guid", None) or entry.link - + # Ignorer les articles déjà connus if known_guids and guid in known_guids: continue @@ -874,16 +901,16 @@ class BlogRSSClient: def _parse_date(self, date_tuple: Optional[tuple]) -> datetime: """ 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)). - + Returns: datetime en UTC. """ if not date_tuple: return datetime.now(timezone.utc) - + # feedparser retourne un tuple struct_time (année, mois, jour, heure, minute, seconde, jour_semaine, jour_année, DST) try: dt = datetime( @@ -902,27 +929,27 @@ class BlogRSSClient: def _html_to_text(self, 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 import re text = re.sub(r"\s+", " ", text).strip() - + return text @@ -1053,25 +1080,25 @@ def fetch_blog_step( ) -> 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. enabled: Si False, retourne une liste vide. - + Returns: Liste des nouveaux articles. """ if not enabled: return [] - + last_guid = blog_state.get_last_guid() articles = rss_client.fetch_and_parse(last_guid=last_guid) - + # Mettre à jour l'état si des articles sont trouvés if articles: blog_state.update_last_guid(articles[0].id) - + return articles ``` @@ -1084,7 +1111,7 @@ L'étape de récupération du blog est insérée **après la récupération Pron 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( @@ -1098,10 +1125,10 @@ def run(self) -> Tuple[Optional[PronoteData], List[PipelineError]]: 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) ... ``` @@ -1219,9 +1246,9 @@ def mock_blog_rss_client(): def test_parse_blog_rss(mock_blog_rss_client): """Test le parsing d'un flux RSS du blog.""" articles = mock_blog_rss_client.fetch_and_parse() - + assert len(articles) == 2 - + # Vérifier le premier article article1 = articles[0] assert article1.title == "Sortie pédagogique" @@ -1230,7 +1257,7 @@ def test_parse_blog_rss(mock_blog_rss_client): 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 = articles[1] assert article2.title == "Réunion de rentrée" @@ -1245,22 +1272,22 @@ def test_parse_blog_rss(mock_blog_rss_client): def test_blog_deduplication(tmp_path): """Test la déduplication des articles du blog.""" from pronote_sync.sync.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_last_guid() is None - + # Simuler la récupération de 2 articles state.update_last_guid("https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625") assert state.get_last_guid() == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625" - + # Simuler une nouvelle récupération : seul le nouvel article doit être retourné client = BlogRSSClient(rss_url="file://tests/fixtures/blog_rss.xml") new_articles = client.fetch_and_parse(last_guid=state.get_last_guid()) - + # Seul l'article avec p=1626 doit être retourné (car p=1625 est déjà connu) assert len(new_articles) == 1 assert new_articles[0].id == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626" @@ -1276,25 +1303,25 @@ Les articles du blog sont intégrés dans la section **"Informations diverses"** # 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) ``` @@ -1313,7 +1340,7 @@ def _format_message(self, message: XmppMessage) -> str: | **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 @@ -1338,14 +1365,14 @@ Les flux iCal générés par Pronote ont des **spécificités importantes** à p Professeur : M. Dupont Salle : 204 Groupe : Classe entière - - Contenu pédagogique : + + Contenu pédagogique : Résoudre des équations du second degré. - Pour le 10/09/2026 : + Pour le 10/09/2026 : Exercices 1 à 5 page 42. - Donné le 05/09/2026 : + Donné le 05/09/2026 : Exercices 1 à 5 page 42. @@ -1410,12 +1437,12 @@ def resolve_target_day( ) -> 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". @@ -1423,15 +1450,15 @@ def resolve_target_day( - 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) @@ -1440,7 +1467,7 @@ def resolve_target_day( 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 @@ -1450,7 +1477,7 @@ def resolve_target_day( holiday_label = event.label resume_date = event.to_date break - + return (tomorrow, "no-school", resume_date, holiday_label) ``` @@ -1478,14 +1505,14 @@ 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. @@ -1494,7 +1521,7 @@ def fetch_ical(url: str, timeout: int = 20) -> str: "accept": "text/calendar, */*;q=0.5", "user-agent": "pronote-sync", } - + # Gestion des URLs file:// pour les tests if url.startswith("file://"): import pathlib @@ -1503,7 +1530,7 @@ def fetch_ical(url: str, timeout: int = 20) -> str: 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( @@ -1520,23 +1547,23 @@ def fetch_ical(url: str, timeout: int = 20) -> str: raise requests.exceptions.RequestException( f"Échec de la récupération de {safe_url}: {safe_error}" ) from e - + # 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. """ @@ -1577,10 +1604,10 @@ 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). """ @@ -1596,11 +1623,11 @@ 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). """ @@ -1613,11 +1640,11 @@ def collect_homeworks(lessons: List[Lesson], target_date: date) -> List[Homework """ 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. """ @@ -1684,12 +1711,16 @@ 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** : +**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). @@ -1704,10 +1735,10 @@ 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. """ @@ -1727,7 +1758,7 @@ def generate_deterministic_uid( ) -> 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. @@ -1735,7 +1766,7 @@ def generate_deterministic_uid( teachers: Liste des professeurs. rooms: Liste des salles. group: Groupe (optionnel). - + Returns: UID déterministe (hash SHA-1 des champs clés). """ @@ -1765,10 +1796,10 @@ 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). """ @@ -1776,7 +1807,7 @@ def split_header_and_body(description: str) -> Tuple[str, str]: strong_start = description.find("") if strong_start == -1: return description, "" - + header = description[:strong_start].strip() body = description[strong_start:] return header, body @@ -1786,37 +1817,37 @@ 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"): @@ -1827,26 +1858,26 @@ def parse_header(header: str) -> dict: 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(.+?)(?:|$)", @@ -1858,7 +1889,7 @@ def parse_body(body: str) -> Tuple[Optional[str], List[dict]]: # 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( @@ -1866,165 +1897,165 @@ def parse_body(body: str) -> Tuple[Optional[str], List[dict]]: 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** : + + **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( @@ -2034,29 +2065,29 @@ def parse_ical(raw_ical: str) -> tuple[List[Lesson], List[HomeworkModel], List[S to_date=component.get("dtend").dt, )) continue - + # Cours annulés ou déplacés status = LessonStatus.NORMAL if "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, @@ -2070,7 +2101,7 @@ def parse_ical(raw_ical: str) -> tuple[List[Lesson], List[HomeworkModel], List[S homework_blocks=raw_homeworks, # Stockage temporaire pour déduplication globale ) lessons.append(lesson) - + return lessons, homeworks, school_events ``` @@ -2093,7 +2124,7 @@ logger = logging.getLogger(__name__) class PronoteClient: """Client pour interagir avec Pronote via pronotepy.""" - + def __init__( self, username: Optional[str] = None, @@ -2106,26 +2137,26 @@ class PronoteClient: self.ent = ent self.ical_url = ical_url self._client: Optional[Client] = None - + def _get_client(self) -> Client: """Initialise et retourne le client pronotepy.""" if self._client is None: if not all([self.username, self.password, self.ent]): raise ValueError("Username, password et ENT sont requis pour pronotepy") - + self._client = Client( self.username, self.password, self.ent, ) return self._client - + def get_messages(self) -> List[Message]: """Récupère les messages des professeurs.""" try: client = self._get_client() discussions = client.get_discussions() - + messages = [] for discussion in discussions: for message in discussion.messages: @@ -2142,13 +2173,13 @@ class PronoteClient: except PronoteAPIError as e: logger.error(f"Échec de la récupération des messages Pronote: {redact_secrets(str(e))}") return [] - + def get_informations(self) -> List[Message]: """Récupère les informations et sondages.""" try: client = self._get_client() informations = client.get_information_and_surveys() - + messages = [] for info in informations: messages.append(Message( @@ -2164,7 +2195,7 @@ class PronoteClient: except PronoteAPIError as e: logger.error(f"Échec de la récupération des informations Pronote: {redact_secrets(str(e))}") return [] - + def get_agenda_fallback(self) -> tuple[List[Lesson], List[HomeworkModel]]: """ Récupère l'agenda et les devoirs via pronotepy (repli si iCal échoue). @@ -2172,7 +2203,7 @@ class PronoteClient: """ try: client = self._get_client() - + lessons = [] for lesson in client.get_lessons(): lessons.append(Lesson( @@ -2185,7 +2216,7 @@ class PronoteClient: status=LessonStatus.NORMAL, # À adapter selon les données content=lesson.content, )) - + homeworks = [] for hw in client.get_homework(): homeworks.append(HomeworkModel( @@ -2197,12 +2228,12 @@ class PronoteClient: text=hw.description, html=hw.description, # pronotepy ne fournit pas de HTML )) - + return lessons, homeworks except PronoteAPIError as e: logger.error(f"Échec de la récupération de l'agenda via pronotepy: {redact_secrets(str(e))}") return [], [] - + def close(self) -> None: """Fermeture du client.""" if self._client: @@ -2228,7 +2259,7 @@ class AgendaSource(Enum): class PronoteFetcher: """Gère la récupération des données Pronote avec repli.""" - + def __init__( self, ical_url: Optional[str] = None, @@ -2245,7 +2276,7 @@ class PronoteFetcher: self.agenda_source = AgendaSource(agenda_source) self.homework_source = AgendaSource(homework_source) self._pronote_client: Optional[PronoteClient] = None - + def _get_pronote_client(self) -> PronoteClient: if self._pronote_client is None: self._pronote_client = PronoteClient( @@ -2255,7 +2286,7 @@ class PronoteFetcher: ical_url=self.ical_url, ) return self._pronote_client - + def fetch_agenda(self) -> tuple[List[Lesson], List[Homework]]: """Récupère l'agenda selon la source configurée (`agenda_source`).""" if self.agenda_source == AgendaSource.ICAL: @@ -2270,7 +2301,7 @@ class PronoteFetcher: return lessons, homeworks except Exception as e: logger.warning(f"Échec de la récupération iCal pour l'agenda: {redact_secrets(str(e))}") - + # Repli sur pronotepy logger.info("Repli sur pronotepy pour l'agenda.") return self._fetch_agenda_pronotepy() @@ -2301,7 +2332,7 @@ class PronoteFetcher: return homeworks except Exception as e: logger.warning(f"Échec de la récupération iCal pour les devoirs: {redact_secrets(str(e))}") - + # Repli sur pronotepy logger.info("Repli sur pronotepy pour les devoirs.") try: @@ -2310,32 +2341,32 @@ class PronoteFetcher: except Exception as e: logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}") return [] - + def _fetch_agenda_ical(self) -> tuple[List[Lesson], List[Homework]]: """Récupère l'agenda depuis iCal.""" if not self.ical_url: raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal") - + raw_ical = fetch_ical(self.ical_url) lessons, homeworks, _ = parse_ical(raw_ical) return lessons, homeworks - + def _fetch_agenda_pronotepy(self) -> tuple[List[Lesson], List[Homework]]: """Récupère l'agenda depuis pronotepy.""" client = self._get_pronote_client() lessons, homeworks = client.get_agenda_fallback() return lessons, homeworks - + def fetch_messages(self) -> List[Message]: """Récupère les messages (toujours via pronotepy).""" client = self._get_pronote_client() return client.get_messages() - + def fetch_informations(self) -> List[Message]: """Récupère les informations (toujours via pronotepy).""" client = self._get_pronote_client() return client.get_informations() - + def close(self) -> None: """Fermeture des ressources.""" if self._pronote_client: @@ -2360,6 +2391,14 @@ class PronoteFetcher: ## 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. @@ -2444,7 +2483,7 @@ class SchoolEvent(BaseModel): 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 @@ -2461,7 +2500,7 @@ class TheoreticalLesson(BaseModel): 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 @@ -2480,7 +2519,7 @@ class Homework(BaseModel): 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 @@ -2505,7 +2544,7 @@ class Message(BaseModel): 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 @@ -2529,7 +2568,7 @@ class AgendaChange(BaseModel): None, description="Cours théorique concerné (pour REMOVED/MODIFIED)" ) details: str = Field(default="", description="Détails du changement") - + class Config: frozen = True @@ -2540,7 +2579,7 @@ class AgendaDiff(BaseModel): """ 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 @@ -2554,23 +2593,23 @@ class XmppMessage(BaseModel): """ target_date: date = Field(..., description="Date cible") synthesis: Optional[str] = Field( - None, + None, description="Synthèse IA (optionnelle). 3-5 phrases, ton chaleureux et sobre." ) homeworks: List[Homework] = Field( - default_factory=list, + default_factory=list, description="Liste **brute** des devoirs (non modifiée par l'IA)" ) changes: List[AgendaChange] = Field( - default_factory=list, + default_factory=list, description="Liste des changements d'agenda" ) messages: List[Message] = Field( - default_factory=list, + default_factory=list, description="Liste des messages/informations importants" ) external_info: ExternalInfo = Field( - default_factory=ExternalInfo, + default_factory=ExternalInfo, description="Informations externes (blog, messages Pronote)" ) @@ -2597,11 +2636,11 @@ class PronoteData(BaseModel): 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, + default_factory=list, description="Liste des événements scolaires" ) messages: List[Message] = Field( - default_factory=list, + default_factory=list, description="Liste des messages/informations" ) target_date: date = Field(..., description="Date cible (J+1 ou prochain jour scolaire)") @@ -2710,11 +2749,11 @@ 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, @@ -2730,7 +2769,7 @@ class CalDAVClient: 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( @@ -2738,7 +2777,7 @@ class CalDAVClient: username=self.username, password=self.password, ) - + # Récupérer ou créer le calendrier try: self._calendar = self._client.calendar(name=self.calendar_name) @@ -2756,19 +2795,19 @@ class CalDAVClient: ) # 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, @@ -2776,13 +2815,13 @@ class CalDAVClient: ) -> 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 "" @@ -2794,7 +2833,7 @@ class CalDAVClient: if lesson.content: description += f"\nContenu : {lesson.content}" event.add("description", vText(description)) - + # Statut if lesson.status == LessonStatus.CANCELLED: event.add("status", "CANCELLED") @@ -2802,10 +2841,10 @@ class CalDAVClient: 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: @@ -2813,72 +2852,72 @@ class CalDAVClient: 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. """ @@ -2887,39 +2926,39 @@ class CalDAVClient: 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( @@ -2933,14 +2972,14 @@ class CalDAVClient: """ 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. """ @@ -2949,14 +2988,14 @@ class CalDAVClient: 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( @@ -2973,29 +3012,29 @@ class CalDAVClient: 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: @@ -3014,7 +3053,7 @@ class CalDAVClient: else: # Ajouter un nouvel événement new_event = self._build_event(lesson) - + if not self.dry_run: try: self._calendar.add_event(new_event) @@ -3026,18 +3065,18 @@ class CalDAVClient: 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: @@ -3055,7 +3094,7 @@ class CalDAVClient: else: # Ajouter new_event = self._build_homework_event(homework) - + if not self.dry_run: try: self._calendar.add_event(new_event) @@ -3067,18 +3106,18 @@ class CalDAVClient: 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: @@ -3096,7 +3135,7 @@ class CalDAVClient: else: # Ajouter new_event = self._build_school_event_event(school_event) - + if not self.dry_run: try: self._calendar.add_event(new_event) @@ -3108,7 +3147,7 @@ class CalDAVClient: 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. @@ -3121,7 +3160,7 @@ class CalDAVClient: } | { 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: @@ -3135,7 +3174,7 @@ class CalDAVClient: else: result.removed += 1 logger.info(f"[DRY-RUN] Suppression de {uid}") - + # Définir le statut final if result.errors: result.status = CalDAVSyncStatus.FAILED @@ -3143,7 +3182,7 @@ class CalDAVClient: result.status = CalDAVSyncStatus.SKIPPED return result - + def close(self) -> None: """Fermeture de la connexion.""" self._client = None @@ -3185,7 +3224,7 @@ 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] = { @@ -3198,7 +3237,7 @@ class SyncState: "sync_history": [], } self._load() - + def _load(self) -> None: """Charge l'état depuis le fichier.""" if self.state_file.exists(): @@ -3220,7 +3259,7 @@ class SyncState: }, "sync_history": [], } - + def _save(self) -> None: """Sauvegarde l'état dans le fichier.""" # Convertir les sets en listes pour JSON @@ -3235,7 +3274,7 @@ class SyncState: 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], @@ -3258,17 +3297,17 @@ class SyncState: "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 = { @@ -3313,19 +3352,19 @@ 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, @@ -3333,11 +3372,11 @@ class TheoreticalAgendaProvider(Protocol): ) -> 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. """ @@ -3362,35 +3401,35 @@ 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(), @@ -3401,7 +3440,7 @@ class ICalTheoreticalAgendaProvider: 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() @@ -3409,7 +3448,7 @@ class ICalTheoreticalAgendaProvider: lesson for lesson in self._lessons if lesson.day_of_week == day_of_week ] - + def get_lessons_for_range( self, start_date: date, @@ -3417,7 +3456,7 @@ class ICalTheoreticalAgendaProvider: ) -> 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: @@ -3448,27 +3487,27 @@ 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, @@ -3479,7 +3518,7 @@ class CSVTheoreticalAgendaProvider: 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 = { @@ -3492,12 +3531,12 @@ class CSVTheoreticalAgendaProvider: "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() @@ -3505,7 +3544,7 @@ class CSVTheoreticalAgendaProvider: lesson for lesson in self._lessons if lesson.day_of_week == day_of_week ] - + def get_lessons_for_range( self, start_date: date, @@ -3513,7 +3552,7 @@ class CSVTheoreticalAgendaProvider: ) -> 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: @@ -3582,13 +3621,13 @@ 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 @@ -3596,12 +3635,12 @@ class AgendaComparator: 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, @@ -3609,14 +3648,14 @@ class AgendaComparator: ) -> 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. @@ -3625,17 +3664,17 @@ class AgendaComparator: 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 @@ -3645,33 +3684,33 @@ class AgendaComparator: 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( @@ -3694,7 +3733,7 @@ class AgendaComparator: 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 @@ -3702,7 +3741,7 @@ class AgendaComparator: self._match_lesson(real, [theoretical]) is not None for real in real_lessons ) - + if not has_match: changes.append(AgendaChange( type=AgendaChangeType.REMOVED, @@ -3710,9 +3749,9 @@ class AgendaComparator: 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, @@ -3720,23 +3759,23 @@ class AgendaComparator: ) -> 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, @@ -3745,12 +3784,12 @@ class AgendaComparator: ) -> 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. """ @@ -3808,10 +3847,10 @@ class SynthesisProvider(Protocol): 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). @@ -3914,10 +3953,10 @@ Exemple de format attendu : 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, @@ -3928,12 +3967,12 @@ Exemple de format attendu : "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", @@ -3941,19 +3980,19 @@ Exemple de format attendu : 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}") @@ -3985,11 +4024,11 @@ 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", @@ -3999,13 +4038,13 @@ class LiteLLMSynthesisProvider: 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 @@ -4016,7 +4055,7 @@ class LiteLLMSynthesisProvider: """Génère une synthèse IA via litellm.""" try: user_prompt = self._build_prompt(input_data) - + response = litellm.completion( model=self.model, messages=[ @@ -4026,16 +4065,16 @@ class LiteLLMSynthesisProvider: 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}") @@ -4056,21 +4095,21 @@ 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( @@ -4078,7 +4117,7 @@ def get_synthesis_provider(settings: AISettings, provider: Optional[str] = None) 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", @@ -4227,10 +4266,10 @@ class Channel(Protocol): 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. """ @@ -4258,9 +4297,9 @@ class XmppChannel(Channel): Canal XMPP pour l'envoi des messages. Utilise slixmpp en mode asynchrone. """ - + name = "xmpp" - + def __init__( self, jid: str, @@ -4275,71 +4314,71 @@ class XmppChannel(Channel): 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 :") @@ -4351,7 +4390,7 @@ class XmppChannel(Channel): elif change.type == "modified": lines.append(f" ~ {change.lesson.subject} ({change.details})") lines.append("") - + # Liste brute des devoirs if message.homeworks: lines.append("📚 Devoirs :") @@ -4359,46 +4398,46 @@ class XmppChannel(Channel): 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: @@ -4410,8 +4449,8 @@ 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é + + **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. """ @@ -4431,7 +4470,7 @@ class SyncXmppChannel: 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: @@ -4461,11 +4500,11 @@ _CHANNEL_FACTORIES: Dict[str, Type[Channel]] = { 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é. """ @@ -4527,7 +4566,7 @@ class ErrorSeverity(Enum): class PipelineError(Exception): """Erreur dans le pipeline.""" - + def __init__( self, message: str, @@ -4544,14 +4583,14 @@ class PipelineError(Exception): 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) ``` @@ -4624,7 +4663,7 @@ class PipelineRunner: def run(self) -> Tuple[Optional[PronoteData], List[PipelineError]]: """ Exécute le pipeline complet. - + Returns: Tuple (PronoteData final, liste des erreurs). """ @@ -4633,7 +4672,7 @@ class PipelineRunner: sync_result: Optional[CalDAVSyncResult] = None synthesis_result: Optional[SynthesisResult] = None blog_articles: List["BlogArticle"] = [] - + try: # Étape 1: Récupération Pronote try: @@ -4646,7 +4685,7 @@ class PipelineRunner: 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) @@ -4654,7 +4693,7 @@ class PipelineRunner: 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: @@ -4670,7 +4709,7 @@ class PipelineRunner: )) 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( @@ -4684,7 +4723,7 @@ class PipelineRunner: step="compare", )) logger.warning(f"Étape 'compare' échouée (non bloquante): {e.message}") - + # Étape 4: Synchronisation CalDAV try: sync_result = caldav_sync_step( @@ -4705,7 +4744,7 @@ class PipelineRunner: 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: @@ -4722,7 +4761,7 @@ class PipelineRunner: 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, @@ -4735,7 +4774,7 @@ class PipelineRunner: 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) @@ -4745,9 +4784,9 @@ class PipelineRunner: 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] @@ -4759,11 +4798,11 @@ class PipelineRunner: 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 @@ -4788,18 +4827,18 @@ from .errors import PipelineError, ErrorSeverity, PipelineCriticalError def fetch_step(fetcher: PronoteFetcher) -> Tuple[List[Lesson], List[Homework], List[SchoolEvent], List[Message]]: """ Étape de récupération des données Pronote. - + **Logique de precedence** : - Si `agenda_source=ical` et `homework_source=ical`, un seul fetch iCal suffit (les devoirs sont extraits du même flux). - Si `agenda_source=ical` et `homework_source=pronotepy`, deux sources distinctes sont utilisées. - La déduplication globale est effectuée après fusion des résultats. - + Args: fetcher: Fetcher Pronote configuré. - + Returns: Tuple (lessons, homeworks, school_events, messages). - + Raises: PipelineCriticalError: Si aucune source n'est disponible. PipelineError: Si une source échoue mais qu'une autre est disponible. @@ -4808,7 +4847,7 @@ def fetch_step(fetcher: PronoteFetcher) -> Tuple[List[Lesson], List[Homework], L # Récupérer l'agenda (cours + événements scolaires) lessons, agenda_homeworks = fetcher.fetch_agenda() school_events = [] # À récupérer depuis iCal ou autre source - + # Récupérer les devoirs selon la source configurée # Si la source est iCal et que l'agenda a déjà été récupéré depuis iCal, # les devoirs sont déjà inclus dans agenda_homeworks (via parsing iCal). @@ -4823,20 +4862,20 @@ def fetch_step(fetcher: PronoteFetcher) -> Tuple[List[Lesson], List[Homework], L else: # Utiliser les devoirs déjà extraits de l'agenda iCal homeworks = agenda_homeworks - + # Récupérer les messages et informations (toujours via pronotepy) messages = fetcher.fetch_messages() informations = fetcher.fetch_informations() messages.extend(informations) - + if not lessons and not homeworks: raise PipelineCriticalError( message="Aucun cours ou devoir récupéré depuis Pronote", step="fetch", ) - + return lessons, homeworks, school_events, messages - + except Exception as e: raise PipelineError( message=f"Échec de la récupération Pronote: {redact_secrets(str(e))}", @@ -4864,31 +4903,31 @@ def fetch_blog_step( ) -> List[BlogArticle]: """ Étape de récupération des articles du blog du collège. - + Args: rss_client: Client RSS configuré. blog_state: État local pour la déduplication. enabled: Si False, retourne une liste vide. - + Returns: Liste des nouveaux articles. - + Raises: PipelineError: Si la récupération échoue (non bloquante pour le pipeline). """ if not enabled: return [] - + try: last_guid = blog_state.get_last_guid() articles = rss_client.fetch_and_parse(last_guid=last_guid) - + # Mettre à jour l'état si des articles sont trouvés if articles: blog_state.update_last_guid(articles[0].id) - + return articles - + except Exception as e: raise PipelineError( message=f"Échec de la récupération du blog: {e}", @@ -4971,11 +5010,11 @@ CATEGORIES:Cours DESCRIPTION:EDUCATION MUSICALE Professeur : DURAND G. Salle : S002 Education musicale -Contenu pédagogique : +Contenu pédagogique : Pratique vocale -Pour le 10/09/2026 : +Pour le 10/09/2026 : Réviser les chansons apprises -Donné le 03/09/2026 : +Donné le 03/09/2026 : Apporter le cahier de chants END:VEVENT BEGIN:VEVENT @@ -4989,7 +5028,7 @@ DESCRIPTION:Français Professeur : Mme Martin Salle : 205 Groupe : Classe entière -Contenu pédagogique : +Contenu pédagogique : Étude d'un texte littéraire. END:VEVENT STATUS:CANCELLED @@ -5134,9 +5173,9 @@ CATEGORIES:Cours DESCRIPTION:Matière : Mathématiques Professeur : M. Dupont Salle : 204 -Contenu pédagogique : +Contenu pédagogique : Résoudre des équations du second degré. -Pour le 10/09/2026 : +Pour le 10/09/2026 : Exercices 1 à 5 page 42. END:VEVENT END:VCALENDAR""" @@ -5154,7 +5193,7 @@ def mock_pronotepy_lessons(): self.teachers = teachers self.rooms = rooms self.content = content - + return [ MockLesson( id=12345, @@ -5174,7 +5213,7 @@ def mock_pronotepy_lessons(): 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", @@ -5191,7 +5230,7 @@ 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 @@ -5202,7 +5241,7 @@ 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 @@ -5215,7 +5254,7 @@ 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 @@ -5229,7 +5268,7 @@ 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( ical_url="https://test.ent/pronote/ical/test.ics", @@ -5272,10 +5311,10 @@ def sample_settings(): 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"] @@ -5288,10 +5327,10 @@ def test_parse_ical_lesson(parsed_lessons): def test_parse_ical_homework(parsed_lessons): """Test le parsing des devoirs depuis iCal.""" lessons, homeworks, school_events = parsed_lessons - + 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 @@ -5307,7 +5346,7 @@ def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, from pronote_sync.sync.caldav import CalDAVClient from pronote_sync.sync.diff import AgendaComparator from pronote_sync.sources.theoretical.file import CSVTheoreticalAgendaProvider - + # Configurer le fetcher Pronote fetcher = PronoteFetcher( ical_url=sample_settings.pronote.ical_url, @@ -5317,7 +5356,7 @@ def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, agenda_source=sample_settings.pronote.agenda_source, homework_source=sample_settings.pronote.homework_source, ) - + # Configurer le client CalDAV caldav_client = CalDAVClient( url=sample_settings.caldav.url, @@ -5325,13 +5364,13 @@ def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, 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, @@ -5341,10 +5380,10 @@ def test_pipeline_full(mock_requests_get, mock_caldav_client, 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 @@ -5358,13 +5397,13 @@ def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, 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 @@ -5379,7 +5418,7 @@ def test_collect_homeworks(): from pronote_sync.models.agenda import Lesson, LessonStatus from pronote_sync.models.homework 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( @@ -5406,7 +5445,7 @@ def test_collect_homeworks(): ), ], ) - + # Cours 2 : contient un bloc "Donné le" pour un autre devoir lesson2 = Lesson( id="lesson-2", @@ -5426,13 +5465,13 @@ def test_collect_homeworks(): ), ], ) - + # 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) @@ -5575,7 +5614,7 @@ 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. @@ -5587,18 +5626,18 @@ def run_command(cmd: list, description: str) -> bool: 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 @@ -5611,18 +5650,18 @@ def check_secrets_in_code(): (["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) @@ -5634,12 +5673,12 @@ def check_secrets_in_git(): ["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 @@ -5649,7 +5688,7 @@ def check_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": @@ -5657,7 +5696,7 @@ def check_fixtures(): if "icalsecurise=" in content: print(f"❌ Token trouvé dans {fixture_file}") return False - + print("✅ Fixtures OK") return True @@ -5665,27 +5704,27 @@ def check_fixtures(): 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 diff --git a/pronote_sync/utils/logging.py b/pronote_sync/utils/logging.py index 5ffbad3..04939a5 100644 --- a/pronote_sync/utils/logging.py +++ b/pronote_sync/utils/logging.py @@ -22,26 +22,15 @@ class RedactingFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: """Formate un enregistrement de log en masquant les secrets. - Le message et chaque argument textuel de l'enregistrement sont rédigés - avant le formatage final effectué par :class:`logging.Formatter`. + La rédaction est appliquée à la chaîne finale (message, arguments et + traceback inclus) produite par :class:`logging.Formatter`. :param record: Enregistrement de log à formater. :return: Message formaté, avec les secrets remplacés par ``REDACTED``. :rtype: str """ - record.msg = redact_secrets(str(record.msg)) - args = record.args - if args: - if isinstance(args, tuple): - record.args = tuple( - redact_secrets(arg) if isinstance(arg, str) else arg for arg in args - ) - else: - record.args = { - key: redact_secrets(value) if isinstance(value, str) else value - for key, value in args.items() - } - return super().format(record) + formatted = super().format(record) + return redact_secrets(formatted) def setup_logging(level: str = "INFO") -> None: diff --git a/pronote_sync/utils/redaction.py b/pronote_sync/utils/redaction.py index 21724d0..ff920a5 100644 --- a/pronote_sync/utils/redaction.py +++ b/pronote_sync/utils/redaction.py @@ -10,10 +10,26 @@ from __future__ import annotations import re from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit -_SENSITIVE_QUERY_KEYS = frozenset({"icalsecurise", "token", "key", "password", "secret"}) +_SENSITIVE_QUERY_KEYS = frozenset( + { + "icalsecurise", + "token", + "key", + "password", + "secret", + "api_key", + "apikey", + "access_token", + "auth", + "authorization", + } +) _URL_PATTERN = re.compile(r"https?://[^\s]+") _ISOLATED_SECRET_PATTERN = re.compile( - r"\b(icalsecurise|token|password|secret|key)\s*=\s*[^\s&]+", + r"\b(icalsecurise|access_token|api_key|apikey|authorization|token|password|secret|key|auth)" + r"(\s*['\"]?\s*[:=]\s*)" + r"(['\"]?)" + r"([^\s&'\"]+)", re.IGNORECASE, ) _REDACTED = "REDACTED" @@ -21,15 +37,30 @@ _REDACTED_URL = "REDACTED_URL" def redact_url(url: str) -> str: - """Masque les paramètres sensibles dans une URL. + """Masque les identifiants et les paramètres sensibles d'une URL. - :param url: URL pouvant contenir des paramètres sensibles (ex: ``icalsecurise``). - :return: URL avec les paramètres sensibles remplacés par ``REDACTED``, + Les informations d'authentification du netloc (``utilisateur:motdepasse@hôte``) + sont masquées, ainsi que les paramètres sensibles de la requête + (ex: ``icalsecurise``). + + :param url: URL pouvant contenir des informations sensibles (ex: ``icalsecurise``). + :return: URL avec les éléments sensibles remplacés par ``REDACTED``, ou ``REDACTED_URL`` si le traitement échoue. :rtype: str """ try: parts = urlsplit(url) + if parts.username is not None or parts.password is not None: + # Netloc sûr : utilisateur:REDACTED@hôte:port. Le deux-point est + # conservé même en l'absence de mot de passe explicite. + userinfo = parts.username or "" + userinfo += ":REDACTED" + host = parts.hostname or "" + if parts.port is not None: + netloc = f"{userinfo}@{host}:{parts.port}" + else: + netloc = f"{userinfo}@{host}" + parts = parts._replace(netloc=netloc) query: list[tuple[str, str]] = parse_qsl(parts.query, keep_blank_values=True) redacted_query = [ (key, _REDACTED if key.lower() in _SENSITIVE_QUERY_KEYS else value) @@ -44,15 +75,15 @@ def redact_secrets(text: str) -> str: """Masque les secrets présents dans un texte arbitraire. Les URLs sont d'abord traitées par :func:`redact_url`, puis les affectations - isolées de type ``cle=valeur`` (ex: ``icalsecurise=XXX``) sont masquées, - sans distinction de casse. + isolées de type ``cle=valeur`` ou ``cle:valeur`` (ex: ``icalsecurise=XXX``, + ``"token": "XXX"``) sont masquées, sans distinction de casse. :param text: Texte pouvant contenir des URLs ou des secrets en clair. :return: Texte avec les secrets remplacés par ``REDACTED``. :rtype: str """ redacted = _URL_PATTERN.sub(lambda match: redact_url(match.group(0)), text) - return _ISOLATED_SECRET_PATTERN.sub(r"\1=REDACTED", redacted) + return _ISOLATED_SECRET_PATTERN.sub(r"\1\2\3REDACTED", redacted) def redact_exception(exc: Exception) -> str: