spb/crea-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   creaka/db.py4# Desc:   Stockage SQLite — fiches créateurs canoniques, comptes rattachés,5#         journal des synchronisations (fraîcheur §13-14, opt-out §15)6# ==============================================================================7"""Base de données Créa-Ka (SQLite, WAL).89- `creators` : une ligne = une fiche canonique (document JSON + colonnes10  indexées pour la recherche).11- `accounts` : un compte (plateforme+handle) pointe vers SA fiche — c'est12  l'index qui permet la déduplication inter-sources (§12.2).13- `sync_log` : statistiques par passage de connecteur (§16 logging structuré).1415Politique de grâce (§13) : on ne supprime jamais brutalement — `last_seen`16horodate chaque passage ; un créateur disparu passe `inactive`, un opt-out17passe `opted_out` (masqué, jamais ré-agrégé).18"""19from __future__ import annotations2021import json22import sqlite323import sys24import time25from dataclasses import asdict26from datetime import datetime, timedelta, timezone27from pathlib import Path2829from . import ethics30from .dedup import merge_creators31from .identity import needs_review32from .normalize import slugify33from .schema import Creator, PlatformAccount, now_iso3435ROOT = Path(__file__).resolve().parent.parent36DB_PATH = ROOT / "data" / "creaka.db"3738_SCHEMA = """39CREATE TABLE IF NOT EXISTS creators (40  id TEXT PRIMARY KEY,41  display_name TEXT NOT NULL,42  status TEXT NOT NULL DEFAULT 'active',43  is_minor INTEGER NOT NULL DEFAULT 0,44  region TEXT, city TEXT,45  niches TEXT, languages TEXT,46  creator_type TEXT, primary_platform TEXT,47  audience_tier TEXT, total_reach INTEGER,48  bio TEXT,49  first_seen TEXT, last_seen TEXT, updated_at TEXT,50  doc TEXT NOT NULL51);52CREATE TABLE IF NOT EXISTS accounts (53  platform TEXT NOT NULL,54  handle TEXT NOT NULL,55  creator_id TEXT NOT NULL REFERENCES creators(id) ON DELETE CASCADE,56  url TEXT NOT NULL,57  followers INTEGER,58  verified INTEGER,59  confidence REAL NOT NULL,60  signal TEXT,61  needs_review INTEGER NOT NULL DEFAULT 0,62  last_checked TEXT,63  metrics TEXT,64  PRIMARY KEY (platform, handle)65);66CREATE INDEX IF NOT EXISTS idx_accounts_creator ON accounts(creator_id);67CREATE INDEX IF NOT EXISTS idx_creators_status ON creators(status);68CREATE INDEX IF NOT EXISTS idx_creators_tier ON creators(audience_tier);69CREATE TABLE IF NOT EXISTS snapshots (70  day TEXT NOT NULL,71  platform TEXT NOT NULL,72  handle TEXT NOT NULL,73  creator_id TEXT NOT NULL,74  followers INTEGER,75  engagement REAL,76  PRIMARY KEY (day, platform, handle)77);78CREATE INDEX IF NOT EXISTS idx_snapshots_creator ON snapshots(creator_id, day);79CREATE TABLE IF NOT EXISTS sync_log (80  id INTEGER PRIMARY KEY AUTOINCREMENT,81  ts TEXT NOT NULL,82  source TEXT NOT NULL,83  creators INTEGER, accounts INTEGER, avg_confidence REAL,84  added INTEGER, updated INTEGER, errors INTEGER, seconds REAL,85  alert TEXT86);87"""888990def connect(path: Path | str = DB_PATH) -> sqlite3.Connection:91    Path(path).parent.mkdir(parents=True, exist_ok=True)92    con = sqlite3.connect(path, check_same_thread=False)93    con.row_factory = sqlite3.Row94    con.execute("PRAGMA journal_mode=WAL")95    con.execute("PRAGMA foreign_keys=ON")96    con.executescript(_SCHEMA)97    # migrations légères (bases existantes)98    try:99        con.execute("ALTER TABLE accounts ADD COLUMN metrics TEXT")100    except sqlite3.OperationalError:101        pass102    return con103104105# --- (dé)sérialisation ----------------------------------------------------------106107def creator_to_doc(cr: Creator) -> str:108    return json.dumps(asdict(cr), ensure_ascii=False)109110111def creator_from_row(row: sqlite3.Row) -> Creator:112    doc = json.loads(row["doc"])113    doc["platforms"] = [PlatformAccount(**a) for a in doc.get("platforms", [])]114    cr = Creator(**doc)115    return cr116117118def _public_dict(row: sqlite3.Row, accounts: list[sqlite3.Row]) -> dict:119    """Fiche publique (API) : comptes « à vérifier » EXCLUS (§12.1)."""120    doc = json.loads(row["doc"])121    doc["id"] = row["id"]122    doc["first_seen"] = row["first_seen"]123    doc["last_seen"] = row["last_seen"]124    doc["updated_at"] = row["updated_at"]125    doc["platforms"] = [126        {"platform": a["platform"], "handle": a["handle"], "url": a["url"],127         "followers": a["followers"],128         "verified": bool(a["verified"]) if a["verified"] is not None else None,129         "confidence": a["confidence"], "last_checked": a["last_checked"],130         "metrics": json.loads(a["metrics"]) if a["metrics"] else {}}131        for a in accounts if not a["needs_review"]]132    # champs internes / sensibles jamais exposés tels quels133    doc.pop("source", None)134    doc.pop("external_id", None)135    doc.pop("legal_name", None)136    return doc137138139# --- écriture -------------------------------------------------------------------140141def _unique_id(con: sqlite3.Connection, base: str) -> str:142    cid, n = base, 2143    while con.execute("SELECT 1 FROM creators WHERE id=?", (cid,)).fetchone():144        cid = f"{base}-{n}"145        n += 1146    return cid147148149def _write_creator(con: sqlite3.Connection, cid: str, cr: Creator,150                   first_seen: str, ts: str) -> None:151    # comptes retirés de la fiche : jamais en silence (§13 — historique) ;152    # merge_accounts fait l'union, donc une perte ici est anormale → journalisée153    old_keys = {f"{r['platform']}:{r['handle']}" for r in con.execute(154        "SELECT platform, handle FROM accounts WHERE creator_id=?", (cid,))}155    lost = old_keys - {a.key for a in cr.platforms}156    if lost:157        print(f"[crea-ka] ⚠ {cid} : compte(s) retiré(s) de la fiche : "158              f"{', '.join(sorted(lost))}", file=sys.stderr)159    con.execute(160        """INSERT INTO creators (id, display_name, status, is_minor, region, city,161             niches, languages, creator_type, primary_platform, audience_tier,162             total_reach, bio, first_seen, last_seen, updated_at, doc)163           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)164           ON CONFLICT(id) DO UPDATE SET165             display_name=excluded.display_name, status=excluded.status,166             is_minor=excluded.is_minor, region=excluded.region, city=excluded.city,167             niches=excluded.niches, languages=excluded.languages,168             creator_type=excluded.creator_type,169             primary_platform=excluded.primary_platform,170             audience_tier=excluded.audience_tier, total_reach=excluded.total_reach,171             bio=excluded.bio, last_seen=excluded.last_seen,172             updated_at=excluded.updated_at, doc=excluded.doc""",173        (cid, cr.display_name, cr.status, int(cr.is_minor), cr.region, cr.city,174         ",".join(cr.niches), ",".join(cr.languages), cr.creator_type,175         cr.primary_platform, cr.audience_tier, cr.total_reach, cr.bio,176         first_seen, ts, ts, creator_to_doc(cr)))177    con.execute("DELETE FROM accounts WHERE creator_id=?", (cid,))178    for acc in cr.platforms:179        con.execute(180            """INSERT INTO accounts (platform, handle, creator_id, url, followers,181                 verified, confidence, signal, needs_review, last_checked, metrics)182               VALUES (?,?,?,?,?,?,?,?,?,?,?)183               ON CONFLICT(platform, handle) DO UPDATE SET184                 creator_id=excluded.creator_id, url=excluded.url,185                 followers=excluded.followers, verified=excluded.verified,186                 confidence=excluded.confidence, signal=excluded.signal,187                 needs_review=excluded.needs_review,188                 last_checked=excluded.last_checked, metrics=excluded.metrics""",189            (acc.platform, acc.handle, cid, acc.url, acc.followers,190             None if acc.verified is None else int(acc.verified),191             acc.confidence, acc.signal, int(needs_review(acc)), acc.last_checked,192             json.dumps(acc.metrics, ensure_ascii=False) if acc.metrics else None))193        # historisation (§13-14) : un point par jour et par compte — matière194        # première des courbes de croissance et des insights (7 j / 30 j)195        if acc.followers:196            con.execute(197                """INSERT INTO snapshots (day, platform, handle, creator_id,198                     followers, engagement)199                   VALUES (?,?,?,?,?,?)200                   ON CONFLICT(day, platform, handle) DO UPDATE SET201                     creator_id=excluded.creator_id,202                     followers=excluded.followers,203                     engagement=COALESCE(excluded.engagement,204                                         snapshots.engagement)""",205                (ts[:10], acc.platform, acc.handle, cid, acc.followers,206                 (acc.metrics or {}).get("engagement_rate_pct")))207208209def _find_existing(con: sqlite3.Connection, cr: Creator) -> str | None:210    """Fiche existante possédant déjà un des comptes entrants.211212    Même plateforme + même handle = MÊME compte : la table accounts n'admet213    qu'un propriétaire par compte (UNIQUE platform,handle). Le rapprochement se214    fait donc sur l'identité exacte du compte, sans condition de confiance —215    exiger `confidence >= STRONG_THRESHOLD` des deux côtés (version d'avant le216    2026-08-20) faisait rater les fiches à signal faible (mention 0.60, ex.217    snowball-ig) : sync_source créait alors une SECONDE fiche et l'upsert des218    comptes lui réassignait le compte, laissant l'originale orpheline219    (691 doublons constatés). Ce n'est PAS une fusion sur le nom (§12.2) :220    l'égalité stricte plateforme+handle identifie le compte lui-même.221    """222    for acc in cr.platforms:223        row = con.execute(224            "SELECT creator_id FROM accounts WHERE platform=? AND handle=?",225            (acc.platform, acc.handle)).fetchone()226        if row:227            return row["creator_id"]228    return None229230231def sync_source(con: sqlite3.Connection, source_id: str,232                creators: list[Creator], *, started: float | None = None,233                errors: int = 0) -> dict:234    """Synchronise le lot d'une source : ajouts, fusions, mises à jour, opt-out.235236    `started` (time.time() du début du passage) → durée réelle persistée dans237    sync_log.seconds ; `errors` → erreurs non bloquantes signalées par le238    connecteur (§16 logging structuré).239    """240    ts = now_iso()241    added = updated = skipped_optout = 0242    confs: list[float] = []243    for cr in creators:244        cr = ethics.scrub(cr)245        # opt-out : jamais ré-agrégé (§15) ; si une fiche existe, la masquer246        if ethics.is_opted_out(cr):247            skipped_optout += 1248            existing = _find_existing(con, cr)249            if existing:250                con.execute("UPDATE creators SET status='opted_out', updated_at=? "251                            "WHERE id=?", (ts, existing))252            continue253        confs.extend(a.confidence or 0 for a in cr.platforms)254        existing_id = _find_existing(con, cr)255        if existing_id:256            row = con.execute("SELECT * FROM creators WHERE id=?",257                              (existing_id,)).fetchone()258            current = creator_from_row(row)259            if current.status == "opted_out":260                continue  # masqué : on ne ré-agrège pas261            # _find_existing garantit un compte IDENTIQUE (plateforme+handle)262            # partagé : c'est le même compte, la fusion est toujours sûre —263            # pas une fusion sur le nom (§12.2)264            merged = merge_creators(current, cr)265            _write_creator(con, existing_id, merged, row["first_seen"], ts)266            updated += 1267            continue268        cid = _unique_id(con, slugify(cr.display_name))269        _write_creator(con, cid, cr, ts, ts)270        added += 1271    con.commit()272    n_acc = sum(len(c.platforms) for c in creators)273    stats = {"source": source_id, "creators": len(creators), "accounts": n_acc,274             "added": added, "updated": updated, "optout_skipped": skipped_optout,275             "avg_confidence": round(sum(confs) / len(confs), 3) if confs else None}276    alert = None277    if not creators:278        alert = "0 créateur retourné — source possiblement bloquée"279        stats["alert"] = alert280    seconds = round(time.time() - started, 1) if started else 0.0281    stats["seconds"] = seconds282    if errors:283        stats["errors"] = errors284    con.execute(285        "INSERT INTO sync_log (ts, source, creators, accounts, avg_confidence,"286        " added, updated, errors, seconds, alert) VALUES (?,?,?,?,?,?,?,?,?,?)",287        (ts, source_id, len(creators), n_acc, stats["avg_confidence"],288         added, updated, errors, seconds, alert))289    con.commit()290    return stats291292293def archive_missing(con: sqlite3.Connection, *, hours: int = 72) -> int:294    """Archivage des disparus (§13, politique de grâce).295296    Un créateur qu'AUCUNE source de découverte n'a revu depuis ~3 passages297    complets (cadence quotidienne §14 → 72 h) passe `status='inactive'`.298    Réversible : dès qu'une source le revoit, la fusion (sync_source →299    _write_creator) réécrit son statut actif et son `last_seen`.300    Les opt-out ne sont JAMAIS touchés (status != 'active').301    """302    cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)) \303        .strftime("%Y-%m-%dT%H:%M:%SZ")304    ts = now_iso()305    cur = con.execute(306        "UPDATE creators SET status='inactive', updated_at=? "307        "WHERE status='active' AND last_seen IS NOT NULL AND last_seen<?",308        (ts, cutoff))309    con.commit()310    if cur.rowcount:311        print(f"[crea-ka] archivage §13 : {cur.rowcount} fiche(s) non revue(s) "312              f"depuis {hours} h → inactive (réversible au retour)")313    return cur.rowcount314315316def log_failure(con: sqlite3.Connection, source_id: str, message: str) -> None:317    con.execute(318        "INSERT INTO sync_log (ts, source, creators, accounts, errors, alert)"319        " VALUES (?,?,0,0,1,?)", (now_iso(), source_id, message[:500]))320    con.commit()321322323def apply_optout(con: sqlite3.Connection, *, name: str | None = None,324                 account: str | None = None) -> int:325    """Masque immédiatement les fiches correspondant à une demande de retrait."""326    ids: set[str] = set()327    if account and ":" in account:328        platform, handle = account.lower().split(":", 1)329        for row in con.execute(330                "SELECT creator_id FROM accounts WHERE platform=? AND handle=?",331                (platform, handle.lstrip("@"))):332            ids.add(row["creator_id"])333    if name:334        from .dedup import name_key335        for row in con.execute("SELECT id, display_name FROM creators"):336            if name_key(row["display_name"]) == name_key(name):337                ids.add(row["id"])338    ts = now_iso()339    for cid in ids:340        con.execute("UPDATE creators SET status='opted_out', updated_at=? WHERE id=?",341                    (ts, cid))342    con.commit()343    return len(ids)344345346# --- lecture (API) ----------------------------------------------------------------347348_TIERS = ("nano", "micro", "macro", "mega")349350351def search(con: sqlite3.Connection, *, q: str = "", niche: str = "",352           region: str = "", langue: str = "", plateforme: str = "",353           tier: str = "", sort: str = "reach", limit: int = 60,354           offset: int = 0) -> dict:355    """Recherche annuaire : filtres niche/région/langue/plateforme/taille + texte.356357    Les fiches `opted_out` sont TOUJOURS exclues ; les fiches de mineurs sont358    exclues de l'annuaire public (prudence extrême, §15).359    """360    where = ["status='active'", "is_minor=0"]361    params: list = []362    if q:363        where.append("(lower(display_name) LIKE ? OR lower(bio) LIKE ?)")364        needle = f"%{q.lower()}%"365        params += [needle, needle]366    if niche:367        where.append("(','||niches||',') LIKE ?")368        params.append(f"%,{niche},%")369    if region:370        where.append("region=?")371        params.append(region)372    if langue:373        where.append("(','||languages||',') LIKE ?")374        params.append(f"%,{langue},%")375    if tier in _TIERS:376        where.append("audience_tier=?")377        params.append(tier)378    if plateforme:379        where.append("id IN (SELECT creator_id FROM accounts WHERE platform=? "380                     "AND needs_review=0)")381        params.append(plateforme)382    order = {"reach": "total_reach IS NULL, total_reach DESC",383             "nom": "display_name COLLATE NOCASE ASC",384             "recent": "updated_at DESC"}.get(sort, "total_reach IS NULL, total_reach DESC")385    sql_where = " AND ".join(where)386    total = con.execute(f"SELECT COUNT(*) c FROM creators WHERE {sql_where}",387                        params).fetchone()["c"]388    rows = con.execute(389        f"SELECT * FROM creators WHERE {sql_where} ORDER BY {order} LIMIT ? OFFSET ?",390        params + [max(1, min(200, limit)), max(0, offset)]).fetchall()391    items = []392    for row in rows:393        accounts = con.execute(394            "SELECT * FROM accounts WHERE creator_id=? ORDER BY followers DESC",395            (row["id"],)).fetchall()396        items.append(_public_dict(row, accounts))397    return {"total": total, "count": len(items), "offset": offset, "items": items}398399400def follower_history(con: sqlite3.Connection, cid: str,401                     days: int = 90) -> dict:402    """Historique d'audience d'une fiche (snapshots quotidiens, §13-14).403404    Retourne les séries par plateforme (jour → abonnés) + la série totale405    (somme des plateformes connues ce jour-là), prêtes pour une sparkline.406    """407    cutoff = (datetime.now(timezone.utc) - timedelta(days=days)) \408        .strftime("%Y-%m-%d")409    rows = con.execute(410        "SELECT day, platform, handle, followers FROM snapshots "411        "WHERE creator_id=? AND day>=? ORDER BY day", (cid, cutoff)).fetchall()412    by_platform: dict[str, dict[str, int]] = {}413    for r in rows:414        by_platform.setdefault(r["platform"], {})[r["day"]] = r["followers"]415    total: dict[str, int] = {}416    for series in by_platform.values():417        # report de la dernière valeur connue pour que la somme quotidienne418        # ne s'effondre pas quand une plateforme n'a pas de point ce jour-là419        last = None420        for day in sorted({d for s in by_platform.values() for d in s}):421            last = series.get(day, last)422            if last is not None:423                total[day] = total.get(day, 0) + last424    return {425        "platforms": {p: [{"day": d, "followers": f}426                          for d, f in sorted(s.items())]427                      for p, s in by_platform.items()},428        "total": [{"day": d, "followers": f} for d, f in sorted(total.items())],429    }430431432def get_creator(con: sqlite3.Connection, cid: str) -> dict | None:433    row = con.execute(434        "SELECT * FROM creators WHERE id=? AND status='active' AND is_minor=0",435        (cid,)).fetchone()436    if not row:437        return None438    accounts = con.execute(439        "SELECT * FROM accounts WHERE creator_id=? ORDER BY followers DESC",440        (row["id"],)).fetchall()441    return _public_dict(row, accounts)442443444def stats(con: sqlite3.Connection) -> dict:445    """Statistiques d'ensemble pour l'accueil de l'annuaire."""446    base = "FROM creators WHERE status='active' AND is_minor=0"447    n = con.execute(f"SELECT COUNT(*) c {base}").fetchone()["c"]448    n_acc = con.execute(449        "SELECT COUNT(*) c FROM accounts a JOIN creators c2 ON c2.id=a.creator_id"450        " WHERE a.needs_review=0 AND c2.status='active' AND c2.is_minor=0"451    ).fetchone()["c"]452    by_platform = {r["platform"]: r["c"] for r in con.execute(453        "SELECT a.platform, COUNT(*) c FROM accounts a JOIN creators c2 "454        "ON c2.id=a.creator_id WHERE a.needs_review=0 AND c2.status='active' "455        "AND c2.is_minor=0 GROUP BY a.platform ORDER BY c DESC")}456    by_tier = {r["audience_tier"]: r["c"] for r in con.execute(457        f"SELECT audience_tier, COUNT(*) c {base} GROUP BY audience_tier")}458    niches: dict[str, int] = {}459    for row in con.execute(f"SELECT niches {base}"):460        for nch in (row["niches"] or "").split(","):461            if nch:462                niches[nch] = niches.get(nch, 0) + 1463    regions: dict[str, int] = {}464    for row in con.execute(f"SELECT region {base} AND region IS NOT NULL"):465        regions[row["region"]] = regions.get(row["region"], 0) + 1466    last_sync = con.execute(467        "SELECT ts FROM sync_log ORDER BY id DESC LIMIT 1").fetchone()468    return {"creators": n, "accounts": n_acc, "by_platform": by_platform,469            "by_tier": by_tier,470            "by_niche": dict(sorted(niches.items(), key=lambda x: -x[1])),471            "by_region": dict(sorted(regions.items(), key=lambda x: -x[1])),472            "last_sync": last_sync["ts"] if last_sync else None}473