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. Toutes les erreurs sont journalisées avec des secrets masqués.
"""

from __future__ import annotations

import logging
from datetime import date
from typing import Protocol

import pronotepy

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__)


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_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]:
        """Récupère les cours et les devoirs 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: Tuple ``(cours, devoirs)``.
        :rtype: tuple[list[Lesson], list[Homework]]
        """
        ...


class PronoteClient:
    """Client d'accès à Pronote via ``pronotepy``.

    Encapsule ``pronotepy.Client`` avec une connexion paresseuse : la
    connexion n'est établie qu'à la première méthode de récupération
    appelée. Les erreurs ``pronotepy.PronoteAPIError`` sont journalisées
    avec des secrets masqués et les méthodes de récupération retournent
    alors une valeur vide au lieu de propager l'exception.
    """

    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. 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 ``username``, ``password`` ou ``ent`` est manquant.
        :raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
        """
        if self._client is None:
            username = self._settings.username
            password = self._settings.password
            ent = self._settings.ent
            if username is None or password is None or ent is None:
                raise ValueError("username, password et ent sont requis pour pronotepy")
            try:
                self._client = pronotepy.Client(username, password.get_secret_value(), ent)
            except pronotepy.PronoteAPIError:
                raise
        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) 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) as exc:
            logger.error(
                "Échec de la récupération des informations Pronote : %s",
                redact_exception(exc),
            )
            return []

    def get_agenda_fallback(self, start: date, end: date) -> tuple[list[Lesson], list[Homework]]:
        """Récupère les cours et les devoirs via ``pronotepy``.

        À utiliser uniquement si les sources iCal sont indisponibles ou en
        repli automatique. Les cours annulés sont mappés sur le statut
        ``CANCELLED`` ; **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: Tuple ``(cours, devoirs)`` ; vide en cas d'erreur.
        :rtype: tuple[list[Lesson], list[Homework]]
        """
        try:
            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,
                    )
                )
            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 lessons, homeworks
        except (pronotepy.PronoteAPIError, ValueError) as exc:
            logger.error(
                "Échec de la récupération de l'agenda via pronotepy : %s",
                redact_exception(exc),
            )
            return [], []