`pull_permits_live.py`

151 lines
#!/usr/bin/env python3
"""
pull_permits_live.py — LIVE, up-to-date LA building permits (updated ~daily). $0, stdlib only.

Uses the City's CURRENT open feed `pi9x-tg5x` (refreshed to within ~2 days of today),
filtered to significant commercial/new construction. Writes a clean CSV with, per permit,
a live LA County Assessor OWNER lookup (by APN) and an LADBS permit lookup (by permit #).

IMPORTANT — why this differs from pull_permits.py:
  The City STOPPED publishing contractor/applicant NAMES in the open feed after
  2023-05-19 (privacy/policy change). The only openly-updated feed (`pi9x-tg5x`) is
  LIVE but carries NO name fields. So for current data the party you enrich live is
  the OWNER/DEVELOPER (via the county Assessor by APN) — which for a loan officer is
  the more useful party anyway. The GC name, if needed, is a per-permit LADBS lookup.

Usage:
  python3 pull_permits_live.py                          # commercial >$1M, last 365d -> permits_live.csv
  python3 pull_permits_live.py --min-valuation 5000000 --days 90
  python3 pull_permits_live.py --permit-types "Bldg-New" --out new_live.csv
"""

import argparse
import csv
import datetime as _dt          # only for arg parsing of --days offset input by the user
import sys
import time
import urllib.parse
import urllib.request

DATASET = "pi9x-tg5x"
BASE = "https://data.lacity.org/resource/{}.json".format(DATASET)
PAGE = 50000
# LA County Assessor parcel detail (current owner) by 10-digit AIN (= apn, no dashes):
ASSESSOR = "https://portal.assessor.lacounty.gov/parceldetail/"
# LADBS Online Building Records search (find the permit -> contractor):
LADBS = "https://www.ladbsservices2.lacity.org/OnlineServices/OnlineServices/OnlineBuildingRecords"

OUT_FIELDS = [
    "permit_nbr", "issue_date", "status_desc", "permit_type", "permit_sub_type",
    "use_desc", "valuation", "square_footage",
    "address", "zip_code", "council_district", "apn",
    "assessor_owner_url", "ladbs_permit_url",
]


def build_where(min_valuation, permit_types, since):
    clauses = ["valuation IS NOT NULL"]
    if min_valuation:
        # valuation is stored as TEXT on the live feed -> cast to number to compare
        clauses.append(f"(valuation::number) > {int(min_valuation)}")
    if permit_types:
        quoted = ",".join("'" + t.replace("'", "''") + "'" for t in permit_types)
        clauses.append(f"permit_type in({quoted})")
    if since:
        clauses.append(f"issue_date >= '{since}T00:00:00'")
    return " and ".join(clauses)


def fetch_page(where, offset, app_token=None):
    import json
    params = {"$where": where, "$order": "(valuation::number) DESC", "$limit": PAGE, "$offset": offset}
    if app_token:
        params["$$app_token"] = app_token
    url = BASE + "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    for attempt in range(4):
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                data = json.load(r)
            if isinstance(data, dict):  # Socrata error object
                print("  ! API error:", data.get("message", data), file=sys.stderr)
                return []
            return data
        except Exception as e:
            if attempt == 3:
                print(f"  ! giving up on offset {offset}: {e}", file=sys.stderr)
                return []
            time.sleep(2 * (attempt + 1))
    return []


def shape(row):
    apn = (row.get("apn") or "").strip()
    return {
        "permit_nbr": row.get("permit_nbr", ""),
        "issue_date": (row.get("issue_date") or "")[:10],
        "status_desc": row.get("status_desc", ""),
        "permit_type": row.get("permit_type", ""),
        "permit_sub_type": row.get("permit_sub_type", ""),
        "use_desc": row.get("use_desc", ""),
        "valuation": row.get("valuation", ""),
        "square_footage": row.get("square_footage", ""),
        "address": row.get("primary_address", ""),
        "zip_code": row.get("zip_code", ""),
        "council_district": row.get("cd", ""),
        "apn": apn,
        "assessor_owner_url": (ASSESSOR + apn) if apn else "",
        "ladbs_permit_url": LADBS,
    }


def main():
    ap = argparse.ArgumentParser(description="LIVE LA permits from the current open feed.")
    ap.add_argument("--min-valuation", type=int, default=1_000_000, help="min $ (default 1000000; 0 = off)")
    ap.add_argument("--permit-types", default="Bldg-New,Bldg-Addition",
                    help="comma list; '' = all. e.g. Bldg-New, Bldg-Addition, Bldg-Alter/Repair")
    ap.add_argument("--days", type=int, default=365, help="only permits issued in the last N days (0 = all)")
    ap.add_argument("--since", default="", help="explicit YYYY-MM-DD floor (overrides --days)")
    ap.add_argument("--limit", type=int, default=0, help="cap rows (0 = all)")
    ap.add_argument("--out", default="permits_live.csv")
    ap.add_argument("--app-token", default="")
    args = ap.parse_args()

    since = args.since
    if not since and args.days:
        since = (_dt.date.today() - _dt.timedelta(days=args.days)).isoformat()

    permit_types = [t.strip() for t in args.permit_types.split(",") if t.strip()]
    where = build_where(args.min_valuation if args.min_valuation > 0 else 0, permit_types, since)

    print("LIVE LADBS permit puller  —  cost: $0 (current LA Open Data feed pi9x-tg5x)")
    print(f"  filter : {where}")
    print(f"  output : {args.out}")

    written, offset = 0, 0
    with open(args.out, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=OUT_FIELDS)
        w.writeheader()
        while True:
            rows = fetch_page(where, offset, args.app_token or None)
            if not rows:
                break
            for row in rows:
                w.writerow(shape(row))
                written += 1
                if args.limit and written >= args.limit:
                    break
            print(f"  ... {written} rows", flush=True)
            if (args.limit and written >= args.limit) or len(rows) < PAGE:
                break
            offset += PAGE
            time.sleep(0.3)

    print(f"\nDone. {written} LIVE permits -> {args.out}")
    print("Enrich (free, live): open assessor_owner_url for the CURRENT owner/developer by APN;")
    print("open ladbs_permit_url and search the permit # for the contractor of record.")


if __name__ == "__main__":
    main()