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# web.py : FastAPI JSON API + static frontend serving (frontend/dist)5# — public search/map/detail API (same shapes as immo-ka) and the6# admin API (/api/admin/*) behind an optional token7# (HOMEKA_ADMIN_TOKEN): sources, connectors, brokerage pipeline.8# -----------------------------------------------------------------------------9from __future__ import annotations1011import json12import os13import threading14import time15from pathlib import Path1617from fastapi import BackgroundTasks, Body, FastAPI, HTTPException, Query, Request18from fastapi.middleware.cors import CORSMiddleware19from fastapi.middleware.gzip import GZipMiddleware20from fastapi.responses import FileResponse, PlainTextResponse21from fastapi.staticfiles import StaticFiles2223from . import db, ingest2425ROOT = Path(__file__).resolve().parent.parent26FRONTEND_DIST = ROOT / "frontend" / "dist"27FRONTEND_DIR = FRONTEND_DIST if FRONTEND_DIST.exists() else ROOT / "frontend"2829app = FastAPI(title="Home-Ka API", version="0.1",30 description="US real-estate aggregator — Groupe KA")31app.add_middleware(CORSMiddleware, allow_origins=["*"],32 allow_methods=["*"], allow_headers=["*"])33app.add_middleware(GZipMiddleware, minimum_size=1024)3435_sync_lock = threading.Lock()3637# quality gate + precomputed dedup — instant reads (see immo-ka)38DEDUP_CLAUSE = " AND dup_hidden=0 AND published=1"394041def _row_to_dict(row) -> dict:42 d = dict(row)43 d["features"] = json.loads(d.get("features") or "[]")44 d["images"] = json.loads(d.get("images") or "[]")45 d["details"] = json.loads(d.get("details") or "{}")46 if d.get("first_seen"):47 d["days_on_market"] = max(0, int((time.time() - d["first_seen"]) // 86400))48 return d495051def _filters(sql: str, args: list, *, city=None, state=None, zip_code=None,52 county=None, property_type=None, status=None, source=None,53 price_max=None, price_min=None, beds_min=None, baths_min=None,54 sqft_min=None, q=None) -> str:55 if city:56 sql += " AND city=? COLLATE NOCASE"; args.append(city)57 if state:58 sql += " AND state=?"; args.append(state.upper())59 if zip_code:60 sql += " AND zip_code=?"; args.append(zip_code)61 if county:62 sql += " AND county=? COLLATE NOCASE"; args.append(county)63 if property_type:64 sql += " AND property_type=?"; args.append(property_type)65 if status:66 sql += " AND status=?"; args.append(status)67 else: # for-sale by default: sold/withdrawn only via explicit filter68 sql += " AND status NOT IN ('sold','withdrawn')"69 if source:70 sql += " AND source=?"; args.append(source)71 if price_max is not None:72 sql += " AND list_price IS NOT NULL AND list_price<=?"; args.append(price_max)73 if price_min is not None:74 sql += " AND list_price IS NOT NULL AND list_price>=?"; args.append(price_min)75 if beds_min is not None:76 sql += " AND bedrooms IS NOT NULL AND bedrooms>=?"; args.append(beds_min)77 if baths_min is not None:78 sql += " AND bathrooms IS NOT NULL AND bathrooms>=?"; args.append(baths_min)79 if sqft_min is not None:80 sql += " AND living_area_sqft IS NOT NULL AND living_area_sqft>=?"; args.append(sqft_min)81 if q:82 sql += (" AND (title LIKE ? OR street_address LIKE ? OR city LIKE ?"83 " OR zip_code LIKE ? OR mls_id LIKE ?)")84 args += [f"%{q}%"] * 585 return sql868788@app.get("/api/listings")89def list_listings(90 city: str | None = None, state: str | None = None,91 zip_code: str | None = None, county: str | None = None,92 property_type: str | None = None, status: str | None = None,93 source: str | None = None,94 price_max: float | None = None, price_min: float | None = None,95 beds_min: int | None = None, baths_min: float | None = None,96 sqft_min: float | None = None,97 q: str | None = None,98 active: int = 1,99 sort: str = "recent", # price_asc | price_desc | recent100 limit: int = Query(60, le=2000),101 offset: int = 0,102):103 con = db.connect()104 sql = "SELECT * FROM listings WHERE 1=1"105 args: list = []106 if active in (0, 1):107 sql += " AND active=?"; args.append(active)108 sql = _filters(sql, args, city=city, state=state, zip_code=zip_code,109 county=county, property_type=property_type, status=status,110 source=source, price_max=price_max, price_min=price_min,111 beds_min=beds_min, baths_min=baths_min, sqft_min=sqft_min, q=q)112 sql += DEDUP_CLAUSE113 total = con.execute(f"SELECT COUNT(*) c FROM ({sql})", args).fetchone()["c"]114 order = {115 "price_asc": " ORDER BY list_price IS NULL, list_price ASC",116 "price_desc": " ORDER BY list_price IS NULL, list_price DESC",117 "recent": " ORDER BY first_seen DESC",118 }.get(sort, " ORDER BY first_seen DESC")119 sql += order + " LIMIT ? OFFSET ?"120 args += [limit, offset]121 rows = [_row_to_dict(r) for r in con.execute(sql, args).fetchall()]122 con.close()123 return {"total": total, "count": len(rows), "listings": rows}124125126@app.get("/api/listings.geojson")127def listings_geojson(128 city: str | None = None, state: str | None = None,129 zip_code: str | None = None, county: str | None = None,130 property_type: str | None = None, status: str | None = None,131 source: str | None = None,132 price_max: float | None = None, price_min: float | None = None,133 beds_min: int | None = None, baths_min: float | None = None,134 sqft_min: float | None = None, q: str | None = None,135 bbox: str | None = None,136 limit: int = Query(3000, le=8000),137):138 """Geolocated listings (map markers, light fields). `bbox=west,south,east,north`."""139 con = db.connect()140 sql = ("SELECT uid, title, street_address, city, state, list_price,"141 " price_label, property_type, bedrooms, bathrooms, source, images,"142 " lat, lng FROM listings"143 " WHERE active=1 AND lat IS NOT NULL AND lng IS NOT NULL")144 args: list = []145 if bbox:146 try:147 west, south, east, north = (float(v) for v in bbox.split(","))148 except ValueError:149 raise HTTPException(400, "bbox expected: west,south,east,north")150 sql += " AND lat BETWEEN ? AND ? AND lng BETWEEN ? AND ?"151 args += [south, north, west, east]152 sql = _filters(sql, args, city=city, state=state, zip_code=zip_code,153 county=county, property_type=property_type, status=status,154 source=source, price_max=price_max, price_min=price_min,155 beds_min=beds_min, baths_min=baths_min, sqft_min=sqft_min, q=q)156 sql += DEDUP_CLAUSE157 total_geo = con.execute(f"SELECT COUNT(*) c FROM ({sql})", args).fetchone()["c"]158 sql_all = sql.replace(" AND lat IS NOT NULL AND lng IS NOT NULL", "")159 args_all = list(args)160 if bbox:161 sql_all = sql_all.replace(" AND lat BETWEEN ? AND ? AND lng BETWEEN ? AND ?", "")162 del args_all[0:4]163 total_all = con.execute(f"SELECT COUNT(*) c FROM ({sql_all})", args_all).fetchone()["c"]164 sql += " ORDER BY list_price IS NULL, list_price ASC LIMIT ?"165 args.append(limit)166 features = []167 for r in con.execute(sql, args).fetchall():168 images = json.loads(r["images"] or "[]")169 features.append({170 "type": "Feature",171 "geometry": {"type": "Point", "coordinates": [r["lng"], r["lat"]]},172 "properties": {173 "uid": r["uid"],174 "title": None if r["street_address"] else r["title"],175 "address": r["street_address"],176 "price": r["list_price"], "price_label": r["price_label"],177 "property_type": r["property_type"], "bedrooms": r["bedrooms"],178 "bathrooms": r["bathrooms"], "source": r["source"],179 "city": r["city"], "state": r["state"],180 "image": images[0] if images else None,181 },182 })183 con.close()184 return {"type": "FeatureCollection", "features": features,185 "totalGeocoded": total_geo, "totalMatching": total_all}186187188@app.get("/api/listings/{uid}")189def get_listing(uid: str):190 con = db.connect()191 row = con.execute("SELECT * FROM listings WHERE uid=?", (uid,)).fetchone()192 if row is None:193 con.close()194 raise HTTPException(404, "Listing not found")195 d = _row_to_dict(row)196 d["price_history"] = [dict(r) for r in con.execute(197 "SELECT ts, price FROM price_log WHERE uid=? ORDER BY ts DESC LIMIT 20",198 (uid,)).fetchall()]199 # same MLS number / same property published elsewhere ("Also listed on…")200 d["duplicates"] = [dict(r) for r in con.execute(201 "SELECT uid, source, url, brokerage_name, agent_name, price_label"202 " FROM listings WHERE dup_of=? AND active=1 ORDER BY source",203 (uid,)).fetchall()]204 # PROPERTY: the physical asset behind this listing + its listing history205 d["property"] = None206 if d.get("property_id"):207 p = con.execute("SELECT * FROM properties WHERE id=?",208 (d["property_id"],)).fetchone()209 if p is not None:210 prop = dict(p)211 prop["details"] = json.loads(prop.get("details") or "{}")212 prop["listing_history"] = [dict(r) for r in con.execute(213 "SELECT uid, source, status, list_price, first_seen, last_seen,"214 " active FROM listings WHERE property_id=? ORDER BY first_seen DESC",215 (d["property_id"],)).fetchall()]216 d["property"] = prop217 con.close()218 return d219220221@app.get("/api/properties/{pid}")222def get_property(pid: int):223 con = db.connect()224 p = con.execute("SELECT * FROM properties WHERE id=?", (pid,)).fetchone()225 if p is None:226 con.close()227 raise HTTPException(404, "Property not found")228 prop = dict(p)229 prop["details"] = json.loads(prop.get("details") or "{}")230 prop["listings"] = [_row_to_dict(r) for r in con.execute(231 "SELECT * FROM listings WHERE property_id=? ORDER BY first_seen DESC",232 (pid,)).fetchall()]233 con.close()234 return prop235236237@app.get("/api/facets")238def facets(state: str | None = None):239 """Distinct values for the frontend filters."""240 con = db.connect()241 city_sql = ("SELECT city, COUNT(*) n FROM listings WHERE active=1"242 " AND city<>''" + DEDUP_CLAUSE)243 city_args: list = []244 if state:245 city_sql += " AND state=?"246 city_args.append(state.upper())247 out = {248 "states": [dict(r) for r in con.execute(249 "SELECT state, COUNT(*) n FROM listings WHERE active=1 AND state<>''"250 + DEDUP_CLAUSE + " GROUP BY state ORDER BY n DESC")],251 "cities": [dict(r) for r in con.execute(252 city_sql + " GROUP BY city ORDER BY n DESC LIMIT 400", city_args)],253 "property_types": [r["property_type"] for r in con.execute(254 "SELECT property_type FROM listings WHERE active=1"255 " AND property_type<>''" + DEDUP_CLAUSE +256 " GROUP BY property_type ORDER BY COUNT(*) DESC")],257 "sources": [dict(r) for r in con.execute(258 "SELECT source, COUNT(*) n FROM listings WHERE active=1"259 + DEDUP_CLAUSE + " GROUP BY source ORDER BY n DESC")],260 }261 con.close()262 return out263264265@app.get("/api/sources")266def sources_public():267 con = db.connect()268 registry = db.get_sources(con)269 counts = {r["source"]: r["n"] for r in con.execute(270 "SELECT source, COUNT(*) n FROM listings WHERE active=1 GROUP BY source")}271 last = {r["source"]: r["ts"] for r in con.execute(272 "SELECT source, MAX(ts) ts FROM sync_log WHERE ok=1 GROUP BY source")}273 con.close()274 out = []275 for s in registry:276 out.append({"id": s["id"], "name": s["name"],277 "connector_type": s["connector_type"],278 "states": s["states"], "enabled": s["enabled"],279 "active_listings": counts.get(s["id"], 0),280 "last_sync": last.get(s["id"])})281 return {"sources": out}282283284@app.get("/api/stats")285def stats():286 con = db.connect()287 row = con.execute(288 """SELECT COUNT(*) total,289 COUNT(DISTINCT source) sources,290 COUNT(DISTINCT city) cities,291 COUNT(DISTINCT state) states,292 AVG(list_price) avg_price,293 MIN(list_price) min_price,294 MAX(list_price) max_price295 FROM listings WHERE active=1""" + DEDUP_CLAUSE).fetchone()296 props = con.execute("SELECT COUNT(*) c FROM properties").fetchone()["c"]297 log = [dict(r) for r in con.execute(298 "SELECT * FROM sync_log ORDER BY ts DESC LIMIT 20")]299 by_state = [dict(r) for r in con.execute(300 "SELECT state, COUNT(*) n, ROUND(AVG(list_price)) avg_price"301 " FROM listings WHERE active=1 AND state<>''" + DEDUP_CLAUSE +302 " GROUP BY state ORDER BY n DESC")]303 by_type = [dict(r) for r in con.execute(304 "SELECT property_type, COUNT(*) n, ROUND(AVG(list_price)) avg_price"305 " FROM listings WHERE active=1 AND property_type<>''" + DEDUP_CLAUSE +306 " GROUP BY property_type ORDER BY n DESC")]307 from . import quality308 qual = quality.summary(con)309 con.close()310 return {**dict(row), "properties": props, "by_state": by_state,311 "by_type": by_type, "quality": qual, "recent_syncs": log}312313314@app.post("/api/sync")315def trigger_sync(background: BackgroundTasks, source: str | None = None):316 """Trigger a sync (incoming-webhook equivalent)."""317 def _job():318 with _sync_lock:319 ingest.run([source] if source else None)320 background.add_task(_job)321 return {"status": "started", "source": source or "all"}322323324# =============================================================================325# Admin API — /admin/sources, /admin/brokerages, /admin/connectors326# Optional bearer: set HOMEKA_ADMIN_TOKEN to require X-Admin-Token (or ?token=).327# =============================================================================328329def _check_admin(request: Request) -> None:330 required = os.environ.get("HOMEKA_ADMIN_TOKEN", "")331 if not required:332 return333 given = request.headers.get("X-Admin-Token") or request.query_params.get("token")334 if given != required:335 raise HTTPException(401, "admin token required")336337338@app.get("/api/admin/overview")339def admin_overview(request: Request):340 _check_admin(request)341 con = db.connect()342 now = time.time()343 listings = con.execute(344 "SELECT COUNT(*) c FROM listings WHERE active=1").fetchone()["c"]345 props = con.execute("SELECT COUNT(*) c FROM properties").fetchone()["c"]346 sources_on = con.execute(347 "SELECT COUNT(*) c FROM sources WHERE enabled=1").fetchone()["c"]348 errors_24h = con.execute(349 "SELECT COUNT(*) c FROM sync_log WHERE ok=0 AND ts>?",350 (now - 86400,)).fetchone()["c"]351 brok = [dict(r) for r in con.execute(352 "SELECT partnership_status, COUNT(*) n FROM brokerages"353 " GROUP BY partnership_status ORDER BY n DESC")]354 freshest = con.execute(355 "SELECT MAX(ts) ts FROM sync_log WHERE ok=1").fetchone()["ts"]356 con.close()357 return {"active_listings": listings, "properties": props,358 "enabled_sources": sources_on, "errors_24h": errors_24h,359 "brokerages_by_status": brok, "last_successful_sync": freshest}360361362@app.get("/api/admin/sources")363def admin_sources(request: Request):364 """Active sources: listing counts, errors, last sync, data freshness,365 connector type, geographic coverage."""366 _check_admin(request)367 con = db.connect()368 registry = {s["id"]: s for s in db.get_sources(con)}369 # custom-coded connectors without a DB row still show up370 from . import connectors as creg371 for sid in creg.CUSTOM:372 registry.setdefault(sid, {373 "id": sid, "name": sid, "connector_type": "custom", "config": {},374 "enabled": 1, "authority": 2, "states": [], "notes": "",375 "brokerage_id": None})376 counts = {r["source"]: dict(r) for r in con.execute(377 "SELECT source, COUNT(*) n, SUM(published) published,"378 " SUM(CASE WHEN lat IS NULL THEN 1 ELSE 0 END) no_geo"379 " FROM listings WHERE active=1 GROUP BY source")}380 lastlog: dict[str, dict] = {}381 for r in con.execute(382 "SELECT * FROM sync_log WHERE id IN"383 " (SELECT MAX(id) FROM sync_log GROUP BY source)"):384 lastlog[r["source"]] = dict(r)385 err7d = {r["source"]: r["c"] for r in con.execute(386 "SELECT source, COUNT(*) c FROM sync_log WHERE ok=0 AND ts>?"387 " GROUP BY source", (time.time() - 7 * 86400,))}388 con.close()389 out = []390 for sid, s in sorted(registry.items()):391 c = counts.get(sid, {})392 log = lastlog.get(sid, {})393 cfg = dict(s.get("config") or {})394 for k in list(cfg): # never expose credentials/config secrets395 if any(w in k.lower() for w in ("token", "secret", "pass", "key")):396 cfg[k] = "•••"397 out.append({398 "id": sid, "name": s.get("name") or sid,399 "connector_type": s.get("connector_type"),400 "enabled": s.get("enabled", 1), "authority": s.get("authority"),401 "states": s.get("states") or [], "notes": s.get("notes") or "",402 "brokerage_id": s.get("brokerage_id"), "config": cfg,403 "active_listings": c.get("n", 0),404 "published": c.get("published", 0),405 "no_geo": c.get("no_geo", 0),406 "last_sync": log.get("ts"), "last_ok": log.get("ok"),407 "last_message": log.get("message"),408 "last_found": log.get("found"), "last_added": log.get("added"),409 "last_removed": log.get("removed"),410 "errors_7d": err7d.get(sid, 0),411 "freshness_hours": (round((time.time() - log["ts"]) / 3600, 1)412 if log.get("ts") else None),413 })414 return {"sources": out}415416417@app.get("/api/admin/connectors")418def admin_connectors(request: Request):419 """Registered connector families + custom connectors + instance counts."""420 _check_admin(request)421 from . import connectors as creg422 con = db.connect()423 inst = {r["connector_type"]: r["n"] for r in con.execute(424 "SELECT connector_type, COUNT(*) n FROM sources GROUP BY connector_type")}425 con.close()426 families = []427 for fam, cls in sorted(creg.FAMILIES.items()):428 families.append({429 "family": fam, "class": cls.__name__,430 "module": cls.__module__,431 "doc": (cls.__doc__ or "").strip().split("\n")[0],432 "instances": inst.get(fam, 0),433 })434 custom = [{"source_id": sid, "class": cls.__name__, "module": cls.__module__}435 for sid, cls in sorted(creg.CUSTOM.items())]436 return {"families": families, "custom": custom}437438439@app.get("/api/admin/brokerages")440def admin_brokerages(441 request: Request,442 state: str | None = None,443 status: str | None = None,444 feed_type: str | None = None,445 idx_provider: str | None = None,446 q: str | None = None,447 min_priority: float | None = None,448 inspected: int | None = None,449 sort: str = "priority", # priority | feed | name | inspected450 limit: int = Query(50, le=500),451 offset: int = 0,452):453 """Potential brokerage partners: scores, detected stack, contacts,454 partnership status."""455 _check_admin(request)456 con = db.connect()457 sql = "SELECT * FROM brokerages WHERE 1=1"458 args: list = []459 if state:460 sql += " AND states LIKE ?"; args.append(f'%"{state.upper()}"%')461 if status:462 sql += " AND partnership_status=?"; args.append(status)463 if feed_type:464 sql += " AND possible_feed_type=?"; args.append(feed_type)465 if idx_provider:466 sql += " AND idx_provider LIKE ?"; args.append(f"%{idx_provider}%")467 if q:468 sql += " AND (name LIKE ? OR website LIKE ?)"; args += [f"%{q}%"] * 2469 if min_priority is not None:470 sql += " AND priority_score>=?"; args.append(min_priority)471 if inspected == 1:472 sql += " AND last_inspected IS NOT NULL"473 elif inspected == 0:474 sql += " AND last_inspected IS NULL"475 total = con.execute(f"SELECT COUNT(*) c FROM ({sql})", args).fetchone()["c"]476 order = {477 "priority": " ORDER BY priority_score DESC NULLS LAST",478 "feed": " ORDER BY feed_probability_score DESC NULLS LAST",479 "name": " ORDER BY name",480 "inspected": " ORDER BY last_inspected DESC NULLS LAST",481 }.get(sort, " ORDER BY priority_score DESC NULLS LAST")482 rows = []483 for r in con.execute(sql + order + " LIMIT ? OFFSET ?",484 args + [limit, offset]):485 d = dict(r)486 for k in ("states", "cities", "mls_affiliations", "evidence"):487 d[k] = json.loads(d.get(k) or ("[]" if k != "evidence" else "{}"))488 rows.append(d)489 con.close()490 return {"total": total, "count": len(rows), "brokerages": rows}491492493@app.post("/api/admin/brokerages/{bid}/inspect")494def admin_inspect(bid: int, request: Request):495 _check_admin(request)496 from . import discovery497 con = db.connect()498 res = discovery.inspect_one(con, bid)499 con.close()500 return res501502503@app.post("/api/admin/brokerages/{bid}/status")504def admin_brokerage_status(bid: int, request: Request, payload: dict = Body(...)):505 _check_admin(request)506 status = str(payload.get("status") or "")507 allowed = {"prospect", "to-contact", "contacted", "in-discussion",508 "feed-received", "live", "declined"}509 if status not in allowed:510 raise HTTPException(400, f"status must be one of {sorted(allowed)}")511 con = db.connect()512 con.execute("UPDATE brokerages SET partnership_status=?, updated=? WHERE id=?",513 (status, time.time(), bid))514 con.commit()515 con.close()516 return {"id": bid, "status": status}517518519@app.post("/api/admin/discover")520def admin_discover(request: Request, background: BackgroundTasks,521 limit: int = 25):522 """Run a discovery batch (inspect the next prospects) in the background."""523 _check_admin(request)524 from . import discovery525 background.add_task(discovery.run_batch, limit)526 return {"status": "started", "limit": limit}527528529# --- robots + static frontend -------------------------------------------------530531@app.get("/robots.txt")532def robots():533 return PlainTextResponse("User-agent: *\nAllow: /\n"534 "Sitemap: https://www.home-ka.com/sitemap.xml\n")535536537if FRONTEND_DIR.exists():538539 if (FRONTEND_DIR / "assets").is_dir():540 app.mount("/assets", StaticFiles(directory=FRONTEND_DIR / "assets"),541 name="assets")542543 @app.get("/{full_path:path}")544 def spa(full_path: str):545 target = FRONTEND_DIR / full_path546 if full_path and ".." not in full_path and target.is_file():547 return FileResponse(target)548 index = FRONTEND_DIR / "index.html"549 if index.exists():550 return FileResponse(index)551 raise HTTPException(404)552