new-site/scripts/build_registry_outreach_lists.py

#!/usr/bin/env python3
"""Build public-registry outreach lists with EMAIL + a dateable compliance clock.

Implements the "500k+ email leads" verticals from
docs/new-sector-compliance-targets.md sec 10. The winning pattern is always:

    public registry  ->  recurring obligation  ->  automated deficiency check
    ->  EMAIL outreach  ->  fixed-price filing service.

Unlike NPPES (email-poor) and EPA RCRA (no email at all), these registries expose
a real, cold-emailable contact email in a FREE bulk/Socrata feed AND a column we
can turn into a dateable "your X is due/overdue" hook:

  ct    Connecticut SOS Business Registry  -- VERIFIED this session
        data.ct.gov/n7gp-d28j  |  email: business_email_address
        clock: annual_report_due_date (annual report)  -> overdue/due-soon
        ~439,763 ACTIVE + email (303k distinct), ~97,466 OVERDUE annual reports.

  fmcsa FMCSA Company Census (motor carriers) -- VERIFIED this session
        data.transportation.gov/az4n-8mr2  |  email: email_address
        clock: mcs150_date + 24mo (biennial MCS-150 update) -> overdue/due-soon
        ~2,927,276 rows with email (2.79M distinct), ~1,668,135 ACTIVE + email.
        This is the genuine 500k+ email vertical (replaces USPTO trademarks,
        whose owner email is NOT in any free bulk file and whose API is now
        ID.me-auth-walled as of 2026-06-18).

Two clock types are supported per adapter:
  * "due_date"   : a column already holds the obligation's due date.
  * "staleness"  : a column holds the last-filed date; due = last + interval_days.

Adding a registry = one Registry entry (Socrata host, resource, column mapping).

Output (every row has an `email` column, per verify_csv_emails.py's contract):
  out/<key>_overdue.csv          obligation PAST DUE        (hottest leads)
  out/<key>_due_soon.csv         due within --soon-days     (pre-emptive)
  out/<key>_active_emailable.csv all active + email         (broad campaign)

Then MX/SMTP-verify on the NON-sending IP before importing:
  python3 scripts/verify_csv_emails.py --in out/ct_overdue.csv --out out/ct_overdue

Usage:
  python3 scripts/build_registry_outreach_lists.py --registries ct,fmcsa
  python3 scripts/build_registry_outreach_lists.py --registries ct --max-rows 5000
  python3 scripts/build_registry_outreach_lists.py --list
"""
from __future__ import annotations

import argparse
import csv
import datetime
import json
import os
import sys
import time
import urllib.parse
import urllib.request
from collections import Counter
from dataclasses import dataclass

UA = "performancewest-registry-outreach/1.0 (+https://performancewest.net)"


def email_ok(e: str) -> bool:
    e = (e or "").strip()
    return bool(e) and "@" in e and "." in e.split("@")[-1] and " " not in e


@dataclass
class Registry:
    """One public registry that exposes a cold-emailable contact + a clock."""

    key: str
    name: str
    host: str                       # Socrata host, e.g. data.ct.gov
    resource: str                   # Socrata 4x4, e.g. n7gp-d28j
    email_field: str                # primary cold-emailable email column
    name_field: str
    obligation: str                 # human label, e.g. "annual report"
    clock: str                      # "due_date" | "staleness"
    clock_field: str                # due-date col, OR last-filed col for staleness
    interval_days: int = 0          # for staleness: days from last-filed to due
    status_field: str | None = None
    active_values: tuple[str, ...] = ()
    alt_email_field: str | None = None
    id_field: str | None = None
    type_field: str | None = None
    naics_field: str | None = None
    city_field: str | None = None
    state_field: str | None = None
    phone_field: str | None = None
    notes: str = ""


REGISTRIES: dict[str, Registry] = {
    "ct": Registry(
        key="ct",
        name="Connecticut SOS Business Registry",
        host="data.ct.gov",
        resource="n7gp-d28j",
        email_field="business_email_address",
        alt_email_field="category_survey_email_address",
        name_field="name",
        obligation="annual report",
        clock="due_date",
        clock_field="annual_report_due_date",
        status_field="status",
        active_values=("active",),
        id_field="accountnumber",
        type_field="business_type",
        naics_field="naics_code",
        city_field="billingcity",
        state_field="billingstate",
        notes="business_email_address is a real cold-emailable inbox; "
        "annual_report_due_date is the dateable annual-report clock.",
    ),
    "fmcsa": Registry(
        key="fmcsa",
        name="FMCSA Company Census (motor carriers)",
        host="data.transportation.gov",
        resource="az4n-8mr2",
        email_field="email_address",
        name_field="legal_name",
        obligation="MCS-150 biennial update",
        clock="staleness",
        clock_field="mcs150_date",
        interval_days=730,  # biennial
        status_field="status_code",
        active_values=("a",),  # A = active
        id_field="dot_number",
        type_field="carrier_operation",
        city_field="phy_city",
        state_field="phy_state",
        phone_field="phone",
        notes="email_address coverage ~2.9M rows; mcs150_date + 24mo is the "
        "biennial MCS-150 staleness clock. Replaces USPTO (now auth-walled).",
    ),
}


