`scripts/fetch-dre-mlo.py`

133 lines
#!/usr/bin/env python3
"""
fetch-dre-mlo.py — Ingest CA DRE's daily "MLO List" (individual real-estate licensees
with a SAFE Act / NMLS mortgage-loan-originator endorsement) into data/loan-officers.json.

PERMITTED SOURCE (DATA_POLICY.md rule 1 = official_bulk_download):
  Page:  https://www.dre.ca.gov/Licensees/ExamineeLicenseeListDataFiles.html
  File:  https://secure.dre.ca.gov/datafile/mlo_list.xls   (updated daily by CA DRE)

SCOPE (Steve, 2026-08-20): ACTIVE, INDIVIDUAL (Broker/Salesperson), LA-County only.
COMPLIANCE:
  - Business/license fields only. address_line_1/2 + zip are DROPPED (rule 3 — no home address).
  - phone_number is kept: it is published by a government source in this public file (rule 3 carve-out).
  - Every row carries a real NMLS # + CalBRE # + the DRE source_url it came from (rule 4).
  - Merges with (never clobbers) the verified firm-roster officers already in the file; dedup by NMLS #.
Run:  python3 scripts/fetch-dre-mlo.py [--download]   (default reuses /tmp/mlo_list.xls if present)
"""
import datetime
import json
import os
import sys
import urllib.request

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATA = os.path.join(ROOT, "data", "loan-officers.json")
XLS_URL = "https://secure.dre.ca.gov/datafile/mlo_list.xls"
SRC_PAGE = "https://www.dre.ca.gov/Licensees/ExamineeLicenseeListDataFiles.html"
XLS_CACHE = "/tmp/mlo_list.xls"

LA_COUNTY = {
"AGOURA HILLS","ALHAMBRA","ARCADIA","ARTESIA","AVALON","AZUSA","BALDWIN PARK","BELL","BELL GARDENS","BELLFLOWER","BEVERLY HILLS","BRADBURY","BURBANK","CALABASAS","CARSON","CERRITOS","CLAREMONT","COMMERCE","COMPTON","COVINA","CUDAHY","CULVER CITY","DIAMOND BAR","DOWNEY","DUARTE","EL MONTE","EL SEGUNDO","GARDENA","GLENDALE","GLENDORA","HAWAIIAN GARDENS","HAWTHORNE","HERMOSA BEACH","HIDDEN HILLS","HUNTINGTON PARK","INDUSTRY","INGLEWOOD","IRWINDALE","LA CANADA FLINTRIDGE","LA HABRA HEIGHTS","LA MIRADA","LA PUENTE","LA VERNE","LAKEWOOD","LANCASTER","LAWNDALE","LOMITA","LONG BEACH","LOS ANGELES","LYNWOOD","MALIBU","MANHATTAN BEACH","MAYWOOD","MONROVIA","MONTEBELLO","MONTEREY PARK","NORWALK","PALMDALE","PALOS VERDES ESTATES","PARAMOUNT","PASADENA","PICO RIVERA","POMONA","RANCHO PALOS VERDES","REDONDO BEACH","ROLLING HILLS","ROLLING HILLS ESTATES","ROSEMEAD","SAN DIMAS","SAN FERNANDO","SAN GABRIEL","SAN MARINO","SANTA CLARITA","SANTA FE SPRINGS","SANTA MONICA","SIERRA MADRE","SIGNAL HILL","SOUTH EL MONTE","SOUTH GATE","SOUTH PASADENA","TEMPLE CITY","TORRANCE","VERNON","WALNUT","WEST COVINA","WEST HOLLYWOOD","WESTLAKE VILLAGE","WHITTIER",
"NORTH HOLLYWOOD","VAN NUYS","SHERMAN OAKS","STUDIO CITY","ENCINO","TARZANA","WOODLAND HILLS","CANOGA PARK","RESEDA","NORTHRIDGE","GRANADA HILLS","CHATSWORTH","SUN VALLEY","PANORAMA CITY","SYLMAR","PACOIMA","MISSION HILLS","VALLEY VILLAGE","VALENCIA","STEVENSON RANCH","NEWHALL","SAUGUS","CANYON COUNTRY","CASTAIC","MARINA DEL REY","PLAYA VISTA","PLAYA DEL REY","VENICE","WESTCHESTER","PACIFIC PALISADES","BRENTWOOD","HOLLYWOOD","WEST HILLS","WINNETKA","LAKE BALBOA","ARLETA","LA CRESCENTA","MONTROSE","TUJUNGA","ALTADENA","TOPANGA","ACTON","AGUA DULCE","QUARTZ HILL","LITTLEROCK","LAKE HUGHES","LEONA VALLEY","VIEW PARK","WINDSOR HILLS","LADERA HEIGHTS","HACIENDA HEIGHTS","ROWLAND HEIGHTS","VALINDA","WEST PUENTE VALLEY","CITRUS","CHARTER OAK","EAST LOS ANGELES","WEST WHITTIER","SOUTH WHITTIER","WILLOWBROOK","FLORENCE","WALNUT PARK","LENNOX","DEL AIRE","WESTMONT","LAKEVIEW TERRACE","SHADOW HILLS","SUNLAND","PORTER RANCH","HARBOR CITY","WILMINGTON","SAN PEDRO",
}
TITLE = {"Broker": "Real Estate Broker (MLO endorsement)", "Salesperson": "Real Estate Salesperson (MLO endorsement)", "Officer": "Real Estate Officer (MLO endorsement)"}
INDIVIDUAL_TYPES = ("Broker", "Salesperson", "Officer")  # individual MLO-endorsed licensees (excludes 'Corporation')

