← back to Unclaimed Property Platform
scripts/demo_tour.py
139 lines
"""One-command end-to-end demo tour of the platform (synthetic data, $0, stdlib).
python -m scripts.demo_tour
Runs the whole vertical slice and narrates each step so a non-technical evaluator can see
what the system does: ingest (3 formats) + reconciliation -> masked search -> suppression
-> claim lifecycle with reliable outbox delivery. Nothing here touches a real state feed.
"""
from __future__ import annotations
import sys
import tempfile
from pathlib import Path
from uuid import uuid4
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from services.common.rate_limit import SlidingWindowRateLimiter
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.ingestion.naupa3 import build_naupa3_xml
from services.search.service import SearchService
from services.claims.claim_workflow import ClaimStatus, queue_state_submission, transition_claim
from services.claims.sqlite_claim_repo import SqliteClaimRepository
from services.claims.outbox_worker import OutboxWorker
ROOT = Path(__file__).resolve().parents[1]
SAMPLE_CSV = ROOT / "data" / "sample" / "sample_state_feed.csv"
def h(title: str) -> None:
print(f"\n{'='*70}\n {title}\n{'='*70}")
class FlakyStateAdapter:
"""Fails once (state API timeout), then succeeds — shows the outbox retry."""
def __init__(self) -> None:
self.calls = 0
def submit_claim(self, claim, idempotency_key) -> str:
self.calls += 1
if self.calls == 1:
raise RuntimeError("simulated state API timeout")
return "CA-CASE-000123"
def main() -> int:
tmp = Path(tempfile.mkdtemp(prefix="upp-demo-"))
store = FileObjectStore(tmp / "obj")
(tmp / "obj").mkdir(parents=True, exist_ok=True)
repo = SqliteRepository(str(tmp / "demo.db"))
print("National Unclaimed-Property Platform — end-to-end demo (SYNTHETIC data, $0)")
print("No real state feed is touched; every record below is fabricated sample data.")
# 1) INGEST 3 FORMATS -------------------------------------------------------
h("1. AUTHORIZED-FEED INGESTION — one pipeline, three formats")
store.write_bytes("incoming/state.csv", SAMPLE_CSV.read_bytes())
r_csv = ingest_authorized_feed(
FeedDefinition("SAMPLE", "incoming/state.csv", expected_count=8), store, repo)
print(f" CSV feed : accepted={r_csv['accepted']} rejected={r_csv['rejected']} "
f"reconciliation={r_csv['reconciliation']}")
naupa2 = "\n".join([
build_naupa2_line({"record_type": "PR", "source_property_id": "N2-1",
"holder_name": "Second Synthetic Bank", "owner_last": "NGUYEN",
"owner_first": "MINH", "city": "LAKESIDE", "state": "SM",
"zip": "000030000", "property_type": "CHECK", "amount_cents": "7500"}),
]) + "\n"
store.write_bytes("incoming/state.naupa2", naupa2.encode())
r_n2 = ingest_authorized_feed(
FeedDefinition("SAMPLE", "incoming/state.naupa2", format_name="naupa2_v1"), store, repo)
print(f" NAUPA II (fixed): accepted={r_n2['accepted']} ($75.00 parsed from cents)")
store.write_bytes("incoming/state.xml", build_naupa3_xml([
{"property_id": "N3-1", "holder_name": "Third Synthetic Co", "first_name": "Robert",
"last_name": "Garcia-Marquez", "city": "Rivertown", "state": "SM", "zip": "00002",
"property_type": "SECURITIES", "amount": "5300.00"}]))
r_n3 = ingest_authorized_feed(
FeedDefinition("SAMPLE", "incoming/state.xml", format_name="naupa3_v1"), store, repo)
print(f" NAUPA III (XML) : accepted={r_n3['accepted']} (namespace-tolerant, XXE-guarded)")
print(f" → total properties in catalog: {repo.count_properties('SAMPLE')}")
# 2) MASKED SEARCH ----------------------------------------------------------
h("2. FREE PUBLIC SEARCH — masked results, rate-limited, recall-fixed")
svc = SearchService(repository=repo,
limiter=SlidingWindowRateLimiter(max_requests=100, window_seconds=60))
for q in ("Oneil", "Nguyen"):
hits = svc.search(q, client_key="demo")
for hitrow in hits[:2]:
print(f" search '{q}' → {hitrow['owner_name_masked']:<12} "
f"{hitrow['owner_city'] or '-':<12} {hitrow['amount_band'] or '-':<12} "
f"[{hitrow['jurisdiction']}]")
print(" (note: 'Oneil' finds \"O'Neil\"; names are masked; exact amount never exposed)")
# 3) SUPPRESSION ------------------------------------------------------------
h("3. STATE SUPPRESSION — near-real-time takedown, fully audited")
pid = repo._property_id("SAMPLE", "SP-0001")
print(f" before: 'Catherine' returns {len(svc.search('Catherine', client_key='d2'))} hit(s)")
repo.suppress_property("SAMPLE", "SP-0001", actor="state:CA", reason="privacy request")
print(f" after suppress by state:CA → {len(svc.search('Catherine', client_key='d3'))} hit(s) "
f"(data retained, {len(repo.audit_trail(pid))} audit event(s) written)")
repo.reinstate_property("SAMPLE", "SP-0001", actor="state:CA")
print(f" after reinstate → {len(svc.search('Catherine', client_key='d4'))} hit(s) again")
# 4) CLAIM LIFECYCLE + OUTBOX ----------------------------------------------
h("4. CLAIM WORKFLOW — state machine + reliable outbox delivery")
claims = SqliteClaimRepository(repo.conn)
claim = claims.create_claim("SAMPLE", public_property_reference=pid, claimant_id=uuid4())
print(f" created claim {str(claim.claim_id)[:8]}… status={claim.status.value}")
for target in (ClaimStatus.IDENTITY_PENDING, ClaimStatus.EVIDENCE_PENDING,
ClaimStatus.READY_FOR_SUBMISSION):
transition_claim(claims, claim.claim_id, target, actor_id="worker",
idempotency_key=f"k-{target.value}")
print(f" → {target.value}")
queue_state_submission(claims, claim.claim_id, actor_id="worker", idempotency_key="sub")
print(" → submitting (outbox event written in same transaction)")
worker = OutboxWorker(claims, FlakyStateAdapter(), max_attempts=3)
d1 = worker.drain_once()
print(f" outbox drain #1: {d1} (state API timed out — claim safely held, will retry)")
d2 = worker.drain_once()
final = claims.get_for_update(claim.claim_id)
print(f" outbox drain #2: {d2} → claim {final.status.value}, "
f"state_case_id={final.state_case_id}")
print(" (a claimant could never self-approve — APPROVED/PAID require a state:… actor)")
h("DEMO COMPLETE")
print(" Ingested 3 formats, served masked search, suppressed+reinstated under audit,")
print(" and drove a claim to the state with at-least-once outbox delivery — all on")
print(" synthetic data, $0, no network. See README.md for the business/legal path")
print(" that actually unblocks a real deployment (state data-use agreements).")
return 0
if __name__ == "__main__":
raise SystemExit(main())