spb/api-ka

1# ============================================2# Projet   : API-KA3# Fichier  : src/collectors/base_collector.py4# Node     : m3u96b5# Author   : Simon-Pierre Boucher6# Contact  : contact@spboucher.ai7# Date     : 2026-08-168# ============================================9"""Classe abstraite des collecteurs KA : fetch → validate → checksum → insert → backup → log run.1011Chaque collecteur relance automatiquement en cas d'échec (3 tentatives,12backoff 30s → 2min → 10min). Après épuisement des tentatives, une entrée13``status=failed`` est écrite dans ``collection_runs`` et une alerte est14ajoutée à ``logs/alerts.log``.15"""1617from __future__ import annotations1819import abc20import datetime21import hashlib22import json23from collections.abc import Sequence24from typing import Any2526import httpx27from sqlalchemy import select2829from src.config import get_settings30from src.database.db import session_scope31from src.database.models import DATA_MODELS, CollectionRun32from src.utils.backup import backup_service33from src.utils.logger import alert, get_logger34from src.utils.retry import DEFAULT_ATTEMPTS, DEFAULT_DELAYS, retry_call3536FETCH_TIMEOUT_SECONDS = 60.0373839class BaseCollector(abc.ABC):40    """Collecteur abstrait ; chaque sous-classe définit l'attribut ``service``.4142    La collecte pagine l'API source jusqu'à épuisement :43    - ``items_key`` : clé de la liste d'items dans la réponse JSON44      (``None`` = la réponse est directement la liste ou un objet unique) ;45    - ``pagination`` : ``"offset"`` (paramètres limit/offset) ou ``"page"``46      (paramètres page/per_page) ;47    - ``page_size`` : taille de page, bornée par le maximum de l'API source.48    """4950    service: str51    items_key: str | None = None52    pagination: str = "offset"53    page_size: int = 50054    max_pages: int = 20005556    def __init__(57        self,58        attempts: int = DEFAULT_ATTEMPTS,59        delays: Sequence[float] = DEFAULT_DELAYS,60    ) -> None:61        if self.service not in DATA_MODELS:62            raise ValueError(f"Service inconnu : {self.service!r}")63        self.attempts = attempts64        self.delays = tuple(delays)65        self.logger = get_logger(f"apika.collector.{self.service}")6667    @property68    def model(self) -> type:69        """Modèle SQLAlchemy de la table de données du service."""70        return DATA_MODELS[self.service]7172    @property73    def source_url(self) -> str:74        """URL source du service, lue depuis le .env."""75        url = get_settings().source_urls.get(self.service, "")76        if not url:77            raise RuntimeError(78                f"Variable source manquante pour {self.service} "79                f"(voir .env : {self.service.upper()}_SOURCE_URL)"80            )81        return url8283    def fetch(self) -> Any:84        """Récupère TOUTES les données de la source en paginant jusqu'à épuisement.8586        Returns:87            Liste complète des items bruts du service.88        """89        items: list[Any] = []90        total: int | None = None91        with httpx.Client(92            timeout=FETCH_TIMEOUT_SECONDS, follow_redirects=True93        ) as client:94            for page_index in range(self.max_pages):95                if self.pagination == "page":96                    params: dict[str, int] = {97                        "page": page_index + 1,98                        "per_page": self.page_size,99                    }100                else:101                    params = {102                        "limit": self.page_size,103                        "offset": page_index * self.page_size,104                    }105                response = client.get(self.source_url, params=params)106                response.raise_for_status()107                payload = response.json()108109                if isinstance(payload, dict) and self.items_key:110                    batch = payload.get(self.items_key) or []111                    raw_total = payload.get("total")112                    if isinstance(raw_total, int):113                        total = raw_total114                elif isinstance(payload, list):115                    batch = payload116                else:117                    batch = [payload]118119                items.extend(batch)120                if not batch or len(batch) < self.page_size:121                    break122                if total is not None and len(items) >= total:123                    break124        return items125126    def validate(self, payload: Any) -> list[dict[str, Any]]:127        """Valide le payload et le normalise en liste d'objets JSON.128129        Raises:130            ValueError: Si le payload est vide ou ``None``.131        """132        if payload is None:133            raise ValueError(f"Payload vide (None) pour {self.service}")134        items = payload if isinstance(payload, list) else [payload]135        if not items:136            raise ValueError(f"Payload vide (liste vide) pour {self.service}")137        return [item if isinstance(item, dict) else {"value": item} for item in items]138139    @staticmethod140    def checksum(item: dict[str, Any]) -> str:141        """SHA-256 du JSON canonique d'un enregistrement (déduplication)."""142        canonical = json.dumps(143            item, sort_keys=True, ensure_ascii=False, separators=(",", ":")144        )145        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()146147    def save(self, items: list[dict[str, Any]], date_key: datetime.date) -> int:148        """Insère les enregistrements nouveaux (dédupliqués par checksum).149150        Returns:151            Nombre d'enregistrements réellement insérés.152        """153        model = self.model154        collected_at = datetime.datetime.now(tz=datetime.UTC)155        with session_scope() as session:156            existing: set[str] = set(157                session.execute(158                    select(model.checksum).where(159                        model.source == self.service, model.date_key == date_key160                    )161                ).scalars()162            )163            inserted = 0164            for item in items:165                digest = self.checksum(item)166                if digest in existing:167                    continue168                existing.add(digest)169                session.add(170                    model(171                        payload=item,172                        source=self.service,173                        collected_at=collected_at,174                        date_key=date_key,175                        checksum=digest,176                    )177                )178                inserted += 1179        return inserted180181    def run(self, date_key: datetime.date | None = None) -> dict[str, Any]:182        """Exécute le pipeline complet et journalise le run dans collection_runs.183184        Returns:185            Résumé du run : service, date_key, status, records_count, error.186        """187        settings = get_settings()188        date_key = date_key or datetime.date.today()189        started_at = datetime.datetime.now(tz=datetime.UTC)190        retries = {"count": 0}191        status = "success"192        error_message: str | None = None193        records_count = 0194195        def _pipeline() -> int:196            payload = self.fetch()197            items = self.validate(payload)198            return self.save(items, date_key)199200        def _on_retry(attempt: int, exc: BaseException, delay: float) -> None:201            retries["count"] = attempt202203        try:204            records_count = retry_call(205                _pipeline,206                attempts=self.attempts,207                delays=self.delays,208                on_retry=_on_retry,209            )210            if retries["count"] > 0:211                status = "retried"212            backup_service(self.service, date_key)213        except Exception as exc:214            status = "failed"215            error_message = str(exc)216            alert(217                f"[{self.service}] collecte échouée pour {date_key.isoformat()} "218                f"après {self.attempts} relances : {exc}"219            )220221        finished_at = datetime.datetime.now(tz=datetime.UTC)222        duration = (finished_at - started_at).total_seconds()223        with session_scope() as session:224            session.add(225                CollectionRun(226                    service=self.service,227                    date_key=date_key,228                    status=status,229                    records_count=records_count,230                    duration_seconds=duration,231                    error_message=error_message,232                    node=settings.node_name,233                    started_at=started_at,234                    finished_at=finished_at,235                )236            )237238        summary = {239            "service": self.service,240            "date_key": date_key.isoformat(),241            "status": status,242            "records_count": records_count,243            "duration_seconds": duration,244            "error": error_message,245        }246        self.logger.info("Run de collecte terminé", extra=summary)247        return summary248