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 logging
from datetime import date
from typing import Any, Protocol

import pronotepy
import pronotepy.ent as pronotepy_ent
import requests

from pronote_sync.config.settings import PronoteSettings
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.utils.redaction import redact_exception

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


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) -> None:
        """Initialise le client Pronote sans se connecter.

        :param settings: Paramètres d'accès à Pronote (username, password, ent).
        """
        self._settings: PronoteSettings = settings
        self._client: pronotepy.Client | None = None

    def _connect(self) -> pronotepy.Client:
        """Crée et connecte le client ``pronotepy`` (connexion paresseuse).

        Le client est créé une seule fois puis réutilisé pour les appels
        suivants. Le nom d'ENT est résolu via :func:`_resolve_ent` et 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 ``pronote_url``, ``username``, ``password``
            ou ``ent`` est manquant, ou si l'ENT est inconnu.
        :raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
        """
        if self._client is None:
            pronote_url = self._settings.pronote_url
            username = self._settings.username
            password = self._settings.password
            ent = self._settings.ent
            if pronote_url is None or username is None or password is None or ent is None:
                raise ValueError(
                    "pronote_url, username, password et ent sont requis pour pronotepy"
                )
            resolver = _resolve_ent(ent)
            client_class: type[pronotepy.Client] = (
                pronotepy.ParentClient
                if self._settings.account_type == "parent"
                else pronotepy.Client
            )
            self._client = client_class(
                pronote_url=pronote_url,
                username=username,
                password=password.get_secret_value(),
                ent=resolver,
            )
        return self._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 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 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
            lessons.append(
                Lesson(
                    id=lesson.id,
                    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 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