def clean(s):
    return s.replace("\x00", "").strip() if isinstance(s, str) else ("" if s is None else str(s))

def titlecase(s):
    s = " ".join(clean(s).split())
    return s.title() if s and s.isupper() else s

def main():
    if "--download" in sys.argv or not os.path.exists(XLS_CACHE):
        print(f"downloading {XLS_URL} ...")
        req = urllib.request.Request(XLS_URL, headers={"User-Agent": "Mozilla/5.0"})
        with urllib.request.urlopen(req, timeout=90) as r, open(XLS_CACHE, "wb") as f:
            f.write(r.read())
    from python_calamine import CalamineWorkbook
    rows = CalamineWorkbook.from_path(XLS_CACHE).get_sheet_by_index(0).to_python()
    hdr = [clean(x) for x in rows[0]]
    ix = {h: i for i, h in enumerate(hdr)}
    now = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    dre = []
    seen = set()
    for r in rows[1:]:
        g = lambda k: clean(r[ix[k]]) if k in ix and ix[k] < len(r) else ""
        lt, st, ns = g("calbre_license_type"), g("calbre_license_status"), g("nmls_status")
        nmls, city = g("nmls_id"), g("city_name").upper()
        if lt not in INDIVIDUAL_TYPES: continue                    # individuals only (excludes 'Corporation')
        if not (st == "Licensed" and ns == "APPROVED"): continue  # active only
        if not nmls or city not in LA_COUNTY: continue            # LA County only
        if nmls in seen: continue
        seen.add(nmls)
        first, last, suf = titlecase(g("firstname_secondary_name")), titlecase(g("lastname_primary_name")), clean(g("name_suffix"))
        name = " ".join(p for p in [first, last, suf] if p).strip() or last
        exp = r[ix["calbre_expiration_date"]] if "calbre_expiration_date" in ix else ""
        exp = exp.isoformat() if hasattr(exp, "isoformat") else clean(exp)
        calbre = g("calbre_license_id")
        dre.append({
            "id": "nmls-" + nmls,
            "name": name,
            "company": titlecase(g("sponsor_name")),          # usually blank for a sole broker → honest "—"
            "nmls_id": nmls,
            "title": TITLE.get(lt, lt),
            "business_phone": g("phone_number"),               # govt-published in the public file
            "business_email": "",                              # DRE file carries none
            "city": titlecase(city),
            "state": "CA",
            "calbre_license_id": calbre,
            "calbre_license_type": lt,
            "license_status": st,
            "license_expires": exp,
            "source": "CA DRE — MLO List (daily)",
            "source_url": SRC_PAGE,
            "identifier": f"NMLS #{nmls} · CalBRE #{calbre}",
            "retrieved_at": now,
            # NOTE: address_line_1/2 + zip_code deliberately NOT stored (DATA_POLICY §3, no home address).
        })

    cur = json.load(open(DATA))
    existing = cur.get("officers", [])
    existing_nmls = {o.get("nmls_id") for o in existing}
    added = [d for d in dre if d["nmls_id"] not in existing_nmls]   # keep verified firm-roster rows intact
    officers = existing + added
    officers.sort(key=lambda o: (o.get("city") or "~", o.get("name") or ""))

    meta = cur["meta"]
    meta["source"] = "CA DRE MLO List (official bulk download) + firm-published public rosters (first-party public web)"
    meta["access_method"] = "official_bulk_download + first_party_public_web"
    meta["count"] = len(officers)
    meta["retrieved_at"] = now
    meta["scope"] = "California — Los Angeles County; active, individual MLO-endorsed licensees (Broker/Salesperson/Officer)"
    meta["policy"] = ("Individual mortgage loan officers are added ONLY from officially-permitted sources: "
        "the CA DRE daily MLO List (official_bulk_download) and firms' OWN public loan-officer rosters "
        "(first_party_public_web). NMLS Consumer Access is NEVER scraped. Every record carries a real public "
        "NMLS license number + source_url; business/license fields only — no home/mailing addresses, no personal "
        "emails. Phone is the government-published contact from the DRE public file. See docs/DATA_POLICY.md.")
    firm_sources = [s for s in meta.get("sources", []) if s.get("access_method") == "first_party_public_web"]
    meta["sources"] = ([{
        "source": "CA DRE — MLO List (updated daily)",
        "page_url": SRC_PAGE,
        "file_url": XLS_URL,
        "access_method": "official_bulk_download",
        "scope": "LA County, active, individual (Broker/Salesperson/Officer)",
        "count": sum(1 for o in officers if str(o.get("source", "")).startswith("CA DRE")),
    }] + firm_sources)

    # PII guard: HARD-ABORT before writing if any address/zip field leaked (protects unattended runs).
    leaked = [o["nmls_id"] for o in officers if any(k in o for k in ("address_line_1", "address_line_2", "zip_code", "zip"))]
    if leaked:
        sys.stderr.write(f"ABORT — PII leak (address/zip) in {len(leaked)} records, e.g. {leaked[:5]}; nothing written.\n")
        sys.exit(2)
    json.dump({"meta": meta, "officers": officers}, open(DATA, "w"), indent=2)
    print(f"DRE LA-County active individuals matched: {len(dre)} | new added (dedup by NMLS): {len(added)}")
    print(f"firm-roster officers preserved: {len(existing)} | TOTAL officers: {len(officers)}")
    print("PII guard (address/zip fields present): clean ✓")

if __name__ == "__main__":
    main()