# ----------------------------------------------------------------------------- # Fabri-Ka — Agrégateur de produits québécois # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # db.py : persistance SQLite — upsert avec hash de contenu, cycle de vie # avec délai de grâce, FTS5 pour la recherche plein-texte. # ----------------------------------------------------------------------------- from __future__ import annotations import json import sqlite3 import time from pathlib import Path from .schema import Product DB_PATH = Path(__file__).resolve().parent.parent / "data" / "fabrika.db" # Nombre de syncs consécutives où un produit doit être absent avant désactivation MISS_GRACE = 2 _SCHEMA = """ CREATE TABLE IF NOT EXISTS stores ( id TEXT PRIMARY KEY, -- domaine canonique name TEXT, url TEXT, platform TEXT, catalog_endpoint TEXT, city TEXT, region TEXT, origin_class TEXT, -- A|B|C|D|E origin_confidence REAL, origin_evidence TEXT, categories TEXT, -- JSON socials TEXT, -- JSON discovery_sources TEXT, -- JSON language TEXT, enabled INTEGER DEFAULT 1, last_sync REAL, last_status TEXT, product_count INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS products ( uid TEXT PRIMARY KEY, store_id TEXT NOT NULL, external_id TEXT NOT NULL, url TEXT, title TEXT, description TEXT, price REAL, price_max REAL, compare_at_price REAL, currency TEXT, images TEXT, -- JSON category TEXT, product_type TEXT, tags TEXT, -- JSON vendor TEXT, available INTEGER, content_hash TEXT, first_seen REAL, last_seen REAL, active INTEGER DEFAULT 1, miss_count INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_products_store ON products(store_id); CREATE INDEX IF NOT EXISTS idx_products_cat ON products(category); CREATE INDEX IF NOT EXISTS idx_products_active ON products(active); CREATE TABLE IF NOT EXISTS sync_log ( ts REAL, store_id TEXT, found INTEGER, added INTEGER, updated INTEGER, removed INTEGER, status TEXT, message TEXT ); CREATE VIRTUAL TABLE IF NOT EXISTS products_fts USING fts5( uid UNINDEXED, title, description, tags, vendor, store_name, tokenize = 'unicode61 remove_diacritics 2' ); """ def connect() -> sqlite3.Connection: DB_PATH.parent.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(DB_PATH, timeout=60) con.execute("PRAGMA busy_timeout=60000") con.row_factory = sqlite3.Row con.executescript(_SCHEMA) return con def upsert_store(con: sqlite3.Connection, s: dict) -> None: con.execute(""" INSERT INTO stores (id, name, url, platform, catalog_endpoint, city, region, origin_class, origin_confidence, origin_evidence, categories, socials, discovery_sources, language, enabled) VALUES (:id,:name,:url,:platform,:catalog_endpoint,:city,:region, :origin_class,:origin_confidence,:origin_evidence, :categories,:socials,:discovery_sources,:language,:enabled) ON CONFLICT(id) DO UPDATE SET name=excluded.name, url=excluded.url, platform=excluded.platform, catalog_endpoint=excluded.catalog_endpoint, city=excluded.city, region=excluded.region, origin_class=excluded.origin_class, origin_confidence=excluded.origin_confidence, origin_evidence=excluded.origin_evidence, categories=excluded.categories, socials=excluded.socials, discovery_sources=excluded.discovery_sources, language=excluded.language, enabled=excluded.enabled """, { "id": s["id"], "name": s.get("name") or s["id"], "url": s.get("url") or f"https://{s['id']}", "platform": s.get("platform") or "", "catalog_endpoint": s.get("catalog_endpoint") or "", "city": s.get("city") or "", "region": s.get("region") or "", "origin_class": s.get("origin_class") or "E", "origin_confidence": s.get("origin_confidence") or 0.0, "origin_evidence": s.get("origin_evidence") or "", "categories": json.dumps(s.get("categories") or [], ensure_ascii=False), "socials": json.dumps(s.get("socials") or [], ensure_ascii=False), "discovery_sources": json.dumps(s.get("discovery_sources") or [], ensure_ascii=False), "language": s.get("language") or "", "enabled": 1 if s.get("enabled", True) else 0, }) def sync_products(con: sqlite3.Connection, store_id: str, store_name: str, products: list[Product]) -> tuple[int, int, int]: """Diff complet d'une boutique : upsert + désactivation avec délai de grâce. Retourne (ajoutés, modifiés, retirés).""" now = time.time() existing = {r["uid"]: (r["content_hash"], r["active"], r["miss_count"]) for r in con.execute("SELECT uid, content_hash, active, miss_count " "FROM products WHERE store_id=?", (store_id,))} seen, added, updated = set(), 0, 0 for p in products: p.finalize() uid, chash = p.uid, p.content_hash() if uid in seen: # doublon dans la même source continue seen.add(uid) row = p.to_row() row.update(images=json.dumps(p.images), tags=json.dumps(p.tags, ensure_ascii=False), content_hash=chash, last_seen=now, available=None if p.available is None else int(p.available)) if uid not in existing: row["first_seen"] = now con.execute("""INSERT INTO products (uid, store_id, external_id, url, title, description, price, price_max, compare_at_price, currency, images, category, product_type, tags, vendor, available, content_hash, first_seen, last_seen, active, miss_count) VALUES (:uid,:store_id,:external_id,:url,:title, :description,:price,:price_max,:compare_at_price,:currency,:images,:category, :product_type,:tags,:vendor,:available,:content_hash,:first_seen,:last_seen,1,0)""", row) con.execute("INSERT INTO products_fts (uid, title, description, tags, vendor, store_name) " "VALUES (?,?,?,?,?,?)", (uid, p.title, p.description, " ".join(p.tags), p.vendor, store_name)) added += 1 else: old_hash, old_active, _ = existing[uid] if old_hash != chash or not old_active: con.execute("""UPDATE products SET url=:url, title=:title, description=:description, price=:price, price_max=:price_max, compare_at_price=:compare_at_price, currency=:currency, images=:images, category=:category, product_type=:product_type, tags=:tags, vendor=:vendor, available=:available, content_hash=:content_hash, last_seen=:last_seen, active=1, miss_count=0 WHERE uid=:uid""", row) con.execute("DELETE FROM products_fts WHERE uid=?", (uid,)) con.execute("INSERT INTO products_fts (uid, title, description, tags, vendor, store_name) " "VALUES (?,?,?,?,?,?)", (uid, p.title, p.description, " ".join(p.tags), p.vendor, store_name)) updated += 1 else: con.execute("UPDATE products SET last_seen=?, miss_count=0 WHERE uid=?", (now, uid)) removed = 0 for uid, (_, active, miss) in existing.items(): if uid in seen or not active: continue if miss + 1 >= MISS_GRACE: con.execute("UPDATE products SET active=0, miss_count=? WHERE uid=?", (miss + 1, uid)) con.execute("DELETE FROM products_fts WHERE uid=?", (uid,)) removed += 1 else: con.execute("UPDATE products SET miss_count=? WHERE uid=?", (miss + 1, uid)) con.execute("UPDATE stores SET last_sync=?, product_count=" "(SELECT COUNT(*) FROM products WHERE store_id=? AND active=1) WHERE id=?", (now, store_id, store_id)) return added, updated, removed def log_sync(con, store_id, found, added, updated, removed, status, message=""): con.execute("INSERT INTO sync_log VALUES (?,?,?,?,?,?,?,?)", (time.time(), store_id, found, added, updated, removed, status, message[:300])) con.execute("UPDATE stores SET last_status=? WHERE id=?", (status, store_id))