spb/api-ka

1# ============================================2# Projet   : API-KA3# Fichier  : src/scheduler/backfill.py4# Node     : m3u96b5# Author   : Simon-Pierre Boucher6# Contact  : contact@spboucher.ai7# Date     : 2026-08-168# ============================================9"""Rattrapage automatique des dates manquées (7 derniers jours par défaut).1011Une date est « manquée » pour un service si aucun run réussi (``success`` ou12``retried``) n'existe dans ``collection_runs`` ET qu'aucune donnée n'est13présente dans la table du service pour cette date.14"""1516from __future__ import annotations1718import argparse19import datetime20from typing import Any2122from sqlalchemy import select2324from src.collectors import get_collector25from src.config import SERVICES, verify_node26from src.database.db import init_db, session_scope27from src.database.models import DATA_MODELS, CollectionRun28from src.utils.logger import get_logger2930SUCCESS_STATUSES = ("success", "retried")313233def missing_dates(service: str, days: int = 7) -> list[datetime.date]:34    """Liste les dates manquées d'un service sur les ``days`` derniers jours.3536    Le jour courant est exclu : il est couvert par le job quotidien lui-même.37    Les dates antérieures à la première activité du service (tout premier run,38    même échoué, ou première donnée) ne sont jamais considérées manquées —39    on ne fabrique pas d'historique antérieur à la naissance de la plateforme.40    """41    if service not in SERVICES:42        raise ValueError(f"Service inconnu : {service}")43    today = datetime.date.today()44    candidates = [today - datetime.timedelta(days=i) for i in range(1, days + 1)]45    model = DATA_MODELS[service]46    with session_scope() as session:47        all_run_dates = set(48            session.execute(49                select(CollectionRun.date_key).where(CollectionRun.service == service)50            ).scalars()51        )52        ok_run_dates = set(53            session.execute(54                select(CollectionRun.date_key).where(55                    CollectionRun.service == service,56                    CollectionRun.status.in_(SUCCESS_STATUSES),57                )58            ).scalars()59        )60        data_dates = set(61            session.execute(62                select(model.date_key).where(model.source == service).distinct()63            ).scalars()64        )65    activity_dates = all_run_dates | data_dates66    if not activity_dates:67        return []68    first_activity = min(activity_dates)69    return sorted(70        d71        for d in candidates72        if d >= first_activity and d not in ok_run_dates and d not in data_dates73    )747576def backfill_date(77    date_key: datetime.date, services: list[str] | None = None78) -> list[dict[str, Any]]:79    """Relance la collecte d'une date précise pour les services donnés (défaut : tous)."""80    results: list[dict[str, Any]] = []81    for service in services or list(SERVICES):82        results.append(get_collector(service).run(date_key=date_key))83    return results848586def run_backfill(days: int = 7) -> list[dict[str, Any]]:87    """Rattrape toutes les dates manquées des ``days`` derniers jours, tous services.8889    Returns:90        Résumés des runs de rattrapage exécutés.91    """92    logger = get_logger("apika.backfill")93    results: list[dict[str, Any]] = []94    for service in SERVICES:95        for date_key in missing_dates(service, days=days):96            logger.info(97                "Backfill d'une date manquée",98                extra={"service": service, "date_key": date_key.isoformat()},99            )100            results.append(get_collector(service).run(date_key=date_key))101    if not results:102        logger.info("Backfill : aucune date manquée", extra={"days": days})103    return results104105106def main() -> None:107    """Point d'entrée CLI : ``python -m src.scheduler.backfill --date 2026-08-15``."""108    parser = argparse.ArgumentParser(description="Backfill API-KA (m3u96b)")109    parser.add_argument(110        "--date", type=datetime.date.fromisoformat, help="Date à rattraper"111    )112    parser.add_argument(113        "--days", type=int, default=7, help="Fenêtre de rattrapage (jours)"114    )115    parser.add_argument("--service", choices=SERVICES, help="Limiter à un service")116    args = parser.parse_args()117118    verify_node()119    init_db()120    if args.date:121        backfill_date(args.date, [args.service] if args.service else None)122    else:123        run_backfill(days=args.days)124125126if __name__ == "__main__":127    main()128