Python 49.6%
TypeScript 25.5%
CSS 24.1%
1# -----------------------------------------------------------------------------2# Home-Ka — US real-estate aggregator (Groupe KA)3# Author: Simon-Pierre Boucher — contact@spboucher.ai4# db.py : SQLite persistence — PROPERTY vs LISTING separation, upsert with5# change detection, lifecycle with grace period (2 syncs), drift6# detection, price history, detail-page cache, sources registry,7# brokerage pipeline. Ported from immo-ka/db.py.8# -----------------------------------------------------------------------------9from __future__ import annotations1011import json12import sqlite313import statistics14import time15from pathlib import Path1617from .schema import Listing1819DB_PATH = Path(__file__).resolve().parent.parent / "data" / "homeka.db"2021# Consecutive syncs a listing must be absent from its source before being22# deactivated (grace period against one-off fetch failures).23MISS_GRACE = 22425# Drift: if a source returns <= DRIFT_RATIO × its historical median26# (median >= DRIFT_MIN_BASE listings), alert and suspend removals.27DRIFT_RATIO = 0.2528DRIFT_MIN_BASE = 829DRIFT_HISTORY = 53031_SCHEMA = """32CREATE TABLE IF NOT EXISTS properties (33 id INTEGER PRIMARY KEY AUTOINCREMENT,34 addr_key TEXT UNIQUE, -- normalized address key (propertymatch.py)35 apn TEXT, -- assessor parcel number when known36 street_address TEXT,37 unit TEXT,38 city TEXT,39 state TEXT, -- 2-letter USPS40 zip_code TEXT,41 county TEXT,42 property_type TEXT,43 year_built INTEGER,44 living_area_sqft REAL,45 lot_size_sqft REAL,46 lat REAL,47 lng REAL,48 details TEXT, -- JSON (assessor/public-record enrichment)49 first_seen REAL,50 last_seen REAL51);52CREATE INDEX IF NOT EXISTS idx_prop_state_city ON properties(state, city);53CREATE INDEX IF NOT EXISTS idx_prop_zip ON properties(zip_code);54CREATE INDEX IF NOT EXISTS idx_prop_apn ON properties(apn);5556CREATE TABLE IF NOT EXISTS listings (57 uid TEXT PRIMARY KEY, -- source:external_id58 source TEXT NOT NULL,59 external_id TEXT NOT NULL,60 property_id INTEGER, -- -> properties.id (property matching)61 url TEXT,62 title TEXT,63 street_address TEXT,64 unit TEXT,65 city TEXT,66 state TEXT,67 zip_code TEXT,68 county TEXT,69 property_type TEXT,70 property_subtype TEXT,71 list_price REAL,72 price_label TEXT,73 bedrooms INTEGER,74 bathrooms_full INTEGER,75 bathrooms_half INTEGER,76 bathrooms REAL,77 living_area_sqft REAL,78 lot_size_sqft REAL,79 year_built INTEGER,80 apn TEXT,81 mls_id TEXT,82 mls_name TEXT,83 status TEXT DEFAULT 'active',84 listed_at TEXT,85 brokerage_name TEXT,86 office_name TEXT,87 agent_name TEXT,88 agent_phone TEXT,89 agent_email TEXT,90 description TEXT,91 features TEXT, -- JSON (list of source texts)92 details TEXT, -- JSON (structured fields, RESO passthrough)93 images TEXT, -- JSON94 lat REAL,95 lng REAL,96 geocode_failed INTEGER DEFAULT 0,97 content_hash TEXT,98 first_seen REAL,99 last_seen REAL,100 updated_at REAL,101 miss_count INTEGER DEFAULT 0,102 active INTEGER DEFAULT 1,103 dup_hidden INTEGER DEFAULT 0,104 dup_of TEXT,105 published INTEGER DEFAULT 1, -- quality gate (homeka/quality.py)106 quality REAL107);108CREATE INDEX IF NOT EXISTS idx_listings_source ON listings(source);109CREATE INDEX IF NOT EXISTS idx_listings_city ON listings(city);110CREATE INDEX IF NOT EXISTS idx_listings_state ON listings(state);111CREATE INDEX IF NOT EXISTS idx_listings_type ON listings(property_type);112CREATE INDEX IF NOT EXISTS idx_listings_active ON listings(active);113CREATE INDEX IF NOT EXISTS idx_listings_extid ON listings(external_id);114CREATE INDEX IF NOT EXISTS idx_listings_mls ON listings(mls_id);115CREATE INDEX IF NOT EXISTS idx_listings_prop ON listings(property_id);116CREATE INDEX IF NOT EXISTS idx_listings_geo ON listings(lat, lng);117CREATE INDEX IF NOT EXISTS idx_listings_duphidden ON listings(dup_hidden);118CREATE INDEX IF NOT EXISTS idx_listings_dupof ON listings(dup_of);119120-- Registered sources: one row = one connector INSTANCE (config-driven).121-- connector_type selects the family (reso, rets, xml, json, jsonld, csv,122-- sftp, arcgis, custom); config is the family-specific JSON (base URL,123-- credentials env-var names, field mapping, throttle...). Custom-coded124-- connectors (connectors/custom_broker/*.py) register themselves by id125-- and need no row, but one can exist to carry metadata.126CREATE TABLE IF NOT EXISTS sources (127 id TEXT PRIMARY KEY,128 name TEXT,129 connector_type TEXT,130 config TEXT, -- JSON131 enabled INTEGER DEFAULT 1,132 authority INTEGER DEFAULT 2, -- 0=direct feed, 1=partner API, 2=site, 3=classifieds133 brokerage_id INTEGER, -- -> brokerages.id when source came from a partnership134 states TEXT, -- JSON list of covered states135 notes TEXT,136 created REAL,137 updated REAL138);139140-- US brokerage pipeline: discovery -> inspection -> scoring -> partnership.141CREATE TABLE IF NOT EXISTS brokerages (142 id INTEGER PRIMARY KEY AUTOINCREMENT,143 name TEXT NOT NULL,144 website TEXT UNIQUE,145 states TEXT, -- JSON list146 cities TEXT, -- JSON list147 estimated_agents INTEGER,148 estimated_listings INTEGER,149 mls_affiliations TEXT, -- JSON list150 idx_present INTEGER, -- 1/0/NULL (unknown until inspected)151 idx_provider TEXT, -- kvCORE, iHomefinder, IDX Broker, ...152 reso_detected INTEGER,153 possible_feed_type TEXT, -- reso | rets | json-api | xml | csv | idx-partner | crawl154 contact_page TEXT,155 partnership_contact TEXT,156 technical_contact TEXT,157 feed_probability_score REAL, -- 0-100158 priority_score REAL, -- 0-100 (size × feed probability × coverage)159 partnership_status TEXT DEFAULT 'prospect',160 -- prospect | to-contact | contacted | in-discussion | feed-received | live | declined161 source_id TEXT, -- -> sources.id once a connector exists162 last_inspected REAL,163 inspect_error TEXT,164 evidence TEXT, -- JSON (fingerprints, URLs, raw hints)165 created REAL,166 updated REAL167);168CREATE INDEX IF NOT EXISTS idx_brok_status ON brokerages(partnership_status);169CREATE INDEX IF NOT EXISTS idx_brok_priority ON brokerages(priority_score);170171CREATE TABLE IF NOT EXISTS sync_log (172 id INTEGER PRIMARY KEY AUTOINCREMENT,173 source TEXT,174 ts REAL,175 found INTEGER,176 added INTEGER,177 updated INTEGER,178 removed INTEGER,179 ok INTEGER,180 message TEXT,181 stats TEXT -- JSON: null-field rates, missed, alert...182);183CREATE INDEX IF NOT EXISTS idx_synclog_source ON sync_log(source, ts);184185CREATE TABLE IF NOT EXISTS detail_cache (186 source TEXT NOT NULL,187 external_id TEXT NOT NULL,188 key TEXT,189 payload TEXT,190 fetched_at REAL,191 PRIMARY KEY (source, external_id)192);193194CREATE TABLE IF NOT EXISTS price_log (195 uid TEXT NOT NULL,196 ts REAL NOT NULL,197 price REAL198);199CREATE INDEX IF NOT EXISTS idx_price_log_uid ON price_log(uid);200201CREATE TABLE IF NOT EXISTS geocode_cache (202 address TEXT PRIMARY KEY,203 lat REAL,204 lng REAL,205 provider TEXT,206 failed INTEGER DEFAULT 0,207 ts REAL208);209"""210211_SCHEMA_READY = False # schema/migrations run ONCE per process (write lock)212213214def _init_schema(con: sqlite3.Connection) -> None:215 con.executescript(_SCHEMA)216 con.commit()217218219def connect() -> sqlite3.Connection:220 global _SCHEMA_READY221 DB_PATH.parent.mkdir(parents=True, exist_ok=True)222 con = sqlite3.connect(DB_PATH, timeout=60)223 con.row_factory = sqlite3.Row224 # WAL + busy_timeout: concurrent access (watcher + web) without locks.225 con.execute("PRAGMA journal_mode=WAL")226 con.execute("PRAGMA busy_timeout=120000")227 con.execute("PRAGMA synchronous=NORMAL")228 if not _SCHEMA_READY:229 _init_schema(con)230 _SCHEMA_READY = True231 return con232233234# ---------------------------------------------------------------------------235# Source sync (ported from immo-ka: change detection, grace, drift)236# ---------------------------------------------------------------------------237238def _drift_alert(con: sqlite3.Connection, source: str, found: int,239 null_price_rate: float) -> str | None:240 """Connector drift detection (volume or extracted-price collapse)."""241 hist = con.execute(242 "SELECT found, stats FROM sync_log WHERE source=? AND ok=1"243 " ORDER BY ts DESC LIMIT ?", (source, DRIFT_HISTORY)).fetchall()244 if len(hist) < 3:245 return None246 med_found = statistics.median(r["found"] for r in hist)247 if med_found >= DRIFT_MIN_BASE and found <= DRIFT_RATIO * med_found:248 return (f"drift: {found} listing(s) found vs a median of "249 f"{med_found:.0f} — removals suspended, check the connector")250 if found >= DRIFT_MIN_BASE and null_price_rate >= 0.8:251 rates = []252 for r in hist:253 try:254 rates.append(json.loads(r["stats"] or "{}")["null_price_rate"])255 except (KeyError, ValueError, TypeError):256 continue257 if rates and statistics.median(rates) <= 0.3:258 return (f"drift: {null_price_rate:.0%} of listings without a price "259 f"(usually {statistics.median(rates):.0%}) — "260 "the source format probably changed")261 return None262263264_LISTING_COLS = (265 "uid", "source", "external_id", "property_id", "url", "title",266 "street_address", "unit", "city", "state", "zip_code", "county",267 "property_type", "property_subtype", "list_price", "price_label",268 "bedrooms", "bathrooms_full", "bathrooms_half", "bathrooms",269 "living_area_sqft", "lot_size_sqft", "year_built", "apn", "mls_id",270 "mls_name", "status", "listed_at", "brokerage_name", "office_name",271 "agent_name", "agent_phone", "agent_email", "description", "features",272 "details", "images", "lat", "lng", "content_hash",273)274275276def _listing_params(lst: Listing, property_id: int | None, h: str,277 now: float) -> dict:278 return dict(279 uid=lst.uid, source=lst.source, external_id=lst.external_id,280 property_id=property_id, url=lst.url, title=lst.title,281 street_address=lst.street_address, unit=lst.unit, city=lst.city,282 state=lst.state, zip_code=lst.zip_code, county=lst.county,283 property_type=lst.property_type, property_subtype=lst.property_subtype,284 list_price=lst.list_price, price_label=lst.price_label,285 bedrooms=lst.bedrooms, bathrooms_full=lst.bathrooms_full,286 bathrooms_half=lst.bathrooms_half, bathrooms=lst.bathrooms,287 living_area_sqft=lst.living_area_sqft, lot_size_sqft=lst.lot_size_sqft,288 year_built=lst.year_built, apn=lst.apn, mls_id=lst.mls_id,289 mls_name=lst.mls_name, status=lst.status, listed_at=lst.listed_at,290 brokerage_name=lst.brokerage_name, office_name=lst.office_name,291 agent_name=lst.agent_name, agent_phone=lst.agent_phone,292 agent_email=lst.agent_email, description=lst.description,293 features=json.dumps(lst.features, ensure_ascii=False),294 details=json.dumps(lst.details, ensure_ascii=False, default=str),295 images=json.dumps(lst.images, ensure_ascii=False),296 lat=lst.lat, lng=lst.lng, content_hash=h, now=now,297 )298299300def sync_source(con: sqlite3.Connection, source: str,301 listings: list[Listing]) -> dict:302 """Synchronize one source's listings.303304 - new listing -> insert (+ property match/create)305 - changed listing -> update (content_hash comparison)306 - vanished listing -> miss_count += 1, then active=0 after MISS_GRACE307 consecutive runs (sold or withdrawn); the matched308 PROPERTY row is kept forever309 - drift detected -> alert logged, removals suspended310 """311 from . import propertymatch312 now = time.time()313 added = updated = 0314 seen_uids = set()315316 n = len(listings)317 null_price = sum(1 for l in listings if l.list_price is None)318 null_addr = sum(1 for l in listings if not l.street_address)319 null_price_rate = round(null_price / n, 3) if n else 0.0320321 alert = _drift_alert(con, source, n, null_price_rate)322323 for lst in listings:324 seen_uids.add(lst.uid)325 h = lst.content_hash()326 row = con.execute(327 "SELECT content_hash, list_price, property_id FROM listings WHERE uid=?",328 (lst.uid,)).fetchone()329 property_id = row["property_id"] if row else None330 if property_id is None:331 property_id = propertymatch.match_or_create(con, lst, now)332 params = _listing_params(lst, property_id, h, now)333 if row is None:334 cols = ", ".join(_LISTING_COLS)335 named = ", ".join(f":{c}" for c in _LISTING_COLS)336 con.execute(337 f"INSERT INTO listings ({cols}, first_seen, last_seen,"338 f" updated_at, miss_count, active)"339 f" VALUES ({named}, :now, :now, :now, 0, 1)", params)340 if lst.list_price is not None:341 con.execute("INSERT INTO price_log (uid, ts, price) VALUES (?,?,?)",342 (lst.uid, now, lst.list_price))343 added += 1344 elif row["content_hash"] != h:345 # COALESCE: never null-overwrite geocoded coordinates or346 # enrichment-filled year/areas when the list feed lacks them347 con.execute(348 """UPDATE listings SET url=:url, title=:title,349 street_address=:street_address, unit=:unit, city=:city,350 state=:state, zip_code=:zip_code, county=:county,351 property_type=:property_type, property_subtype=:property_subtype,352 list_price=:list_price, price_label=:price_label,353 bedrooms=:bedrooms, bathrooms_full=:bathrooms_full,354 bathrooms_half=:bathrooms_half, bathrooms=:bathrooms,355 living_area_sqft=COALESCE(:living_area_sqft, living_area_sqft),356 lot_size_sqft=COALESCE(:lot_size_sqft, lot_size_sqft),357 year_built=COALESCE(:year_built, year_built),358 apn=:apn, mls_id=:mls_id, mls_name=:mls_name,359 status=:status, listed_at=:listed_at,360 brokerage_name=:brokerage_name, office_name=:office_name,361 agent_name=:agent_name, agent_phone=:agent_phone,362 agent_email=:agent_email, description=:description,363 features=:features, details=:details, images=:images,364 lat=COALESCE(:lat, lat), lng=COALESCE(:lng, lng),365 property_id=:property_id,366 content_hash=:content_hash, last_seen=:now,367 updated_at=:now, miss_count=0, active=1368 WHERE uid=:uid""", params)369 if lst.list_price != row["list_price"]:370 con.execute("INSERT INTO price_log (uid, ts, price) VALUES (?,?,?)",371 (lst.uid, now, lst.list_price))372 updated += 1373 else:374 con.execute(375 "UPDATE listings SET last_seen=?, miss_count=0, active=1 WHERE uid=?",376 (now, lst.uid))377 if property_id:378 propertymatch.refresh_property(con, property_id, lst, now)379380 # Listings of this source no longer present: grace period, then381 # deactivation (sold/withdrawn). Suspended when drift is detected.382 # The PROPERTY row keeps existing — only the listing goes off-market.383 removed = missed = 0384 if not alert:385 for r in con.execute(386 "SELECT uid, miss_count FROM listings WHERE source=? AND active=1",387 (source,)).fetchall():388 if r["uid"] in seen_uids:389 continue390 missed += 1391 if r["miss_count"] + 1 >= MISS_GRACE:392 con.execute(393 "UPDATE listings SET active=0, status='withdrawn',"394 " miss_count=?, updated_at=? WHERE uid=?",395 (r["miss_count"] + 1, now, r["uid"]))396 removed += 1397 else:398 con.execute("UPDATE listings SET miss_count=miss_count+1 WHERE uid=?",399 (r["uid"],))400401 stats = {402 "null_price_rate": null_price_rate,403 "null_address_rate": round(null_addr / n, 3) if n else 0.0,404 "missed": missed,405 }406 if alert:407 stats["alert"] = alert408 con.execute(409 "INSERT INTO sync_log (source, ts, found, added, updated, removed, ok,"410 " message, stats) VALUES (?,?,?,?,?,?,1,?,?)",411 (source, now, n, added, updated, removed, alert or "ok",412 json.dumps(stats, ensure_ascii=False)))413 con.commit()414 out = {"source": source, "found": n, "added": added,415 "updated": updated, "removed": removed}416 if alert:417 out["alert"] = alert418 return out419420421def log_failure(con: sqlite3.Connection, source: str, message: str) -> None:422 con.execute(423 "INSERT INTO sync_log (source, ts, found, added, updated, removed, ok, message)"424 " VALUES (?,?,0,0,0,0,0,?)", (source, time.time(), message))425 con.commit()426427428# ---------------------------------------------------------------------------429# Deduplication — precomputed like immo-ka (dup_hidden/dup_of columns).430# US rules:431# 1) same MLS number (mls_id + state) published by several sources432# 2) same PROPERTY (property_id) listed by several sources at ±1% price433# The most authoritative source wins (sources.authority: direct feed first),434# then the record with an agent, then the smallest uid.435# ---------------------------------------------------------------------------436437def _authority_map(con: sqlite3.Connection) -> dict[str, int]:438 return {r["id"]: r["authority"] if r["authority"] is not None else 2439 for r in con.execute("SELECT id, authority FROM sources")}440441442def refresh_dedup(con: sqlite3.Connection) -> int:443 auth = _authority_map(con)444 con.execute("UPDATE listings SET dup_hidden=0, dup_of=NULL")445 groups: dict[tuple, list] = {}446 for r in con.execute(447 "SELECT uid, source, mls_id, state, property_id, list_price,"448 " agent_name, brokerage_name FROM listings WHERE active=1"):449 keys = []450 if r["mls_id"]:451 keys.append(("mls", r["mls_id"], r["state"] or ""))452 if r["property_id"] and r["list_price"]:453 keys.append(("prop", r["property_id"]))454 anonymous = 0 if (r["agent_name"] or r["brokerage_name"]) else 1455 entry = (auth.get(r["source"], 2), anonymous, r["uid"],456 r["source"], r["list_price"])457 for k in keys:458 groups.setdefault(k, []).append(entry)459 hidden: dict[str, str] = {}460 for key, rows in groups.items():461 uniq = {e[2]: e for e in rows}462 rows = list(uniq.values())463 if len(rows) < 2:464 continue465 if key[0] == "prop":466 # property-level: only cluster listings at ±1% price467 rows.sort(key=lambda e: (e[4] is None, e[4] or 0))468 cluster: list = []469 for e in rows:470 if e[4] is None:471 continue472 if cluster and e[4] > cluster[0][4] * 1.01:473 _pick_cluster(cluster, hidden)474 cluster = []475 cluster.append(e)476 _pick_cluster(cluster, hidden)477 else:478 _pick_cluster(rows, hidden)479 for uid, dup_of in hidden.items():480 con.execute("UPDATE listings SET dup_hidden=1, dup_of=? WHERE uid=?",481 (dup_of, uid))482 con.commit()483 return len(hidden)484485486def _pick_cluster(cluster: list, hidden: dict[str, str]) -> None:487 if len(cluster) < 2:488 return489 sources = [e[3] for e in cluster]490 if len(set(sources)) != len(sources):491 return # same source twice = probably distinct units: be careful492 keep = min(cluster, key=lambda e: (e[0], e[1], e[2])) # authority, anon, uid493 for e in cluster:494 if e[2] != keep[2] and e[2] not in hidden:495 hidden[e[2]] = keep[2]496497498# ---------------------------------------------------------------------------499# Detail-page cache500# ---------------------------------------------------------------------------501502def get_cached_detail(con: sqlite3.Connection, source: str,503 external_id: str, key: str) -> dict | None:504 row = con.execute(505 "SELECT key, payload FROM detail_cache WHERE source=? AND external_id=?",506 (source, external_id)).fetchone()507 if row and row["key"] == key and row["payload"]:508 try:509 return json.loads(row["payload"])510 except ValueError:511 return None512 return None513514515def get_stale_detail(con: sqlite3.Connection, source: str,516 external_id: str) -> dict | None:517 row = con.execute(518 "SELECT payload FROM detail_cache WHERE source=? AND external_id=?",519 (source, external_id)).fetchone()520 if row and row["payload"]:521 try:522 return json.loads(row["payload"])523 except ValueError:524 return None525 return None526527528def put_cached_detail(con: sqlite3.Connection, source: str,529 external_id: str, key: str, payload: dict) -> None:530 con.execute(531 "INSERT INTO detail_cache (source, external_id, key, payload, fetched_at)"532 " VALUES (?,?,?,?,?)"533 " ON CONFLICT(source, external_id) DO UPDATE SET"534 " key=excluded.key, payload=excluded.payload, fetched_at=excluded.fetched_at",535 (source, external_id, key, json.dumps(payload, ensure_ascii=False),536 time.time()))537 con.commit()538539540# ---------------------------------------------------------------------------541# Sources registry (config-driven connector instances)542# ---------------------------------------------------------------------------543544def upsert_source(con: sqlite3.Connection, sid: str, name: str,545 connector_type: str, config: dict | None = None,546 enabled: int = 1, authority: int = 2,547 brokerage_id: int | None = None,548 states: list[str] | None = None, notes: str = "") -> None:549 now = time.time()550 con.execute(551 """INSERT INTO sources (id, name, connector_type, config, enabled,552 authority, brokerage_id, states, notes, created, updated)553 VALUES (?,?,?,?,?,?,?,?,?,?,?)554 ON CONFLICT(id) DO UPDATE SET name=excluded.name,555 connector_type=excluded.connector_type, config=excluded.config,556 enabled=excluded.enabled, authority=excluded.authority,557 brokerage_id=excluded.brokerage_id, states=excluded.states,558 notes=excluded.notes, updated=excluded.updated""",559 (sid, name, connector_type,560 json.dumps(config or {}, ensure_ascii=False), enabled, authority,561 brokerage_id, json.dumps(states or [], ensure_ascii=False),562 notes, now, now))563 con.commit()564565566def get_sources(con: sqlite3.Connection, enabled_only: bool = False) -> list[dict]:567 sql = "SELECT * FROM sources"568 if enabled_only:569 sql += " WHERE enabled=1"570 out = []571 for r in con.execute(sql + " ORDER BY id"):572 d = dict(r)573 d["config"] = json.loads(d.get("config") or "{}")574 d["states"] = json.loads(d.get("states") or "[]")575 out.append(d)576 return out577