Repository Public
1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (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# cache des pages détail, cache de géocodage.7# -----------------------------------------------------------------------------8from __future__ import annotations910import json11import sqlite312import statistics13import time14from pathlib import Path1516from .schema import Listing1718DB_PATH = Path(__file__).resolve().parent.parent / "data" / "louka.db"1920# Nombre d'exécutions consécutives où une annonce doit être absente de la21# source avant d'être désactivée (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 annonces), on alerte et on suspend les retraits.26DRIFT_RATIO = 0.2527DRIFT_MIN_BASE = 828DRIFT_HISTORY = 52930_SCHEMA = """31CREATE TABLE IF NOT EXISTS listings (32 uid TEXT PRIMARY KEY,33 source TEXT NOT NULL,34 external_id TEXT NOT NULL,35 url TEXT,36 title TEXT,37 address TEXT,38 sector TEXT,39 city TEXT,40 unit_type TEXT,41 price REAL,42 price_label TEXT,43 availability TEXT,44 availability_date TEXT,45 area_sqft REAL,46 pets TEXT,47 furnished INTEGER,48 description TEXT,49 amenities TEXT, -- JSON (liste de textes source)50 details TEXT, -- JSON (champs structurés : inclusions, parking…)51 images TEXT, -- JSON52 lat REAL,53 lng REAL,54 geocode_failed INTEGER DEFAULT 0,55 content_hash TEXT,56 first_seen REAL,57 last_seen REAL,58 updated_at REAL,59 miss_count INTEGER DEFAULT 0,60 active INTEGER DEFAULT 161);62CREATE INDEX IF NOT EXISTS idx_listings_source ON listings(source);63CREATE INDEX IF NOT EXISTS idx_listings_city ON listings(city);64CREATE INDEX IF NOT EXISTS idx_listings_active ON listings(active);6566CREATE TABLE IF NOT EXISTS sync_log (67 id INTEGER PRIMARY KEY AUTOINCREMENT,68 source TEXT,69 ts REAL,70 found INTEGER,71 added INTEGER,72 updated INTEGER,73 removed INTEGER,74 ok INTEGER,75 message TEXT,76 stats TEXT -- JSON : taux de champs null, missed, alerte…77);7879CREATE TABLE IF NOT EXISTS detail_cache (80 source TEXT NOT NULL,81 external_id TEXT NOT NULL,82 key TEXT, -- hash du contenu « liste » de l'annonce83 payload TEXT, -- JSON opaque propre au connecteur84 fetched_at REAL,85 PRIMARY KEY (source, external_id)86);8788CREATE TABLE IF NOT EXISTS geocode_cache (89 address TEXT PRIMARY KEY, -- adresse normalisée (clé de cache)90 lat REAL,91 lng REAL,92 provider TEXT,93 failed INTEGER DEFAULT 0,94 ts REAL95);96"""9798# Colonnes ajoutées après la v1 — migration automatique des bases existantes.99_MIGRATIONS = {100 "listings": {101 "availability_date": "TEXT",102 "area_sqft": "REAL",103 "pets": "TEXT",104 "furnished": "INTEGER",105 "details": "TEXT",106 "geocode_failed": "INTEGER DEFAULT 0",107 "miss_count": "INTEGER DEFAULT 0",108 },109 "sync_log": {110 "stats": "TEXT",111 },112}113114115def connect() -> sqlite3.Connection:116 DB_PATH.parent.mkdir(parents=True, exist_ok=True)117 con = sqlite3.connect(DB_PATH)118 con.row_factory = sqlite3.Row119 con.executescript(_SCHEMA)120 for table, cols in _MIGRATIONS.items():121 existing = {r["name"] for r in con.execute(f"PRAGMA table_info({table})")}122 for col, decl in cols.items():123 if col not in existing:124 con.execute(f"ALTER TABLE {table} ADD COLUMN {col} {decl}")125 con.commit()126 return con127128129# ---------------------------------------------------------------------------130# Synchronisation d'une source131# ---------------------------------------------------------------------------132133def _drift_alert(con: sqlite3.Connection, source: str, found: int,134 null_price_rate: float) -> str | None:135 """Détecte une dérive du connecteur (chute du volume ou des prix extraits).136137 Retourne un message d'alerte, ou None si tout est normal.138 """139 hist = con.execute(140 "SELECT found, stats FROM sync_log WHERE source=? AND ok=1"141 " ORDER BY ts DESC LIMIT ?", (source, DRIFT_HISTORY)).fetchall()142 if len(hist) < 3:143 return None144 med_found = statistics.median(r["found"] for r in hist)145 if med_found >= DRIFT_MIN_BASE and found <= DRIFT_RATIO * med_found:146 return (f"dérive: {found} annonce(s) trouvée(s) contre une médiane de "147 f"{med_found:.0f} — retraits suspendus, vérifier le connecteur")148 if found >= DRIFT_MIN_BASE and null_price_rate >= 0.8:149 rates = []150 for r in hist:151 try:152 rates.append(json.loads(r["stats"] or "{}")["null_price_rate"])153 except (KeyError, ValueError, TypeError):154 continue155 if rates and statistics.median(rates) <= 0.3:156 return (f"dérive: {null_price_rate:.0%} des annonces sans prix "157 f"(habituellement {statistics.median(rates):.0%}) — "158 "le format de la source a probablement changé")159 return None160161162def sync_source(con: sqlite3.Connection, source: str,163 listings: list[Listing]) -> dict:164 """Synchronise les annonces d'une source.165166 - nouvelle annonce -> insertion167 - annonce modifiée -> mise à jour (comparaison de content_hash)168 - annonce disparue -> miss_count += 1, puis active=0 après MISS_GRACE169 exécutions consécutives (délai de grâce)170 - dérive détectée -> alerte consignée, retraits suspendus171 """172 now = time.time()173 added = updated = 0174 seen_uids = set()175176 n = len(listings)177 null_price = sum(1 for l in listings if l.price is None)178 null_addr = sum(1 for l in listings if not l.address)179 null_price_rate = round(null_price / n, 3) if n else 0.0180181 alert = _drift_alert(con, source, n, null_price_rate)182183 for lst in listings:184 seen_uids.add(lst.uid)185 h = lst.content_hash()186 row = con.execute("SELECT content_hash FROM listings WHERE uid=?",187 (lst.uid,)).fetchone()188 params = dict(189 uid=lst.uid, source=lst.source, external_id=lst.external_id,190 url=lst.url, title=lst.title, address=lst.address,191 sector=lst.sector, city=lst.city, unit_type=lst.unit_type,192 price=lst.price, price_label=lst.price_label,193 availability=lst.availability,194 availability_date=lst.availability_date,195 area_sqft=lst.area_sqft, pets=lst.pets,196 furnished=(None if lst.furnished is None else int(lst.furnished)),197 description=lst.description,198 amenities=json.dumps(lst.amenities, ensure_ascii=False),199 details=json.dumps(lst.details, ensure_ascii=False),200 images=json.dumps(lst.images, ensure_ascii=False),201 lat=lst.lat, lng=lst.lng, content_hash=h, now=now,202 )203 if row is None:204 con.execute(205 """INSERT INTO listings (uid, source, external_id, url, title,206 address, sector, city, unit_type, price, price_label,207 availability, availability_date, area_sqft, pets, furnished,208 description, amenities, details, images, lat, lng,209 content_hash, first_seen, last_seen, updated_at,210 miss_count, active)211 VALUES (:uid,:source,:external_id,:url,:title,:address,212 :sector,:city,:unit_type,:price,:price_label,:availability,213 :availability_date,:area_sqft,:pets,:furnished,214 :description,:amenities,:details,:images,:lat,:lng,215 :content_hash,:now,:now,:now,0,1)""", params)216 added += 1217 elif row["content_hash"] != h:218 # COALESCE : ne jamais écraser des coordonnées géocodées par null219 con.execute(220 """UPDATE listings SET url=:url, title=:title,221 address=:address, sector=:sector, city=:city,222 unit_type=:unit_type, price=:price,223 price_label=:price_label, availability=:availability,224 availability_date=:availability_date,225 area_sqft=:area_sqft, pets=:pets, furnished=:furnished,226 description=:description, amenities=:amenities,227 details=:details, images=:images,228 lat=COALESCE(:lat, lat), lng=COALESCE(:lng, lng),229 content_hash=:content_hash, last_seen=:now,230 updated_at=:now, miss_count=0, active=1231 WHERE uid=:uid""", params)232 updated += 1233 else:234 con.execute(235 "UPDATE listings SET last_seen=?, miss_count=0, active=1 WHERE uid=?",236 (now, lst.uid))237238 # Annonces de cette source qui n'apparaissent plus : délai de grâce,239 # puis désactivation. Suspendu si une dérive est détectée.240 removed = missed = 0241 if not alert:242 for r in con.execute(243 "SELECT uid, miss_count FROM listings WHERE source=? AND active=1",244 (source,)).fetchall():245 if r["uid"] in seen_uids:246 continue247 missed += 1248 if r["miss_count"] + 1 >= MISS_GRACE:249 con.execute(250 "UPDATE listings SET active=0, miss_count=?, updated_at=?"251 " WHERE uid=?", (r["miss_count"] + 1, now, r["uid"]))252 removed += 1253 else:254 con.execute("UPDATE listings SET miss_count=miss_count+1 WHERE uid=?",255 (r["uid"],))256257 stats = {258 "null_price_rate": null_price_rate,259 "null_address_rate": round(null_addr / n, 3) if n else 0.0,260 "missed": missed,261 }262 if alert:263 stats["alert"] = alert264 con.execute(265 "INSERT INTO sync_log (source, ts, found, added, updated, removed, ok,"266 " message, stats) VALUES (?,?,?,?,?,?,1,?,?)",267 (source, now, n, added, updated, removed, alert or "ok",268 json.dumps(stats, ensure_ascii=False)))269 con.commit()270 out = {"source": source, "found": n, "added": added,271 "updated": updated, "removed": removed}272 if alert:273 out["alert"] = alert274 return out275276277def log_failure(con: sqlite3.Connection, source: str, message: str) -> None:278 con.execute(279 "INSERT INTO sync_log (source, ts, found, added, updated, removed, ok, message)"280 " VALUES (?,?,0,0,0,0,0,?)", (source, time.time(), message))281 con.commit()282283284# ---------------------------------------------------------------------------285# Cache des pages détail (« détail si nouveau/modifié »)286# ---------------------------------------------------------------------------287288def get_cached_detail(con: sqlite3.Connection, source: str,289 external_id: str, key: str) -> dict | None:290 """Payload détail mis en cache si la clé (hash liste) n'a pas changé."""291 row = con.execute(292 "SELECT key, payload FROM detail_cache WHERE source=? AND external_id=?",293 (source, external_id)).fetchone()294 if row and row["key"] == key and row["payload"]:295 try:296 return json.loads(row["payload"])297 except ValueError:298 return None299 return None300301302def put_cached_detail(con: sqlite3.Connection, source: str,303 external_id: str, key: str, payload: dict) -> None:304 con.execute(305 "INSERT INTO detail_cache (source, external_id, key, payload, fetched_at)"306 " VALUES (?,?,?,?,?)"307 " ON CONFLICT(source, external_id) DO UPDATE SET"308 " key=excluded.key, payload=excluded.payload, fetched_at=excluded.fetched_at",309 (source, external_id, key, json.dumps(payload, ensure_ascii=False),310 time.time()))311 con.commit()312