college-infos/pronote_sync/sources/pronote/client.py

"""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