← back to Unclaimed Property Platform
scripts/serve_ca.py
143 lines
#!/usr/bin/env python3
"""
serve_ca.py — masked public search over the real California unclaimed-property DB.
Serves db/ca_unclaimed.sqlite on http://127.0.0.1:8799 : a name-search box + results
table showing MASKED owner names, city/state, masked amount band, and holder. This is
the platform's public-facing projection — it never returns the raw owner name or a
confirmable exact amount to an anonymous searcher (anti-oracle: you can't binary-search
to confirm a stranger's identity).
Stdlib only. Read-only against the DB. Usage: python3 scripts/serve_ca.py [port]
"""
from __future__ import annotations
import html
import json
import sqlite3
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse, parse_qs
ROOT = Path(__file__).resolve().parents[1]
DB = ROOT / "db" / "ca_unclaimed.sqlite"
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8799
MAX_ROWS = 100
PAGE = """<!doctype html><html><head><meta charset=utf-8>
<title>California Unclaimed Property — masked search</title>
<meta name=viewport content="width=device-width,initial-scale=1">
<style>
body{{font:15px/1.5 -apple-system,Segoe UI,Roboto,sans-serif;margin:0;background:#0d1117;color:#e6edf3}}
header{{padding:22px 26px;border-bottom:1px solid #30363d;background:#161b22}}
h1{{margin:0;font-size:19px;font-weight:650}} .sub{{color:#8b949e;font-size:13px;margin-top:4px}}
main{{max-width:1000px;margin:0 auto;padding:22px 26px}}
form{{display:flex;gap:10px;margin-bottom:18px}}
input[type=text]{{flex:1;padding:11px 13px;border:1px solid #30363d;border-radius:8px;background:#0d1117;color:#e6edf3;font-size:15px}}
button{{padding:11px 20px;border:0;border-radius:8px;background:#238636;color:#fff;font-weight:600;cursor:pointer}}
table{{width:100%;border-collapse:collapse;font-size:14px}}
th,td{{text-align:left;padding:9px 11px;border-bottom:1px solid #21262d}}
th{{color:#8b949e;font-weight:600;font-size:12px;text-transform:uppercase;letter-spacing:.04em}}
td.amt{{font-variant-numeric:tabular-nums;color:#3fb950;font-weight:600}}
.mask{{font-family:ui-monospace,Menlo,monospace;letter-spacing:.06em}}
.meta{{color:#8b949e;font-size:13px;margin:10px 0 16px}}
.empty{{color:#8b949e;padding:30px 0}}
.badge{{display:inline-block;background:#1f6feb22;color:#58a6ff;border:1px solid #1f6feb55;border-radius:5px;padding:1px 7px;font-size:11px;margin-left:8px}}
</style></head><body>
<header><h1>California Unclaimed Property <span class=badge>MASKED · SAMPLE</span></h1>
<div class=sub>{count} records · public data from the CA State Controller · names masked, exact amounts withheld</div></header>
<main>
<form method=get action="/"><input type=text name=q placeholder="Search an owner name (e.g. a last name)" value="{q}" autofocus>
<button type=submit>Search</button></form>
{results}
</main></body></html>"""
def _band(amount: float | None) -> str:
if amount is None:
return "—"
if amount < 100:
return "Under $100"
if amount < 1000:
return "$100–$1,000"
if amount < 10000:
return "$1,000–$10,000"
return "$10,000+"
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a): # quiet
pass
def _db(self):
c = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
c.row_factory = sqlite3.Row
return c
def do_GET(self):
u = urlparse(self.path)
if u.path == "/api/search":
return self._api(parse_qs(u.query).get("q", [""])[0].strip())
if u.path != "/":
self.send_error(404)
return
q = parse_qs(u.query).get("q", [""])[0].strip()
c = self._db()
total = c.execute("SELECT COUNT(*) FROM ca_property").fetchone()[0]
results = ""
if q:
norm = q.upper()
rows = c.execute(
"""SELECT owner_name_masked, owner_city, owner_state, amount, holder_name
FROM ca_property WHERE owner_name_normalized LIKE ?
ORDER BY amount DESC LIMIT ?""",
(f"%{norm}%", MAX_ROWS)).fetchall()
hits = c.execute(
"SELECT COUNT(*) FROM ca_property WHERE owner_name_normalized LIKE ?",
(f"%{norm}%",)).fetchone()[0]
if rows:
body = "".join(
f"<tr><td class=mask>{html.escape(r['owner_name_masked'] or '')}</td>"
f"<td>{html.escape((r['owner_city'] or '').title())}, {html.escape(r['owner_state'] or '')}</td>"
f"<td class=amt>{_band(r['amount'])}</td>"
f"<td>{html.escape((r['holder_name'] or '')[:44])}</td></tr>"
for r in rows)
more = f" (showing top {MAX_ROWS} by value)" if hits > MAX_ROWS else ""
results = (f"<div class=meta>{hits:,} match(es) for “{html.escape(q)}”{more}. "
f"To claim, verify identity on claimit.ca.gov.</div>"
"<table><tr><th>Owner (masked)</th><th>Last known city</th>"
"<th>Amount band</th><th>Reported by</th></tr>" + body + "</table>")
else:
results = f"<div class=empty>No matches for “{html.escape(q)}”.</div>"
c.close()
out = PAGE.format(count=f"{total:,}", q=html.escape(q), results=results)
self._send(out.encode(), "text/html; charset=utf-8")
def _api(self, q: str):
c = self._db()
rows = []
if q:
rows = [dict(r) for r in c.execute(
"""SELECT owner_name_masked, owner_city, owner_state, amount_band, holder_name
FROM ca_property WHERE owner_name_normalized LIKE ? ORDER BY amount DESC LIMIT ?""",
(f"%{q.upper()}%", MAX_ROWS))]
c.close()
self._send(json.dumps({"query": q, "count": len(rows), "results": rows}).encode(),
"application/json")
def _send(self, body: bytes, ctype: str):
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
if __name__ == "__main__":
if not DB.exists():
print(f"DB not found: {DB} — run scripts/load_ca.py first", file=sys.stderr)
raise SystemExit(1)
print(f"CA masked search → http://127.0.0.1:{PORT} (DB: {DB.name})")
ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()