← back to Unclaimed Property Platform
auto-save: 2026-08-01T11:06:15 (4 files) — services/ingestion/ingest.py services/ingestion/naupa2.py tests/test_cycle3_naupa_and_blocking.py tests/test_cycle4_fairness.py
c0f1c688212757335a3dd265eb27f50ac4d5e4ff · 2026-08-01 11:06:17 -0700 · Steve Abrams
Files touched
M services/ingestion/ingest.pyA services/ingestion/naupa2.pyA tests/test_cycle3_naupa_and_blocking.pyA tests/test_cycle4_fairness.py
Diff
commit c0f1c688212757335a3dd265eb27f50ac4d5e4ff
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Aug 1 11:06:17 2026 -0700
auto-save: 2026-08-01T11:06:15 (4 files) — services/ingestion/ingest.py services/ingestion/naupa2.py tests/test_cycle3_naupa_and_blocking.py tests/test_cycle4_fairness.py
---
services/ingestion/ingest.py | 18 +++++-
services/ingestion/naupa2.py | 108 ++++++++++++++++++++++++++++++++
tests/test_cycle3_naupa_and_blocking.py | 102 ++++++++++++++++++++++++++++++
tests/test_cycle4_fairness.py | 68 ++++++++++++++++++++
4 files changed, 294 insertions(+), 2 deletions(-)
diff --git a/services/ingestion/ingest.py b/services/ingestion/ingest.py
index a5f4539..00e1b70 100644
--- a/services/ingestion/ingest.py
+++ b/services/ingestion/ingest.py
@@ -138,6 +138,19 @@ def parse_csv_feed(data: bytes, jurisdiction: str) -> Iterable[CanonicalProperty
)
+# 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 == "naupa2_v1":
+ from services.ingestion.naupa2 import parse_naupa2_feed
+ return parse_naupa2_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)
@@ -159,9 +172,10 @@ def ingest_authorized_feed(feed: FeedDefinition, object_store: ObjectStore,
accepted = rejected = 0
try:
- if feed.format_name != "state_csv_v1":
+ parser = _get_parser(feed.format_name)
+ if parser is None:
raise NotImplementedError(f"Unsupported format: {feed.format_name}")
- for record in parse_csv_feed(data, feed.jurisdiction):
+ for record in parser(data, feed.jurisdiction):
if not record.source_property_id or not record.owner_name_raw:
rejected += 1
continue
diff --git a/services/ingestion/naupa2.py b/services/ingestion/naupa2.py
new file mode 100644
index 0000000..4897fea
--- /dev/null
+++ b/services/ingestion/naupa2.py
@@ -0,0 +1,108 @@
+"""NAUPA II fixed-width parser adapter.
+
+NAUPA II is the legacy fixed-width reporting format (NAUPA III moves to schema-validated
+XML — a future adapter, wired the same way). A real state's layout carries dozens of
+fields and code tables; this is a documented, representative PROPERTY-DETAIL subset
+sufficient to prove the adapter-registry pattern end-to-end on synthetic data.
+
+Registered lazily by services.ingestion.ingest._get_parser under format_name 'naupa2_v1'.
+Emits the same CanonicalProperty as the CSV adapter, so ingestion/dedup/masking/search are
+identical regardless of source format.
+"""
+from __future__ import annotations
+
+import hashlib
+from decimal import Decimal
+from typing import Iterable
+
+from services.common.normalize import (
+ amount_band, mask_name, normalize_business, normalize_postal, normalize_text,
+)
+
+# (field_name, start, length) — 0-indexed. Fixed record width = 213.
+LAYOUT: list[tuple[str, int, int]] = [
+ ("record_type", 0, 2), # 'PR' = property detail
+ ("source_property_id", 2, 18),
+ ("holder_name", 20, 40),
+ ("owner_last", 60, 30),
+ ("owner_first", 90, 20),
+ ("address", 110, 40),
+ ("city", 150, 28),
+ ("state", 178, 2),
+ ("zip", 180, 9),
+ ("property_type", 189, 12),
+ ("amount_cents", 201, 12), # right-justified, zero-padded CENTS (implied 2 decimals)
+]
+RECORD_WIDTH = 213
+_PROPERTY_RECORD_TYPE = "PR"
+
+
+def _field(line: str, start: int, length: int) -> str:
+ return line[start:start + length].strip()
+
+
+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 build_naupa2_line(fields: dict) -> str:
+ """Assemble one fixed-width line from a field dict (used by tests/tools).
+
+ Guarantees column alignment: numeric amount_cents is right-justified zero-padded,
+ everything else is left-justified space-padded, each truncated to its width.
+ """
+ line = [" "] * RECORD_WIDTH
+ for name, start, length in LAYOUT:
+ val = str(fields.get(name, ""))
+ if name == "amount_cents":
+ val = val.rjust(length, "0")[:length]
+ else:
+ val = val.ljust(length)[:length]
+ line[start:start + length] = list(val)
+ return "".join(line)
+
+
+def _amount_from_cents(cents_field: str) -> Decimal | None:
+ digits = cents_field.strip()
+ if not digits or not digits.isdigit():
+ return None
+ return (Decimal(digits) / Decimal(100)).quantize(Decimal("0.01"))
+
+
+def parse_naupa2_feed(data: bytes, jurisdiction: str) -> Iterable["object"]:
+ from services.ingestion.ingest import CanonicalProperty # lazy: avoid import cycle
+
+ text = data.decode("utf-8", errors="replace")
+ for raw_line in text.splitlines():
+ if not raw_line.strip():
+ continue
+ if _field(raw_line, 0, 2) != _PROPERTY_RECORD_TYPE:
+ # header/trailer/holder records are skipped by this property-detail adapter
+ continue
+
+ rec = {name: _field(raw_line, start, length) for name, start, length in LAYOUT}
+ owner_name = f"{rec['owner_first']} {rec['owner_last']}".strip()
+ is_business = _looks_like_business(owner_name)
+ amount = _amount_from_cents(rec["amount_cents"])
+ raw_hash = hashlib.sha256(raw_line.encode("utf-8")).hexdigest()
+ owner_norm = (
+ normalize_business(owner_name) if is_business else normalize_text(owner_name)
+ )
+ yield CanonicalProperty(
+ jurisdiction=jurisdiction,
+ source_property_id=rec["source_property_id"],
+ holder_name_raw=rec["holder_name"],
+ 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=raw_line,
+ raw_record_hash=raw_hash,
+ city_normalized=normalize_text(rec["city"]) or None,
+ region=normalize_text(rec["state"]) or None,
+ postal_code=normalize_postal(rec["zip"]),
+ property_type=normalize_text(rec["property_type"]) or None,
+ )
diff --git a/tests/test_cycle3_naupa_and_blocking.py b/tests/test_cycle3_naupa_and_blocking.py
new file mode 100644
index 0000000..ced8e31
--- /dev/null
+++ b/tests/test_cycle3_naupa_and_blocking.py
@@ -0,0 +1,102 @@
+"""Cycle 3 tests — NAUPA II fixed-width adapter parity + candidate-generation blocking.
+
+Run: python -m tests.test_cycle3_naupa_and_blocking
+
+Proves:
+ 1. The NAUPA II adapter ingests via the SAME ingest_authorized_feed pipeline (registry
+ dispatch), yields the canonical shape, parses cents->dollars, and masks.
+ 2. Header/trailer records are skipped; property records accepted.
+ 3. Same-file idempotency holds for the NAUPA path too.
+ 4. blocking_keys() puts spelling variants of the same owner in a shared bucket (so they
+ get compared) — the candidate-generation step that makes matching sub-O(n²).
+"""
+from __future__ import annotations
+
+import sys
+import tempfile
+from decimal import Decimal
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from services.common.sqlite_repo import FileObjectStore, SqliteRepository
+from services.ingestion.ingest import FeedDefinition, ingest_authorized_feed
+from services.ingestion.naupa2 import build_naupa2_line
+from services.matching.entity_match import MatchInput, blocking_keys, entity_match
+
+
+def ok(msg: str) -> None:
+ print(f" ✓ {msg}")
+
+
+def _naupa2_bytes() -> bytes:
+ lines = [
+ build_naupa2_line({"record_type": "HD", "holder_name": "SYNTHETIC STATE FILE"}), # header, skipped
+ build_naupa2_line({
+ "record_type": "PR", "source_property_id": "SP-N001",
+ "holder_name": "First Synthetic Bank", "owner_last": "ONEIL",
+ "owner_first": "CATHERINE", "address": "100 TEST ST", "city": "SPRINGFIELD",
+ "state": "SM", "zip": "000010000", "property_type": "CHECK",
+ "amount_cents": "4250", # $42.50
+ }),
+ build_naupa2_line({
+ "record_type": "PR", "source_property_id": "SP-N002",
+ "holder_name": "Made-Up Securities", "owner_last": "",
+ "owner_first": "ACME WIDGETS INC", "address": "300 NOWHERE BLVD",
+ "city": "RIVERTOWN", "state": "SM", "zip": "000020000",
+ "property_type": "SECURITIES", "amount_cents": "530000", # $5,300.00
+ }),
+ build_naupa2_line({"record_type": "TR", "holder_name": "TRAILER 2 RECORDS"}), # trailer, skipped
+ ]
+ return ("\n".join(lines) + "\n").encode("utf-8")
+
+
+def main() -> int:
+ tmp = Path(tempfile.mkdtemp(prefix="upp-cycle3-"))
+ store = FileObjectStore(tmp / "obj")
+ (tmp / "obj").mkdir(parents=True, exist_ok=True)
+ store.write_bytes("incoming/naupa2_feed.dat", _naupa2_bytes())
+ repo = SqliteRepository(str(tmp / "p.db"))
+
+ print("1) NAUPA II ingest via the shared pipeline (registry dispatch)")
+ feed = FeedDefinition("SAMPLE", "incoming/naupa2_feed.dat", format_name="naupa2_v1")
+ r = ingest_authorized_feed(feed, store, repo)
+ assert r["status"] == "completed", r
+ assert r["accepted"] == 2, f"expected 2 property records, got {r['accepted']}"
+ assert r["rejected"] == 0, r
+ ok(f"accepted={r['accepted']} (2 PR records; HD/TR skipped)")
+
+ print("2) cents parsed to dollars; masking applied")
+ row = repo.conn.execute(
+ "SELECT amount FROM property WHERE source_property_id='SP-N001'").fetchone()
+ assert Decimal(str(row["amount"])) == Decimal("42.50"), row["amount"]
+ hits = repo.masked_search("Oneil")
+ assert hits and "ONEIL" not in (hits[0]["owner_name_masked"] or "").upper(), hits
+ ok(f"amount 4250c -> $42.50; masked hit {hits[0]['owner_name_masked']}")
+
+ print("3) idempotency on the NAUPA path")
+ n_before = repo.count_properties("SAMPLE")
+ r2 = ingest_authorized_feed(feed, store, repo)
+ assert r2["status"] == "duplicate", r2
+ assert repo.count_properties("SAMPLE") == n_before
+ ok(f"re-ingest -> duplicate; count stable at {n_before}")
+
+ print("4) blocking keys put same-owner variants in a shared bucket")
+ a = MatchInput("Catherine O'Neil", postal_code="00001", region="SM")
+ b = MatchInput("Kathryn ONeill", postal_code="00001", region="SM")
+ c = MatchInput("Jonathan Doe", postal_code="99999", region="XX")
+ ka, kb, kc = blocking_keys(a), blocking_keys(b), blocking_keys(c)
+ phon_a = {k for k in ka if k.startswith("ph:")}
+ phon_b = {k for k in kb if k.startswith("ph:")}
+ assert phon_a & phon_b, f"variants should share a phonetic block key: {phon_a} vs {phon_b}"
+ assert not (phon_a & {k for k in kc if k.startswith('ph:')}), "unrelated names shouldn't share phonetic key"
+ # and the blocked pair actually scores as a match candidate
+ assert entity_match(a, b).disposition in ("manual_review", "likely_duplicate_candidate")
+ ok(f"variants share {sorted(phon_a & phon_b)}; distinct owner excluded from that bucket")
+
+ print("\nALL CYCLE-3 ASSERTIONS PASSED ✅")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/tests/test_cycle4_fairness.py b/tests/test_cycle4_fairness.py
new file mode 100644
index 0000000..ca2a87f
--- /dev/null
+++ b/tests/test_cycle4_fairness.py
@@ -0,0 +1,68 @@
+"""Fairness / coverage tests for candidate generation (referenced by entity_match docstring).
+
+The fairness property that matters at the BLOCKING stage is coverage: no name distribution
+may be systematically un-indexable, or those owners become structurally un-findable (they'd
+never enter a candidate set, so their property is never matched to them). This is distinct
+from scoring fairness (calibration parity), which belongs to the trained production model.
+
+These tests assert the coverage floor holds across diverse name distributions AND document
+the known Soundex limitation so it can't be forgotten. Stdlib only, $0.
+
+Run: python -m tests.test_cycle4_fairness
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from services.common.normalize import phonetic_key
+from services.matching.entity_match import MatchInput, blocking_keys
+
+# Deliberately diverse surnames — Anglo, Polish, Vietnamese, Arabic, Hispanic, hyphenated.
+DIVERSE_NAMES = [
+ "Catherine O'Neil", "Kathryn ONeill", "Grzegorz Brzęczyszczykiewicz",
+ "Nguyễn Thị Hương", "محمد بن سلمان", "José García-Márquez",
+ "Xochitl Ramírez", "Þórunn Jónsdóttir", "李伟", "O", " ",
+]
+
+
+def ok(msg: str) -> None:
+ print(f" ✓ {msg}")
+
+
+def main() -> int:
+ print("1) Coverage floor: every non-empty name yields >=1 blocking key")
+ unindexable = []
+ for name in DIVERSE_NAMES:
+ keys = blocking_keys(MatchInput(name, postal_code="12345", region="SM"))
+ # A record with an address is ALWAYS indexable via zip/region even when the name
+ # script produces no phonetic/token key — that's the coverage guarantee.
+ if not keys:
+ unindexable.append(name)
+ assert not unindexable, f"these names produced NO blocking key (un-findable): {unindexable}"
+ ok(f"all {len(DIVERSE_NAMES)} names indexable (name-key or geo-key)")
+
+ print("2) A name with NO geo still needs a name-derived key when it has letters")
+ # A latin-script name with no address must still be indexable by a name key.
+ keys = blocking_keys(MatchInput("Catherine O'Neil"))
+ assert any(k.startswith(("ph:", "tok0:")) for k in keys), keys
+ ok(f"name-only record indexable via {[k for k in keys if k.startswith(('ph:','tok0:'))][:1]}")
+
+ print("3) Known limitation is real, not hidden: non-latin scripts get no phonetic key")
+ # This DOCUMENTS the gap the docstring warns about — Soundex is latin-only. The test
+ # asserts the CURRENT behavior so a future Double-Metaphone upgrade visibly changes it.
+ cjk = phonetic_key("李伟")
+ assert cjk == "", f"expected empty phonetic key for CJK today, got {cjk!r}"
+ # ...but such a record is STILL indexable when it has geo (see test 1), so coverage holds.
+ cjk_keys = blocking_keys(MatchInput("李伟", postal_code="12345"))
+ assert any(k.startswith("zip:") for k in cjk_keys), cjk_keys
+ ok("CJK name has no phonetic key today (limitation logged) but stays geo-indexable")
+
+ print("\nALL FAIRNESS/COVERAGE ASSERTIONS PASSED ✅")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
← a02e4b2 Cycle 3 complete: blocking_keys, Sec2.5 encryption marker, r
·
back to Unclaimed Property Platform
·
Cycle 3: NAUPA II adapter + candidate-blocking + guardrail t 9fbefae →