# ----------------------------------------------------------------------------- # Lou-Ka — Location court terme # db.py : persistance SQLite SÉPARÉE (data/louka_ct.db) — upsert avec détection # de changements, cycle de vie avec délai de grâce, cache détail. # ----------------------------------------------------------------------------- from __future__ import annotations import json import sqlite3 import time from pathlib import Path from .schema import StListing DB_PATH = Path(__file__).resolve().parent.parent.parent / "data" / "louka_ct.db" MISS_GRACE = 2 # absences consécutives tolérées avant désactivation _SCHEMA = """ CREATE TABLE IF NOT EXISTS st_listings ( uid TEXT PRIMARY KEY, source TEXT NOT NULL, external_id TEXT NOT NULL, url TEXT, title TEXT, property_type TEXT, address TEXT, city TEXT, region TEXT, price_night REAL, price_label TEXT, capacity REAL, bedrooms REAL, beds REAL, bathrooms REAL, pets TEXT, citq TEXT, rating REAL, reviews INTEGER, description TEXT, amenities TEXT, -- JSON details TEXT, -- JSON images TEXT, -- JSON lat REAL, lng REAL, content_hash TEXT, first_seen REAL, last_seen REAL, updated_at REAL, miss_count INTEGER DEFAULT 0, active INTEGER DEFAULT 1 ); CREATE INDEX IF NOT EXISTS idx_st_source ON st_listings(source); CREATE INDEX IF NOT EXISTS idx_st_region ON st_listings(region); CREATE INDEX IF NOT EXISTS idx_st_active ON st_listings(active); CREATE INDEX IF NOT EXISTS idx_st_geo ON st_listings(lat, lng); CREATE TABLE IF NOT EXISTS st_sync_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT, ts REAL, found INTEGER, added INTEGER, updated INTEGER, removed INTEGER, ok INTEGER, message TEXT ); CREATE TABLE IF NOT EXISTS st_detail_cache ( source TEXT NOT NULL, external_id TEXT NOT NULL, key TEXT, payload TEXT, fetched_at REAL, PRIMARY KEY (source, external_id) ); """ def connect() -> sqlite3.Connection: DB_PATH.parent.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(DB_PATH) con.row_factory = sqlite3.Row con.executescript(_SCHEMA) con.commit() con.execute("PRAGMA journal_mode=WAL") con.execute("PRAGMA busy_timeout=30000") return con def sync_source(con: sqlite3.Connection, source: str, listings: list[StListing]) -> dict: """Synchronise les hébergements d'une source (ajout / màj / retrait).""" now = time.time() added = updated = 0 seen = set() for lst in listings: seen.add(lst.uid) h = lst.content_hash() row = con.execute("SELECT content_hash FROM st_listings WHERE uid=?", (lst.uid,)).fetchone() params = dict( uid=lst.uid, source=lst.source, external_id=str(lst.external_id), url=lst.url, title=lst.title, property_type=lst.property_type, address=lst.address, city=lst.city, region=lst.region, price_night=lst.price_night, price_label=lst.price_label, capacity=lst.capacity, bedrooms=lst.bedrooms, beds=lst.beds, bathrooms=lst.bathrooms, pets=lst.pets, citq=lst.citq, rating=lst.rating, reviews=lst.reviews, description=lst.description, amenities=json.dumps(lst.amenities, ensure_ascii=False), details=json.dumps(lst.details, ensure_ascii=False), images=json.dumps(lst.images, ensure_ascii=False), lat=lst.lat, lng=lst.lng, content_hash=h, now=now, ) if row is None: con.execute( """INSERT INTO st_listings (uid, source, external_id, url, title, property_type, address, city, region, price_night, price_label, capacity, bedrooms, beds, bathrooms, pets, citq, rating, reviews, description, amenities, details, images, lat, lng, content_hash, first_seen, last_seen, updated_at, miss_count, active) VALUES (:uid,:source,:external_id,:url,:title, :property_type,:address,:city,:region,:price_night, :price_label,:capacity,:bedrooms,:beds,:bathrooms,:pets, :citq,:rating,:reviews,:description,:amenities,:details, :images,:lat,:lng,:content_hash,:now,:now,:now,0,1)""", params) added += 1 elif row["content_hash"] != h: con.execute( """UPDATE st_listings SET url=:url, title=:title, property_type=:property_type, address=:address, city=:city, region=:region, price_night=:price_night, price_label=:price_label, capacity=:capacity, bedrooms=:bedrooms, beds=:beds, bathrooms=:bathrooms, pets=:pets, citq=:citq, rating=:rating, reviews=:reviews, description=:description, amenities=:amenities, details=:details, images=:images, lat=COALESCE(:lat, lat), lng=COALESCE(:lng, lng), content_hash=:content_hash, last_seen=:now, updated_at=:now, miss_count=0, active=1 WHERE uid=:uid""", params) updated += 1 else: con.execute("UPDATE st_listings SET last_seen=?, miss_count=0," " active=1 WHERE uid=?", (now, lst.uid)) removed = 0 for r in con.execute("SELECT uid, miss_count FROM st_listings" " WHERE source=? AND active=1", (source,)).fetchall(): if r["uid"] in seen: continue if r["miss_count"] + 1 >= MISS_GRACE: con.execute("UPDATE st_listings SET active=0, miss_count=?," " updated_at=? WHERE uid=?", (r["miss_count"] + 1, now, r["uid"])) removed += 1 else: con.execute("UPDATE st_listings SET miss_count=miss_count+1" " WHERE uid=?", (r["uid"],)) con.execute("INSERT INTO st_sync_log (source, ts, found, added, updated," " removed, ok, message) VALUES (?,?,?,?,?,?,1,'ok')", (source, now, len(listings), added, updated, removed)) con.commit() return {"source": source, "found": len(listings), "added": added, "updated": updated, "removed": removed} def log_failure(con: sqlite3.Connection, source: str, message: str) -> None: con.execute("INSERT INTO st_sync_log (source, ts, found, added, updated," " removed, ok, message) VALUES (?,?,0,0,0,0,0,?)", (source, time.time(), message)) con.commit() def get_cached_detail(con: sqlite3.Connection, source: str, external_id: str, key: str) -> dict | None: row = con.execute("SELECT key, payload FROM st_detail_cache" " WHERE source=? AND external_id=?", (source, external_id)).fetchone() if row and row["key"] == key and row["payload"]: try: return json.loads(row["payload"]) except ValueError: return None return None def put_cached_detail(con: sqlite3.Connection, source: str, external_id: str, key: str, payload: dict) -> None: con.execute("INSERT INTO st_detail_cache (source, external_id, key," " payload, fetched_at) VALUES (?,?,?,?,?)" " ON CONFLICT(source, external_id) DO UPDATE SET" " key=excluded.key, payload=excluded.payload," " fetched_at=excluded.fetched_at", (source, external_id, key, json.dumps(payload, ensure_ascii=False), time.time())) con.commit()