← back to Unclaimed Property Platform
ingest: CA adapter + streaming loader — real California DB built (3.76M records, $11.1B)
eeef3190ffdd7e087485c5011ba09db1a14fc4c8 · 2026-08-07 07:54:15 -0700 · steve@designerwallcoverings.com
Files touched
M .gitignoreA scripts/load_ca.pyM services/ingestion/ingest.py
Diff
commit eeef3190ffdd7e087485c5011ba09db1a14fc4c8
Author: steve@designerwallcoverings.com <steve@designerwallcoverings.com>
Date: Fri Aug 7 07:54:15 2026 -0700
ingest: CA adapter + streaming loader — real California DB built (3.76M records, $11.1B)
---
.gitignore | 4 ++
scripts/load_ca.py | 153 +++++++++++++++++++++++++++++++++++++++++++
services/ingestion/ingest.py | 53 ++++++++++++++-
3 files changed, 208 insertions(+), 2 deletions(-)
diff --git a/.gitignore b/.gitignore
index 49d956a..89ebb43 100644
--- a/.gitignore
+++ b/.gitignore
@@ -40,3 +40,7 @@ raw/
data/**/*.txt
data/**/*.xml
data/**/*.zip
+
+# Real authorized public data (California SCO bulk download) — NEVER commit (PII-adjacent + huge)
+data/ca-source/
+*.zip
diff --git a/scripts/load_ca.py b/scripts/load_ca.py
new file mode 100644
index 0000000..2ec7b40
--- /dev/null
+++ b/scripts/load_ca.py
@@ -0,0 +1,153 @@
+#!/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:]))
diff --git a/services/ingestion/ingest.py b/services/ingestion/ingest.py
index 39f7be3..78b9409 100644
--- a/services/ingestion/ingest.py
+++ b/services/ingestion/ingest.py
@@ -86,8 +86,14 @@ def _reconcile(expected: int | None, parsed: int) -> tuple[str, int]:
# --- No-scrape / synthetic-only red line, ENFORCED IN CODE (Security finding 2.1) ---------
# Crossing the line now requires editing THIS allowlist — a greppable, reviewable, human-
# gateable change — not just passing a different URI or plugging in a networked ObjectStore.
-ALLOWED_SOURCE_PREFIXES = ("raw/", "incoming/", "data/sample/")
-ALLOWED_JURISDICTIONS = frozenset({"SAMPLE"})
+#
+# CA (California) is an AUTHORIZED PUBLIC SOURCE, added 2026-08-07 with human sign-off:
+# the CA State Controller PUBLISHES its full unclaimed-property database as free CSV,
+# expressly for owner outreach (sco.ca.gov "Download Unclaimed Property Records"). The
+# file is downloaded by hand and placed under data/ca-source/ — the platform still never
+# scrapes or fetches from a portal (the "://" network guard below stays in force).
+ALLOWED_SOURCE_PREFIXES = ("raw/", "incoming/", "data/sample/", "data/ca-source/")
+ALLOWED_JURISDICTIONS = frozenset({"SAMPLE", "CA"})
def _assert_authorized(feed: "FeedDefinition") -> None:
@@ -152,6 +158,47 @@ def parse_csv_feed(data: bytes, jurisdiction: str) -> Iterable[CanonicalProperty
)
+def parse_ca_csv_feed(data: bytes, jurisdiction: str) -> Iterable[CanonicalProperty]:
+ """California State Controller public unclaimed-property CSV -> canonical model.
+
+ Real schema (25 cols): PROPERTY_ID, PROPERTY_TYPE, CASH_REPORTED, ...,
+ OWNER_NAME, OWNER_CITY, OWNER_STATE, OWNER_ZIP, CURRENT_CASH_BALANCE, HOLDER_NAME, ...
+ The SCO file is already the PUBLIC projection — no SSN/DOB/account number — so
+ raw_payload here carries no restricted identifiers.
+ """
+ reader = csv.DictReader(io.StringIO(data.decode("utf-8-sig")))
+ for row in reader:
+ serialized = json.dumps(row, sort_keys=True)
+ raw_hash = hashlib.sha256(serialized.encode()).hexdigest()
+ owner_name = (row.get("OWNER_NAME") or "").strip()
+ is_business = _looks_like_business(owner_name)
+ # CURRENT_CASH_BALANCE is the claimable amount; fall back to CASH_REPORTED.
+ try:
+ amount = parse_decimal(row.get("CURRENT_CASH_BALANCE") or row.get("CASH_REPORTED"))
+ except ValueError:
+ amount = None
+ owner_norm = (
+ normalize_business(owner_name) if is_business else normalize_text(owner_name)
+ )
+ yield CanonicalProperty(
+ jurisdiction=jurisdiction,
+ source_property_id=(row.get("PROPERTY_ID") or "").strip(),
+ holder_name_raw=(row.get("HOLDER_NAME") or "").strip(),
+ owner_type="business" if is_business else "person",
+ owner_name_raw=owner_name,
+ owner_name_normalized=owner_norm,
+ owner_name_masked=mask_name(owner_name),
+ amount=amount,
+ amount_band=amount_band(amount),
+ raw_payload=serialized,
+ raw_record_hash=raw_hash,
+ city_normalized=normalize_text(row.get("OWNER_CITY")) or None,
+ region=normalize_text(row.get("OWNER_STATE")) or None,
+ postal_code=normalize_postal(row.get("OWNER_ZIP")),
+ property_type=normalize_text(row.get("PROPERTY_TYPE")) or None,
+ )
+
+
# Adapter registry: format_name -> parser(data, jurisdiction) -> Iterable[CanonicalProperty].
# ingest_authorized_feed stays format-agnostic; each format is just a parser. NAUPA II is
# imported LAZILY so naupa2's top-level `from ...ingest import CanonicalProperty` can't form
@@ -159,6 +206,8 @@ def parse_csv_feed(data: bytes, jurisdiction: str) -> Iterable[CanonicalProperty
def _get_parser(format_name: str):
if format_name == "state_csv_v1":
return parse_csv_feed
+ if format_name == "ca_csv_v1":
+ return parse_ca_csv_feed
if format_name == "naupa2_v1":
from services.ingestion.naupa2 import parse_naupa2_feed
return parse_naupa2_feed
← 42ad8e0 outreach: CA has a free public CSV bulk download — no PRA le
·
back to Unclaimed Property Platform
·
ca: masked search UI (serve_ca.py :8799) + disk-safe full-CA 9f2367a →