← back to Labj Cre Banks
ingest/socrata/ingest.py
263 lines
#!/usr/bin/env python3
"""
LA Socrata CRE-prospect ingester.
Pulls two free, keyless LA City open-data (Socrata / SODA) datasets and turns them
into de-duplicated CRE advertiser-prospect firm records:
1. LADBS Building Permits (d9aa-v8bm) — Commercial + Apartment permits, recent,
with a named contractor -> role=Contractor (+ applicant person, if owner).
2. Listing of Active Businesses (6rrh-rzua) — NAICS-filtered to CRE-relevant firms
(RE brokers, property managers, RE activities, commercial/multifamily builders,
architecture/engineering) -> role by NAICS.
Output: a Postgres staging table `socrata_cre_prospects` in the local `realestate`
DB (same DB as rentv_licensed_targets) via TSV + psql \\copy + upsert, PLUS a JSONL
staging file. Firms already in the curated LABJ set (labj_master.json) are skipped so
this surfaces only NET-NEW names.
Zero third-party deps: urllib (fetch) + psql (load). Free ($0) + keyless; set
SOCRATA_APP_TOKEN to avoid throttling on large pulls.
Usage:
python3 ingest.py # bounded default pull, load to PG
python3 ingest.py --businesses-only
python3 ingest.py --permits-only
python3 ingest.py --max-businesses 20000 --permit-months 18 --min-valuation 250000
python3 ingest.py --dry-run # fetch + normalize + dedup, no DB write
"""
import argparse, json, os, re, subprocess, sys, urllib.parse, urllib.request, datetime
BASE = "https://data.lacity.org/resource"
PERMITS_ID = "d9aa-v8bm"
BUSINESS_ID = "6rrh-rzua"
APP_TOKEN = os.environ.get("SOCRATA_APP_TOKEN", "")
PGDB = os.environ.get("PGDATABASE", "realestate")
PGHOST = os.environ.get("PGHOST", "/tmp")
HERE = os.path.dirname(os.path.abspath(__file__))
CURATED = os.path.join(HERE, "..", "..", "data", "labj_master.json")
TABLE = "socrata_cre_prospects"
# NAICS prefix -> role for the Active Businesses pull. "Lessors" (5311) is huge and
# low-signal per-record, so it's OFF by default (enable with --include-lessors).
NAICS_ROLE = {
"5312": "Broker", # offices of real estate agents & brokers
"5313": "Property Manager", # property managers / other RE activities
"2362": "Developer/Builder", # nonresidential building construction
"23611": "Developer/Builder",# residential (incl. multifamily) construction
"5413": "Architect/Engineer",# architectural, engineering & related services
}
LESSOR_PREFIX = "5311"
SUF = re.compile(r"\b(inc|llc|llp|lp|corp|corporation|company|co|ltd|the)\b")
def norm(s):
s = SUF.sub(" ", re.sub(r"[^a-z0-9 ]", " ", (s or "").lower()))
return re.sub(r"\s+", " ", s).strip()
def clean(v):
"""TSV-safe single-line string."""
if v is None:
return ""
return re.sub(r"[\t\r\n]+", " ", str(v)).strip()
def soda_get(dataset, params, cap):
"""Paged SODA fetch (limit/offset). Returns list of dict rows, up to cap."""
rows, offset, page = [], 0, 5000
while len(rows) < cap:
q = dict(params)
q["$limit"] = min(page, cap - len(rows))
q["$offset"] = offset
url = f"{BASE}/{dataset}.json?" + urllib.parse.urlencode(q, safe="()' ,>=<")
req = urllib.request.Request(url, headers={"Accept": "application/json",
**({"X-App-Token": APP_TOKEN} if APP_TOKEN else {})})
with urllib.request.urlopen(req, timeout=60) as r:
batch = json.load(r)
if not batch:
break
rows.extend(batch)
offset += len(batch)
if len(batch) < q["$limit"]:
break
return rows
def role_for_naics(naics):
n = (naics or "").strip()
for pref, role in NAICS_ROLE.items():
if n.startswith(pref):
return role
if n.startswith(LESSOR_PREFIX):
return "Property Owner (Lessor)"
return "Other RE"
def fetch_businesses(cap, include_lessors):
prefixes = list(NAICS_ROLE.keys()) + ([LESSOR_PREFIX] if include_lessors else [])
where = " OR ".join(f"starts_with(naics,'{p}')" for p in prefixes)
where = f"({where}) AND business_name IS NOT NULL"
rows = soda_get(BUSINESS_ID, {
"$select": "location_account,business_name,street_address,city,zip_code,naics,"
"primary_naics_description,council_district,location_start_date",
"$where": where, "$order": "location_account",
}, cap)
out = []
for r in rows:
out.append({
"source": "la_active_businesses", "dataset_id": BUSINESS_ID,
"role": role_for_naics(r.get("naics")),
"firm_name": clean(r.get("business_name")), "contact_name": "",
"address": clean(r.get("street_address")), "city": clean(r.get("city")),
"state": "CA", "zip": clean(r.get("zip_code")),
"naics": clean(r.get("naics")), "naics_desc": clean(r.get("primary_naics_description")),
"permit_type": "", "permit_sub_type": "", "valuation": "",
"issue_date": clean(r.get("location_start_date"))[:10],
"work_description": "", "council_district": clean(r.get("council_district")),
"src_id": clean(r.get("location_account")),
})
return out
def permits_max_date():
"""This dataset (d9aa-v8bm) is a historical snapshot; window relative to ITS max
issue_date, not today, so the ingester always gets the freshest available permits.
(A live LADBS feed should replace this ID when sourced — see README.)"""
rows = soda_get(PERMITS_ID, {"$select": "max(issue_date)"}, 1)
m = (rows[0].get("max_issue_date") if rows else "") or "2023-05-19T00:00:00"
return datetime.date.fromisoformat(m[:10])
def fetch_permits(months, min_valuation, cap):
cutoff = permits_max_date() - datetime.timedelta(days=int(months * 30))
where = ("(permit_sub_type='Commercial' OR permit_sub_type='Apartment') "
"AND contractors_business_name IS NOT NULL "
f"AND issue_date > '{cutoff.isoformat()}T00:00:00'")
rows = soda_get(PERMITS_ID, {
"$select": "pcis_permit,permit_type,permit_sub_type,issue_date,valuation,work_description,"
"address_start,street_direction,street_name,street_suffix,zip_code,"
"contractors_business_name,contractor_address,contractor_city,contractor_state,"
"applicant_first_name,applicant_last_name,applicant_relationship,council_district",
"$where": where, "$order": "issue_date DESC",
}, cap)
out = []
for r in rows:
try:
val = float(re.sub(r"[^0-9.]", "", r.get("valuation") or "") or 0)
except ValueError:
val = 0
if val < min_valuation:
continue
appl = " ".join(x for x in [r.get("applicant_first_name"), r.get("applicant_last_name")] if x).strip()
out.append({
"source": "ladbs_permits", "dataset_id": PERMITS_ID, "role": "Contractor",
"firm_name": clean(r.get("contractors_business_name")),
"contact_name": clean(appl) + (f" ({clean(r.get('applicant_relationship'))})" if r.get("applicant_relationship") else ""),
"address": clean(r.get("contractor_address")), "city": clean(r.get("contractor_city")),
"state": clean(r.get("contractor_state")) or "CA", "zip": "",
"naics": "", "naics_desc": "",
"permit_type": clean(r.get("permit_type")), "permit_sub_type": clean(r.get("permit_sub_type")),
"valuation": str(int(val)),
"issue_date": clean(r.get("issue_date"))[:10],
"work_description": clean(r.get("work_description"))[:300],
"council_district": clean(r.get("council_district")),
"src_id": clean(r.get("pcis_permit")),
})
return out
COLS = ["dedup_key", "source", "dataset_id", "role", "firm_name", "contact_name",
"address", "city", "state", "zip", "naics", "naics_desc", "permit_type",
"permit_sub_type", "valuation", "issue_date", "work_description",
"council_district", "src_id"]
def load_pg(records, tsv_path):
with open(tsv_path, "w") as f:
for r in records:
f.write("\t".join(clean(r.get(c, "")) for c in COLS) + "\n")
ddl = f"""
CREATE TABLE IF NOT EXISTS {TABLE} (
dedup_key text PRIMARY KEY, source text, dataset_id text, role text,
firm_name text, contact_name text, address text, city text, state text, zip text,
naics text, naics_desc text, permit_type text, permit_sub_type text,
valuation text, issue_date text, work_description text, council_district text,
src_id text, first_seen timestamptz DEFAULT now(), last_seen timestamptz DEFAULT now()
);
CREATE TEMP TABLE _stg (LIKE {TABLE} INCLUDING DEFAULTS);
ALTER TABLE _stg DROP COLUMN first_seen, DROP COLUMN last_seen;
\\copy _stg ({','.join(COLS)}) FROM '{tsv_path}' WITH (FORMAT text, DELIMITER E'\\t');
INSERT INTO {TABLE} ({','.join(COLS)})
SELECT {','.join(COLS)} FROM _stg
ON CONFLICT (dedup_key) DO UPDATE SET last_seen = now(),
firm_name = EXCLUDED.firm_name, address = EXCLUDED.address,
role = EXCLUDED.role, naics_desc = EXCLUDED.naics_desc;
SELECT source, role, count(*) FROM {TABLE} GROUP BY 1,2 ORDER BY 1,3 DESC;
"""
p = subprocess.run(["psql", "-h", PGHOST, "-d", PGDB, "-v", "ON_ERROR_STOP=1"],
input=ddl, text=True, capture_output=True)
if p.returncode != 0:
print("PG load failed:\n", p.stderr, file=sys.stderr)
sys.exit(1)
print(p.stdout)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--businesses-only", action="store_true")
ap.add_argument("--permits-only", action="store_true")
ap.add_argument("--max-businesses", type=int, default=15000)
ap.add_argument("--max-permits", type=int, default=20000)
ap.add_argument("--permit-months", type=int, default=24)
ap.add_argument("--min-valuation", type=int, default=250000)
ap.add_argument("--include-lessors", action="store_true")
ap.add_argument("--dry-run", action="store_true")
a = ap.parse_args()
recs = []
if not a.permits_only:
print(f"Fetching Active Businesses (CRE NAICS, cap {a.max_businesses})…", flush=True)
recs += fetch_businesses(a.max_businesses, a.include_lessors)
if not a.businesses_only:
print(f"Fetching LADBS permits (Commercial/Apartment, last {a.permit_months}mo, ≥${a.min_valuation:,})…", flush=True)
recs += fetch_permits(a.permit_months, a.min_valuation, a.max_permits)
print(f"Raw records: {len(recs)}")
# dedup vs curated LABJ set + within-run
curated = set()
try:
for r in json.load(open(CURATED))["records"]:
curated.add(norm(r["company_name"]))
for al in (r.get("aliases") or "").split(";"):
if al.strip():
curated.add(norm(al))
except Exception as e:
print("warn: could not read curated set:", e)
seen, out, skip_cur = set(), [], 0
for r in recs:
nf = norm(r["firm_name"])
if not nf:
continue
if nf in curated:
skip_cur += 1
continue
key = f"{r['source']}|{nf}|{r['zip'] or r['city']}"
if key in seen:
continue
seen.add(key)
r["dedup_key"] = key
out.append(r)
print(f"After dedup: {len(out)} net-new (skipped {skip_cur} already-curated, "
f"{len(recs)-len(out)-skip_cur} intra-run dups)")
jsonl = os.path.join(HERE, "socrata_prospects.jsonl")
with open(jsonl, "w") as f:
for r in out:
f.write(json.dumps(r) + "\n")
print(f"Wrote staging JSONL: {jsonl}")
if a.dry_run:
from collections import Counter
c = Counter((r["source"], r["role"]) for r in out)
for (s, role), n in c.most_common():
print(f" {n:>6} {s} · {role}")
print("(dry-run — no DB write)")
return
load_pg(out, os.path.join(HERE, "socrata_prospects.tsv"))
if __name__ == "__main__":
main()