veye-lalwa/pipeline/collecteurs/conseil_constitutionnel.py

"""Collecteur du Conseil constitutionnel.

Deux pages, deux usages :

- **Affaires en instance**, filtrées sur les saisines DC par le paramètre
  `?id=32246` — sans lui, la page ne rend que des questions prioritaires de
  constitutionnalité et aucune affaire DC. C'est là que se lit le sort des
  textes adoptés mais non promulgués.
- **Décisions rendues**, pour détecter le passage d'une affaire en instance à
  une décision, et donc le déblocage d'une promulgation.

Vérifié le 25 juillet 2026 : la page des affaires en instance rendait les sept
affaires DC attendues par le corpus, et une de plus que lui — l'affaire
n° 2026-907 DC sur la programmation militaire, que le rapport mentionnait sans
numéro. C'est précisément ce qu'un pipeline de veille doit rattraper.
"""

from __future__ import annotations

import re
from datetime import date

from selectolax.parser import HTMLParser, Node

from pipeline.collecteurs.base import Collecteur, ResultatCollecte
from pipeline.collecteurs.http import ClientHttp
from pipeline.journal import logger
from pipeline.modeles import DecisionCC, ResultatCC
from pipeline.parseurs.dates_fr import lire_date, sans_accents

log = logger("collecteur.conseil_constitutionnel")

BASE = "https://www.conseil-constitutionnel.fr"
URL_AFFAIRES_DC = f"{BASE}/decisions/affaires-en-instances?id=32246"
URL_DECISIONS = f"{BASE}/decisions"

_AFFAIRE = re.compile(r"\b(\d{4}-\d{1,4})\s*DC\b")
_DECISION_TITRE = re.compile(
    r"D[ée]cision\s+n°\s*(\d{4}-\d{1,4})\s*DC\s+du\s+([^-–—]+?)\s{2,}(.*)", re.IGNORECASE
)

# Formulations employées par le Conseil dans ses intitulés de décision, testées
# du plus spécifique au plus général. L'ordre compte : « non conformité
# partielle » contient « conformité », et « conformité » ne contient pas
# « conforme » — c'est bien « Conformité » et « Conformité - réserve » que la
# page publie, pas « conforme ».
_RESULTATS = (
    ("non conformite partielle", ResultatCC.NON_CONFORMITE_PARTIELLE),
    ("non conformite totale", ResultatCC.NON_CONFORMITE_TOTALE),
    ("non conformite", ResultatCC.NON_CONFORMITE_TOTALE),
    ("non conforme", ResultatCC.NON_CONFORMITE_TOTALE),
    ("conformite - reserve", ResultatCC.CONFORME_AVEC_RESERVES),
    ("conformite reserve", ResultatCC.CONFORME_AVEC_RESERVES),
    ("conformite avec reserve", ResultatCC.CONFORME_AVEC_RESERVES),
    ("conforme avec reserve", ResultatCC.CONFORME_AVEC_RESERVES),
    ("conformite", ResultatCC.CONFORME),
    ("conforme", ResultatCC.CONFORME),
)

# Le tableau des affaires DC n'a pas les mêmes colonnes que celui des QPC :
# les positions ne sont donc pas fiables, seuls les en-têtes le sont.
_COLONNES = {
    "affaire": ("affaire",),
    "disposition": ("disposition",),
    "saisissants": ("auteur de la saisine", "auteur"),
    "date_saisine": ("date de la saisine",),
    "date_decision": ("date de lecture de la decision", "decision rendue le"),
}