def socrata_get(host: str, resource: str, params: dict) -> list[dict]:
    url = f"https://{host}/resource/{resource}.json?{urllib.parse.urlencode(params)}"
    req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=180) as r:
        return json.loads(r.read().decode("utf-8"))


def socrata_count(host: str, resource: str, where: str | None = None) -> int:
    params = {"$select": "count(1)"}
    if where:
        params["$where"] = where
    rows = socrata_get(host, resource, params)
    return int(rows[0].get("count_1", 0)) if rows else 0


def parse_date(val: str | None) -> datetime.date | None:
    if not val:
        return None
    val = val.strip()
    for fmt in ("%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d",
                "%m/%d/%Y", "%m/%d/%Y %H:%M:%S",
                "%Y%m%d %H%M", "%Y%m%d"):  # FMCSA mcs150_date = 'YYYYMMDD HHMM'
        try:
            return datetime.datetime.strptime(val, fmt).date()
        except ValueError:
            continue
    try:
        return datetime.datetime.strptime(val[:10], "%Y-%m-%d").date()
    except ValueError:
        return None


def due_date_for(reg: Registry, raw: str | None) -> datetime.date | None:
    """Resolve the obligation's due date from the registry's clock column."""
    d = parse_date(raw)
    if d is None:
        return None
    if reg.clock == "staleness":
        return d + datetime.timedelta(days=reg.interval_days)
    return d  # due_date


def fetch_active_emailable(reg: Registry, max_rows: int | None, page: int = 50000):
    """Yield raw rows: active records that carry a usable email (server-filtered)."""
    cols = [c for c in {
        reg.id_field, reg.name_field, reg.email_field, reg.alt_email_field,
        reg.status_field, reg.clock_field, reg.type_field, reg.naics_field,
        reg.city_field, reg.state_field, reg.phone_field,
    } if c]

    where = [f"{reg.email_field} IS NOT NULL"]
    if reg.status_field and reg.active_values:
        ors = " OR ".join(f"upper({reg.status_field})='{v.upper()}'" for v in reg.active_values)
        where.append(f"({ors})")
    where_s = " AND ".join(where)

    offset, pulled = 0, 0
    while True:
        params = {
            "$select": ",".join(cols),
            "$where": where_s,
            "$order": (reg.id_field or reg.name_field),
            "$limit": page,
            "$offset": offset,
        }
        rows = socrata_get(reg.host, reg.resource, params)
        if not rows:
            break
        for r in rows:
            email = (r.get(reg.email_field) or "").strip()
            if not email_ok(email) and reg.alt_email_field:
                email = (r.get(reg.alt_email_field) or "").strip()
            if not email_ok(email):
                continue
            yield r, email.lower()
            pulled += 1
            if max_rows and pulled >= max_rows:
                return
        offset += len(rows)
        if len(rows) < page:
            break


HEADER = ["registry", "entity_id", "name", "email", "phone", "entity_type",
          "naics", "city", "biz_state", "status", "obligation",
          "due_date", "days_overdue", "deficiency", "last_filed"]


def enrich(reg: Registry, r: dict, email: str, today: datetime.date) -> dict:
    raw = r.get(reg.clock_field) if reg.clock_field else None
    last_filed = parse_date(raw) if reg.clock == "staleness" else None
    due = due_date_for(reg, raw)
    if due:
        days_overdue = (today - due).days
        deficiency = "overdue" if days_overdue > 0 else "upcoming"
    else:
        days_overdue, deficiency = None, "no_date"
    return {
        "registry": reg.key,
        "entity_id": r.get(reg.id_field, "") if reg.id_field else "",
        "name": r.get(reg.name_field, ""),
        "email": email,
        "phone": r.get(reg.phone_field, "") if reg.phone_field else "",
        "entity_type": r.get(reg.type_field, "") if reg.type_field else "",
        "naics": r.get(reg.naics_field, "") if reg.naics_field else "",
        "city": r.get(reg.city_field, "") if reg.city_field else "",
        "biz_state": r.get(reg.state_field, "") if reg.state_field else "",
        "status": r.get(reg.status_field, "") if reg.status_field else "",
        "obligation": reg.obligation,
        "due_date": due.isoformat() if due else "",
        "days_overdue": days_overdue if days_overdue is not None else "",
        "deficiency": deficiency,
        "last_filed": last_filed.isoformat() if last_filed else "",
    }


