`scripts/ingest-permits.py`

101 lines
#!/usr/bin/env python3
"""
Construction-pipeline layer: ingest NEW-RESIDENTIAL building permits from free,
no-key city open-data APIs (Socrata SODA) into the building_permits table.
Leading indicator — these homes are being built months before they list.
$0, no key. West Coast metros first. Reversible (INSERT ... ON CONFLICT DO NOTHING).
"""
import json, urllib.request, urllib.parse, subprocess, sys

DBURL = "postgresql://macstudio3@localhost/homesonspec?host=/tmp"
UA = "HomesOnSpecBot/0.1 (+https://homesonspec.com/bot; contact: data@homesonspec.com)"

def num(v):
    try:
        f = float(str(v).replace("$", "").replace(",", "")); return f if f == f else None
    except: return None

# each city: (source, base_url, where, mapper(row)->dict)
def sf_map(r):
    loc = r.get("location") or {}
    lat = num(r.get("latitude")) or (num(loc.get("latitude")) if isinstance(loc, dict) else None)
    lon = num(r.get("longitude")) or (num(loc.get("longitude")) if isinstance(loc, dict) else None)
    if lat is None and isinstance(loc, dict) and loc.get("coordinates"):
        lon, lat = loc["coordinates"][0], loc["coordinates"][1]
    st = " ".join(x for x in [r.get("street_number"), r.get("street_name"), r.get("street_suffix")] if x)
    return dict(permit_number=r.get("permit_number") or r.get("permit_creation_date","")+st, address=st, city="San Francisco",
        state="CA", zip=r.get("zipcode"), lat=lat, lon=lon, permit_type=r.get("permit_type_definition"),
        work_class="New Construction", valuation=num(r.get("revised_cost") or r.get("estimated_cost")),
        status=r.get("status"), filed_date=(r.get("filed_date") or "")[:10] or None,
        issued_date=(r.get("issued_date") or "")[:10] or None, applicant=None)

def sea_map(r):
    return dict(permit_number=r.get("permitnum"), address=r.get("originaladdress1"), city=r.get("originalcity") or "Seattle",
        state=r.get("originalstate") or "WA", zip=r.get("originalzip"), lat=num(r.get("latitude")), lon=num(r.get("longitude")),
        permit_type=r.get("permittypedesc"), work_class=r.get("housingunitsadded") and "New Residential" or "New",
        valuation=num(r.get("estprojectcost")), status=r.get("statuscurrent"),
        filed_date=(r.get("applieddate") or "")[:10] or None, issued_date=(r.get("issueddate") or "")[:10] or None,
        applicant=r.get("contractorcompanyname"))

def la_map(r):  # LADBS hbkd-qubn — addresses (no lat/lon; geocode later), all permit types
    st = " ".join(x for x in [r.get("address_start"), r.get("street_name"), r.get("street_suffix")] if x)
    pt = " / ".join(x for x in [r.get("permit_type"), r.get("permit_sub_type")] if x)
    return dict(permit_number=r.get("pcis_permit"), address=st, city="Los Angeles", state="CA", zip=r.get("zip_code"),
        lat=None, lon=None, permit_type=pt, work_class=r.get("permit_sub_type"), valuation=num(r.get("valuation")),
        status="Issued" if r.get("issue_date") else None, filed_date=None,
        issued_date=(r.get("issue_date") or "")[:10] or None,
        applicant=r.get("contractors_business_name") or " ".join(x for x in [r.get("applicant_first_name"), r.get("applicant_last_name")] if x))

# (source, url, where, mapper, order_field, max_rows)  — LA = ALL permit types (commercial + residential)
CITIES = [
    ("la", "https://data.lacity.org/resource/hbkd-qubn.json",
     "issue_date > '2020-01-01'", la_map, "issue_date", 600000),
    ("sf",   "https://data.sfgov.org/resource/i98e-djp9.json",
     "permit_type_definition like '%new construction%' AND filed_date > '2022-01-01'", sf_map, "filed_date", 8000),
    ("seattle", "https://data.seattle.gov/resource/76t5-zqzr.json",
     "permitclassmapped='Residential' AND permittypedesc like '%New%' AND issueddate > '2022-01-01'", sea_map, "issueddate", 8000),
]

def esc(v):
    if v is None: return "NULL"
    if isinstance(v, (int, float)): return str(v)
    return "'" + str(v).replace("'", "''") + "'"

def main():
    total = 0
    only = sys.argv[1] if len(sys.argv) > 1 else None
    for source, base, where, mapper, order, max_rows in CITIES:
        if only and source != only: continue
        got = 0; offset = 0
        while offset < max_rows:
            q = {"$limit": 1000, "$offset": offset, "$where": where, "$order": order + " DESC"}
            url = base + "?" + urllib.parse.urlencode(q)
            try:
                req = urllib.request.Request(url, headers={"User-Agent": UA})
                rows = json.load(urllib.request.urlopen(req, timeout=60))
            except Exception as e:
                print(f"  {source} offset {offset} error: {e}"); break
            if not rows: break
            vals = []
            for r in rows:
                m = mapper(r)
                if not m.get("permit_number"): continue
                pid = f"{source}-{m['permit_number']}"
                cols = ["id","source","permit_number","address","city","state","zip","lat","lon","permit_type","work_class","valuation","status","filed_date","issued_date","applicant"]
                row = [pid, source, m["permit_number"], m.get("address"), m.get("city"), m.get("state"), m.get("zip"),
                       m.get("lat"), m.get("lon"), m.get("permit_type"), m.get("work_class"), m.get("valuation"),
                       m.get("status"), m.get("filed_date"), m.get("issued_date"), m.get("applicant")]
                vals.append("(" + ",".join(esc(x) for x in row) + ")")
            if vals:
                sql = ('insert into building_permits (id,source,permit_number,address,city,state,zip,lat,lon,permit_type,work_class,valuation,status,filed_date,issued_date,applicant) values '
                       + ",".join(vals) + " on conflict (id) do nothing;")
                subprocess.run(["psql", DBURL, "-q", "-c", sql], capture_output=True)
            got += len(rows); offset += 1000
            if len(rows) < 1000: break
        print(f"  {source}: fetched {got} new-residential permits")
        total += got
    print(f"total permit rows fetched: {total}")

if __name__ == "__main__":
    main()