spb/fabri-ka

1# -----------------------------------------------------------------------------2# Fabri-Ka — Agrégateur de produits québécois3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# db.py : persistance SQLite — upsert avec hash de contenu, cycle de vie5#         avec délai de grâce, FTS5 pour la recherche plein-texte.6# -----------------------------------------------------------------------------7from __future__ import annotations89import json10import sqlite311import time12from pathlib import Path1314from .schema import Product1516DB_PATH = Path(__file__).resolve().parent.parent / "data" / "fabrika.db"1718# Nombre de syncs consécutives où un produit doit être absent avant désactivation19MISS_GRACE = 22021_SCHEMA = """22CREATE TABLE IF NOT EXISTS stores (23    id            TEXT PRIMARY KEY,      -- domaine canonique24    name          TEXT,25    url           TEXT,26    platform      TEXT,27    catalog_endpoint TEXT,28    city          TEXT,29    region        TEXT,30    origin_class  TEXT,                  -- A|B|C|D|E31    origin_confidence REAL,32    origin_evidence TEXT,33    categories    TEXT,                  -- JSON34    socials       TEXT,                  -- JSON35    discovery_sources TEXT,              -- JSON36    language      TEXT,37    enabled       INTEGER DEFAULT 1,38    last_sync     REAL,39    last_status   TEXT,40    product_count INTEGER DEFAULT 041);42CREATE TABLE IF NOT EXISTS products (43    uid          TEXT PRIMARY KEY,44    store_id     TEXT NOT NULL,45    external_id  TEXT NOT NULL,46    url          TEXT,47    title        TEXT,48    description  TEXT,49    price        REAL,50    price_max    REAL,51    compare_at_price REAL,52    currency     TEXT,53    images       TEXT,   -- JSON54    category     TEXT,55    product_type TEXT,56    tags         TEXT,   -- JSON57    vendor       TEXT,58    available    INTEGER,59    details      TEXT,   -- JSON : avis, dimensions, matériaux, variantes…60    content_hash TEXT,61    first_seen   REAL,62    last_seen    REAL,63    active       INTEGER DEFAULT 1,64    miss_count   INTEGER DEFAULT 065);66CREATE INDEX IF NOT EXISTS idx_products_store ON products(store_id);67CREATE INDEX IF NOT EXISTS idx_products_cat ON products(category);68CREATE INDEX IF NOT EXISTS idx_products_active ON products(active);69CREATE INDEX IF NOT EXISTS idx_products_pub_cover ON products(category, store_id, price, first_seen) WHERE active=1 AND listing_status='published';70CREATE INDEX IF NOT EXISTS idx_products_fs_cover ON products(first_seen, store_id, category, price);71CREATE TABLE IF NOT EXISTS sync_log (72    ts REAL, store_id TEXT, found INTEGER, added INTEGER,73    updated INTEGER, removed INTEGER, status TEXT, message TEXT74);75CREATE VIRTUAL TABLE IF NOT EXISTS products_fts USING fts5(76    uid UNINDEXED, title, description, tags, vendor, store_name,77    tokenize = 'unicode61 remove_diacritics 2'78);79"""808182def connect() -> sqlite3.Connection:83    DB_PATH.parent.mkdir(parents=True, exist_ok=True)84    # L'initialisation du schéma (executescript) peut être refusée pendant85    # qu'un autre processus écrit (sync massive) : on retente avec délai86    # croissant plutôt que d'échouer immédiatement.87    last_exc = None88    for attempt in range(8):89        con = sqlite3.connect(DB_PATH, timeout=60)90        con.execute("PRAGMA busy_timeout=60000")91        con.row_factory = sqlite3.Row92        try:93            con.executescript(_SCHEMA)94            # migrations additives (colonnes ajoutées après la création)95            cols = {r[1] for r in con.execute("PRAGMA table_info(products)")}96            if "details" not in cols:97                con.execute("ALTER TABLE products ADD COLUMN details TEXT")98            if "listing_status" not in cols:99                con.execute("ALTER TABLE products ADD COLUMN listing_status TEXT DEFAULT 'published'")100                con.execute("CREATE INDEX IF NOT EXISTS idx_products_listing ON products(listing_status)")101            if "price_on_request" not in cols:102                con.execute("ALTER TABLE products ADD COLUMN price_on_request INTEGER DEFAULT 0")103            scols = {r[1] for r in con.execute("PRAGMA table_info(stores)")}104            for col, typ in (("shipping_info", "TEXT"), ("email", "TEXT"),105                             ("phone", "TEXT"), ("postal_prefix", "TEXT"),106                             ("lat", "REAL"), ("lng", "REAL"),107                             ("store_kind", "TEXT"),108                             # vague 4 : certifications structurées (JSON) +109                             # détection des ateliers présumés fermés110                             ("certifications", "TEXT"),111                             ("fail_streak", "INTEGER DEFAULT 0"),112                             ("presumed_closed", "INTEGER DEFAULT 0")):113                if col not in scols:114                    con.execute(f"ALTER TABLE stores ADD COLUMN {col} {typ}")115            # auto-inscription des artisans (« Inscrire mon atelier ») :116            # quarantaine de validation, jamais publiée sans geste manuel117            con.execute("""CREATE TABLE IF NOT EXISTS inscriptions (118                id INTEGER PRIMARY KEY AUTOINCREMENT,119                ts REAL, ip TEXT, name TEXT, metier TEXT, description TEXT,120                address TEXT, website TEXT, email TEXT, phone TEXT,121                socials TEXT, photos TEXT,122                status TEXT DEFAULT 'pending',   -- pending|approved|rejected123                reviewed_at REAL, review_note TEXT)""")124            return con125        except sqlite3.OperationalError as exc:126            last_exc = exc127            con.close()128            time.sleep(1.5 * (attempt + 1))129    raise last_exc130131132def upsert_store(con: sqlite3.Connection, s: dict) -> None:133    con.execute("""134        INSERT INTO stores (id, name, url, platform, catalog_endpoint, city, region,135                            origin_class, origin_confidence, origin_evidence,136                            categories, socials, discovery_sources, language, enabled,137                            phone, postal_prefix, store_kind)138        VALUES (:id,:name,:url,:platform,:catalog_endpoint,:city,:region,139                :origin_class,:origin_confidence,:origin_evidence,140                :categories,:socials,:discovery_sources,:language,:enabled,141                :phone,:postal_prefix,:store_kind)142        ON CONFLICT(id) DO UPDATE SET143            name=excluded.name, url=excluded.url, platform=excluded.platform,144            catalog_endpoint=excluded.catalog_endpoint,145            -- city vient du géocodage (pas du registre) : ne jamais l'écraser par du vide146            city=CASE WHEN excluded.city<>'' THEN excluded.city ELSE stores.city END,147            region=excluded.region, origin_class=excluded.origin_class,148            origin_confidence=excluded.origin_confidence,149            origin_evidence=excluded.origin_evidence,150            categories=excluded.categories, socials=excluded.socials,151            discovery_sources=excluded.discovery_sources,152            language=excluded.language, enabled=excluded.enabled,153            phone=CASE WHEN excluded.phone<>'' THEN excluded.phone ELSE stores.phone END,154            postal_prefix=CASE WHEN excluded.postal_prefix<>'' THEN excluded.postal_prefix155                               ELSE stores.postal_prefix END,156            store_kind=CASE WHEN excluded.store_kind<>'' THEN excluded.store_kind157                            ELSE stores.store_kind END158    """, {159        "id": s["id"], "name": s.get("name") or s["id"], "url": s.get("url") or f"https://{s['id']}",160        "platform": s.get("platform") or "", "catalog_endpoint": s.get("catalog_endpoint") or "",161        "city": s.get("city") or "", "region": s.get("region") or "",162        "origin_class": s.get("origin_class") or "E",163        "origin_confidence": s.get("origin_confidence") or 0.0,164        "origin_evidence": s.get("origin_evidence") or "",165        "categories": json.dumps(s.get("categories") or [], ensure_ascii=False),166        "socials": json.dumps(s.get("socials") or [], ensure_ascii=False),167        "discovery_sources": json.dumps(s.get("discovery_sources") or [], ensure_ascii=False),168        "language": s.get("language") or "", "enabled": 1 if s.get("enabled", True) else 0,169        "phone": s.get("phone") or "", "postal_prefix": s.get("postal_prefix") or "",170        "store_kind": s.get("store_kind") or "",171    })172173174def sync_products(con: sqlite3.Connection, store_id: str, store_name: str,175                  products: list[Product]) -> tuple[int, int, int]:176    """Diff complet d'une boutique : upsert + désactivation avec délai de grâce.177178    Retourne (ajoutés, modifiés, retirés)."""179    now = time.time()180    existing = {r["uid"]: (r["content_hash"], r["active"], r["miss_count"])181                for r in con.execute("SELECT uid, content_hash, active, miss_count "182                                     "FROM products WHERE store_id=?", (store_id,))}183    seen, added, updated = set(), 0, 0184    # FTS5 : uid est UNINDEXED -> chaque DELETE WHERE uid=? balaie toute la185    # table. On accumule et on supprime par lots (voir fin de fonction).186    fts_del, fts_add = [], []187    for p in products:188        p.finalize()189        uid, chash = p.uid, p.content_hash()190        if uid in seen:      # doublon dans la même source191            continue192        seen.add(uid)193        row = p.to_row()194        row.update(images=json.dumps(p.images), tags=json.dumps(p.tags, ensure_ascii=False),195                   details=json.dumps(p.details, ensure_ascii=False) if p.details else None,196                   content_hash=chash, last_seen=now,197                   available=None if p.available is None else int(p.available))198        if uid not in existing:199            row["first_seen"] = now200            con.execute("""INSERT INTO products (uid, store_id, external_id, url, title,201                description, price, price_max, compare_at_price, currency, images, category,202                product_type, tags, vendor, available, details, content_hash, first_seen,203                last_seen, active, miss_count, listing_status, price_on_request)204                VALUES (:uid,:store_id,:external_id,:url,:title,205                :description,:price,:price_max,:compare_at_price,:currency,:images,:category,206                :product_type,:tags,:vendor,:available,:details,:content_hash,:first_seen,207                :last_seen,1,0,:listing_status,:price_on_request)""", row)208            con.execute("INSERT INTO products_fts (uid, title, description, tags, vendor, store_name) "209                        "VALUES (?,?,?,?,?,?)",210                        (uid, p.title, p.description, " ".join(p.tags), p.vendor, store_name))211            added += 1212        else:213            old_hash, old_active, _ = existing[uid]214            if old_hash != chash or not old_active:215                con.execute("""UPDATE products SET url=:url, title=:title, description=:description,216                    price=:price, price_max=:price_max, compare_at_price=:compare_at_price,217                    currency=:currency, images=:images, category=:category,218                    product_type=:product_type, tags=:tags, vendor=:vendor,219                    available=:available, details=:details, content_hash=:content_hash,220                    last_seen=:last_seen, listing_status=:listing_status,221                    price_on_request=:price_on_request,222                    active=1, miss_count=0 WHERE uid=:uid""", row)223                fts_del.append(uid)224                fts_add.append((uid, p.title, p.description, " ".join(p.tags),225                                p.vendor, store_name))226                updated += 1227            else:228                con.execute("UPDATE products SET last_seen=?, miss_count=0 WHERE uid=?", (now, uid))229230    removed = 0231    for uid, (_, active, miss) in existing.items():232        if uid in seen or not active:233            continue234        if miss + 1 >= MISS_GRACE:235            con.execute("UPDATE products SET active=0, miss_count=? WHERE uid=?", (miss + 1, uid))236            fts_del.append(uid)237            removed += 1238        else:239            con.execute("UPDATE products SET miss_count=? WHERE uid=?", (miss + 1, uid))240241    # purge/réinsertion FTS par lots (1 balayage par lot de 500 au lieu de N)242    for i in range(0, len(fts_del), 500):243        chunk = fts_del[i:i + 500]244        con.execute("DELETE FROM products_fts WHERE uid IN (%s)"245                    % ",".join("?" * len(chunk)), chunk)246    if fts_add:247        con.executemany("INSERT INTO products_fts (uid, title, description, tags, "248                        "vendor, store_name) VALUES (?,?,?,?,?,?)", fts_add)249250    con.execute("UPDATE stores SET last_sync=?, product_count="251                "(SELECT COUNT(*) FROM products WHERE store_id=? AND active=1 "252                "AND listing_status='published') WHERE id=?",253                (now, store_id, store_id))254    return added, updated, removed255256257def log_sync(con, store_id, found, added, updated, removed, status, message=""):258    con.execute("INSERT INTO sync_log VALUES (?,?,?,?,?,?,?,?)",259                (time.time(), store_id, found, added, updated, removed, status, message[:300]))260    con.execute("UPDATE stores SET last_status=? WHERE id=?", (status, store_id))261