← back to Unclaimed Property Platform
services/ingestion/ingest.py
264 lines
"""Authorized-feed ingestion with idempotent ETL + immutable raw archive.
Accepts an AUTHORIZED file (state CSV here; NAUPA II fixed-width and NAUPA III XML
adapters plug in the same way). It NEVER scrapes and NEVER fetches from a state portal —
the source is always a file already handed to us under a data-use agreement.
Idempotency: a re-run of the same file short-circuits at the batch checksum; a
re-delivered record upserts in place (see SqliteRepository).
"""
from __future__ import annotations
import csv
import hashlib
import io
import json
from dataclasses import dataclass
from decimal import Decimal
from typing import Iterable, Protocol
from services.common.normalize import (
amount_band, mask_name, normalize_business, normalize_postal, normalize_text,
parse_decimal,
)
class ObjectStore(Protocol):
def read_bytes(self, uri: str) -> bytes: ...
def write_bytes(self, uri: str, data: bytes) -> None: ...
class Repository(Protocol):
def batch_exists(self, jurisdiction: str, checksum: str) -> bool: ...
def begin_batch(self, jurisdiction: str, source_uri: str, checksum: str,
parser_version: str) -> str: ...
def upsert_property(self, batch_id: str, record: "CanonicalProperty") -> None: ...
def finish_batch(self, batch_id: str, accepted: int, rejected: int, status: str,
expected: int | None = None, reconciliation: str | None = None) -> None: ...
@dataclass
class CanonicalProperty:
jurisdiction: str
source_property_id: str
holder_name_raw: str
owner_type: str
owner_name_raw: str
owner_name_normalized: str
owner_name_masked: str
amount: Decimal | None
amount_band: str | None
# PROD-SECURITY (Sec 2.5): raw_payload MUST be encrypted at rest before persistence.
# Use AES-256-GCM with a per-jurisdiction DEK wrapped by KMS (AWS/GCP).
# The matcher operates on decrypted + normalized values that are NEVER written
# back — only raw_payload and owner_name_raw carry PII and need encryption.
# Masked/normalized fields (owner_name_masked, owner_name_normalized) are safe
# to store unencrypted since they cannot be reverse-mapped to a real person.
raw_payload: str
raw_record_hash: str
city_normalized: str | None = None
region: str | None = None
postal_code: str | None = None
property_type: str | None = None
@dataclass(frozen=True)
class FeedDefinition:
jurisdiction: str
source_uri: str
format_name: str = "state_csv_v1"
parser_version: str = "2026.07.1"
expected_count: int | None = None # state-declared control total (NAUPA trailer in prod)
def _reconcile(expected: int | None, parsed: int) -> tuple[str, int]:
"""Compare records PARSED (accepted+rejected) against the state's control total.
Returns (status, delta). 'short' = silent-truncation signal (parsed < expected)."""
if expected is None:
return "unknown", 0
if parsed < expected:
return "short", expected - parsed
if parsed > expected:
return "over", parsed - expected
return "ok", 0
# --- 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.
#
# 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:
if feed.jurisdiction not in ALLOWED_JURISDICTIONS:
raise PermissionError(
f"jurisdiction {feed.jurisdiction!r} is not an authorized synthetic feed; "
f"real-feed ingestion is human-gated, not autonomous"
)
if "://" in feed.source_uri:
raise PermissionError(
"URL/network sources are forbidden — the platform ingests files handed to it "
"under a data-use agreement; it never scrapes or fetches from a portal"
)
if not any(feed.source_uri.startswith(p) for p in ALLOWED_SOURCE_PREFIXES):
raise PermissionError(
f"source_uri {feed.source_uri!r} is outside the synthetic-only allowlist"
)
def _looks_like_business(name: str) -> bool:
from services.common.normalize import CORPORATE_SUFFIXES
return any(tok in CORPORATE_SUFFIXES for tok in normalize_text(name).split())
def parse_csv_feed(data: bytes, jurisdiction: str) -> Iterable[CanonicalProperty]:
"""Map a state's CSV columns into the canonical model.
Expected columns: property_id, holder_name, owner_name, address, city, state, zip,
property_type, amount. A real adapter maps each state's own field names + code tables.
"""
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)
# A malformed amount must NOT abort the whole batch (C2). Bad amount -> None; the
# raw value is preserved in raw_payload for later correction.
try:
amount = parse_decimal(row.get("amount"))
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("city")) or None,
region=normalize_text(row.get("state")) or None,
postal_code=normalize_postal(row.get("zip")),
property_type=normalize_text(row.get("property_type")) or None,
)
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
# an import cycle at module-load time.
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
if format_name == "naupa3_v1":
from services.ingestion.naupa3 import parse_naupa3_feed
return parse_naupa3_feed
return None
def ingest_authorized_feed(feed: FeedDefinition, object_store: ObjectStore,
repository: Repository) -> dict:
_assert_authorized(feed) # fail closed before any read (no-scrape red line)
data = object_store.read_bytes(feed.source_uri)
checksum = hashlib.sha256(data).hexdigest()
# Idempotency gate 1: identical file already processed.
if repository.batch_exists(feed.jurisdiction, checksum):
return {"status": "duplicate", "accepted": 0, "rejected": 0}
# Immutable raw archive (forensic replay).
raw_archive_uri = f"raw/{feed.jurisdiction}/{checksum}.bin"
object_store.write_bytes(raw_archive_uri, data)
batch_id = repository.begin_batch(
jurisdiction=feed.jurisdiction, source_uri=feed.source_uri,
checksum=checksum, parser_version=feed.parser_version,
)
accepted = rejected = 0
try:
parser = _get_parser(feed.format_name)
if parser is None:
raise NotImplementedError(f"Unsupported format: {feed.format_name}")
for record in parser(data, feed.jurisdiction):
if not record.source_property_id or not record.owner_name_raw:
rejected += 1
continue
try:
repository.upsert_property(batch_id, record)
accepted += 1
except Exception:
rejected += 1
status = "completed_with_errors" if rejected else "completed"
recon, delta = _reconcile(feed.expected_count, accepted + rejected)
repository.finish_batch(batch_id, accepted, rejected, status,
expected=feed.expected_count, reconciliation=recon)
except Exception:
repository.finish_batch(batch_id, accepted, rejected, "failed")
raise
return {"status": status, "batch_id": batch_id,
"accepted": accepted, "rejected": rejected,
"reconciliation": recon, "reconciliation_delta": delta}