# ----------------------------------------------------------------------------- # Home-Ka — US real-estate aggregator (Groupe KA) # Author: Simon-Pierre Boucher — contact@spboucher.ai # connectors/public_data/arcgis.py : county assessor / GIS parcel connector. # # US counties publish parcel/assessor layers on ArcGIS FeatureServers (open # data portals) — the lowest-level public source there is. This connector # feeds the PROPERTIES table (physical assets: APN, address, year built, # areas, assessed values), NOT the listings table: it enriches property # matching and keeps properties alive after listings go off-market. # # {"id": "county_travis_tx", "connector_type": "arcgis", "config": { # "layer_url": "https://services.arcgis.com/.../FeatureServer/0", # "where": "1=1", "out_fields": "*", "page_size": 2000, # "max_records": 200000, # "mapping": {"apn": "PROP_ID", "street_address": "SITUS_ADDR", # "city": "SITUS_CITY", "zip_code": "SITUS_ZIP", # "year_built": "YEAR_BUILT", "living_area_sqft": "LIVING_AREA", # "assessed_value": "ASSESSED_VAL"}, # "static": {"state": "TX", "county": "Travis"}}} # ----------------------------------------------------------------------------- from __future__ import annotations import json import time from ..base import BaseConnector from .._maputil import dig from ...schema import Listing # properties-table columns a mapping can target; everything else goes to details _PROP_FIELDS = {"apn", "street_address", "unit", "city", "state", "zip_code", "county", "property_type", "year_built", "living_area_sqft", "lot_size_sqft", "lat", "lng"} class ArcGISParcelConnector(BaseConnector): family = "arcgis" request_delay = 0.4 is_public_records = True # ingest.py routes fetch_records() → properties def fetch(self) -> list[Listing]: return [] # public records never create listings def fetch_records(self) -> list[dict]: """Yield property records {column: value, 'details': {...}}.""" cfg = self.config layer = (cfg.get("layer_url") or "").rstrip("/") if not layer: raise RuntimeError(f"{self.source_id}: layer_url missing") mapping = cfg.get("mapping") or {} static = cfg.get("static") or {} page = int(cfg.get("page_size", 2000)) max_records = int(cfg.get("max_records", 200_000)) offset = 0 out: list[dict] = [] while offset < max_records: params = { "f": "json", "where": cfg.get("where", "1=1"), "outFields": cfg.get("out_fields", "*"), "resultOffset": offset, "resultRecordCount": page, # polygons are heavy: ask for centroids only by default "returnGeometry": "true" if cfg.get("return_geometry") else "false", "returnCentroid": "true", "outSR": 4326, } data = self.get(f"{layer}/query", params=params).json() feats = data.get("features") or [] if not feats: break for f in feats: attrs = f.get("attributes") or {} rec: dict = {"details": {}} for field, path in mapping.items(): if "{" in str(path): # template: "{LOCN} {LOCS} {LOCT}" safe = {k: ("" if v is None else v) for k, v in attrs.items()} try: v = " ".join(str(path).format(**safe).split()) except (KeyError, IndexError): v = None else: v = dig(attrs, path) if v in (None, ""): continue if field in _PROP_FIELDS: rec[field] = v else: rec["details"][field] = v for field, v in static.items(): rec.setdefault(field, v) geom = f.get("centroid") or f.get("geometry") or {} if "lat" not in rec and geom.get("y") is not None: rec["lat"], rec["lng"] = geom.get("y"), geom.get("x") if rec.get("apn") or rec.get("street_address"): out.append(rec) if not data.get("exceededTransferLimit") and len(feats) < page: break offset += len(feats) return out def sync_records(con, source_id: str, records: list[dict]) -> dict: """Upsert public-records rows into `properties` (match APN, then address).""" from ... import propertymatch from ...normalize import clean_title, normalize_zip now = time.time() added = updated = 0 for rec in records: # assessor exports: ZIP as float ("34446.0"), SHOUTING city names if rec.get("zip_code") is not None: rec["zip_code"] = normalize_zip(rec["zip_code"]) if rec.get("city"): rec["city"] = clean_title(str(rec["city"]).strip().title()) if rec.get("street_address"): rec["street_address"] = str(rec["street_address"]).strip().title() state = str(rec.get("state") or "") apn = str(rec.get("apn") or "").replace(" ", "") # address+unit key FIRST — APNs are not unit-unique for condos # (whole buildings share one assessor account in several counties) key = propertymatch.addr_key(str(rec.get("street_address") or ""), str(rec.get("unit") or ""), str(rec.get("city") or ""), state, str(rec.get("zip_code") or "")) row = None if key: row = con.execute("SELECT id, details FROM properties" " WHERE addr_key=?", (key,)).fetchone() elif apn and state: row = con.execute("SELECT id, details FROM properties" " WHERE apn=? AND state=?", (apn, state)).fetchone() details = rec.get("details") or {} if row is None: if not key and not apn: continue con.execute( """INSERT OR IGNORE INTO properties (addr_key, apn, street_address, unit, city, state, zip_code, county, property_type, year_built, living_area_sqft, lot_size_sqft, lat, lng, details, first_seen, last_seen) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (key, apn or None, rec.get("street_address"), rec.get("unit"), rec.get("city"), state, rec.get("zip_code"), rec.get("county"), rec.get("property_type"), rec.get("year_built"), rec.get("living_area_sqft"), rec.get("lot_size_sqft"), rec.get("lat"), rec.get("lng"), json.dumps(details, ensure_ascii=False, default=str), now, now)) added += 1 else: old = json.loads(row["details"] or "{}") old.update(details) con.execute( """UPDATE properties SET apn=COALESCE(NULLIF(apn,''), ?), year_built=COALESCE(year_built, ?), living_area_sqft=COALESCE(living_area_sqft, ?), lot_size_sqft=COALESCE(lot_size_sqft, ?), lat=COALESCE(lat, ?), lng=COALESCE(lng, ?), details=?, last_seen=? WHERE id=?""", (apn or None, rec.get("year_built"), rec.get("living_area_sqft"), rec.get("lot_size_sqft"), rec.get("lat"), rec.get("lng"), json.dumps(old, ensure_ascii=False, default=str), now, row["id"])) updated += 1 con.commit() return {"source": source_id, "found": len(records), "added": added, "updated": updated, "removed": 0, "kind": "public-records"}