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 ``requests``, le parse 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]
import requests
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 ``requests`` avec les en-têtes de
        requête conditionnelle fournis (``ETag``/``Last-Modified``), puis
        parsé par ``feedparser``. 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:
            # Téléchargement HTTP explicite via requests : feedparser 6.x
            # n'accepte aucun paramètre de transport ; les requêtes
            # conditionnelles sont gérées avec les en-têtes HTTP standards.
            headers: dict[str, str] = {"user-agent": "pronote-sync"}
            if etag is not None:
                headers["If-None-Match"] = etag
            if last_modified is not None:
                headers["If-Modified-Since"] = last_modified
            response = requests.get(self.rss_url, headers=headers, 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 response.status_code == 304:
                return BlogRSSFetchResult(
                    articles=(),
                    etag=etag,
                    last_modified=last_modified,
                    not_modified=True,
                )

            # Les statuts 4xx/5xx lèvent une exception HTTP, attrapée par le
            # gestionnaire général et dégradée en résultat vide.
            response.raise_for_status()

            response_etag: str | None = response.headers.get("ETag", None)
            if response_etag is None:
                response_etag = response.headers.get("etag", None)
            response_last_modified: str | None = response.headers.get("Last-Modified", None)
            if response_last_modified is None:
                response_last_modified = response.headers.get("last-modified", None)

            # feedparser ne reçoit que le contenu brut de la réponse.
            feed = feedparser.parse(response.content)

            # Flux invalide (XML malformé, etc.) : avertissement puis résultat
            # vide, sans propager l'exception brute. Les validateurs de cache
            # d'entrée sont conservés : on ne fait pas confiance aux en-têtes
            # d'une réponse au contenu invalide.
            if getattr(feed, "bozo", None):
                bozo_exception = getattr(feed, "bozo_exception", None)
                if bozo_exception is not None:
                    error_msg = f"Flux RSS invalide : {redact_exception(bozo_exception)}"
                    logger.warning(
                        "Flux RSS du blog invalide (%s), ignoré : %s",
                        redact_exception(bozo_exception),
                        redact_url(self.rss_url),
                    )
                else:
                    error_msg = "Flux RSS invalide"
                    logger.warning(
                        "Flux RSS du blog invalide, ignoré : %s",
                        redact_url(self.rss_url),
                    )
                return BlogRSSFetchResult(
                    articles=(),
                    etag=etag,
                    last_modified=last_modified,
                    not_modified=False,
                    error=error_msg,
                )

            articles: list[BlogArticle] = []
            # Déduplication silencieuse des GUID déjà connus (exécutions
            # précédentes) et détection des doublons au sein de la réponse.
            known_set = set(known_guids) if known_guids is not None else None
            seen_in_feed: set[str] = 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 known_set is not None and guid in known_set:
                    # Déduplication normale (GUID connu d'une exécution
                    # précédente) : aucun journal n'est nécessaire.
                    continue

                if guid in seen_in_feed:
                    logger.warning(
                        "Entrée RSS en double dans le flux, ignorée : %s",
                        redact_url(self.rss_url),
                    )
                    continue

                seen_in_feed.add(guid)

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

            # 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:
            error_msg = redact_exception(exc)
            logger.error(
                "Échec de la récupération du flux RSS du blog %s : %s",
                redact_url(self.rss_url),
                error_msg,
            )
            return BlogRSSFetchResult(
                articles=(),
                etag=etag,
                last_modified=last_modified,
                not_modified=False,
                error=error_msg,
            )

    @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()