"""Client d'accès à Pronote via ``pronotepy``. Ce module fournit l'encapsulation du client ``pronotepy`` pour la source Pronote : récupération des messages des professeurs, des informations et sondages, ainsi que des cours et devoirs en mode repli lorsque le flux iCal échoue. Les erreurs des méthodes dégradées (messages, informations) sont journalisées avec des secrets masqués ; les erreurs de récupération des cours et des devoirs se propagent pour déclencher le repli iCal. """ from __future__ import annotations import json import logging from datetime import date from pathlib import Path from typing import Any, Protocol from uuid import uuid4 import pronotepy import pronotepy.ent as pronotepy_ent import requests from pronote_sync.config.settings import PronoteSettings from pronote_sync.errors import PronoteAuthRotationError from pronote_sync.models.agenda import Lesson, LessonStatus from pronote_sync.models.homework import Homework from pronote_sync.models.message import Message, MessageType from pronote_sync.sources.pronote.auth_state import PronoteAuthState from pronote_sync.utils.redaction import redact_exception, redact_secrets from pronote_sync.utils.uid import generate_deterministic_uid, normalize_pronote_uid logger = logging.getLogger(__name__) def _get_ent_callable(name: str) -> Any: """Retourne le callable ``pronotepy`` associé à un nom d'ENT. L'accès par :func:`getattr` évite les erreurs ``attr-defined`` de mypy sur les attributs non exportés explicitement par ``pronotepy.ent``. :param name: Nom de l'attribut dans ``pronotepy.ent``. :return: Callable ``pronotepy`` associé. :rtype: Any """ return getattr(pronotepy_ent, name) _ENT_NAMES: list[str] = [ "monbureaunumerique", "ent_elyco", "bordeaux", "ent_creuse", "occitanie_montpellier", "paris_classe_numerique", "ile_de_france", "ent_hdf", "ac_orleans_tours", "ac_poitiers", "ac_rennes", "laclasse_educonnect", "ent77", "ent_ecollege78", "ent_essonne", "val_doise", "val_de_marne", "ent_var", "atrium_sud", "laclasse_lyon", "eclat_bfc", "cas_arsene76", "cas_ent27", "cas_kosmos", "ent_creuse_educonnect", "ent_mayotte", "ent_somme", "ent_94", "extranet_colleges_somme", "ac_reunion", ] _ENT_RESOLVERS: dict[str, Any] = {name: _get_ent_callable(name) for name in _ENT_NAMES} def _resolve_ent(ent_name: str) -> Any: """Résout un nom d'ENT en callable ``pronotepy``. :param ent_name: Nom de l'ENT tel que configuré (ex. ``"bordeaux"``). :return: Callable ``pronotepy`` associé à l'ENT. :raises ValueError: Si le nom d'ENT n'est pas reconnu. """ resolver = _ENT_RESOLVERS.get(ent_name) if resolver is None: supported = ", ".join(sorted(_ENT_RESOLVERS.keys())) raise ValueError(f"ENT inconnu : {ent_name!r}. ENT supportés : {supported}") return resolver def _collect_auth_secrets(client: PronoteClient) -> list[str]: """Collecte toutes les valeurs sensibles d'authentification pour la redaction. Rassemble le mot de passe, le PIN QR, le contenu du fichier QR (jeton, login, url) et les credentials persistés (token, username) afin de les transmettre comme ``extra_secrets`` aux fonctions de masquage. Une valeur vide ou ``None`` est ignorée. :param client: Le client Pronote dont on collecte les secrets. :return: Liste des valeurs sensibles à expurger des logs. :rtype: list[str] """ secrets: list[str] = [] settings = client._settings # Mot de passe if settings.password is not None: secrets.append(settings.password.get_secret_value()) # PIN QR if settings.qr_pin is not None: secrets.append(settings.qr_pin.get_secret_value()) # Contenu du fichier QR (jeton, login, url) if settings.qr_code_file is not None: try: qr_path = Path(settings.qr_code_file) qr_data: Any = json.loads(qr_path.read_text(encoding="utf-8")) for key in ("jeton", "login", "url"): val = qr_data.get(key) if isinstance(val, str): secrets.append(val) except Exception as exc: logger.debug( "Impossible de lire le fichier QR %s : %s", redact_secrets(settings.qr_code_file), redact_exception(exc), ) # Credentials persistés (token, username du fichier d'état) if client._auth_state is not None: creds = client._auth_state.load() if creds is not None: for val in creds.values(): if isinstance(val, str): secrets.append(val) return [s for s in secrets if s] class PronoteClientProtocol(Protocol): """Interface du client Pronote consommée par la logique de repli.""" def get_messages(self) -> list[Message]: """Récupère les messages des discussions Pronote. :return: Liste des messages des professeurs. :rtype: list[Message] """ ... def get_informations(self) -> list[Message]: """Récupère les informations et sondages Pronote. :return: Liste des informations et sondages. :rtype: list[Message] """ ... def get_lessons(self, start: date, end: date) -> list[Lesson]: """Récupère les cours via ``pronotepy`` (repli iCal). :param start: Date de début de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse). :return: Liste des cours. :rtype: list[Lesson] """ ... def get_homeworks(self, start: date, end: date) -> list[Homework]: """Récupère les devoirs via ``pronotepy``. :param start: Date de début de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse). :return: Liste des devoirs. :rtype: list[Homework] """ ... class PronoteClient: """Client d'accès à Pronote via ``pronotepy``. Encapsule ``pronotepy.Client`` ou ``pronotepy.ParentClient`` selon le type de compte, avec une connexion paresseuse : la connexion n'est établie qu'à la première méthode de récupération appelée. Les erreurs des méthodes dégradées (``get_messages()``, ``get_informations()``) sont journalisées avec des secrets masqués et retournent une valeur vide ; ``get_lessons()`` et ``get_homeworks()`` laissent les exceptions se propager pour déclencher le repli iCal. """ def __init__( self, settings: PronoteSettings, auth_state: PronoteAuthState | None = None, ) -> None: """Initialise le client Pronote sans se connecter. :param settings: Paramètres d'accès à Pronote (username, password, ent, mode d'authentification, fichier QR et PIN). :param auth_state: Gestionnaire de persistance du token d'authentification (optionnel ; requis en mode ``qr_token`` pour conserver le token entre les exécutions). """ self._settings: PronoteSettings = settings self._auth_state: PronoteAuthState | None = auth_state self._client: pronotepy.Client | None = None def _connect(self) -> pronotepy.Client: """Crée et connecte le client ``pronotepy`` (connexion paresseuse). En mode ``password``, utilise l'authentification classique (URL, username, password, ENT). En mode ``qr_token``, utilise le token persisté via :class:`PronoteAuthState`, ou procède à l'enrôlement initial par QR code si aucun token n'est présent. :return: Le client ``pronotepy`` connecté. :rtype: pronotepy.Client :raises ValueError: Si les credentials requis sont manquants. :raises PronoteAuthRotationError: Si le token persisté est invalide (rotation requise) ou si l'enrôlement QR échoue. :raises pronotepy.PronoteAPIError: Si la connexion échoue. """ if self._client is not None: return self._client if self._settings.auth_mode == "qr_token": self._client = self._connect_qr_token() else: self._client = self._connect_password() return self._client def _connect_password(self) -> pronotepy.Client: """Connecte le client ``pronotepy`` en mode ``password``. Le nom d'ENT, s'il est configuré, est résolu via :func:`_resolve_ent` ; en l'absence d'ENT, ``ent=None`` est transmis à ``pronotepy`` pour une connexion directe. Le type de compte (``student`` ou ``parent``) détermine la classe de client utilisée. L'erreur de connexion est relancée sans journalisation, la méthode publique appelante étant responsable de la journaliser. :return: Le client ``pronotepy`` connecté. :rtype: pronotepy.Client :raises ValueError: Si ``url``, ``username`` ou ``password`` est manquant, ou si l'ENT fourni est inconnu. :raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue. """ url = self._settings.url username = self._settings.username password = self._settings.password ent = self._settings.ent if url is None or username is None or password is None: raise ValueError("url, username et password sont requis pour pronotepy") resolver = _resolve_ent(ent) if ent is not None else None client_class: type[pronotepy.Client] = ( pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client ) self._client = client_class( pronote_url=url, username=username, password=password.get_secret_value(), ent=resolver, ) return self._client def _connect_qr_token(self) -> pronotepy.Client: """Connecte via token persisté ou enrôlement par QR code. En premier lieu, les credentials persistés (``pronote_url``, username, ``password``/token, ``uuid``) sont rejoués via ``pronotepy.Client.token_login`` si :class:`PronoteAuthState` est disponible et fournit un état. En cas d'échec du login par token (exception ou client non connecté), une :class:`PronoteAuthRotationError` est levée immédiatement, sans repli vers l'enrôlement QR : la rotation du token doit être déclenchée par l'opérateur. L'enrôlement par QR code n'est tenté que lorsqu'aucun credential n'est persisté (premier login) ; le nouveau token est ensuite persisté immédiatement. :return: Le client ``pronotepy`` connecté. :rtype: pronotepy.Client :raises PronoteAuthRotationError: Si le token persisté est invalide (expiré ou refusé par Pronote), ou si l'enrôlement QR échoue (fichier QR ou PIN manquant, fichier QR invalide ou expiré). """ client_class: type[pronotepy.Client] = ( pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client ) # Login par token avec les credentials persistés if self._auth_state is not None: creds = self._auth_state.load() if creds is not None: try: client = client_class.token_login(**creds) if client.logged_in: self._auth_state.save(client.export_credentials()) return client # logged_in est False — le token est invalide raise PronoteAuthRotationError( "Le token d'authentification Pronote est invalide (non connecté). " "Action requise : supprimez le fichier .pronote_auth_state.json " "et relancez avec un nouveau QR code." ) from None except PronoteAuthRotationError: raise except Exception as exc: logger.error( "Échec du login par token pronotepy : %s", redact_exception(exc, extra_secrets=_collect_auth_secrets(self)), ) # Token expiré/invalide — pas de repli vers l'enrôlement QR raise PronoteAuthRotationError( "Le token d'authentification Pronote est expiré ou invalide. " "Action requise : supprimez le fichier .pronote_auth_state.json " "et relancez avec un nouveau QR code (PRONOTE_QR_CODE_FILE + " "PRONOTE_QR_PIN)." ) from None # Enrôlement : premier login via QR code (aucun credential persisté) client = self._enroll_qr_code(client_class) # Persister le token rotaté immédiatement if self._auth_state is not None: self._auth_state.save(client.export_credentials()) return client def _enroll_qr_code(self, client_class: type[pronotepy.Client]) -> pronotepy.Client: """Procède à l'enrôlement initial via QR code pronotepy. Le fichier QR JSON doit contenir les clés ``login``, ``jeton`` et ``url``. Le PIN et le contenu du fichier ne sont jamais journalisés ; les erreurs propagées sont expurgées. :param client_class: Classe de client pronotepy à utiliser. :return: Le client ``pronotepy`` connecté après enrôlement. :rtype: pronotepy.Client :raises PronoteAuthRotationError: Si le fichier QR ou le PIN est manquant, si le fichier QR est illisible ou incomplet, ou si le login par QR code échoue (PIN invalide ou QR code expiré). """ qr_file = self._settings.qr_code_file qr_pin = self._settings.qr_pin if qr_file is None or qr_pin is None: raise PronoteAuthRotationError( "Enrôlement QR requis : PRONOTE_QR_CODE_FILE et PRONOTE_QR_PIN sont " "nécessaires pour le premier login en mode qr_token. Supprimez le " "fichier .pronote_auth_state.json si présent et relancez avec un " "QR code frais." ) from None # Read and validate QR code JSON try: qr_path = Path(qr_file) qr_data: Any = json.loads(qr_path.read_text(encoding="utf-8")) except Exception as exc: logger.error( "Fichier QR invalide %s : %s", redact_secrets(qr_file, extra_secrets=_collect_auth_secrets(self)), redact_exception(exc, extra_secrets=_collect_auth_secrets(self)), ) raise PronoteAuthRotationError( "Impossible de lire le fichier QR code : " f"{redact_secrets(qr_file, extra_secrets=_collect_auth_secrets(self))}" ) from None # Validate required keys for key in ("login", "jeton", "url"): if key not in qr_data: raise PronoteAuthRotationError( f"Le fichier QR code ne contient pas la clé requise : {key}" ) from None pin_value = qr_pin.get_secret_value() app_uuid = f"pronote-sync-{uuid4().hex}" try: client = client_class.qrcode_login( qr_code=qr_data, pin=pin_value, uuid=app_uuid, ) except Exception as exc: logger.error( "Échec de l'enrôlement QR : %s", redact_exception(exc, extra_secrets=_collect_auth_secrets(self)), ) raise PronoteAuthRotationError( "Échec de l'enrôlement par QR code : PIN invalide ou QR code expiré. " "Générez un nouveau QR code dans l'application Pronote et mettez à " "jour PRONOTE_QR_CODE_FILE." ) from None return client def get_messages(self) -> list[Message]: """Récupère les messages des discussions Pronote. Chaque message d'une discussion est mappé sur un modèle :class:`Message` de type ``DISCUSSION``, le sujet de la discussion servant de titre. :return: Liste des messages des professeurs ; vide en cas d'erreur. :rtype: list[Message] """ try: client = self._connect() messages: list[Message] = [] for discussion in client.discussions(): for message in discussion.messages: messages.append( Message( id=message.id, type=MessageType.DISCUSSION, title=discussion.subject, content=message.content, author=message.author or "", date=message.created, read=message.seen, ) ) return messages except ( pronotepy.PronoteAPIError, ValueError, requests.RequestException, ConnectionError, TimeoutError, ) as exc: logger.error( "Échec de la récupération des messages Pronote : %s", redact_exception(exc), ) return [] def get_informations(self) -> list[Message]: """Récupère les informations et sondages Pronote. Chaque entrée est mappée sur un modèle :class:`Message` de type ``SURVEY`` si c'est un sondage, ``INFORMATION`` sinon. :return: Liste des informations et sondages ; vide en cas d'erreur. :rtype: list[Message] """ try: client = self._connect() messages: list[Message] = [] for info in client.information_and_surveys(): messages.append( Message( id=info.id, type=MessageType.SURVEY if info.survey else MessageType.INFORMATION, title=info.title or "", content=info.content(), author=info.author, date=info.creation_date, read=info.read, ) ) return messages except ( pronotepy.PronoteAPIError, ValueError, requests.RequestException, ConnectionError, TimeoutError, ) as exc: logger.error( "Échec de la récupération des informations Pronote : %s", redact_exception(exc), ) return [] def get_lessons(self, start: date, end: date) -> list[Lesson]: """Récupère les cours via ``pronotepy`` (repli iCal). Les UIDs des cours sont normalisés comme ceux du flux iCal via :func:`normalize_pronote_uid` afin que la même leçon produise le même identifiant quelle que soit la source ; en l'absence d'UID exploitable, un UID déterministe est généré via :func:`generate_deterministic_uid`. Les exceptions ne sont pas attrapées : elles se propagent afin que l'appelant puisse détecter l'échec et déclencher le repli (ou une erreur explicite). :param start: Date de début de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse). :return: Liste des cours. :rtype: list[Lesson] :raises PronoteAuthRotationError: Si le token persisté est invalide et qu'aucun ré-enrôlement n'est possible (fichier QR ou PIN manquant). :raises pronotepy.PronoteAPIError: Si l'API Pronote échoue. :raises ValueError: Si la configuration ou l'ENT est invalide. :raises requests.RequestException: Si une requête réseau échoue. :raises ConnectionError: Si la connexion réseau échoue. :raises TimeoutError: Si la requête réseau expire. """ client = self._connect() lessons: list[Lesson] = [] for lesson in client.lessons(start, end): content = lesson.content raw_uid = lesson.id if raw_uid: uid = normalize_pronote_uid(raw_uid) else: uid = generate_deterministic_uid( start=lesson.start, end=lesson.end, subject=lesson.subject.name if lesson.subject is not None else "", teachers=list(lesson.teacher_names or ()), rooms=list(lesson.classrooms or ()), group=lesson.group_name, ) lessons.append( Lesson( id=uid, start=lesson.start, end=lesson.end, subject=lesson.subject.name if lesson.subject is not None else "", teachers=tuple(lesson.teacher_names or ()), rooms=tuple(lesson.classrooms or ()), group=lesson.group_name, status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL), content=content.description if content is not None else None, ) ) return lessons def get_homeworks(self, start: date, end: date) -> list[Homework]: """Récupère les devoirs via ``pronotepy``. Les exceptions ne sont pas attrapées : elles se propagent afin que l'appelant puisse détecter l'échec et déclencher le repli (ou une erreur explicite). **pronotepy** ne fournissant ni la date de distribution ni les professeurs des devoirs, ces champs restent vides. :param start: Date de début de la fenêtre (incluse). :param end: Date de fin de la fenêtre (incluse). :return: Liste des devoirs. :rtype: list[Homework] :raises PronoteAuthRotationError: Si le token persisté est invalide et qu'aucun ré-enrôlement n'est possible (fichier QR ou PIN manquant). :raises pronotepy.PronoteAPIError: Si l'API Pronote échoue. :raises ValueError: Si la configuration ou l'ENT est invalide. :raises requests.RequestException: Si une requête réseau échoue. :raises ConnectionError: Si la connexion réseau échoue. :raises TimeoutError: Si la requête réseau expire. """ client = self._connect() homeworks: list[Homework] = [] for hw in client.homework(start, end): homeworks.append( Homework( id=hw.id, subject=hw.subject.name, teachers=(), assigned_on=None, due_on=hw.date, text=hw.description, html=hw.description, ) ) return homeworks