`enrich_live_contractor.py`

147 lines
#!/usr/bin/env python3
"""
enrich_live_contractor.py — attach LIVE contractor + license to current permits. $0, stdlib.

Breakthrough: LADBS's public "Permit & Inspection Report" detail page is a plain GET
keyed by the three permit-number segments:
  https://.../OnlineServices/PermitReport/PcisPermitDetail?id1=<5>&id2=<5>&id3=<5>
It renders "Contractor <name>; Lic. No.: <license>" — so we CAN recover contractor names
for CURRENT permits (which the open data feed strips), for free, no browser, no login.

Reads permits_live.csv (from pull_permits_live.py), fetches each permit's detail page
(cached + rate-limited to stay polite to the public endpoint), extracts contractor +
license, and writes permits_live_enriched.csv with SOS/CSLB lookup links.

Usage:
  python3 enrich_live_contractor.py                     # permits_live.csv -> permits_live_enriched.csv
  python3 enrich_live_contractor.py --in x.csv --delay 1.5 --limit 50
"""

import argparse
import csv
import html as _html
import json
import os
import re
import sys
import time
import urllib.parse
import urllib.request

DETAIL = "https://www.ladbsservices2.lacity.org/OnlineServices/PermitReport/PcisPermitDetail"
CACHE = ".ladbs_detail_cache.json"
CA_SOS = "https://bizfileonline.sos.ca.gov/search/business"
CSLB = "https://www.cslb.ca.gov/OnlineServices/CheckLicenseII/CheckLicense.aspx"

# Name ends at the first field delimiter: ";" (licensed GC) OR ", Inspector"/"Inspector
# Information"/", Owner" (owner-builder permits carry NO "; Lic. No.:" after the name).
RE_NAME = re.compile(r"Contractor\s+(.{2,90}?)\s*(?:;|,?\s*Inspector\b|,\s*Owner\b)", re.I)
RE_LIC = re.compile(r"Lic\.?\s*No\.?:\s*([A-Za-z0-9\-]+)", re.I)


def load_cache():
    if os.path.exists(CACHE):
        try:
            return json.load(open(CACHE))
        except Exception:
            return {}
    return {}


def save_cache(c):
    json.dump(c, open(CACHE, "w"))


def flatten(html):
    return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", html)).strip()


def fetch_detail(seg1, seg2, seg3):
    q = urllib.parse.urlencode({"id1": seg1, "id2": seg2, "id3": seg3})
    req = urllib.request.Request(DETAIL + "?" + q, headers={"User-Agent": "Mozilla/5.0", "Accept": "text/html"})
    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=45) as r:
                return r.read().decode("utf-8", "ignore")
        except Exception as e:
            if attempt == 2:
                print(f"  ! fetch failed {seg1}-{seg2}-{seg3}: {e}", file=sys.stderr)
                return ""
            time.sleep(2 * (attempt + 1))
    return ""


def parse_contractor(html):
    t = re.sub(r"\s+", " ", _html.unescape(flatten(html)))   # decode &#39; &amp; &nbsp;
    m = RE_NAME.search(t)
    if not m:
        return "", ""
    name = re.sub(r"\s+", " ", m.group(1)).strip(" .,")
    # owner self-permitted: report "Owner-Builder", no contractor license
    if name.lower().startswith("owner-builder"):
        return "Owner-Builder", ""
    # license lives in the ~60 chars right after the name (the GC's "; Lic. No.: <n>")
    tail = t[m.end() - 1: m.end() + 60]
    lm = RE_LIC.search(tail)
    return name, (lm.group(1).strip() if lm else "")


def sos_url(name):
    return CA_SOS + "?" + urllib.parse.urlencode({"searchType": "BUSINESS", "q": name}) if name else ""


def cslb_url(lic):
    return CSLB + "?" + urllib.parse.urlencode({"LicNum": lic}) if lic else CSLB


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--in", dest="infile", default="permits_live.csv")
    ap.add_argument("--out", default="permits_live_enriched.csv")
    ap.add_argument("--delay", type=float, default=1.0, help="seconds between fetches (be polite)")
    ap.add_argument("--limit", type=int, default=0, help="cap rows (0 = all)")
    args = ap.parse_args()

    rows = list(csv.DictReader(open(args.infile, newline="", encoding="utf-8")))
    if args.limit:
        rows = rows[: args.limit]
    cache = load_cache()

    out_fields = list(rows[0].keys()) + ["contractor_name", "contractor_license",
                                         "ca_sos_lookup_url", "cslb_lookup_url"] if rows else []
    hits = 0
    with open(args.out, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=out_fields)
        w.writeheader()
        for i, r in enumerate(rows, 1):
            pn = (r.get("permit_nbr") or "").strip()
            segs = pn.split("-")
            name, lic = "", ""
            if len(segs) == 3:
                key = pn
                if key in cache:
                    name, lic = cache[key]["name"], cache[key]["lic"]
                else:
                    html = fetch_detail(*segs)
                    name, lic = parse_contractor(html)
                    cache[key] = {"name": name, "lic": lic}
                    save_cache(cache)
                    time.sleep(args.delay)  # polite pacing on the public endpoint
            if name:
                hits += 1
            r = dict(r)
            r["contractor_name"] = name
            r["contractor_license"] = lic
            r["ca_sos_lookup_url"] = sos_url(name)
            r["cslb_lookup_url"] = cslb_url(lic)
            w.writerow(r)
            print(f"  [{i}/{len(rows)}] {pn}  ->  {name or '(none)'}"
                  + (f"  Lic {lic}" if lic else ""), flush=True)

    print(f"\nLive contractor enrichment  —  cost: $0 (public LADBS detail GET, cached)")
    print(f"  {hits}/{len(rows)} permits resolved to a contractor -> {args.out}")


if __name__ == "__main__":
    main()