`scripts/backfill-address-from-csv.py`

120 lines
#!/usr/bin/env python3
"""
backfill-address-from-csv.py — enrich staged prospect JSONs with the full
address / state / category fields that the email crawler dropped at ingest.

The email crawler (crawl-csv-emails.py) only kept title/city/phone/website/
email from the raw Google-Places CSV. This backfills the richer location
fields straight from the same source CSV, keyed by website host (fallback:
title|city), WITHOUT disturbing the expensive email / instagram / linkedin
enrichment already in the JSON.

  python3 scripts/backfill-address-from-csv.py            # all prospect groups
  python3 scripts/backfill-address-from-csv.py --dry-run  # report only

Each prospects-<slug>.json records its own `source` CSV basename, which we
resolve against data/sources/. Groups whose source CSV is missing are skipped
and reported. Nothing is uploaded to Constant Contact — local staging only.
"""
import csv, sys, os, re, json, glob, argparse, urllib.parse

csv.field_size_limit(10_000_000)
HERE = os.path.dirname(__file__)
SRC_DIR = os.path.join(HERE, "..", "data", "sources")
DATA_DIR = os.path.join(HERE, "..", "public", "data")

# CSV column -> JSON field. Only filled if the JSON value is empty (non-destructive).
FIELD_MAP = {
    "address": "address",            # full formatted address, e.g. "970 W 190th St Ste 250, Torrance, CA 90502"
    "state": "state",
    "postalCode": "postal",
    "neighborhood": "neighborhood",
    "categoryName": "category",
    "url": "mapsUrl",                # Google Maps place URL
}

def host_of(u):
    u = (u or "").strip()
    if not u:
        return ""
    try:
        return urllib.parse.urlparse(u if u.startswith(("http://", "https://")) else "http://" + u).netloc.lower().replace("www.", "")
    except Exception:
        return ""

def key_of(rec):
    return host_of(rec.get("website")) or ((rec.get("title") or "").strip().lower() + "|" + (rec.get("city") or "").strip().lower())

def build_lut(csv_path):
    lut = {}
    with open(csv_path, newline="", encoding="utf-8", errors="replace") as fh:
        for row in csv.DictReader(fh):
            k = key_of(row)
            if k and k not in lut:
                lut[k] = row
    return lut

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--glob", default="prospects-*.json")
    a = ap.parse_args()

    srcmap = {os.path.basename(f): f for f in glob.glob(os.path.join(SRC_DIR, "*.csv"))}

    # Global host->row fallback across EVERY present CSV (rows with an address).
    # Used when a group's own source CSV is missing, or to fill records absent
    # from their own crawl. LA firms overlap across the architect/ID lists.
    glut = {}
    for f in srcmap.values():
        with open(f, newline="", encoding="utf-8", errors="replace") as fh:
            for row in csv.DictReader(fh):
                k = host_of(row.get("website"))
                if k and k not in glut and (row.get("address") or "").strip():
                    glut[k] = row

    total_files = total_matched = 0

    for jf in sorted(glob.glob(os.path.join(DATA_DIR, a.glob))):
        with open(jf) as _jf:
            d = json.load(_jf)
        biz = d.get("businesses", [])
        src = d.get("source")
        path = srcmap.get(src)
        base = os.path.basename(jf)
        lut = build_lut(path) if path else {}
        via = "own CSV" if path else "cross-CSV fallback (own source missing)"
        matched = 0
        for b in biz:
            r = lut.get(key_of(b)) or glut.get(host_of(b.get("website")))
            if not r:
                continue
            got = False
            for col, field in FIELD_MAP.items():
                val = (r.get(col) or "").strip()
                if val and not (b.get(field) or "").strip():
                    b[field] = val
                    got = True
            # fill phone if the crawler had none but Places did
            if not (b.get("phone") or "").strip() and (r.get("phone") or "").strip():
                b["phone"] = r["phone"].strip()
                got = True
            # a machine-diallable phone for tel: links
            if not (b.get("phoneUnformatted") or "").strip() and (r.get("phoneUnformatted") or "").strip():
                b["phoneUnformatted"] = r["phoneUnformatted"].strip()
            if got:
                matched += 1
        with_addr = sum(1 for b in biz if (b.get("address") or "").strip())
        d["with_address"] = with_addr
        total_files += 1
        total_matched += matched
        print(f"{'· ' if a.dry_run else '✓ '}{base:44s} {matched:5d} enriched · {with_addr:5d}/{len(biz):5d} now have address  [{via}]")
        if not a.dry_run:
            with open(jf, "w") as fh:
                json.dump(d, fh)

    print(f"\n{'DRY-RUN — no files written. ' if a.dry_run else ''}{total_files} groups, {total_matched} records enriched.  [$0 local]")

if __name__ == "__main__":
    main()