def write_csv(path: str, rows: list[dict]) -> None:
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=HEADER)
        w.writeheader()
        w.writerows(rows)


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--registries", default="ct,fmcsa",
                    help="comma list of registry keys (default: ct,fmcsa)")
    ap.add_argument("--out-dir", default="/tmp/registry_out")
    ap.add_argument("--soon-days", type=int, default=90)
    ap.add_argument("--max-rows", type=int, default=None,
                    help="cap rows per registry (quick test)")
    ap.add_argument("--dedupe-email", action="store_true",
                    help="keep one row per email (most-overdue first)")
    ap.add_argument("--list", action="store_true", dest="do_list")
    args = ap.parse_args()

    if args.do_list:
        print("Configured registries (email + dateable clock):\n")
        for k, reg in REGISTRIES.items():
            print(f"  {k:6} {reg.name}")
            print(f"         {reg.host}/{reg.resource}  email={reg.email_field!r}")
            print(f"         clock={reg.clock}({reg.clock_field}"
                  + (f"+{reg.interval_days}d" if reg.clock == 'staleness' else "")
                  + f")  obligation={reg.obligation!r}\n")
        return 0

    os.makedirs(args.out_dir, exist_ok=True)
    today = datetime.date.today()
    keys = [k.strip().lower() for k in args.registries.split(",") if k.strip()]
    grand = Counter()

    for k in keys:
        reg = REGISTRIES.get(k)
        if not reg:
            print(f"!! unknown registry {k!r} (see --list)", file=sys.stderr)
            continue

        print(f"\n=== {reg.name} [{k}] :: {reg.host}/{reg.resource} ===")
        try:
            print(f"  dataset rows: {socrata_count(reg.host, reg.resource):,}")
        except Exception as e:
            print(f"  (count failed: {e})")

        t0 = time.time()
        rows: list[dict] = []
        seen: set[str] = set()
        for r, email in fetch_active_emailable(reg, args.max_rows):
            er = enrich(reg, r, email, today)
            if args.dedupe_email:
                if email in seen:
                    continue
                seen.add(email)
            rows.append(er)
            if len(rows) % 100000 == 0:
                print(f"  pulled {len(rows):,} ({len(rows)/max(time.time()-t0,1e-6):.0f}/s)")

        overdue = [r for r in rows if r["deficiency"] == "overdue"]
        overdue.sort(key=lambda r: -(r["days_overdue"] or 0))

        def days_until(r: dict) -> int | None:
            d = parse_date(r["due_date"])
            return (d - today).days if d else None

        soon = [r for r in rows if r["deficiency"] == "upcoming"
                and (du := days_until(r)) is not None and 0 <= du <= args.soon_days]
        soon.sort(key=days_until)

        rank = {"overdue": 0, "upcoming": 1, "no_date": 2}
        rows.sort(key=lambda r: (rank.get(r["deficiency"], 3), -(r["days_overdue"] or 0)))

        p_over = os.path.join(args.out_dir, f"{k}_overdue.csv")
        p_soon = os.path.join(args.out_dir, f"{k}_due_soon.csv")
        p_all = os.path.join(args.out_dir, f"{k}_active_emailable.csv")
        write_csv(p_over, overdue)
        write_csv(p_soon, soon)
        write_csv(p_all, rows)

        uniq = len({r["email"] for r in rows})
        grand["rows"] += len(rows); grand["overdue"] += len(overdue); grand["soon"] += len(soon)
        print(f"  active+emailable: {len(rows):,}  (distinct emails {uniq:,})")
        print(f"  OVERDUE {reg.obligation}: {len(overdue):,}  -> {p_over}")
        print(f"  due soon (<= {args.soon_days}d): {len(soon):,}  -> {p_soon}")
        print(f"  all active emailable:  {len(rows):,}  -> {p_all}")
        print(f"  pulled in {time.time()-t0:.0f}s")

    print("\n=== TOTALS ===")
    print(f"  active+emailable: {grand['rows']:,}")
    print(f"  overdue:          {grand['overdue']:,}")
    print(f"  due soon:         {grand['soon']:,}")
    print("\nNext: MX/SMTP-verify on the NON-sending IP before import, e.g.")
    print(f"  python3 scripts/verify_csv_emails.py --in {args.out_dir}/{keys[0]}_overdue.csv "
          f"--out {args.out_dir}/{keys[0]}_overdue")
    return 0


if __name__ == "__main__":
    sys.exit(main())