college-infos/pronote_sync/sources/blog/rss.py

"""Client de récupération et de parsing du flux RSS du blog du collège.

Ce module définit :class:`BlogRSSClient`, un client sans état qui
télécharge le flux RSS du blog via ``feedparser``, déduplique les entrées
par GUID et les convertit en :class:`~pronote_sync.models.blog.BlogArticle`.

Le résultat d'une récupération est un
:class:`~pronote_sync.sources.blog.result.BlogRSSFetchResult` : les
nouveaux articles (triés par date de publication décroissante, puis par
identifiant croissant) accompagnés des en-têtes HTTP ``ETag`` et
``Last-Modified`` de la réponse. Toute erreur de récupération ou de
parsing est journalisée (URL et exception rédigées) puis dégradée en
résultat vide : une liste vide est un succès valide, pas une panne.
"""

from __future__ import annotations

import logging
import re
from datetime import UTC, datetime
from html import unescape

import feedparser  # type: ignore[import-untyped]
from bs4 import BeautifulSoup

from pronote_sync.models.blog import BlogArticle
from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.utils.redaction import redact_exception, redact_url

logger = logging.getLogger(__name__)


class BlogRSSClient:
    """Client de récupération et de parsing du flux RSS du blog du collège.

    Client sans état : aucune E/S n'est effectuée à la construction et
    aucune donnée n'est conservée entre deux appels à
    :meth:`fetch_and_parse`. Toute erreur de récupération ou de parsing
    est journalisée puis dégradée en résultat vide.

    :param rss_url: URL du flux RSS du blog du collège.
    :param timeout: Timeout HTTP en secondes (défaut : 20).
    """

    def __init__(self, rss_url: str, timeout: int = 20) -> None:
        """Initialise le client RSS du blog.

        Aucune opération d'E/S n'est réalisée ici : le téléchargement et
        le parsing n'ont lieu qu'à l'appel de :meth:`fetch_and_parse`.

        :param rss_url: URL du flux RSS du blog du collège.
        :param timeout: Timeout HTTP en secondes (défaut : 20).
        """
        self.rss_url = rss_url
        self.timeout = timeout

    def fetch_and_parse(
        self,
        *,
        known_guids: frozenset[str] | None = None,
        etag: str | None = None,
        last_modified: str | None = None,
    ) -> BlogRSSFetchResult:
        """Télécharge et parse le flux RSS du blog en nouveaux articles.

        Le flux est téléchargé par ``feedparser`` avec les en-têtes de
        requête conditionnelle fournis (``ETag``/``Last-Modified``). Si le
        serveur répond ``304 Not Modified``, le résultat est vide avec
        ``not_modified=True`` et les en-têtes passés en entrée sont
        restitués tels quels. Chaque entrée est dédupliquée par GUID,
        convertie en :class:`~pronote_sync.models.blog.BlogArticle`, puis
        l'ensemble est trié par date de publication décroissante puis par
        identifiant croissant. Toute erreur est journalisée (URL et
        exception rédigées) et dégradée en résultat vide : aucune
        exception n'est propagée.

        :param known_guids: Ensemble des GUID d'articles déjà traités ; les
            entrées correspondantes sont ignorées. ``None`` pour tout
            conserver (défaut).
        :param etag: Valeur de l'en-tête ``ETag`` mémorisée pour la requête
            conditionnelle, ou ``None`` (défaut).
        :param last_modified: Valeur de l'en-tête ``Last-Modified`` mémorisée
            pour la requête conditionnelle, ou ``None`` (défaut).
        :return: Résultat de la récupération : nouveaux articles (tuple vide
            si aucun nouvel article, réponse ``304`` ou erreur), en-têtes de
            cache de la réponse et indicateur ``not_modified``.
        :rtype: :class:`~pronote_sync.sources.blog.result.BlogRSSFetchResult`
        """
        try:
            feed = feedparser.parse(
                self.rss_url,
                etag=etag,
                modified=last_modified,
                request_timeout=self.timeout,
            )

            # Réponse 304 Not Modified : rien n'a changé, on restitue les
            # en-têtes mémorisés tels quels pour les conserver.
            if getattr(feed, "status", None) == 304:
                return BlogRSSFetchResult(
                    articles=(),
                    etag=etag,
                    last_modified=last_modified,
                    not_modified=True,
                )

            response_etag: str | None = getattr(feed, "etag", None)
            response_last_modified: str | None = getattr(feed, "modified", None)
            if response_last_modified is None:
                headers = getattr(feed, "headers", None)
                if headers is not None:
                    # Les clés des en-têtes sont en minuscules côté feedparser.
                    response_last_modified = headers.get("last-modified") or None

            # Flux invalide (erreur HTTP, XML malformé, etc.) : avertissement
            # puis résultat vide, sans propager l'exception brute.
            if getattr(feed, "bozo", None):
                bozo_exception = getattr(feed, "bozo_exception", None)
                if bozo_exception is not None:
                    logger.warning(
                        "Flux RSS du blog invalide (%s), ignoré : %s",
                        redact_exception(bozo_exception),
                        redact_url(self.rss_url),
                    )
                else:
                    logger.warning(
                        "Flux RSS du blog invalide, ignoré : %s",
                        redact_url(self.rss_url),
                    )
                return BlogRSSFetchResult(
                    articles=(),
                    etag=response_etag,
                    last_modified=response_last_modified,
                    not_modified=False,
                )

            articles: list[BlogArticle] = []
            seen_guids: set[str] = set(known_guids) if known_guids is not None else set()
            for entry in getattr(feed, "entries", []):
                guid_source = entry.get("id") or entry.get("link")
                if not guid_source:
                    logger.warning(
                        "Entrée RSS sans GUID ni lien, ignorée : %s",
                        redact_url(self.rss_url),
                    )
                    continue
                guid = str(guid_source)

                if guid in seen_guids:
                    logger.warning(
                        "Entrée RSS déjà traitée ou en double, ignorée : %s",
                        redact_url(self.rss_url),
                    )
                    continue

                published_at = self._parse_date(
                    entry.get("published_parsed") or entry.get("pubdate_parsed")
                )
                if published_at is None:
                    logger.warning(
                        "Entrée RSS sans date de publication valide, ignorée : %s",
                        redact_url(self.rss_url),
                    )
                    continue

                updated_at = self._parse_date(entry.get("updated_parsed"))

                raw_content = entry.get("content")
                if raw_content:
                    content_html = str(raw_content[0].get("value") or "")
                else:
                    content_html = str(entry.get("description") or "")

                tags = entry.get("tags")
                category_value = tags[0].get("term") if tags else None
                if not category_value:
                    category_value = entry.get("category")
                category = str(category_value) if category_value else None

                author_value = entry.get("author")
                author = str(author_value) if author_value else None

                title = str(entry.get("title") or guid)
                url = str(entry.get("link") or guid)

                articles.append(
                    BlogArticle(
                        id=guid,
                        title=title,
                        url=url,
                        published_at=published_at,
                        updated_at=updated_at,
                        category=category,
                        author=author,
                        content_html=content_html,
                        content_text=self._html_to_text(content_html),
                    )
                )

                seen_guids.add(guid)

            # Tri stable : d'abord par identifiant croissant, puis par date de
            # publication décroissante ; l'ordre par identifiant est conservé
            # entre articles de même date.
            articles.sort(key=lambda article: article.id)
            articles.sort(key=lambda article: article.published_at, reverse=True)

            return BlogRSSFetchResult(
                articles=tuple(articles),
                etag=response_etag,
                last_modified=response_last_modified,
                not_modified=False,
            )
        except Exception as exc:
            logger.error(
                "Échec de la récupération du flux RSS du blog %s : %s",
                redact_url(self.rss_url),
                redact_exception(exc),
            )
            return BlogRSSFetchResult(articles=(), not_modified=False)

    @staticmethod
    def _parse_date(date_tuple: tuple[int, ...] | None) -> datetime | None:
        """Convertit un tuple de date ``struct_time`` en :class:`datetime` UTC.

        :param date_tuple: Tuple horodaté au format ``time.struct_time``
            (indices 0 à 5 : année, mois, jour, heure, minute, seconde), ou
            ``None`` si absent.
        :return: Date/heure consciente du fuseau UTC, ou ``None`` si le
            tuple est absent, vide ou invalide.
        :rtype: datetime | None
        """
        if not date_tuple:
            return None
        try:
            return datetime(
                date_tuple[0],
                date_tuple[1],
                date_tuple[2],
                date_tuple[3],
                date_tuple[4],
                date_tuple[5],
                tzinfo=UTC,
            )
        except (ValueError, IndexError):
            return None

    @staticmethod
    def _html_to_text(html: str) -> str:
        """Convertit du HTML en texte brut nettoyé.

        Le HTML est parsé avec BeautifulSoup, les balises sont remplacées
        par des espaces, les entités HTML sont décodées et les suites
        d'espaces sont unifiées.

        :param html: Contenu HTML à convertir.
        :return: Texte brut sans balises, entités décodées et espaces
            unifiés ; chaîne vide si ``html`` est vide.
        :rtype: str
        """
        if not html:
            return ""
        soup = BeautifulSoup(html, "html.parser")
        text = soup.get_text(separator=" ", strip=True)
        text = unescape(text)
        return re.sub(r"\s+", " ", text).strip()