`enrich_contractors.py`

139 lines
#!/usr/bin/env python3
"""
enrich_contractors.py — dedupe permits.csv down to UNIQUE contractors and attach
free entity-lookup links (school project). $0, standard library only.

Why this shape: the CA Secretary of State bizfile site is behind Imperva/Incapsula
bot protection (blocks plain HTTP), OpenCorporates now needs a paid token, and the
CSLB license page is session-gated. So there is no reliable *free* HTTP endpoint to
auto-scrape entity data. The high-leverage move instead:

  1. collapse thousands of permit rows to the handful of UNIQUE contractors,
  2. rank them by total project value (the "top contractors" list you actually want),
  3. hand each one ready-to-click lookups for CA SOS (by name) AND CSLB (by license).

You then click a small, deduped worklist — not thousands of rows.

Usage:
  python3 enrich_contractors.py                        # reads permits.csv -> contractors.csv
  python3 enrich_contractors.py --in sample.csv --out firms.csv
  python3 enrich_contractors.py --min-permits 2        # only firms with >=2 permits
"""

import argparse
import csv
import re
import urllib.parse

CA_SOS_SEARCH = "https://bizfileonline.sos.ca.gov/search/business"
CSLB_LOOKUP = "https://www.cslb.ca.gov/OnlineServices/CheckLicenseII/CheckLicense.aspx"


def norm_name(name):
    """Normalize a business name for grouping (uppercase, collapse space/punct)."""
    n = (name or "").upper().strip()
    n = re.sub(r"[.,]", " ", n)
    n = re.sub(r"\s+", " ", n)
    return n.strip()


def money(v):
    try:
        return float(v)
    except (TypeError, ValueError):
        return 0.0


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


def cslb_url(license_no):
    # CSLB gates the direct license-detail page, so link the public lookup and
    # carry the number in its own column for a one-paste check.
    return CSLB_LOOKUP + "?" + urllib.parse.urlencode({"LicNum": license_no}) if license_no else CSLB_LOOKUP


def main():
    ap = argparse.ArgumentParser(description="Dedupe permits to unique contractors + entity lookups.")
    ap.add_argument("--in", dest="infile", default="permits.csv", help="input permits CSV (from pull_permits.py)")
    ap.add_argument("--out", default="contractors.csv", help="output unique-contractor CSV")
    ap.add_argument("--min-permits", type=int, default=1, help="only keep firms with >= this many permits")
    args = ap.parse_args()

    firms = {}  # norm_name -> aggregate dict
    total_rows = 0
    with open(args.infile, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            name = (row.get("contractors_business_name") or "").strip()
            if not name:
                continue
            key = norm_name(name)
            # "OWNER-BUILDER" is a placeholder (owner pulled their own permit), not a
            # firm — it aggregates many unrelated owners, so drop it from the ranking.
            if key in ("OWNER-BUILDER", "OWNER BUILDER", "OWNER/BUILDER"):
                continue
            total_rows += 1
            fm = firms.setdefault(key, {
                "contractor": name, "permits": 0, "total_valuation": 0.0,
                "first_seen": "", "last_seen": "", "licenses": set(),
                "city": row.get("contractor_city", ""), "state": row.get("contractor_state", ""),
                "top_projects": [],  # (valuation, address)
            })
            fm["permits"] += 1
            val = money(row.get("valuation"))
            fm["total_valuation"] += val
            d = (row.get("issue_date") or "")[:10]
            if d:
                fm["first_seen"] = min(fm["first_seen"], d) if fm["first_seen"] else d
                fm["last_seen"] = max(fm["last_seen"], d) if fm["last_seen"] else d
            lic = (row.get("license") or "").strip()
            if lic and lic.strip("0"):  # ignore "0"/"00000" placeholders
                fm["licenses"].add(lic)
            fm["top_projects"].append((val, row.get("address", "")))

    rows_out = []
    for fm in firms.values():
        if fm["permits"] < args.min_permits:
            continue
        top = sorted(fm["top_projects"], reverse=True)[:3]
        best_license = sorted(fm["licenses"])[0] if fm["licenses"] else ""
        rows_out.append({
            "contractor": fm["contractor"],
            "permits": fm["permits"],
            "total_valuation": int(fm["total_valuation"]),
            "first_seen": fm["first_seen"],
            "last_seen": fm["last_seen"],
            "licenses": ";".join(sorted(fm["licenses"])),
            "city": fm["city"],
            "state": fm["state"],
            "top_projects": " | ".join(f"${int(v):,} {a}" for v, a in top if a),
            "ca_sos_lookup_url": sos_url(fm["contractor"]),
            "cslb_license": best_license,
            "cslb_lookup_url": cslb_url(best_license),
        })

    rows_out.sort(key=lambda r: r["total_valuation"], reverse=True)

    cols = ["contractor", "permits", "total_valuation", "first_seen", "last_seen",
            "licenses", "city", "state", "top_projects",
            "ca_sos_lookup_url", "cslb_license", "cslb_lookup_url"]
    with open(args.out, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=cols)
        w.writeheader()
        w.writerows(rows_out)

    print("Contractor enrichment  —  cost: $0 (local dedupe + free lookup links)")
    print(f"  {total_rows} permit rows  ->  {len(rows_out)} unique contractors")
    print(f"  written to {args.out} (sorted by total project value)")
    if rows_out:
        print("\n  Top firms by total build value:")
        for r in rows_out[:5]:
            print(f"    ${r['total_valuation']:>14,}  {r['permits']:>3} permits  {r['contractor']}")
        print("\n  Enrich: open each row's ca_sos_lookup_url (legal entity + agent) and")
        print("  cslb_lookup_url (license status/classification). Small deduped worklist.")


if __name__ == "__main__":
    main()