`scripts/load_ca.py`

154 lines
#!/usr/bin/env python3
"""
load_ca.py — build the REAL California unclaimed-property database.

Streams the authorized CA State Controller public CSVs (data/ca-source/*.csv) through the
platform's CA ingestion adapter (services.ingestion.parse_ca_csv_feed) and bulk-loads the
normalized + masked records into db/ca_unclaimed.sqlite, with a search index on the
normalized owner name.

Design:
  * STREAMING per file (csv.DictReader) — never holds a whole file's rows in memory.
  * BATCHED executemany (BATCH rows/commit) — millions of rows load in minutes, not hours.
  * Index built AFTER the bulk load (far faster than maintaining it per-insert).
  * Idempotent-ish: PROPERTY_ID is the PRIMARY KEY, so a re-run UPSERTs in place.
  * Stores the MASKED + NORMALIZED projection for search; keeps raw owner name + city/zip
    (the CA file is already the public projection — no SSN/DOB/account #).

Usage:
    python3 scripts/load_ca.py                         # load every data/ca-source/*.csv
    python3 scripts/load_ca.py From_500_To_Beyond_1_of_4.csv   # one/several files
    python3 scripts/load_ca.py --limit 50000           # smoke test: first N rows/file
"""
from __future__ import annotations

import sqlite3
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from services.ingestion.ingest import parse_ca_csv_feed  # noqa: E402

SRC_DIR = ROOT / "data" / "ca-source"
DB_PATH = ROOT / "db" / "ca_unclaimed.sqlite"
BATCH = 5000

DDL = """
CREATE TABLE IF NOT EXISTS ca_property (
    property_id           TEXT NOT NULL,
    owner_type            TEXT,
    owner_name_raw        TEXT NOT NULL,
    owner_name_normalized TEXT,
    owner_name_masked     TEXT,
    owner_city            TEXT,
    owner_state           TEXT,
    owner_zip             TEXT,
    holder_name           TEXT,
    amount                REAL,
    amount_band           TEXT,
    property_type         TEXT,
    -- co-owners share a PROPERTY_ID, so the identity is the (property, owner) PAIR,
    -- otherwise a 2-owner property would overwrite itself and lose a claimant.
    PRIMARY KEY (property_id, owner_name_raw)
) WITHOUT ROWID;
"""
UPSERT = """
INSERT INTO ca_property
    (property_id, owner_type, owner_name_raw, owner_name_normalized, owner_name_masked,
     owner_city, owner_state, owner_zip, holder_name, amount, amount_band, property_type)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(property_id, owner_name_raw) DO UPDATE SET
    owner_name_normalized=excluded.owner_name_normalized,
    owner_name_masked=excluded.owner_name_masked, owner_city=excluded.owner_city,
    owner_state=excluded.owner_state, owner_zip=excluded.owner_zip,
    holder_name=excluded.holder_name, amount=excluded.amount,
    amount_band=excluded.amount_band, property_type=excluded.property_type;
"""


def _row(rec) -> tuple:
    return (
        rec.source_property_id, rec.owner_type, rec.owner_name_raw,
        rec.owner_name_normalized, rec.owner_name_masked, rec.city_normalized,
        rec.region, rec.postal_code, rec.holder_name_raw,
        float(rec.amount) if rec.amount is not None else None,
        rec.amount_band, rec.property_type,
    )


def load_file(conn: sqlite3.Connection, path: Path, limit: int | None) -> tuple[int, int]:
    data = path.read_bytes()
    accepted = skipped = 0
    batch: list[tuple] = []
    cur = conn.cursor()
    for rec in parse_ca_csv_feed(data, "CA"):
        if not rec.source_property_id or not rec.owner_name_raw:
            skipped += 1
            continue
        batch.append(_row(rec))
        if len(batch) >= BATCH:
            cur.executemany(UPSERT, batch)
            conn.commit()
            accepted += len(batch)
            batch.clear()
            if accepted % 250_000 == 0:
                print(f"    …{accepted:,} rows", flush=True)
        if limit and accepted + len(batch) >= limit:
            break
    if batch:
        cur.executemany(UPSERT, batch)
        conn.commit()
        accepted += len(batch)
    return accepted, skipped


def main(argv: list[str]) -> int:
    limit = None
    files_args = []
    it = iter(argv)
    for a in it:
        if a == "--limit":
            limit = int(next(it))
        else:
            files_args.append(a)

    files = ([SRC_DIR / f for f in files_args] if files_args
             else sorted(SRC_DIR.glob("*.csv")))
    files = [f for f in files if f.exists()]
    if not files:
        print(f"No CSVs found under {SRC_DIR} (download the SCO ZIP first).", file=sys.stderr)
        return 1

    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA synchronous=NORMAL;")
    conn.executescript(DDL)

    total_acc = total_skip = 0
    for f in files:
        print(f"Loading {f.name} …", flush=True)
        acc, skip = load_file(conn, f, limit)
        total_acc += acc
        total_skip += skip
        print(f"  {f.name}: +{acc:,} rows ({skip:,} skipped)", flush=True)

    print("Building search index on owner_name_normalized …", flush=True)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_owner_norm ON ca_property(owner_name_normalized);")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_amount ON ca_property(amount);")
    conn.commit()

    total = conn.execute("SELECT COUNT(*) FROM ca_property").fetchone()[0]
    dollars = conn.execute("SELECT SUM(amount) FROM ca_property").fetchone()[0] or 0
    conn.close()
    print(f"\nDONE. db/ca_unclaimed.sqlite now holds {total:,} California records "
          f"(this run: +{total_acc:,}, {total_skip:,} skipped).")
    print(f"Total unclaimed value in the DB: ${dollars:,.2f}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))