spb/sorti-ka

1# -----------------------------------------------------------------------------2# Sorti-Ka — Agrégateur de sorties & événements (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# db.py : persistance SQLite — upsert avec détection de changements,5#         cycle de vie avec délai de grâce (2 syncs), détection de dérive,6#         déduplication inter-sources par empreinte (patron Lou-Ka : louka/db.py)7# -----------------------------------------------------------------------------8from __future__ import annotations910import json11import sqlite312import statistics13import time14from pathlib import Path1516from .schema import Event1718DB_PATH = Path(__file__).resolve().parent.parent / "data" / "sortika.db"1920# Nombre d'exécutions consécutives où un événement doit être absent de la21# source avant d'être désactivé (délai de grâce contre les ratés ponctuels).22MISS_GRACE = 22324# Dérive : si une source retourne <= DRIFT_RATIO × sa médiane historique25# (médiane >= DRIFT_MIN_BASE événements), on alerte et on suspend les retraits.26DRIFT_RATIO = 0.2527DRIFT_MIN_BASE = 828DRIFT_HISTORY = 52930_SCHEMA = """31CREATE TABLE IF NOT EXISTS events (32    uid           TEXT PRIMARY KEY,33    source        TEXT NOT NULL,34    external_id   TEXT NOT NULL,35    url           TEXT,36    title         TEXT,37    description   TEXT,38    categories    TEXT,   -- JSON (taxonomie canonique)39    raw_categories TEXT,  -- JSON (libellés source)40    audience      TEXT,41    venue         TEXT,42    address       TEXT,43    city          TEXT,44    region        TEXT,45    tourist_region TEXT,46    postal_code   TEXT,47    lat           REAL,48    lng           REAL,49    start_date    TEXT,50    end_date      TEXT,51    is_free       INTEGER,          -- 1 gratuit / 0 payant / NULL inconnu52    price_min     REAL,53    price_label   TEXT,54    organizer     TEXT,55    website       TEXT,56    image         TEXT,57    dedup_key     TEXT,58    content_hash  TEXT,59    first_seen    REAL,60    last_seen     REAL,61    updated_at    REAL,62    miss_count    INTEGER DEFAULT 0,63    active        INTEGER DEFAULT 164);65CREATE INDEX IF NOT EXISTS idx_events_source  ON events(source);66CREATE INDEX IF NOT EXISTS idx_events_active  ON events(active);67CREATE INDEX IF NOT EXISTS idx_events_region  ON events(region);68CREATE INDEX IF NOT EXISTS idx_events_start   ON events(start_date);69CREATE INDEX IF NOT EXISTS idx_events_dedup   ON events(dedup_key);7071CREATE TABLE IF NOT EXISTS users (72    id          INTEGER PRIMARY KEY AUTOINCREMENT,73    hub_sub     TEXT UNIQUE,          -- clé stable du hub KA ("ka:<sub>")74    ka_id       TEXT UNIQUE,          -- identifiant membre du groupe (hub)75    email       TEXT,76    name        TEXT,77    picture     TEXT,78    created_at  REAL,79    last_login  REAL80);8182CREATE TABLE IF NOT EXISTS sync_log (83    id        INTEGER PRIMARY KEY AUTOINCREMENT,84    source    TEXT NOT NULL,85    ts        REAL NOT NULL,86    fetched   INTEGER,87    added     INTEGER,88    updated   INTEGER,89    removed   INTEGER,90    error     TEXT91);92"""939495def connect(path: Path | str | None = None) -> sqlite3.Connection:96    p = Path(path) if path else DB_PATH97    p.parent.mkdir(parents=True, exist_ok=True)98    con = sqlite3.connect(p)99    con.row_factory = sqlite3.Row100    # plusieurs connecteurs peuvent synchroniser en parallèle (backfill) :101    # attendre le verrou plutôt que d'échouer, et journal WAL pour les lecteurs102    con.execute("PRAGMA busy_timeout=60000")103    try:104        con.execute("PRAGMA journal_mode=WAL")105    except sqlite3.OperationalError:106        pass107    con.executescript(_SCHEMA)108    return con109110111def _event_row(ev: Event, now: float) -> dict:112    return {113        "uid": ev.uid, "source": ev.source, "external_id": ev.external_id,114        "url": ev.url, "title": ev.title, "description": ev.description,115        "categories": json.dumps(ev.categories, ensure_ascii=False),116        "raw_categories": json.dumps(ev.raw_categories, ensure_ascii=False),117        "audience": ev.audience, "venue": ev.venue, "address": ev.address,118        "city": ev.city, "region": ev.region, "tourist_region": ev.tourist_region,119        "postal_code": ev.postal_code, "lat": ev.lat, "lng": ev.lng,120        "start_date": ev.start_date, "end_date": ev.end_date,121        "is_free": (None if ev.is_free is None else int(ev.is_free)),122        "price_min": ev.price_min, "price_label": ev.price_label,123        "organizer": ev.organizer, "website": ev.website, "image": ev.image,124        "dedup_key": ev.dedup_key(), "content_hash": ev.content_hash(),125        "now": now,126    }127128129def sync_source(con: sqlite3.Connection, source: str, events: list[Event]) -> dict:130    """Synchronise une source : ajouts / mises à jour / retraits avec grâce."""131    now = time.time()132    stats = {"source": source, "fetched": len(events),133             "added": 0, "updated": 0, "unchanged": 0, "removed": 0}134    alert = None135136    # Dérive : volume anormalement bas → on suspend les retraits.137    hist = [r["fetched"] for r in con.execute(138        "SELECT fetched FROM sync_log WHERE source=? AND error IS NULL "139        "ORDER BY ts DESC LIMIT ?", (source, DRIFT_HISTORY))]140    if hist:141        med = statistics.median(hist)142        if med >= DRIFT_MIN_BASE and len(events) <= med * DRIFT_RATIO:143            alert = (f"dérive : {len(events)} événements reçus vs médiane {med:.0f} "144                     f"— retraits suspendus (connecteur probablement cassé)")145146    seen: set[str] = set()147    for ev in events:148        row = _event_row(ev, now)149        seen.add(ev.uid)150        cur = con.execute("SELECT content_hash FROM events WHERE uid=?", (ev.uid,))151        old = cur.fetchone()152        if old is None:153            con.execute(154                """INSERT INTO events (uid, source, external_id, url, title,155                   description, categories, raw_categories, audience, venue,156                   address, city, region, tourist_region, postal_code, lat, lng,157                   start_date, end_date, is_free, price_min, price_label,158                   organizer, website, image, dedup_key, content_hash,159                   first_seen, last_seen, updated_at, miss_count, active)160                   VALUES (:uid,:source,:external_id,:url,:title,:description,161                   :categories,:raw_categories,:audience,:venue,:address,:city,162                   :region,:tourist_region,:postal_code,:lat,:lng,:start_date,163                   :end_date,:is_free,:price_min,:price_label,:organizer,164                   :website,:image,:dedup_key,:content_hash,165                   :now,:now,:now,0,1)""", row)166            stats["added"] += 1167        elif old["content_hash"] != row["content_hash"]:168            con.execute(169                """UPDATE events SET url=:url, title=:title,170                   description=:description, categories=:categories,171                   raw_categories=:raw_categories, audience=:audience,172                   venue=:venue, address=:address, city=:city, region=:region,173                   tourist_region=:tourist_region, postal_code=:postal_code,174                   lat=:lat, lng=:lng, start_date=:start_date,175                   end_date=:end_date, is_free=:is_free, price_min=:price_min,176                   price_label=:price_label, organizer=:organizer,177                   website=:website, image=:image, dedup_key=:dedup_key,178                   content_hash=:content_hash, last_seen=:now, updated_at=:now,179                   miss_count=0, active=1 WHERE uid=:uid""", row)180            stats["updated"] += 1181        else:182            con.execute("UPDATE events SET last_seen=?, miss_count=0, active=1 "183                        "WHERE uid=?", (now, ev.uid))184            stats["unchanged"] += 1185186    # Retraits (avec délai de grâce), sauf en cas de dérive détectée.187    if not alert:188        for r in con.execute(189                "SELECT uid, miss_count FROM events WHERE source=? AND active=1",190                (source,)):191            if r["uid"] in seen:192                continue193            miss = r["miss_count"] + 1194            if miss >= MISS_GRACE:195                con.execute("UPDATE events SET active=0, miss_count=?, "196                            "updated_at=? WHERE uid=?", (miss, now, r["uid"]))197                stats["removed"] += 1198            else:199                con.execute("UPDATE events SET miss_count=? WHERE uid=?",200                            (miss, r["uid"]))201202    con.execute("INSERT INTO sync_log (source, ts, fetched, added, updated, "203                "removed, error) VALUES (?,?,?,?,?,?,NULL)",204                (source, now, stats["fetched"], stats["added"],205                 stats["updated"], stats["removed"]))206    con.commit()207    if alert:208        stats["alert"] = alert209    return stats210211212def log_failure(con: sqlite3.Connection, source: str, error: str) -> None:213    con.execute("INSERT INTO sync_log (source, ts, fetched, added, updated, "214                "removed, error) VALUES (?,?,0,0,0,0,?)",215                (source, time.time(), error[:500]))216    con.commit()217