# ----------------------------------------------------------------------------- # Home-Ka — US real-estate aggregator (Groupe KA) # Author: Simon-Pierre Boucher — contact@spboucher.ai # web.py : FastAPI JSON API + static frontend serving (frontend/dist) # — public search/map/detail API (same shapes as immo-ka) and the # admin API (/api/admin/*) behind an optional token # (HOMEKA_ADMIN_TOKEN): sources, connectors, brokerage pipeline. # ----------------------------------------------------------------------------- from __future__ import annotations import json import os import threading import time from pathlib import Path from fastapi import BackgroundTasks, Body, FastAPI, HTTPException, Query, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware from fastapi.responses import FileResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles from . import db, ingest ROOT = Path(__file__).resolve().parent.parent FRONTEND_DIST = ROOT / "frontend" / "dist" FRONTEND_DIR = FRONTEND_DIST if FRONTEND_DIST.exists() else ROOT / "frontend" app = FastAPI(title="Home-Ka API", version="0.1", description="US real-estate aggregator — Groupe KA") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) app.add_middleware(GZipMiddleware, minimum_size=1024) _sync_lock = threading.Lock() # quality gate + precomputed dedup — instant reads (see immo-ka) DEDUP_CLAUSE = " AND dup_hidden=0 AND published=1" def _row_to_dict(row) -> dict: d = dict(row) d["features"] = json.loads(d.get("features") or "[]") d["images"] = json.loads(d.get("images") or "[]") d["details"] = json.loads(d.get("details") or "{}") if d.get("first_seen"): d["days_on_market"] = max(0, int((time.time() - d["first_seen"]) // 86400)) return d def _filters(sql: str, args: list, *, city=None, state=None, zip_code=None, county=None, property_type=None, status=None, source=None, price_max=None, price_min=None, beds_min=None, baths_min=None, sqft_min=None, q=None) -> str: if city: sql += " AND city=? COLLATE NOCASE"; args.append(city) if state: sql += " AND state=?"; args.append(state.upper()) if zip_code: sql += " AND zip_code=?"; args.append(zip_code) if county: sql += " AND county=? COLLATE NOCASE"; args.append(county) if property_type: sql += " AND property_type=?"; args.append(property_type) if status: sql += " AND status=?"; args.append(status) else: # for-sale by default: sold/withdrawn only via explicit filter sql += " AND status NOT IN ('sold','withdrawn')" if source: sql += " AND source=?"; args.append(source) if price_max is not None: sql += " AND list_price IS NOT NULL AND list_price<=?"; args.append(price_max) if price_min is not None: sql += " AND list_price IS NOT NULL AND list_price>=?"; args.append(price_min) if beds_min is not None: sql += " AND bedrooms IS NOT NULL AND bedrooms>=?"; args.append(beds_min) if baths_min is not None: sql += " AND bathrooms IS NOT NULL AND bathrooms>=?"; args.append(baths_min) if sqft_min is not None: sql += " AND living_area_sqft IS NOT NULL AND living_area_sqft>=?"; args.append(sqft_min) if q: sql += (" AND (title LIKE ? OR street_address LIKE ? OR city LIKE ?" " OR zip_code LIKE ? OR mls_id LIKE ?)") args += [f"%{q}%"] * 5 return sql @app.get("/api/listings") def list_listings( city: str | None = None, state: str | None = None, zip_code: str | None = None, county: str | None = None, property_type: str | None = None, status: str | None = None, source: str | None = None, price_max: float | None = None, price_min: float | None = None, beds_min: int | None = None, baths_min: float | None = None, sqft_min: float | None = None, q: str | None = None, active: int = 1, sort: str = "recent", # price_asc | price_desc | recent limit: int = Query(60, le=2000), offset: int = 0, ): con = db.connect() sql = "SELECT * FROM listings WHERE 1=1" args: list = [] if active in (0, 1): sql += " AND active=?"; args.append(active) sql = _filters(sql, args, city=city, state=state, zip_code=zip_code, county=county, property_type=property_type, status=status, source=source, price_max=price_max, price_min=price_min, beds_min=beds_min, baths_min=baths_min, sqft_min=sqft_min, q=q) sql += DEDUP_CLAUSE total = con.execute(f"SELECT COUNT(*) c FROM ({sql})", args).fetchone()["c"] order = { "price_asc": " ORDER BY list_price IS NULL, list_price ASC", "price_desc": " ORDER BY list_price IS NULL, list_price DESC", "recent": " ORDER BY first_seen DESC", }.get(sort, " ORDER BY first_seen DESC") sql += order + " LIMIT ? OFFSET ?" args += [limit, offset] rows = [_row_to_dict(r) for r in con.execute(sql, args).fetchall()] con.close() return {"total": total, "count": len(rows), "listings": rows} @app.get("/api/listings.geojson") def listings_geojson( city: str | None = None, state: str | None = None, zip_code: str | None = None, county: str | None = None, property_type: str | None = None, status: str | None = None, source: str | None = None, price_max: float | None = None, price_min: float | None = None, beds_min: int | None = None, baths_min: float | None = None, sqft_min: float | None = None, q: str | None = None, bbox: str | None = None, limit: int = Query(3000, le=8000), ): """Geolocated listings (map markers, light fields). `bbox=west,south,east,north`.""" con = db.connect() sql = ("SELECT uid, title, street_address, city, state, list_price," " price_label, property_type, bedrooms, bathrooms, source, images," " lat, lng FROM listings" " WHERE active=1 AND lat IS NOT NULL AND lng IS NOT NULL") args: list = [] if bbox: try: west, south, east, north = (float(v) for v in bbox.split(",")) except ValueError: raise HTTPException(400, "bbox expected: west,south,east,north") sql += " AND lat BETWEEN ? AND ? AND lng BETWEEN ? AND ?" args += [south, north, west, east] sql = _filters(sql, args, city=city, state=state, zip_code=zip_code, county=county, property_type=property_type, status=status, source=source, price_max=price_max, price_min=price_min, beds_min=beds_min, baths_min=baths_min, sqft_min=sqft_min, q=q) sql += DEDUP_CLAUSE total_geo = con.execute(f"SELECT COUNT(*) c FROM ({sql})", args).fetchone()["c"] sql_all = sql.replace(" AND lat IS NOT NULL AND lng IS NOT NULL", "") args_all = list(args) if bbox: sql_all = sql_all.replace(" AND lat BETWEEN ? AND ? AND lng BETWEEN ? AND ?", "") del args_all[0:4] total_all = con.execute(f"SELECT COUNT(*) c FROM ({sql_all})", args_all).fetchone()["c"] sql += " ORDER BY list_price IS NULL, list_price ASC LIMIT ?" args.append(limit) features = [] for r in con.execute(sql, args).fetchall(): images = json.loads(r["images"] or "[]") features.append({ "type": "Feature", "geometry": {"type": "Point", "coordinates": [r["lng"], r["lat"]]}, "properties": { "uid": r["uid"], "title": None if r["street_address"] else r["title"], "address": r["street_address"], "price": r["list_price"], "price_label": r["price_label"], "property_type": r["property_type"], "bedrooms": r["bedrooms"], "bathrooms": r["bathrooms"], "source": r["source"], "city": r["city"], "state": r["state"], "image": images[0] if images else None, }, }) con.close() return {"type": "FeatureCollection", "features": features, "totalGeocoded": total_geo, "totalMatching": total_all} @app.get("/api/listings/{uid}") def get_listing(uid: str): con = db.connect() row = con.execute("SELECT * FROM listings WHERE uid=?", (uid,)).fetchone() if row is None: con.close() raise HTTPException(404, "Listing not found") d = _row_to_dict(row) d["price_history"] = [dict(r) for r in con.execute( "SELECT ts, price FROM price_log WHERE uid=? ORDER BY ts DESC LIMIT 20", (uid,)).fetchall()] # same MLS number / same property published elsewhere ("Also listed on…") d["duplicates"] = [dict(r) for r in con.execute( "SELECT uid, source, url, brokerage_name, agent_name, price_label" " FROM listings WHERE dup_of=? AND active=1 ORDER BY source", (uid,)).fetchall()] # PROPERTY: the physical asset behind this listing + its listing history d["property"] = None if d.get("property_id"): p = con.execute("SELECT * FROM properties WHERE id=?", (d["property_id"],)).fetchone() if p is not None: prop = dict(p) prop["details"] = json.loads(prop.get("details") or "{}") prop["listing_history"] = [dict(r) for r in con.execute( "SELECT uid, source, status, list_price, first_seen, last_seen," " active FROM listings WHERE property_id=? ORDER BY first_seen DESC", (d["property_id"],)).fetchall()] d["property"] = prop con.close() return d @app.get("/api/properties/{pid}") def get_property(pid: int): con = db.connect() p = con.execute("SELECT * FROM properties WHERE id=?", (pid,)).fetchone() if p is None: con.close() raise HTTPException(404, "Property not found") prop = dict(p) prop["details"] = json.loads(prop.get("details") or "{}") prop["listings"] = [_row_to_dict(r) for r in con.execute( "SELECT * FROM listings WHERE property_id=? ORDER BY first_seen DESC", (pid,)).fetchall()] con.close() return prop @app.get("/api/facets") def facets(state: str | None = None): """Distinct values for the frontend filters.""" con = db.connect() city_sql = ("SELECT city, COUNT(*) n FROM listings WHERE active=1" " AND city<>''" + DEDUP_CLAUSE) city_args: list = [] if state: city_sql += " AND state=?" city_args.append(state.upper()) out = { "states": [dict(r) for r in con.execute( "SELECT state, COUNT(*) n FROM listings WHERE active=1 AND state<>''" + DEDUP_CLAUSE + " GROUP BY state ORDER BY n DESC")], "cities": [dict(r) for r in con.execute( city_sql + " GROUP BY city ORDER BY n DESC LIMIT 400", city_args)], "property_types": [r["property_type"] for r in con.execute( "SELECT property_type FROM listings WHERE active=1" " AND property_type<>''" + DEDUP_CLAUSE + " GROUP BY property_type ORDER BY COUNT(*) DESC")], "sources": [dict(r) for r in con.execute( "SELECT source, COUNT(*) n FROM listings WHERE active=1" + DEDUP_CLAUSE + " GROUP BY source ORDER BY n DESC")], } con.close() return out @app.get("/api/sources") def sources_public(): con = db.connect() registry = db.get_sources(con) counts = {r["source"]: r["n"] for r in con.execute( "SELECT source, COUNT(*) n FROM listings WHERE active=1 GROUP BY source")} last = {r["source"]: r["ts"] for r in con.execute( "SELECT source, MAX(ts) ts FROM sync_log WHERE ok=1 GROUP BY source")} con.close() out = [] for s in registry: out.append({"id": s["id"], "name": s["name"], "connector_type": s["connector_type"], "states": s["states"], "enabled": s["enabled"], "active_listings": counts.get(s["id"], 0), "last_sync": last.get(s["id"])}) return {"sources": out} @app.get("/api/stats") def stats(): con = db.connect() row = con.execute( """SELECT COUNT(*) total, COUNT(DISTINCT source) sources, COUNT(DISTINCT city) cities, COUNT(DISTINCT state) states, AVG(list_price) avg_price, MIN(list_price) min_price, MAX(list_price) max_price FROM listings WHERE active=1""" + DEDUP_CLAUSE).fetchone() props = con.execute("SELECT COUNT(*) c FROM properties").fetchone()["c"] log = [dict(r) for r in con.execute( "SELECT * FROM sync_log ORDER BY ts DESC LIMIT 20")] by_state = [dict(r) for r in con.execute( "SELECT state, COUNT(*) n, ROUND(AVG(list_price)) avg_price" " FROM listings WHERE active=1 AND state<>''" + DEDUP_CLAUSE + " GROUP BY state ORDER BY n DESC")] by_type = [dict(r) for r in con.execute( "SELECT property_type, COUNT(*) n, ROUND(AVG(list_price)) avg_price" " FROM listings WHERE active=1 AND property_type<>''" + DEDUP_CLAUSE + " GROUP BY property_type ORDER BY n DESC")] from . import quality qual = quality.summary(con) con.close() return {**dict(row), "properties": props, "by_state": by_state, "by_type": by_type, "quality": qual, "recent_syncs": log} @app.post("/api/sync") def trigger_sync(background: BackgroundTasks, source: str | None = None): """Trigger a sync (incoming-webhook equivalent).""" def _job(): with _sync_lock: ingest.run([source] if source else None) background.add_task(_job) return {"status": "started", "source": source or "all"} # ============================================================================= # Admin API — /admin/sources, /admin/brokerages, /admin/connectors # Optional bearer: set HOMEKA_ADMIN_TOKEN to require X-Admin-Token (or ?token=). # ============================================================================= def _check_admin(request: Request) -> None: required = os.environ.get("HOMEKA_ADMIN_TOKEN", "") if not required: return given = request.headers.get("X-Admin-Token") or request.query_params.get("token") if given != required: raise HTTPException(401, "admin token required") @app.get("/api/admin/overview") def admin_overview(request: Request): _check_admin(request) con = db.connect() now = time.time() listings = con.execute( "SELECT COUNT(*) c FROM listings WHERE active=1").fetchone()["c"] props = con.execute("SELECT COUNT(*) c FROM properties").fetchone()["c"] sources_on = con.execute( "SELECT COUNT(*) c FROM sources WHERE enabled=1").fetchone()["c"] errors_24h = con.execute( "SELECT COUNT(*) c FROM sync_log WHERE ok=0 AND ts>?", (now - 86400,)).fetchone()["c"] brok = [dict(r) for r in con.execute( "SELECT partnership_status, COUNT(*) n FROM brokerages" " GROUP BY partnership_status ORDER BY n DESC")] freshest = con.execute( "SELECT MAX(ts) ts FROM sync_log WHERE ok=1").fetchone()["ts"] con.close() return {"active_listings": listings, "properties": props, "enabled_sources": sources_on, "errors_24h": errors_24h, "brokerages_by_status": brok, "last_successful_sync": freshest} @app.get("/api/admin/sources") def admin_sources(request: Request): """Active sources: listing counts, errors, last sync, data freshness, connector type, geographic coverage.""" _check_admin(request) con = db.connect() registry = {s["id"]: s for s in db.get_sources(con)} # custom-coded connectors without a DB row still show up from . import connectors as creg for sid in creg.CUSTOM: registry.setdefault(sid, { "id": sid, "name": sid, "connector_type": "custom", "config": {}, "enabled": 1, "authority": 2, "states": [], "notes": "", "brokerage_id": None}) counts = {r["source"]: dict(r) for r in con.execute( "SELECT source, COUNT(*) n, SUM(published) published," " SUM(CASE WHEN lat IS NULL THEN 1 ELSE 0 END) no_geo" " FROM listings WHERE active=1 GROUP BY source")} lastlog: dict[str, dict] = {} for r in con.execute( "SELECT * FROM sync_log WHERE id IN" " (SELECT MAX(id) FROM sync_log GROUP BY source)"): lastlog[r["source"]] = dict(r) err7d = {r["source"]: r["c"] for r in con.execute( "SELECT source, COUNT(*) c FROM sync_log WHERE ok=0 AND ts>?" " GROUP BY source", (time.time() - 7 * 86400,))} con.close() out = [] for sid, s in sorted(registry.items()): c = counts.get(sid, {}) log = lastlog.get(sid, {}) cfg = dict(s.get("config") or {}) for k in list(cfg): # never expose credentials/config secrets if any(w in k.lower() for w in ("token", "secret", "pass", "key")): cfg[k] = "•••" out.append({ "id": sid, "name": s.get("name") or sid, "connector_type": s.get("connector_type"), "enabled": s.get("enabled", 1), "authority": s.get("authority"), "states": s.get("states") or [], "notes": s.get("notes") or "", "brokerage_id": s.get("brokerage_id"), "config": cfg, "active_listings": c.get("n", 0), "published": c.get("published", 0), "no_geo": c.get("no_geo", 0), "last_sync": log.get("ts"), "last_ok": log.get("ok"), "last_message": log.get("message"), "last_found": log.get("found"), "last_added": log.get("added"), "last_removed": log.get("removed"), "errors_7d": err7d.get(sid, 0), "freshness_hours": (round((time.time() - log["ts"]) / 3600, 1) if log.get("ts") else None), }) return {"sources": out} @app.get("/api/admin/connectors") def admin_connectors(request: Request): """Registered connector families + custom connectors + instance counts.""" _check_admin(request) from . import connectors as creg con = db.connect() inst = {r["connector_type"]: r["n"] for r in con.execute( "SELECT connector_type, COUNT(*) n FROM sources GROUP BY connector_type")} con.close() families = [] for fam, cls in sorted(creg.FAMILIES.items()): families.append({ "family": fam, "class": cls.__name__, "module": cls.__module__, "doc": (cls.__doc__ or "").strip().split("\n")[0], "instances": inst.get(fam, 0), }) custom = [{"source_id": sid, "class": cls.__name__, "module": cls.__module__} for sid, cls in sorted(creg.CUSTOM.items())] return {"families": families, "custom": custom} @app.get("/api/admin/brokerages") def admin_brokerages( request: Request, state: str | None = None, status: str | None = None, feed_type: str | None = None, idx_provider: str | None = None, q: str | None = None, min_priority: float | None = None, inspected: int | None = None, sort: str = "priority", # priority | feed | name | inspected limit: int = Query(50, le=500), offset: int = 0, ): """Potential brokerage partners: scores, detected stack, contacts, partnership status.""" _check_admin(request) con = db.connect() sql = "SELECT * FROM brokerages WHERE 1=1" args: list = [] if state: sql += " AND states LIKE ?"; args.append(f'%"{state.upper()}"%') if status: sql += " AND partnership_status=?"; args.append(status) if feed_type: sql += " AND possible_feed_type=?"; args.append(feed_type) if idx_provider: sql += " AND idx_provider LIKE ?"; args.append(f"%{idx_provider}%") if q: sql += " AND (name LIKE ? OR website LIKE ?)"; args += [f"%{q}%"] * 2 if min_priority is not None: sql += " AND priority_score>=?"; args.append(min_priority) if inspected == 1: sql += " AND last_inspected IS NOT NULL" elif inspected == 0: sql += " AND last_inspected IS NULL" total = con.execute(f"SELECT COUNT(*) c FROM ({sql})", args).fetchone()["c"] order = { "priority": " ORDER BY priority_score DESC NULLS LAST", "feed": " ORDER BY feed_probability_score DESC NULLS LAST", "name": " ORDER BY name", "inspected": " ORDER BY last_inspected DESC NULLS LAST", }.get(sort, " ORDER BY priority_score DESC NULLS LAST") rows = [] for r in con.execute(sql + order + " LIMIT ? OFFSET ?", args + [limit, offset]): d = dict(r) for k in ("states", "cities", "mls_affiliations", "evidence"): d[k] = json.loads(d.get(k) or ("[]" if k != "evidence" else "{}")) rows.append(d) con.close() return {"total": total, "count": len(rows), "brokerages": rows} @app.post("/api/admin/brokerages/{bid}/inspect") def admin_inspect(bid: int, request: Request): _check_admin(request) from . import discovery con = db.connect() res = discovery.inspect_one(con, bid) con.close() return res @app.post("/api/admin/brokerages/{bid}/status") def admin_brokerage_status(bid: int, request: Request, payload: dict = Body(...)): _check_admin(request) status = str(payload.get("status") or "") allowed = {"prospect", "to-contact", "contacted", "in-discussion", "feed-received", "live", "declined"} if status not in allowed: raise HTTPException(400, f"status must be one of {sorted(allowed)}") con = db.connect() con.execute("UPDATE brokerages SET partnership_status=?, updated=? WHERE id=?", (status, time.time(), bid)) con.commit() con.close() return {"id": bid, "status": status} @app.post("/api/admin/discover") def admin_discover(request: Request, background: BackgroundTasks, limit: int = 25): """Run a discovery batch (inspect the next prospects) in the background.""" _check_admin(request) from . import discovery background.add_task(discovery.run_batch, limit) return {"status": "started", "limit": limit} # --- robots + static frontend ------------------------------------------------- @app.get("/robots.txt") def robots(): return PlainTextResponse("User-agent: *\nAllow: /\n" "Sitemap: https://www.home-ka.com/sitemap.xml\n") if FRONTEND_DIR.exists(): if (FRONTEND_DIR / "assets").is_dir(): app.mount("/assets", StaticFiles(directory=FRONTEND_DIR / "assets"), name="assets") @app.get("/{full_path:path}") def spa(full_path: str): target = FRONTEND_DIR / full_path if full_path and ".." not in full_path and target.is_file(): return FileResponse(target) index = FRONTEND_DIR / "index.html" if index.exists(): return FileResponse(index) raise HTTPException(404)