class CollecteurConseilConstitutionnel(Collecteur):
    nom = "conseil_constitutionnel"

    def __init__(self, client: ClientHttp) -> None:
        self.client = client

    def _collecter(self) -> ResultatCollecte:
        resultat = ResultatCollecte(collecteur=self.nom)

        try:
            resultat.decisions_cc.extend(self.affaires_en_instance())
        except Exception as erreur:  # noqa: BLE001
            resultat.erreurs.append(f"affaires en instance : {erreur}")

        try:
            rendues = self.decisions_rendues()
        except Exception as erreur:  # noqa: BLE001
            resultat.erreurs.append(f"décisions rendues : {erreur}")
            rendues = []

        # Une décision rendue prime sur l'inscription au rôle : si une affaire
        # figure dans les deux listes, c'est qu'elle vient d'être tranchée.
        par_numero = {d.numero_affaire: d for d in resultat.decisions_cc}
        for decision in rendues:
            par_numero[decision.numero_affaire] = decision
        resultat.decisions_cc = list(par_numero.values())

        return resultat

    # ── Affaires en instance ─────────────────────────────────────────────────
    def affaires_en_instance(self) -> list[DecisionCC]:
        reponse = self.client.get(URL_AFFAIRES_DC)
        if not reponse.a_reussi:
            raise RuntimeError(f"HTTP {reponse.code}")

        document = HTMLParser(reponse.texte)
        tableau = document.css_first("table")
        if tableau is None:
            raise RuntimeError("tableau des affaires introuvable")

        entetes = _index_des_colonnes(tableau)
        affaires: list[DecisionCC] = []
        for ligne in tableau.css("tbody tr"):
            if affaire := _lire_ligne_instance(ligne, entetes):
                affaires.append(affaire)

        log.debug("affaires DC en instance", nombre=len(affaires))
        return affaires

    # ── Décisions rendues ────────────────────────────────────────────────────
    def decisions_rendues(self) -> list[DecisionCC]:
        reponse = self.client.get(URL_DECISIONS)
        if not reponse.a_reussi:
            raise RuntimeError(f"HTTP {reponse.code}")

        document = HTMLParser(reponse.texte)
        decisions: list[DecisionCC] = []

        for article in document.css("article"):
            texte = " ".join(article.text(separator=" ", strip=True).split())
            if " DC " not in texte and " DC " not in texte:
                continue
            if decision := _lire_decision_rendue(texte, _lien(article)):
                decisions.append(decision)

        log.debug("décisions DC rendues", nombre=len(decisions))
        return decisions


def _index_des_colonnes(tableau: Node) -> dict[str, int]:
    """Associe chaque rôle de colonne à sa position, d'après les en-têtes."""
    entetes = [
        sans_accents(th.text(separator=" ", strip=True)).lower()
        for th in tableau.css("th")
    ]
    index: dict[str, int] = {}
    for role, variantes in _COLONNES.items():
        for position, entete in enumerate(entetes):
            if any(entete.startswith(variante) for variante in variantes):
                index[role] = position
                break
    return index


def _lire_ligne_instance(ligne: Node, colonnes: dict[str, int]) -> DecisionCC | None:
    cellules = [" ".join(c.text(separator=" ", strip=True).split()) for c in ligne.css("td")]
    if not cellules:
        return None

    def cellule(role: str) -> str:
        position = colonnes.get(role)
        return cellules[position] if position is not None and position < len(cellules) else ""

    correspondance = _AFFAIRE.search(cellule("affaire") or cellules[0])
    if not correspondance:
        return None

    date_decision = lire_date(cellule("date_decision"))
    saisissants = cellule("saisissants")
    intitule = cellule("disposition")

    return DecisionCC(
        numero_affaire=f"{correspondance.group(1)} DC",
        date_saisine=lire_date(cellule("date_saisine")),
        date_decision=date_decision,
        resultat=ResultatCC.EN_INSTANCE if date_decision is None else None,
        saisissants=saisissants[:300] or None,
        resume=intitule[:500] or None,
        url=URL_AFFAIRES_DC,
    )


def _lire_decision_rendue(texte: str, lien: str | None) -> DecisionCC | None:
    correspondance = _AFFAIRE.search(texte)
    if not correspondance:
        return None

    numero = correspondance.group(1)
    apres = texte[correspondance.end() :]

    date_decision = _date_apres_du(apres)
    resultat = _resultat_du_texte(apres)

    return DecisionCC(
        numero_affaire=f"{numero} DC",
        date_decision=date_decision,
        resultat=resultat,
        resume=" ".join(apres.split())[:500] or None,
        url=lien,
    )


_APRES_DU = re.compile(r"\bdu\s+(\d{1,2}(?:er)?\s+\w+\s+\d{4})", re.IGNORECASE)


def _date_apres_du(fragment: str) -> date | None:
    if trouve := _APRES_DU.search(fragment):
        return lire_date(trouve.group(1))
    return lire_date(fragment)


def _resultat_du_texte(fragment: str) -> ResultatCC | None:
    plat = sans_accents(fragment).lower()
    for forme, resultat in _RESULTATS:
        if forme in plat:
            return resultat
    return None


def _lien(article: Node) -> str | None:
    ancre = article.css_first("a")
    if ancre is None:
        return None
    href = ancre.attributes.get("href") or ""
    if href.startswith("http"):
        return href
    if href.startswith("/"):
        return BASE + href
    return None