API-KA — plateforme centrale : collecte quotidienne des 8 services KA, historisation append-only et API publique sur www.api-ka.com
1# ============================================2# Projet : API-KA3# Fichier : src/monitoring/connector_health.py4# Node : m3u96b5# Author : Simon-Pierre Boucher6# Contact : contact@spboucher.ai7# Date : 2026-08-188# ============================================9"""Supervision centralisée des connecteurs des 9 apps de l'écosystème KA.1011Toutes les 2 h, le job lit le ``GET /api/stats`` de chaque app sœur (LAN,12lecture seule, timeout 5 s) et en déduit un état par source quand le détail13existe (``recent_syncs`` / ``sync_log``), plus un état global par app14(source virtuelle ``_app``). L'échec d'une app n'interrompt jamais le job.1516Règles de classement :1718- ``broken`` : ≥ 3 synchros consécutives en échec ou à 0 résultat19 (ou app injoignable 3 fois de suite) ;20- ``stale`` : aucun sync réussi depuis > 2× la cadence attendue du service21 (au niveau source, la cadence est multipliée par un facteur de22 rotation : les apps synchronisent leurs sources par lots, une23 source individuelle repasse donc moins souvent que l'app) ;24- ``degraded`` : dernier volume < 50 % de la médiane observée, dernier sync en25 échec (< 3), ou app injoignable (< 3) ;26- ``ok`` : sinon.2728Anti-spam : l'état précédent est mémorisé dans la table ``connector_health``29(une ligne par couple service/source, upsert à chaque run) ; une alerte30n'est écrite dans ``logs/alerts.log`` que lors d'une TRANSITION vers31``broken`` ou ``stale``.32"""3334from __future__ import annotations3536import datetime37import math38import statistics39import time40from dataclasses import dataclass, field41from typing import Any42from urllib.parse import urlsplit4344import httpx45from sqlalchemy import select4647from src.config import SERVICES, get_settings48from src.database.db import session_scope49from src.database.models import ConnectorHealth50from src.utils.logger import alert, get_logger5152# Source virtuelle représentant l'état global d'une app.53APP_SOURCE = "_app"5455STATS_TIMEOUT_SECONDS = 5.05657# Fenêtre de journal de sync demandée aux apps sœurs (heures). Les apps ne58# renvoient par défaut que ~20 entrées (~11 min d'activité pour lou-ka et ses59# ~200 sources) alors qu'on polle toutes les 2 h : la quasi-totalité des60# sources n'était jamais observée → fausses alertes « stale » de masse61# (incident du 2026-08-22 : 195 stale, dont 82 lou-ka qui synchronisaient62# pourtant avec succès). Les apps qui ne connaissent pas encore le paramètre63# l'ignorent (FastAPI) et gardent le comportement historique.64SYNC_WINDOW_HOURS = 66566STATUS_OK = "ok"67STATUS_DEGRADED = "degraded"68STATUS_BROKEN = "broken"69STATUS_STALE = "stale"70ALL_STATUSES = (STATUS_OK, STATUS_DEGRADED, STATUS_BROKEN, STATUS_STALE)7172# broken après N échecs (ou 0 résultat) consécutifs.73BROKEN_THRESHOLD = 374# degraded si volume < 50 % de la médiane.75DEGRADED_RATIO = 0.576# stale si aucun succès depuis > 2× la cadence attendue.77STALE_FACTOR = 27879# Cadence de sync attendue par service (heures).80EXPECTED_CADENCE_HOURS: dict[str, float] = {81 "louka": 1,82 "immoka": 4,83 "autoka": 2,84 "foodka": 6,85 "fabrika": 6,86 "sortika": 1,87 "creaka": 24,88 "restoka": 168,89 "jobka": 1,90}919293@dataclass94class SyncEntry:95 """Une entrée de journal de sync exposée par une app sœur."""9697 ts: float98 found: int99 ok: bool100 message: str = ""101102103@dataclass104class SourceState:105 """État précédent d'un connecteur (relu depuis ``connector_health``)."""106107 status: str | None = None108 last_success: datetime.datetime | None = None109 found_last: int | None = None110 median_found: float | None = None111 consecutive_failures: int = 0112 checked_at: datetime.datetime | None = None113114115@dataclass116class Assessment:117 """Résultat du classement d'un connecteur pour le run courant."""118119 status: str120 last_success: datetime.datetime | None121 found_last: int | None122 median_found: float | None123 consecutive_failures: int124 message: str = ""125 entries: list[SyncEntry] = field(default_factory=list)126127128def _utc(ts: float) -> datetime.datetime:129 return datetime.datetime.fromtimestamp(ts, tz=datetime.UTC)130131132def assess_source(133 entries: list[SyncEntry],134 prev: SourceState,135 now: float,136 stale_after_seconds: float,137) -> Assessment:138 """Classe un connecteur : broken > stale > degraded > ok.139140 Fonction pure (testable) : combine les entrées de la fenêtre courante141 avec l'état précédent (streak d'échecs, dernier succès, médiane glissante).142 """143 entries = sorted(entries, key=lambda e: e.ts, reverse=True)144145 # Streak d'échecs : seules les entrées jamais vues (postérieures au dernier146 # passage) alimentent le compteur, du plus ancien au plus récent.147 prev_checked_ts = prev.checked_at.timestamp() if prev.checked_at else None148 new_entries = [149 e for e in entries if prev_checked_ts is None or e.ts > prev_checked_ts150 ]151 # Un « 0 résultat » n'est suspect que si la source a normalement des152 # résultats (médiane historique ≥ 1). Une source vide légitimement153 # (agence sans inventaire en ligne, employeur sans poste ouvert…) qui154 # synchronise ok à 0 n'est PAS en panne — revue du 2026-08-23 : les155 # « brisés » niddamour, courtemanche, place_florimay, atlas_immo étaient156 # tous des vides légitimes vérifiés à la main.157 legit_empty_history = (prev.median_found or 0) < 1158 consecutive_failures = prev.consecutive_failures159 for entry in reversed(new_entries):160 suspicious_zero = entry.ok and entry.found <= 0 and not legit_empty_history161 if not entry.ok or suspicious_zero:162 consecutive_failures += 1163 else:164 consecutive_failures = 0165166 # Dernier sync réussi (ok ET au moins un résultat).167 last_success = prev.last_success168 for entry in entries:169 if entry.ok and entry.found > 0:170 candidate = _utc(entry.ts)171 if last_success is None or candidate > last_success:172 last_success = candidate173 break174175 # Médiane glissante du volume (entrées ok de la fenêtre + médiane passée).176 ok_founds = [float(e.found) for e in entries if e.ok]177 values = ok_founds + (178 [float(prev.median_found)] if prev.median_found is not None else []179 )180 median_found = statistics.median(values) if values else None181182 found_last = entries[0].found if entries else prev.found_last183184 last_message = next((e.message for e in entries if e.message), "")185186 # La dernière synchro observée est-elle saine ? (ok avec des résultats,187 # ou ok à 0 pour une source vide légitimement) — si oui, la source n'est188 # pas « stale » même si son dernier succès AVEC résultats est ancien.189 latest_healthy = bool(entries) and entries[0].ok and (190 entries[0].found > 0 or legit_empty_history191 )192193 if consecutive_failures >= BROKEN_THRESHOLD:194 message = (195 f"{consecutive_failures} synchros consécutives en échec ou à 0 résultat"196 )197 if last_message and last_message != "ok":198 message += f" (dernier message : {last_message})"199 status = STATUS_BROKEN200 elif (201 not latest_healthy202 and last_success is not None203 and (now - last_success.timestamp()) > stale_after_seconds204 ):205 hours = (now - last_success.timestamp()) / 3600206 message = (207 f"aucun sync réussi depuis {hours:.1f} h "208 f"(seuil : {stale_after_seconds / 3600:.0f} h)"209 )210 status = STATUS_STALE211 elif last_success is None:212 status = STATUS_DEGRADED213 message = "aucun sync réussi observé pour l'instant"214 elif entries and not entries[0].ok:215 status = STATUS_DEGRADED216 message = (217 f"dernier sync en échec ({consecutive_failures}/{BROKEN_THRESHOLD})"218 )219 if last_message and last_message != "ok":220 message += f" — {last_message}"221 elif (222 found_last is not None223 and median_found224 and found_last < DEGRADED_RATIO * median_found225 ):226 status = STATUS_DEGRADED227 message = (228 f"volume {found_last} < 50 % de la médiane ({median_found:.0f})"229 )230 else:231 status = STATUS_OK232 message = "ok"233234 return Assessment(235 status=status,236 last_success=last_success,237 found_last=found_last,238 median_found=median_found,239 consecutive_failures=consecutive_failures,240 message=message,241 entries=entries,242 )243244245def assess_unreachable(prev: SourceState, error: str) -> Assessment:246 """Classe l'état global d'une app dont le /api/stats est injoignable."""247 consecutive_failures = prev.consecutive_failures + 1248 status = (249 STATUS_BROKEN if consecutive_failures >= BROKEN_THRESHOLD else STATUS_DEGRADED250 )251 return Assessment(252 status=status,253 last_success=prev.last_success,254 found_last=prev.found_last,255 median_found=prev.median_found,256 consecutive_failures=consecutive_failures,257 message=f"GET /api/stats injoignable ({error})",258 )259260261def parse_sync_entries(payload: dict[str, Any]) -> dict[str, list[SyncEntry]]:262 """Extrait les entrées de sync par source depuis un payload /api/stats.263264 Comprend les deux formats observés dans l'écosystème :265 ``recent_syncs`` (lou-ka, immo-ka, auto-ka, food-ka, resto-ka, job-ka) et266 ``sync_log`` par magasin (fabri-ka, clé ``store_id`` + ``status``).267 """268 raw = payload.get("recent_syncs") or payload.get("sync_log") or []269 by_source: dict[str, list[SyncEntry]] = {}270 for item in raw:271 if not isinstance(item, dict):272 continue273 source = item.get("source") or item.get("store_id")274 if not source:275 continue276 if "ok" in item:277 ok = bool(item.get("ok"))278 else:279 ok = str(item.get("status", "ok")).lower() == "ok"280 try:281 entry = SyncEntry(282 ts=float(item.get("ts") or 0),283 found=int(item.get("found") or 0),284 ok=ok,285 message=str(item.get("message") or ""),286 )287 except (TypeError, ValueError):288 continue289 by_source.setdefault(str(source), []).append(entry)290 return by_source291292293_GLOBAL_TOTAL_KEYS = ("total_active", "total", "creators", "restaurants")294295296def build_app_entry(297 service: str,298 payload: dict[str, Any],299 by_source: dict[str, list[SyncEntry]],300 now: float,301) -> SyncEntry:302 """Synthétise UNE entrée représentant l'état global de l'app.303304 - apps avec journal de sync : agrégat de la fenêtre (ts le plus récent,305 somme des volumes ok) ;306 - crea-ka : ``last_sync`` global + total ``creators`` ;307 - sorti-ka (aucun horodatage exposé) : joignabilité + volume total.308 """309 all_entries = [e for lst in by_source.values() for e in lst]310 if all_entries:311 return SyncEntry(312 ts=max(e.ts for e in all_entries),313 found=sum(e.found for e in all_entries if e.ok),314 ok=any(e.ok for e in all_entries),315 )316317 last_sync = payload.get("last_sync")318 if isinstance(last_sync, str) and last_sync:319 try:320 ts = datetime.datetime.fromisoformat(321 last_sync.replace("Z", "+00:00")322 ).timestamp()323 except ValueError:324 ts = now325 total = _global_total(payload)326 return SyncEntry(ts=ts, found=total, ok=True, message="last_sync global")327328 # Fallback (ex. sorti-ka) : l'app ne publie aucun horodatage de sync.329 return SyncEntry(330 ts=now,331 found=_global_total(payload),332 ok=True,333 message="pas d'horodatage de sync exposé — joignabilité + volume total",334 )335336337def _global_total(payload: dict[str, Any]) -> int:338 for key in _GLOBAL_TOTAL_KEYS:339 value = payload.get(key)340 if isinstance(value, (int, float)):341 return int(value)342 totals = payload.get("totals")343 if isinstance(totals, dict):344 for value in totals.values():345 if isinstance(value, (int, float)):346 return int(value)347 return 0348349350def rotation_factor(351 payload: dict[str, Any], by_source: dict[str, list[SyncEntry]]352) -> int:353 """Facteur de rotation des sources : les apps synchronisent leurs sources354 par lots (fenêtre ~20 entrées pour parfois >100 sources). Une source355 individuelle repasse donc toutes les ``rotation × cadence`` heures environ ;356 le seuil de staleness par source en tient compte pour ne pas générer de357 fausses alertes de masse."""358 window = sum(len(v) for v in by_source.values())359 hint = payload.get("sources")360 if not isinstance(hint, (int, float)):361 totals = payload.get("totals")362 hint = totals.get("stores_live") if isinstance(totals, dict) else 0363 total = max(int(hint or 0), len(by_source))364 if window <= 0 or total <= 0:365 return 1366 return max(1, math.ceil(total / window))367368369def _stats_url(service: str) -> str | None:370 """Dérive l'URL /api/stats de l'app depuis la SOURCE_URL du .env."""371 source_url = get_settings().source_urls.get(service, "")372 if not source_url:373 return None374 parts = urlsplit(source_url)375 if not parts.scheme or not parts.netloc:376 return None377 return f"{parts.scheme}://{parts.netloc}/api/stats?syncs_since_h={SYNC_WINDOW_HOURS}"378379380def _fetch_stats(url: str) -> dict[str, Any]:381 """GET /api/stats (lecture seule, timeout court)."""382 response = httpx.get(url, timeout=STATS_TIMEOUT_SECONDS, follow_redirects=True)383 response.raise_for_status()384 payload = response.json()385 if not isinstance(payload, dict):386 raise ValueError("payload /api/stats inattendu (dict requis)")387 return payload388389390def _state_from_row(row: ConnectorHealth | None) -> SourceState:391 if row is None:392 return SourceState()393 return SourceState(394 status=row.status,395 last_success=_ensure_utc(row.last_success),396 found_last=row.found_last,397 median_found=row.median_found,398 consecutive_failures=row.consecutive_failures or 0,399 checked_at=_ensure_utc(row.checked_at),400 )401402403def _ensure_utc(value: datetime.datetime | None) -> datetime.datetime | None:404 if value is None:405 return None406 if value.tzinfo is None:407 return value.replace(tzinfo=datetime.UTC)408 return value409410411def _apply(412 session: Any,413 rows: dict[str, ConnectorHealth],414 service: str,415 source: str,416 assessment: Assessment,417 now: float,418) -> str:419 """Upsert de la ligne connector_health + alerte sur transition broken/stale."""420 row = rows.get(source)421 previous_status = row.status if row is not None else None422423 checked_at = _utc(now)424 if row is None:425 row = ConnectorHealth(service=service, source=source)426 session.add(row)427 rows[source] = row428 row.checked_at = checked_at429 row.status = assessment.status430 row.last_success = assessment.last_success431 row.found_last = assessment.found_last432 row.median_found = assessment.median_found433 row.consecutive_failures = assessment.consecutive_failures434 row.message = assessment.message435436 if assessment.status in (STATUS_BROKEN, STATUS_STALE) and (437 previous_status != assessment.status438 ):439 alert(440 f"[connecteurs] {service}/{source} : {assessment.status} — "441 f"{assessment.message}"442 )443 return assessment.status444445446def _check_service(service: str, now: float) -> dict[str, Any]:447 """Collecte et classe tous les connecteurs d'un service. Ne lève jamais448 pour cause d'app injoignable (l'échec est un état, pas une exception)."""449 url = _stats_url(service)450 if url is None:451 return {"service": service, "skipped": "aucune SOURCE_URL configurée"}452453 cadence_seconds = EXPECTED_CADENCE_HOURS.get(service, 24) * 3600454 app_stale_after = STALE_FACTOR * cadence_seconds455456 counts = dict.fromkeys(ALL_STATUSES, 0)457458 try:459 payload = _fetch_stats(url)460 except Exception as exc: # httpx, JSON, payload inattendu…461 with session_scope() as session:462 rows = _load_rows(session, service)463 assessment = assess_unreachable(464 _state_from_row(rows.get(APP_SOURCE)), str(exc)465 )466 counts[_apply(session, rows, service, APP_SOURCE, assessment, now)] += 1467 return {"service": service, "url": url, "reachable": False, "counts": counts}468469 by_source = parse_sync_entries(payload)470 app_entry = build_app_entry(service, payload, by_source, now)471 # Marge d'une rotation supplémentaire (+1) : une source qui repasse472 # toutes les ~N h dérive naturellement de quelques dizaines de minutes473 # d'un cycle à l'autre — sans marge, les sources en bord de cycle474 # oscillent entre ok et stale (ex. jobka/lassonde à 2,3 h pour un seuil475 # de 2 h, fabrika en bord de rotation de ~11 h pour un seuil de 24 h).476 source_stale_after = app_stale_after * (rotation_factor(payload, by_source) + 1)477478 with session_scope() as session:479 rows = _load_rows(session, service)480481 assessment = assess_source(482 [app_entry], _state_from_row(rows.get(APP_SOURCE)), now, app_stale_after483 )484 counts[_apply(session, rows, service, APP_SOURCE, assessment, now)] += 1485486 for source, entries in by_source.items():487 assessment = assess_source(488 entries, _state_from_row(rows.get(source)), now, source_stale_after489 )490 counts[_apply(session, rows, service, source, assessment, now)] += 1491492 # Sources connues mais absentes de la fenêtre courante : réévaluer la493 # staleness (le streak et la médiane restent inchangés).494 for source in list(rows):495 if source == APP_SOURCE or source in by_source:496 continue497 assessment = assess_source(498 [], _state_from_row(rows.get(source)), now, source_stale_after499 )500 counts[_apply(session, rows, service, source, assessment, now)] += 1501502 return {503 "service": service,504 "url": url,505 "reachable": True,506 "sources_in_window": len(by_source),507 "counts": counts,508 }509510511def _load_rows(session: Any, service: str) -> dict[str, ConnectorHealth]:512 return {513 row.source: row514 for row in session.execute(515 select(ConnectorHealth).where(ConnectorHealth.service == service)516 ).scalars()517 }518519520def run_connector_health_check(now: float | None = None) -> dict[str, Any]:521 """Point d'entrée du job (toutes les 2 h) : vérifie les 9 apps.522523 L'échec d'une app (réseau, payload) ne fait jamais échouer le job.524 """525 logger = get_logger("apika.monitoring")526 now = now if now is not None else time.time()527 results: list[dict[str, Any]] = []528 for service in SERVICES:529 try:530 results.append(_check_service(service, now))531 except Exception as exc: # filet de sécurité (ex. DB) — jamais de crash532 logger.error(533 "Échec de la supervision d'un service",534 extra={"service": service, "error": str(exc)},535 )536 results.append({"service": service, "error": str(exc)})537538 summary = {539 "checked_at": _utc(now).isoformat(),540 "services": results,541 }542 logger.info("Supervision des connecteurs terminée", extra=summary)543 return summary544545546def main() -> None:547 """Exécution manuelle : ``python -m src.monitoring.connector_health``."""548 import json549550 from src.database.db import init_db551552 init_db()553 print(json.dumps(run_connector_health_check(), ensure_ascii=False, indent=2))554555556if __name__ == "__main__":557 main()558