# ----------------------------------------------------------------------------- # Home-Ka — US real-estate aggregator (Groupe KA) # Author: Simon-Pierre Boucher — contact@spboucher.ai # db.py : SQLite persistence — PROPERTY vs LISTING separation, upsert with # change detection, lifecycle with grace period (2 syncs), drift # detection, price history, detail-page cache, sources registry, # brokerage pipeline. Ported from immo-ka/db.py. # ----------------------------------------------------------------------------- from __future__ import annotations import json import sqlite3 import statistics import time from pathlib import Path from .schema import Listing DB_PATH = Path(__file__).resolve().parent.parent / "data" / "homeka.db" # Consecutive syncs a listing must be absent from its source before being # deactivated (grace period against one-off fetch failures). MISS_GRACE = 2 # Drift: if a source returns <= DRIFT_RATIO × its historical median # (median >= DRIFT_MIN_BASE listings), alert and suspend removals. DRIFT_RATIO = 0.25 DRIFT_MIN_BASE = 8 DRIFT_HISTORY = 5 _SCHEMA = """ CREATE TABLE IF NOT EXISTS properties ( id INTEGER PRIMARY KEY AUTOINCREMENT, addr_key TEXT UNIQUE, -- normalized address key (propertymatch.py) apn TEXT, -- assessor parcel number when known street_address TEXT, unit TEXT, city TEXT, state TEXT, -- 2-letter USPS zip_code TEXT, county TEXT, property_type TEXT, year_built INTEGER, living_area_sqft REAL, lot_size_sqft REAL, lat REAL, lng REAL, details TEXT, -- JSON (assessor/public-record enrichment) first_seen REAL, last_seen REAL ); CREATE INDEX IF NOT EXISTS idx_prop_state_city ON properties(state, city); CREATE INDEX IF NOT EXISTS idx_prop_zip ON properties(zip_code); CREATE INDEX IF NOT EXISTS idx_prop_apn ON properties(apn); CREATE TABLE IF NOT EXISTS listings ( uid TEXT PRIMARY KEY, -- source:external_id source TEXT NOT NULL, external_id TEXT NOT NULL, property_id INTEGER, -- -> properties.id (property matching) url TEXT, title TEXT, street_address TEXT, unit TEXT, city TEXT, state TEXT, zip_code TEXT, county TEXT, property_type TEXT, property_subtype TEXT, list_price REAL, price_label TEXT, bedrooms INTEGER, bathrooms_full INTEGER, bathrooms_half INTEGER, bathrooms REAL, living_area_sqft REAL, lot_size_sqft REAL, year_built INTEGER, apn TEXT, mls_id TEXT, mls_name TEXT, status TEXT DEFAULT 'active', listed_at TEXT, brokerage_name TEXT, office_name TEXT, agent_name TEXT, agent_phone TEXT, agent_email TEXT, description TEXT, features TEXT, -- JSON (list of source texts) details TEXT, -- JSON (structured fields, RESO passthrough) images TEXT, -- JSON lat REAL, lng REAL, geocode_failed INTEGER DEFAULT 0, content_hash TEXT, first_seen REAL, last_seen REAL, updated_at REAL, miss_count INTEGER DEFAULT 0, active INTEGER DEFAULT 1, dup_hidden INTEGER DEFAULT 0, dup_of TEXT, published INTEGER DEFAULT 1, -- quality gate (homeka/quality.py) quality REAL ); CREATE INDEX IF NOT EXISTS idx_listings_source ON listings(source); CREATE INDEX IF NOT EXISTS idx_listings_city ON listings(city); CREATE INDEX IF NOT EXISTS idx_listings_state ON listings(state); CREATE INDEX IF NOT EXISTS idx_listings_type ON listings(property_type); CREATE INDEX IF NOT EXISTS idx_listings_active ON listings(active); CREATE INDEX IF NOT EXISTS idx_listings_extid ON listings(external_id); CREATE INDEX IF NOT EXISTS idx_listings_mls ON listings(mls_id); CREATE INDEX IF NOT EXISTS idx_listings_prop ON listings(property_id); CREATE INDEX IF NOT EXISTS idx_listings_geo ON listings(lat, lng); CREATE INDEX IF NOT EXISTS idx_listings_duphidden ON listings(dup_hidden); CREATE INDEX IF NOT EXISTS idx_listings_dupof ON listings(dup_of); -- Registered sources: one row = one connector INSTANCE (config-driven). -- connector_type selects the family (reso, rets, xml, json, jsonld, csv, -- sftp, arcgis, custom); config is the family-specific JSON (base URL, -- credentials env-var names, field mapping, throttle...). Custom-coded -- connectors (connectors/custom_broker/*.py) register themselves by id -- and need no row, but one can exist to carry metadata. CREATE TABLE IF NOT EXISTS sources ( id TEXT PRIMARY KEY, name TEXT, connector_type TEXT, config TEXT, -- JSON enabled INTEGER DEFAULT 1, authority INTEGER DEFAULT 2, -- 0=direct feed, 1=partner API, 2=site, 3=classifieds brokerage_id INTEGER, -- -> brokerages.id when source came from a partnership states TEXT, -- JSON list of covered states notes TEXT, created REAL, updated REAL ); -- US brokerage pipeline: discovery -> inspection -> scoring -> partnership. CREATE TABLE IF NOT EXISTS brokerages ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, website TEXT UNIQUE, states TEXT, -- JSON list cities TEXT, -- JSON list estimated_agents INTEGER, estimated_listings INTEGER, mls_affiliations TEXT, -- JSON list idx_present INTEGER, -- 1/0/NULL (unknown until inspected) idx_provider TEXT, -- kvCORE, iHomefinder, IDX Broker, ... reso_detected INTEGER, possible_feed_type TEXT, -- reso | rets | json-api | xml | csv | idx-partner | crawl contact_page TEXT, partnership_contact TEXT, technical_contact TEXT, feed_probability_score REAL, -- 0-100 priority_score REAL, -- 0-100 (size × feed probability × coverage) partnership_status TEXT DEFAULT 'prospect', -- prospect | to-contact | contacted | in-discussion | feed-received | live | declined source_id TEXT, -- -> sources.id once a connector exists last_inspected REAL, inspect_error TEXT, evidence TEXT, -- JSON (fingerprints, URLs, raw hints) created REAL, updated REAL ); CREATE INDEX IF NOT EXISTS idx_brok_status ON brokerages(partnership_status); CREATE INDEX IF NOT EXISTS idx_brok_priority ON brokerages(priority_score); CREATE TABLE IF NOT EXISTS sync_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT, ts REAL, found INTEGER, added INTEGER, updated INTEGER, removed INTEGER, ok INTEGER, message TEXT, stats TEXT -- JSON: null-field rates, missed, alert... ); CREATE INDEX IF NOT EXISTS idx_synclog_source ON sync_log(source, ts); CREATE TABLE IF NOT EXISTS detail_cache ( source TEXT NOT NULL, external_id TEXT NOT NULL, key TEXT, payload TEXT, fetched_at REAL, PRIMARY KEY (source, external_id) ); CREATE TABLE IF NOT EXISTS price_log ( uid TEXT NOT NULL, ts REAL NOT NULL, price REAL ); CREATE INDEX IF NOT EXISTS idx_price_log_uid ON price_log(uid); CREATE TABLE IF NOT EXISTS geocode_cache ( address TEXT PRIMARY KEY, lat REAL, lng REAL, provider TEXT, failed INTEGER DEFAULT 0, ts REAL ); """ _SCHEMA_READY = False # schema/migrations run ONCE per process (write lock) def _init_schema(con: sqlite3.Connection) -> None: con.executescript(_SCHEMA) con.commit() def connect() -> sqlite3.Connection: global _SCHEMA_READY DB_PATH.parent.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(DB_PATH, timeout=60) con.row_factory = sqlite3.Row # WAL + busy_timeout: concurrent access (watcher + web) without locks. con.execute("PRAGMA journal_mode=WAL") con.execute("PRAGMA busy_timeout=120000") con.execute("PRAGMA synchronous=NORMAL") if not _SCHEMA_READY: _init_schema(con) _SCHEMA_READY = True return con # --------------------------------------------------------------------------- # Source sync (ported from immo-ka: change detection, grace, drift) # --------------------------------------------------------------------------- def _drift_alert(con: sqlite3.Connection, source: str, found: int, null_price_rate: float) -> str | None: """Connector drift detection (volume or extracted-price collapse).""" hist = con.execute( "SELECT found, stats FROM sync_log WHERE source=? AND ok=1" " ORDER BY ts DESC LIMIT ?", (source, DRIFT_HISTORY)).fetchall() if len(hist) < 3: return None med_found = statistics.median(r["found"] for r in hist) if med_found >= DRIFT_MIN_BASE and found <= DRIFT_RATIO * med_found: return (f"drift: {found} listing(s) found vs a median of " f"{med_found:.0f} — removals suspended, check the connector") if found >= DRIFT_MIN_BASE and null_price_rate >= 0.8: rates = [] for r in hist: try: rates.append(json.loads(r["stats"] or "{}")["null_price_rate"]) except (KeyError, ValueError, TypeError): continue if rates and statistics.median(rates) <= 0.3: return (f"drift: {null_price_rate:.0%} of listings without a price " f"(usually {statistics.median(rates):.0%}) — " "the source format probably changed") return None _LISTING_COLS = ( "uid", "source", "external_id", "property_id", "url", "title", "street_address", "unit", "city", "state", "zip_code", "county", "property_type", "property_subtype", "list_price", "price_label", "bedrooms", "bathrooms_full", "bathrooms_half", "bathrooms", "living_area_sqft", "lot_size_sqft", "year_built", "apn", "mls_id", "mls_name", "status", "listed_at", "brokerage_name", "office_name", "agent_name", "agent_phone", "agent_email", "description", "features", "details", "images", "lat", "lng", "content_hash", ) def _listing_params(lst: Listing, property_id: int | None, h: str, now: float) -> dict: return dict( uid=lst.uid, source=lst.source, external_id=lst.external_id, property_id=property_id, url=lst.url, title=lst.title, street_address=lst.street_address, unit=lst.unit, city=lst.city, state=lst.state, zip_code=lst.zip_code, county=lst.county, property_type=lst.property_type, property_subtype=lst.property_subtype, list_price=lst.list_price, price_label=lst.price_label, bedrooms=lst.bedrooms, bathrooms_full=lst.bathrooms_full, bathrooms_half=lst.bathrooms_half, bathrooms=lst.bathrooms, living_area_sqft=lst.living_area_sqft, lot_size_sqft=lst.lot_size_sqft, year_built=lst.year_built, apn=lst.apn, mls_id=lst.mls_id, mls_name=lst.mls_name, status=lst.status, listed_at=lst.listed_at, brokerage_name=lst.brokerage_name, office_name=lst.office_name, agent_name=lst.agent_name, agent_phone=lst.agent_phone, agent_email=lst.agent_email, description=lst.description, features=json.dumps(lst.features, ensure_ascii=False), details=json.dumps(lst.details, ensure_ascii=False, default=str), images=json.dumps(lst.images, ensure_ascii=False), lat=lst.lat, lng=lst.lng, content_hash=h, now=now, ) def sync_source(con: sqlite3.Connection, source: str, listings: list[Listing]) -> dict: """Synchronize one source's listings. - new listing -> insert (+ property match/create) - changed listing -> update (content_hash comparison) - vanished listing -> miss_count += 1, then active=0 after MISS_GRACE consecutive runs (sold or withdrawn); the matched PROPERTY row is kept forever - drift detected -> alert logged, removals suspended """ from . import propertymatch now = time.time() added = updated = 0 seen_uids = set() n = len(listings) null_price = sum(1 for l in listings if l.list_price is None) null_addr = sum(1 for l in listings if not l.street_address) null_price_rate = round(null_price / n, 3) if n else 0.0 alert = _drift_alert(con, source, n, null_price_rate) for lst in listings: seen_uids.add(lst.uid) h = lst.content_hash() row = con.execute( "SELECT content_hash, list_price, property_id FROM listings WHERE uid=?", (lst.uid,)).fetchone() property_id = row["property_id"] if row else None if property_id is None: property_id = propertymatch.match_or_create(con, lst, now) params = _listing_params(lst, property_id, h, now) if row is None: cols = ", ".join(_LISTING_COLS) named = ", ".join(f":{c}" for c in _LISTING_COLS) con.execute( f"INSERT INTO listings ({cols}, first_seen, last_seen," f" updated_at, miss_count, active)" f" VALUES ({named}, :now, :now, :now, 0, 1)", params) if lst.list_price is not None: con.execute("INSERT INTO price_log (uid, ts, price) VALUES (?,?,?)", (lst.uid, now, lst.list_price)) added += 1 elif row["content_hash"] != h: # COALESCE: never null-overwrite geocoded coordinates or # enrichment-filled year/areas when the list feed lacks them con.execute( """UPDATE listings SET url=:url, title=:title, street_address=:street_address, unit=:unit, city=:city, state=:state, zip_code=:zip_code, county=:county, property_type=:property_type, property_subtype=:property_subtype, list_price=:list_price, price_label=:price_label, bedrooms=:bedrooms, bathrooms_full=:bathrooms_full, bathrooms_half=:bathrooms_half, bathrooms=:bathrooms, living_area_sqft=COALESCE(:living_area_sqft, living_area_sqft), lot_size_sqft=COALESCE(:lot_size_sqft, lot_size_sqft), year_built=COALESCE(:year_built, year_built), apn=:apn, mls_id=:mls_id, mls_name=:mls_name, status=:status, listed_at=:listed_at, brokerage_name=:brokerage_name, office_name=:office_name, agent_name=:agent_name, agent_phone=:agent_phone, agent_email=:agent_email, description=:description, features=:features, details=:details, images=:images, lat=COALESCE(:lat, lat), lng=COALESCE(:lng, lng), property_id=:property_id, content_hash=:content_hash, last_seen=:now, updated_at=:now, miss_count=0, active=1 WHERE uid=:uid""", params) if lst.list_price != row["list_price"]: con.execute("INSERT INTO price_log (uid, ts, price) VALUES (?,?,?)", (lst.uid, now, lst.list_price)) updated += 1 else: con.execute( "UPDATE listings SET last_seen=?, miss_count=0, active=1 WHERE uid=?", (now, lst.uid)) if property_id: propertymatch.refresh_property(con, property_id, lst, now) # Listings of this source no longer present: grace period, then # deactivation (sold/withdrawn). Suspended when drift is detected. # The PROPERTY row keeps existing — only the listing goes off-market. removed = missed = 0 if not alert: for r in con.execute( "SELECT uid, miss_count FROM listings WHERE source=? AND active=1", (source,)).fetchall(): if r["uid"] in seen_uids: continue missed += 1 if r["miss_count"] + 1 >= MISS_GRACE: con.execute( "UPDATE listings SET active=0, status='withdrawn'," " miss_count=?, updated_at=? WHERE uid=?", (r["miss_count"] + 1, now, r["uid"])) removed += 1 else: con.execute("UPDATE listings SET miss_count=miss_count+1 WHERE uid=?", (r["uid"],)) stats = { "null_price_rate": null_price_rate, "null_address_rate": round(null_addr / n, 3) if n else 0.0, "missed": missed, } if alert: stats["alert"] = alert con.execute( "INSERT INTO sync_log (source, ts, found, added, updated, removed, ok," " message, stats) VALUES (?,?,?,?,?,?,1,?,?)", (source, now, n, added, updated, removed, alert or "ok", json.dumps(stats, ensure_ascii=False))) con.commit() out = {"source": source, "found": n, "added": added, "updated": updated, "removed": removed} if alert: out["alert"] = alert return out def log_failure(con: sqlite3.Connection, source: str, message: str) -> None: con.execute( "INSERT INTO sync_log (source, ts, found, added, updated, removed, ok, message)" " VALUES (?,?,0,0,0,0,0,?)", (source, time.time(), message)) con.commit() # --------------------------------------------------------------------------- # Deduplication — precomputed like immo-ka (dup_hidden/dup_of columns). # US rules: # 1) same MLS number (mls_id + state) published by several sources # 2) same PROPERTY (property_id) listed by several sources at ±1% price # The most authoritative source wins (sources.authority: direct feed first), # then the record with an agent, then the smallest uid. # --------------------------------------------------------------------------- def _authority_map(con: sqlite3.Connection) -> dict[str, int]: return {r["id"]: r["authority"] if r["authority"] is not None else 2 for r in con.execute("SELECT id, authority FROM sources")} def refresh_dedup(con: sqlite3.Connection) -> int: auth = _authority_map(con) con.execute("UPDATE listings SET dup_hidden=0, dup_of=NULL") groups: dict[tuple, list] = {} for r in con.execute( "SELECT uid, source, mls_id, state, property_id, list_price," " agent_name, brokerage_name FROM listings WHERE active=1"): keys = [] if r["mls_id"]: keys.append(("mls", r["mls_id"], r["state"] or "")) if r["property_id"] and r["list_price"]: keys.append(("prop", r["property_id"])) anonymous = 0 if (r["agent_name"] or r["brokerage_name"]) else 1 entry = (auth.get(r["source"], 2), anonymous, r["uid"], r["source"], r["list_price"]) for k in keys: groups.setdefault(k, []).append(entry) hidden: dict[str, str] = {} for key, rows in groups.items(): uniq = {e[2]: e for e in rows} rows = list(uniq.values()) if len(rows) < 2: continue if key[0] == "prop": # property-level: only cluster listings at ±1% price rows.sort(key=lambda e: (e[4] is None, e[4] or 0)) cluster: list = [] for e in rows: if e[4] is None: continue if cluster and e[4] > cluster[0][4] * 1.01: _pick_cluster(cluster, hidden) cluster = [] cluster.append(e) _pick_cluster(cluster, hidden) else: _pick_cluster(rows, hidden) for uid, dup_of in hidden.items(): con.execute("UPDATE listings SET dup_hidden=1, dup_of=? WHERE uid=?", (dup_of, uid)) con.commit() return len(hidden) def _pick_cluster(cluster: list, hidden: dict[str, str]) -> None: if len(cluster) < 2: return sources = [e[3] for e in cluster] if len(set(sources)) != len(sources): return # same source twice = probably distinct units: be careful keep = min(cluster, key=lambda e: (e[0], e[1], e[2])) # authority, anon, uid for e in cluster: if e[2] != keep[2] and e[2] not in hidden: hidden[e[2]] = keep[2] # --------------------------------------------------------------------------- # Detail-page cache # --------------------------------------------------------------------------- def get_cached_detail(con: sqlite3.Connection, source: str, external_id: str, key: str) -> dict | None: row = con.execute( "SELECT key, payload FROM 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 get_stale_detail(con: sqlite3.Connection, source: str, external_id: str) -> dict | None: row = con.execute( "SELECT payload FROM detail_cache WHERE source=? AND external_id=?", (source, external_id)).fetchone() if row 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 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() # --------------------------------------------------------------------------- # Sources registry (config-driven connector instances) # --------------------------------------------------------------------------- def upsert_source(con: sqlite3.Connection, sid: str, name: str, connector_type: str, config: dict | None = None, enabled: int = 1, authority: int = 2, brokerage_id: int | None = None, states: list[str] | None = None, notes: str = "") -> None: now = time.time() con.execute( """INSERT INTO sources (id, name, connector_type, config, enabled, authority, brokerage_id, states, notes, created, updated) VALUES (?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, connector_type=excluded.connector_type, config=excluded.config, enabled=excluded.enabled, authority=excluded.authority, brokerage_id=excluded.brokerage_id, states=excluded.states, notes=excluded.notes, updated=excluded.updated""", (sid, name, connector_type, json.dumps(config or {}, ensure_ascii=False), enabled, authority, brokerage_id, json.dumps(states or [], ensure_ascii=False), notes, now, now)) con.commit() def get_sources(con: sqlite3.Connection, enabled_only: bool = False) -> list[dict]: sql = "SELECT * FROM sources" if enabled_only: sql += " WHERE enabled=1" out = [] for r in con.execute(sql + " ORDER BY id"): d = dict(r) d["config"] = json.loads(d.get("config") or "{}") d["states"] = json.loads(d.get("states") or "[]") out.append(d) return out