diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index d124075..936f4a4 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -27,13 +27,13 @@ jobs:
- name: Install dependencies
run: |
python -m pip install --upgrade pip
- pip install -r requirements.txt ruff
+ pip install -r requirements.txt ruff==0.15.20
- name: Lint
- run: ruff check app.py prizm_client.py cache_manager_new.py cache_cli.py test_prizm_api.py
+ run: ruff check app.py prizm_client.py cache_manager_new.py cache_cli.py cohort_queue.py historical_import.py test_prizm_api.py test_cohort_queue.py test_historical_import.py
- name: Test
- run: python -m unittest test_prizm_api.py
+ run: python -m unittest test_prizm_api.py test_cohort_queue.py test_historical_import.py
node:
name: Node tests and build
diff --git a/.gitignore b/.gitignore
index 102e7b1..b6d5a16 100644
--- a/.gitignore
+++ b/.gitignore
@@ -56,3 +56,5 @@ debug_screenshots/
missing_postal_codes.txt
postal_code_export*.csv
postal_code_v2_export*.csv
+
+*.migration.lock
diff --git a/README.md b/README.md
index bde5305..4bc43fd 100644
--- a/README.md
+++ b/README.md
@@ -25,7 +25,7 @@ GET /health
GET /dashboard
```
-The dashboard shows cache totals, recent daily additions, recent failures, searchable cached postal-code data, CSV export, and a manual weekly-report send button. It uses HTTP Basic Auth.
+The root URL and `/dashboard` show cache totals, daily lookup outcomes, failure history (including uncached quota/network errors), searchable and paginated postal-code records with expandable details, complete cache/failure CSV exports, and weekly-report preview. It uses HTTP Basic Auth and fails closed without a configured password or API key. Dashboard credentials provide read-only reporting access; sending an email requires the API key.
Recommended dashboard/reporting variables:
@@ -74,7 +74,9 @@ Useful if you want the full segment reference data without postal-code lookup.
```http
GET /api/cache/entries?status=success&search=V8A&limit=500
-GET /api/cache/export.csv
+GET /api/cache/export.csv?include_expired=1
+GET /api/lookups?failures=1&days=7&limit=100&offset=0
+GET /api/lookups/failures.csv?days=3650
```
### Weekly Report
@@ -192,3 +194,14 @@ python3 -m unittest test_prizm_api.py
- Urban postal codes still depend on the Environics geocoder API. During triage on June 21, 2026, that endpoint returned `403 Quota Exceeded`, so urban postal-code lookups may fail until the public quota replenishes or a licensed/geocoding credential is available.
- `prizm_cache_v2.db`, CSV exports, debug screenshots, and `node_modules` are runtime/local artifacts and are ignored by Docker.
- The old crash was most likely caused by a globally reused Selenium Chrome session combined with Flask debug mode and repeated screenshot/page-source capture. Removing the browser from request handling eliminates that failure path.
+
+## Operational behavior
+
+- Weekly metrics use a trailing seven-day window with UTC daily dates. Lookup success means an API result, not a successful Salesforce update. Cache hits, upstream attempts, upstream successes/failures, and unique postal codes are separate metrics. First captures mean the first successful upstream result in retained lookup-event history; imported/earlier records may predate that history.
+- The daily Salesforce selector no longer retries an assigned household solely because net worth is unavailable. It preserves an existing net-worth value when no historical reference exists. Retryable quota/network errors leave the household eligible for a later attempt.
+- Salesforce household updates now roll back the batch and surface a failed job if any update fails. This exposes errors previously hidden by ignored partial-update results; inspect the job error before retrying a persistently failing batch.
+- The current public reference data includes dwelling type but no net worth. Net worth shown here is historical segment reference data of mixed vintage, not a current postal-code-specific estimate. Missing values remain unavailable. Some dwelling types cannot be represented by the existing Salesforce picklist and map to `None or Unverifiable`.
+- `MAX_BATCH_POSTAL_CODES=10` limits this application's request size; it does not establish an upstream allowance. A claimed three-per-day restriction remains unverified. Confirm licensed access with Environics before adjusting collection volume.
+- Weekly delivery already uses Salesforce's scheduled sender; do not create a second Railway email schedule. Salesforce job completion is not an inbox-delivery receipt.
+
+See [September 6 investigation and upgrade scope](docs/2026-09-06-upgrade.md).
diff --git a/app.py b/app.py
index a40450f..0100a37 100644
--- a/app.py
+++ b/app.py
@@ -13,6 +13,8 @@
from flask import Flask, Response, jsonify, make_response, request
from cache_manager_new import cache_manager
+from cohort_queue import CohortQueue
+from historical_import import merge_history
from prizm_client import PrizmClient, PrizmLookupError, normalize_postal_code
from segment_net_worth import average_household_net_worth, average_household_net_worth_amount
@@ -57,7 +59,9 @@
.muted { color:var(--muted); }
.section-title { display:flex; justify-content:space-between; align-items:end; gap:12px; margin:28px 0 12px; }
.section-title h2 { margin:0; font-size:20px; }
- pre { white-space:pre-wrap; }
+ pre { white-space:pre-wrap; overflow-wrap:anywhere; max-width:540px; }
+ #reportStatus { white-space:pre-wrap; }
+ table { display:block; overflow-x:auto; }
@media (max-width:900px) { .grid { grid-template-columns:repeat(2,minmax(0,1fr)); } main, header { padding-left:18px; padding-right:18px; } table { font-size:13px; } }
@media (max-width:580px) { .grid { grid-template-columns:1fr; } th:nth-child(5),td:nth-child(5),th:nth-child(6),td:nth-child(6){display:none;} }
@@ -72,25 +76,30 @@
–
postal codes cached
–
successful cached codes
–
failed / unassigned cached codes
-
–
lookups recorded this week
+
–
lookups in the trailing 7 days
-
Export and reports
Download the full cache or send the configured weekly email report.
+
Export and reports
Download cached data and failed attempts, or preview the scheduled weekly report.
Based on when each postal code was first cached or refreshed.
+
Daily lookup outcomes
Trailing 7 days, grouped by UTC date. Cache hits do not represent new captures.
+
Loading…
-
Postal codes
Search cached successes, failures, and invalid records.
+
Postal codes
Search all cached records, including expired entries. Net worth is historical segment reference data; unavailable values remain blank.
@@ -98,12 +107,19 @@
-
Postal code
Status
Segment
Name
Home type
Income
Net worth
Cached
Message
+
Postal code
Status
Segment
Name
Home type
Income
Historical net worth
Cached
Message
Loading…
+
+
Failed lookup history
+
Includes quota/network errors that are not stored in the cache. Dates are UTC. Export downloads all retained failures.
+
Loading…
+
@@ -210,7 +245,7 @@ def has_valid_api_key() -> bool:
def has_valid_dashboard_auth() -> bool:
username, password = dashboard_credentials()
if not password:
- return True
+ return False
provided_username, provided_password = parse_basic_auth()
return secrets.compare_digest(provided_username or "", username or "") and secrets.compare_digest(provided_password or "", password)
@@ -226,12 +261,15 @@ def require_authentication():
if request.path == "/health" or request.method == "OPTIONS":
return None
- dashboard_paths = {"/", "/dashboard", "/api/dashboard/summary", "/api/cache/entries", "/api/cache/export.csv", "/api/reports/weekly", "/api/reports/weekly/send"}
+ dashboard_paths = {"/", "/dashboard", "/api/dashboard/summary", "/api/cache/entries", "/api/cache/export.csv", "/api/reports/weekly", "/api/lookups", "/api/lookups/failures.csv", "/api/cohort", "/api/cohort/export.csv"}
if request.path in dashboard_paths:
- if has_valid_dashboard_auth() or has_valid_api_key():
+ if has_valid_dashboard_auth() or (os.environ.get("PRIZM_API_KEY") and has_valid_api_key()):
return None
return dashboard_auth_required()
+ if request.path == "/api/reports/weekly/send" and not os.environ.get("PRIZM_API_KEY"):
+ return jsonify({"error": "Email sending requires a configured API key"}), 503
+
if request.path.startswith("/api/") and has_valid_api_key():
return None
@@ -304,6 +342,7 @@ def get_prizm_code(postal_code: str, endpoint: str = "single", batch_id: Optiona
"home_type": "",
"status": "error",
"message": str(exc),
+ "retryable": True,
}
if should_cache:
@@ -383,6 +422,75 @@ def get_batch_prizm():
)
+def cohort_queue():
+ return CohortQueue(cache_manager.db_path)
+
+
+@app.route("/api/cohort", methods=["GET"])
+def cohort_status():
+ queue = cohort_queue()
+ return jsonify({"summary": queue.summary(), "entries": queue.entries()})
+
+
+@app.route("/api/cohort/export.csv", methods=["GET"])
+def cohort_export():
+ response = Response(cohort_queue().export_csv(), mimetype="text/csv")
+ response.headers["Content-Disposition"] = "attachment; filename=major-donor-coverage.csv"
+ return response
+
+
+@app.route("/api/cache/import-history", methods=["POST"])
+def import_historical_cache():
+ if not os.environ.get("PRIZM_API_KEY") or not has_valid_api_key():
+ return jsonify({"error": "Historical import requires an API key"}), 401
+ if request.content_length is None or request.content_length > 2_000_000:
+ return jsonify({"error": "A historical JSON payload under 2 MB is required"}), 400
+ data = request.get_json(silent=True)
+ if not isinstance(data, dict):
+ return jsonify({"error": "Expected a JSON object"}), 400
+ try:
+ result = merge_history(cache_manager.db_path, data.get("rows"))
+ except (ValueError, TypeError):
+ return jsonify({"error": "Invalid historical data; import rolled back"}), 400
+ return jsonify(result)
+
+
+@app.route("/api/cohort/import", methods=["POST"])
+def import_cohort():
+ if not os.environ.get("PRIZM_API_KEY") or not has_valid_api_key():
+ return jsonify({"error": "Import requires an API key"}), 401
+ if request.content_length is None or request.content_length > 1_000_000:
+ return jsonify({"error": "A CSV body under 1 MB is required"}), 400
+ try:
+ import io
+ summary = cohort_queue().import_accounts(io.StringIO(request.get_data().decode("utf-8-sig")))
+ except (ValueError, UnicodeError):
+ return jsonify({"error": "Invalid Canadian billing-address CSV; cohort unchanged"}), 400
+ return jsonify(summary)
+
+
+@app.route("/api/cohort/capture", methods=["POST"])
+def capture_cohort():
+ # Collection is a mutation and always requires an explicit API key.
+ if not os.environ.get("PRIZM_API_KEY") or not has_valid_api_key():
+ return jsonify({"error": "Collection requires an API key"}), 401
+ queue = cohort_queue()
+ if not queue.summary()["configured"]:
+ return jsonify({"error": "Major-donor cohort has not been imported"}), 503
+ results = []
+ # One code per call keeps Salesforce callouts below transaction time limits.
+ # Its queueable chains ten calls; the database also enforces the daily ceiling.
+ for _ in range(1):
+ code = queue.claim()
+ if code is None:
+ break
+ # Claim persists before network I/O: retries/concurrent requests cannot repeat a code.
+ result = get_prizm_code(code, endpoint="cohort")
+ queue.finish(code, result)
+ results.append(result)
+ return jsonify({"results": results, "summary": queue.summary()})
+
+
@app.route("/api/segments", methods=["GET"])
def get_segments():
return jsonify({"status": "success", "segments": prizm_client.get_all_segments()})
@@ -390,7 +498,27 @@ def get_segments():
@app.route("/api/dashboard/summary", methods=["GET"])
def dashboard_summary():
- return jsonify(cache_manager.get_dashboard_summary())
+ summary = cache_manager.get_dashboard_summary()
+ summary["cohort"] = cohort_queue().summary()
+ return jsonify(summary)
+
+
+@app.route("/api/lookups", methods=["GET"])
+def lookup_events():
+ rows = cache_manager.list_lookup_events(
+ failures_only=request.args.get("failures") == "1",
+ days=request.args.get("days", 7, type=int),
+ limit=request.args.get("limit", 100, type=int),
+ offset=request.args.get("offset", 0, type=int),
+ )
+ return jsonify({"entries": rows, "count": len(rows)})
+
+
+@app.route("/api/lookups/failures.csv", methods=["GET"])
+def export_lookup_failures():
+ response = Response(cache_manager.export_failures_csv(request.args.get("days", 7, type=int)), mimetype="text/csv")
+ response.headers["Content-Disposition"] = "attachment; filename=prizm-failures.csv"
+ return response
@app.route("/api/cache/entries", methods=["GET"])
@@ -483,34 +611,49 @@ def get_debug_html(postal_code):
def build_weekly_report(days: int = 7) -> Dict[str, Any]:
summary = cache_manager.get_dashboard_summary()
- events = summary.get("lookup_events_7d") or cache_manager.get_lookup_event_summary(days)
+ events = cache_manager.get_lookup_event_summary(days)
+ summary["lookup_events_7d"] = events
+ summary["cohort"] = cohort_queue().summary()
stats = summary.get("cache_stats", {})
- daily_counts = summary.get("daily_cache_counts", [])[:days]
- failures = summary.get("recent_failures", [])[:20]
+ daily_counts = events.get("by_day", [])
+ failures = cache_manager.list_lookup_events(failures_only=True, days=days, limit=20)
subject = os.environ.get("WEEKLY_REPORT_SUBJECT", "PRIZM weekly report")
lines = [
"PRIZM weekly report",
"",
+ f"Window: trailing {days} days; daily dates are UTC.",
+ "API success does not prove Salesforce household updates or email delivery.",
f"Lookups recorded: {events.get('lookups', 0)}",
f"Successful lookups: {events.get('successful', 0)}",
f"Failed lookups: {events.get('failed', 0)}",
f"Cache hits: {events.get('cache_hits', 0)}",
f"Upstream attempts: {events.get('upstream_attempts', 0)}",
+ f"Upstream successful: {events.get('upstream_successful', 0)}",
+ f"Upstream failed: {events.get('upstream_failed', 0)}",
+ f"Unique postal codes requested: {events.get('unique_postal_codes', 0)}",
+ f"New captures (first successful lookup in recorded history): {events.get('newly_captured', 0)}",
+ "Historical net worth is a segment reference, not current postal-code wealth.",
"",
f"Total active cached postal codes: {stats.get('valid_entries', 0)}",
f"Cached successes: {(stats.get('status_breakdown') or {}).get('success', 0)}",
f"Cached failed/unassigned: {((stats.get('status_breakdown') or {}).get('error', 0) + (stats.get('status_breakdown') or {}).get('invalid', 0))}",
"",
- "Daily cache additions:",
+ "Daily lookup outcomes (UTC):",
]
+ cohort = summary["cohort"]
+ lines.extend(["", "Major-donor Canadian postal-code coverage:",
+ f"Captured: {cohort['captured']}; pending: {cohort['pending']}; failed: {cohort['failed']}; incomplete: {cohort['incomplete']}",
+ "Daily target: 10 never-attempted codes. Failed/incomplete attempts require review; no automatic repeats.", ""])
if daily_counts:
for row in daily_counts:
- lines.append(f"- {row.get('day')}: {row.get('total', 0)} total, {row.get('successful', 0)} success, {row.get('failed', 0)} failed")
+ lines.append(f"- {row.get('day')}: {row.get('lookups', 0)} lookups, {row.get('cache_hits', 0)} cached, {row.get('upstream_successful', 0)} upstream successes, {row.get('failed', 0)} failures")
else:
lines.append("- None recorded")
- lines.extend(["", "Recent failures/unassigned postal codes:"])
+ if events.get("lookups", 0) and not events.get("upstream_attempts", 0):
+ lines.extend(["", "ATTENTION: All recorded lookups were served without new upstream attempts. Check the enrichment queue for repeated postal codes."])
+ lines.extend(["", "Failed lookup attempts in this reporting window:"])
if failures:
for row in failures:
lines.append(f"- {row.get('postal_code')}: {row.get('message') or row.get('status')}")
diff --git a/cache_manager_new.py b/cache_manager_new.py
index 659412b..bb9d6f4 100644
--- a/cache_manager_new.py
+++ b/cache_manager_new.py
@@ -6,6 +6,7 @@
"""
import csv
+import fcntl
import io
import json
import logging
@@ -16,6 +17,8 @@
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
+from segment_net_worth import average_household_net_worth, average_household_net_worth_amount
+
logger = logging.getLogger(__name__)
@@ -25,7 +28,23 @@ class CacheManager:
def __init__(self, db_path: str = None, cache_duration_days: int = None):
self.db_path = db_path or os.environ.get("PRIZM_CACHE_DB_PATH", "prizm_cache_v2.db")
self.cache_duration_days = cache_duration_days or int(os.environ.get("PRIZM_CACHE_DURATION_DAYS", "90"))
- self._init_database()
+ # Serialize startup migrations across Gunicorn workers. Keep the original
+ # production database before this release adds tables/columns/triggers.
+ with open(self.db_path + ".migration.lock", "a") as lock:
+ fcntl.flock(lock, fcntl.LOCK_EX)
+ if os.environ.get("RAILWAY_ENVIRONMENT_ID") and os.path.exists(self.db_path):
+ snapshot = self.db_path + ".before-cohort-upgrade.db"
+ if not os.path.exists(snapshot):
+ source = sqlite3.connect(self.db_path)
+ target = sqlite3.connect(snapshot + ".tmp")
+ try:
+ source.backup(target)
+ finally:
+ target.close()
+ source.close()
+ os.replace(snapshot + ".tmp", snapshot)
+ logger.info("Saved pre-upgrade database snapshot")
+ self._init_database()
@contextmanager
def _connect(self):
@@ -218,6 +237,7 @@ def _init_database(self):
cursor,
"postal_code_cache",
{
+ "historical_source": "TEXT",
"segment_number": "TEXT",
"segment_name": "TEXT",
"segment_description": "TEXT",
@@ -281,6 +301,27 @@ def _init_database(self):
cursor.execute("CREATE INDEX IF NOT EXISTS idx_lookup_events_source ON lookup_events (source)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_lookup_events_postal_code ON lookup_events (postal_code)")
+ cursor.executescript("""
+ CREATE TABLE IF NOT EXISTS capture_history (
+ postal_code TEXT PRIMARY KEY, captured_at TEXT NOT NULL
+ );
+ INSERT OR IGNORE INTO capture_history
+ SELECT postal_code, COALESCE(cached_at, CURRENT_TIMESTAMP)
+ FROM postal_code_cache WHERE status='success';
+ CREATE TRIGGER IF NOT EXISTS remember_prizm_capture_insert
+ AFTER INSERT ON postal_code_cache WHEN NEW.status='success'
+ BEGIN
+ INSERT OR IGNORE INTO capture_history VALUES
+ (NEW.postal_code, COALESCE(NEW.cached_at, CURRENT_TIMESTAMP));
+ END;
+ CREATE TRIGGER IF NOT EXISTS remember_prizm_capture_update
+ AFTER UPDATE ON postal_code_cache WHEN NEW.status='success'
+ BEGIN
+ INSERT OR IGNORE INTO capture_history VALUES
+ (NEW.postal_code, COALESCE(NEW.cached_at, CURRENT_TIMESTAMP));
+ END;
+ """)
+
conn.commit()
logger.info("Cache database initialized at %s", self.db_path)
@@ -682,6 +723,15 @@ def list_cache_entries(
def _dashboard_row(self, row: sqlite3.Row) -> Dict[str, Any]:
data = self._row_to_response_dict(row)
+ amount = average_household_net_worth_amount(data.get("segment_number"))
+ if amount is not None:
+ if not data.get("average_household_net_worth"):
+ data["average_household_net_worth"] = average_household_net_worth(data.get("segment_number"))
+ if data.get("average_household_net_worth_amount") is None:
+ data["average_household_net_worth_amount"] = amount
+ data["net_worth_source"] = "Historical segment reference" if data.get("average_household_net_worth") else "Unavailable"
+ data["data_source"] = ("Historical import: " + row["historical_source"]
+ if row["historical_source"] and row["cached_at"] is None else "Current dataset")
data.update(
{
"cached_at": row["cached_at"],
@@ -723,7 +773,15 @@ def get_lookup_event_summary(self, days: int = 7) -> Dict[str, Any]:
SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) successful,
SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END) failed,
SUM(CASE WHEN from_cache = 1 THEN 1 ELSE 0 END) cache_hits,
- SUM(CASE WHEN source = 'upstream' THEN 1 ELSE 0 END) upstream_attempts
+ SUM(CASE WHEN source = 'upstream' THEN 1 ELSE 0 END) upstream_attempts,
+ COUNT(DISTINCT postal_code) unique_postal_codes,
+ SUM(CASE WHEN source = 'upstream' AND status = 'success' THEN 1 ELSE 0 END) upstream_successful,
+ SUM(CASE WHEN source = 'upstream' AND status != 'success' THEN 1 ELSE 0 END) upstream_failed,
+ SUM(CASE WHEN source = 'upstream' AND status = 'success' AND NOT EXISTS (
+ SELECT 1 FROM lookup_events prior
+ WHERE prior.postal_code = lookup_events.postal_code
+ AND prior.status = 'success' AND prior.id < lookup_events.id
+ ) THEN 1 ELSE 0 END) newly_captured
FROM lookup_events
WHERE requested_at >= datetime('now', ?)
""",
@@ -738,7 +796,15 @@ def get_lookup_event_summary(self, days: int = 7) -> Dict[str, Any]:
SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) successful,
SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END) failed,
SUM(CASE WHEN from_cache = 1 THEN 1 ELSE 0 END) cache_hits,
- SUM(CASE WHEN source = 'upstream' THEN 1 ELSE 0 END) upstream_attempts
+ SUM(CASE WHEN source = 'upstream' THEN 1 ELSE 0 END) upstream_attempts,
+ COUNT(DISTINCT postal_code) unique_postal_codes,
+ SUM(CASE WHEN source = 'upstream' AND status = 'success' THEN 1 ELSE 0 END) upstream_successful,
+ SUM(CASE WHEN source = 'upstream' AND status != 'success' THEN 1 ELSE 0 END) upstream_failed,
+ SUM(CASE WHEN source = 'upstream' AND status = 'success' AND NOT EXISTS (
+ SELECT 1 FROM lookup_events prior
+ WHERE prior.postal_code = lookup_events.postal_code
+ AND prior.status = 'success' AND prior.id < lookup_events.id
+ ) THEN 1 ELSE 0 END) newly_captured
FROM lookup_events
WHERE requested_at >= datetime('now', ?)
GROUP BY date(requested_at)
@@ -757,11 +823,43 @@ def get_dashboard_summary(self) -> Dict[str, Any]:
"cache_stats": self.get_cache_stats(),
"daily_cache_counts": self.get_daily_cache_counts(30),
"lookup_events_7d": self.get_lookup_event_summary(7),
- "recent_failures": self.list_cache_entries(status="error", limit=50),
+ "recent_failures": self.list_lookup_events(failures_only=True, days=7, limit=50),
}
+ def list_lookup_events(self, failures_only=False, days=7, limit=500, offset=0):
+ conditions = ["requested_at >= datetime('now', ?)"]
+ if failures_only:
+ conditions.append("status != 'success'")
+ with self._connect() as conn:
+ rows = conn.execute(
+ f"SELECT * FROM lookup_events WHERE {' AND '.join(conditions)} "
+ "ORDER BY requested_at DESC, id DESC LIMIT ? OFFSET ?",
+ (f"-{max(1, min(int(days), 3650))} days", max(1, min(int(limit), 5000)), max(0, int(offset))),
+ ).fetchall()
+ return [dict(row) for row in rows]
+
+ @staticmethod
+ def _csv_value(value):
+ if isinstance(value, str) and value.lstrip().startswith(('=', '+', '-', '@')):
+ return "'" + value
+ return value
+
+ def export_failures_csv(self, days=7):
+ output = io.StringIO()
+ fields = ['postal_code', 'requested_at', 'status', 'source', 'message', 'endpoint', 'batch_id']
+ writer = csv.DictWriter(output, fieldnames=fields)
+ writer.writeheader()
+ offset = 0
+ while True:
+ rows = self.list_lookup_events(failures_only=True, days=days, limit=5000, offset=offset)
+ for row in rows:
+ writer.writerow({field: self._csv_value(row.get(field, '')) for field in fields})
+ if len(rows) < 5000:
+ break
+ offset += len(rows)
+ return output.getvalue()
+
def export_cache_csv(self, include_expired: bool = False) -> str:
- rows = self.list_cache_entries(limit=5000, include_expired=include_expired)
output = io.StringIO()
fieldnames = [
"postal_code",
@@ -774,6 +872,8 @@ def export_cache_csv(self, include_expired: bool = False) -> str:
"average_household_income",
"average_household_net_worth",
"average_household_net_worth_amount",
+ "net_worth_source",
+ "data_source",
"education",
"urbanity",
"occupation",
@@ -796,8 +896,14 @@ def export_cache_csv(self, include_expired: bool = False) -> str:
]
writer = csv.DictWriter(output, fieldnames=fieldnames)
writer.writeheader()
- for row in rows:
- writer.writerow({field: row.get(field, "") for field in fieldnames})
+ offset = 0
+ while True:
+ rows = self.list_cache_entries(limit=5000, offset=offset, include_expired=include_expired)
+ for row in rows:
+ writer.writerow({field: self._csv_value(row.get(field, "")) for field in fieldnames})
+ if len(rows) < 5000:
+ break
+ offset += len(rows)
return output.getvalue()
def clear_cache(self) -> bool:
diff --git a/cohort_queue.py b/cohort_queue.py
new file mode 100644
index 0000000..fd1aa95
--- /dev/null
+++ b/cohort_queue.py
@@ -0,0 +1,148 @@
+"""Private cohort collection, independent of Salesforce field completeness and cache TTL."""
+import csv
+import io
+import json
+import re
+import sqlite3
+from contextlib import contextmanager
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+
+def postal_code(value):
+ value = re.sub(r'\s+', '', (value or '').upper())
+ if re.fullmatch(r'[ABCEGHJ-NPRSTVXY][0-9][ABCEGHJ-NPRSTV-Z][0-9][ABCEGHJ-NPRSTV-Z][0-9]', value):
+ return value[:3] + ' ' + value[3:]
+ return None
+
+
+class CohortQueue:
+ def __init__(self, db_path):
+ self.db_path = db_path
+ with self.connect() as db:
+ db.executescript('''
+ CREATE TABLE IF NOT EXISTS capture_history (
+ postal_code TEXT PRIMARY KEY, captured_at TEXT NOT NULL
+ );
+ CREATE TABLE IF NOT EXISTS donor_cohort (
+ postal_code TEXT PRIMARY KEY, account_rows INTEGER NOT NULL
+ );
+ CREATE TABLE IF NOT EXISTS cohort_attempts (
+ postal_code TEXT PRIMARY KEY, day TEXT NOT NULL,
+ started_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ result_json TEXT
+ );
+ CREATE TABLE IF NOT EXISTS cohort_metadata (
+ id INTEGER PRIMARY KEY CHECK(id=1), source_rows INTEGER NOT NULL,
+ excluded_rows INTEGER NOT NULL
+ );
+ ''')
+ # Seed all historical successes, including expired cache and retained events.
+ for query in (
+ "SELECT postal_code, cached_at FROM postal_code_cache WHERE status='success'",
+ "SELECT postal_code, MIN(requested_at) FROM lookup_events WHERE status='success' GROUP BY postal_code",
+ ):
+ for raw, captured in db.execute(query).fetchall():
+ code = postal_code(raw)
+ if code:
+ db.execute('INSERT OR IGNORE INTO capture_history VALUES (?, ?)', (code, str(captured)))
+
+ @contextmanager
+ def connect(self):
+ db = sqlite3.connect(self.db_path, timeout=30)
+ db.row_factory = sqlite3.Row
+ try:
+ with db:
+ yield db
+ finally:
+ db.close()
+
+ def import_accounts(self, stream):
+ reader = csv.DictReader(stream)
+ required = {'Billing Country Code', 'Billing Zip/Postal Code'}
+ if not required.issubset(reader.fieldnames or []):
+ raise ValueError('Salesforce billing country and postal-code columns are required')
+ counts, total, excluded = {}, 0, 0
+ for row in reader:
+ total += 1
+ code = postal_code(row['Billing Zip/Postal Code'])
+ if (row.get('Billing Country Code') or '').strip().upper() != 'CA' or not code:
+ excluded += 1
+ continue
+ counts[code] = counts.get(code, 0) + 1
+ if not counts:
+ raise ValueError('No eligible Canadian billing postal codes; existing cohort preserved')
+ with self.connect() as db:
+ db.execute('BEGIN IMMEDIATE')
+ # Exact membership replacement; retain capture and attempt history permanently.
+ db.execute('DELETE FROM donor_cohort')
+ db.executemany('INSERT INTO donor_cohort VALUES (?, ?)', counts.items())
+ db.execute('INSERT OR REPLACE INTO cohort_metadata VALUES (1, ?, ?)', (total, excluded))
+ return self.summary()
+
+ def claim(self, day=None):
+ day = day or datetime.now(ZoneInfo('America/Vancouver')).date().isoformat()
+ with self.connect() as db:
+ db.execute('BEGIN IMMEDIATE')
+ if db.execute('SELECT COUNT(*) FROM cohort_attempts WHERE day=?', (day,)).fetchone()[0] >= 10:
+ return None
+ row = db.execute('''
+ SELECT c.postal_code FROM donor_cohort c
+ LEFT JOIN capture_history h USING(postal_code)
+ LEFT JOIN cohort_attempts a USING(postal_code)
+ WHERE h.postal_code IS NULL AND a.postal_code IS NULL
+ ORDER BY c.account_rows DESC, c.postal_code LIMIT 1
+ ''').fetchone()
+ if row is None:
+ return None
+ db.execute('INSERT INTO cohort_attempts (postal_code, day) VALUES (?, ?)', (row[0], day))
+ return row[0]
+
+ def finish(self, code, result):
+ with self.connect() as db:
+ db.execute('UPDATE cohort_attempts SET result_json=? WHERE postal_code=?', (json.dumps(result), code))
+ if result.get('status') == 'success':
+ db.execute("INSERT OR IGNORE INTO capture_history VALUES (?, CURRENT_TIMESTAMP)", (code,))
+
+ def entries(self):
+ with self.connect() as db:
+ return [dict(row) for row in db.execute('''
+ SELECT c.postal_code, c.account_rows, h.captured_at, a.started_at,
+ CASE WHEN h.postal_code IS NOT NULL THEN 'captured'
+ WHEN a.result_json IS NOT NULL THEN 'failed'
+ WHEN a.postal_code IS NOT NULL THEN 'incomplete'
+ ELSE 'pending' END AS coverage,
+ a.result_json
+ FROM donor_cohort c LEFT JOIN capture_history h USING(postal_code)
+ LEFT JOIN cohort_attempts a USING(postal_code)
+ ORDER BY c.account_rows DESC, c.postal_code
+ ''')]
+
+ def summary(self):
+ rows = self.entries()
+ with self.connect() as db:
+ meta = db.execute('SELECT source_rows, excluded_rows FROM cohort_metadata').fetchone()
+ return dict(meta or {}, configured=meta is not None, unique_codes=len(rows),
+ eligible_accounts=sum(row['account_rows'] for row in rows),
+ **{key: sum(row['coverage'] == key for row in rows)
+ for key in ('captured', 'pending', 'failed', 'incomplete')})
+
+ def export_csv(self):
+ stream = io.StringIO()
+ writer = csv.DictWriter(stream, fieldnames=['postal_code', 'account_rows', 'coverage', 'captured_at', 'started_at'])
+ writer.writeheader()
+ for row in self.entries():
+ writer.writerow({key: row[key] for key in writer.fieldnames})
+ return stream.getvalue()
+
+
+if __name__ == '__main__':
+ import argparse
+ from cache_manager_new import CacheManager
+ parser = argparse.ArgumentParser(description='Import a private Salesforce account export; never scrape or update Salesforce.')
+ parser.add_argument('csv_file')
+ parser.add_argument('--db', required=True)
+ args = parser.parse_args()
+ CacheManager(args.db)
+ with open(args.csv_file, encoding='utf-8-sig', newline='') as stream:
+ print(json.dumps(CohortQueue(args.db).import_accounts(stream), indent=2))
diff --git a/docs/2026-09-06-upgrade.md b/docs/2026-09-06-upgrade.md
new file mode 100644
index 0000000..640c2e1
--- /dev/null
+++ b/docs/2026-09-06-upgrade.md
@@ -0,0 +1,120 @@
+# PRIZM operational upgrade
+
+## Verified production findings (September 6, 2026)
+
+Railway production runs main 07816f0. The Salesforce org is Wilderness Committee (00DAm0000008bmtMAA), not a sandbox. August 31–September 6: 10 successful cached responses daily, 70 total, zero upstream attempts. Cache: 161 active successes, 2 expired records, newest cached_at July 3. Salesforce daily queueables completed without platform errors.
+
+The daily selector treats missing net worth as incomplete enrichment. At least 140 V0N 1T0 households (segment 66) and 70 V0G 1M0 households (segment 50) already have scores, income and property values but no net worth. The unavailable optional field repeatedly selects those postal codes. Partial DML results are also currently ignored.
+
+Live PRIZM reference rows still include Dwelling Type, but have no net-worth column. All 161 active cache rows have home_type. Net worth is supplemented from a partial legacy segment map; it is not a current postal-code-specific wealth measurement. Certain dwelling strings cannot map exactly to the Salesforce picklist.
+
+The root/dashboard, Basic authentication and cache CSV export already exist. Weekly email is scheduled in Salesforce, Monday 08:00 America/Los_Angeles, to dena@wildernesscommittee.org. August 31 queueable completed; this is not evidence of inbox delivery. Railway SMTP is not configured and is unnecessary for the existing Salesforce sender.
+
+## Final collection policy confirmed by Beaux
+
+Use the supplied 513 Account records as the complete major-donor cohort. Collect
+Canadian billing postal codes only. Daily target is ten **new** codes, not three.
+Never automatically select an already enriched code again, including after cache
+expiry. The supplied export contains 503 eligible accounts, 484 unique codes;
+106 codes already captured cover 112 accounts, leaving 378 codes across 391
+accounts. Nine non-Canadian addresses and one missing Canadian billing postal
+code remain address exceptions. Shipping addresses are not used.
+
+## Implemented locally
+
+- Railway owns a private cohort table and a permanent capture-history table seeded
+ from existing successes, including expired cache and retained lookup events.
+ Database triggers retain future captures even when cache rows are deleted.
+- Atomic claims select never-attempted, never-captured cohort codes, ordered by
+ account frequency then postal code. At most ten claims per Vancouver calendar
+ day across repeated requests, concurrent workers and service restarts.
+- Each API call claims one code. The existing Salesforce scheduler starts a chain
+ of up to ten queueable jobs, each with one callout and matching household updates.
+ This avoids accumulating ten upstream lookups inside one callout transaction.
+ The general Salesforce candidate selector is removed. An empty or unconfigured
+ cohort never falls back to general household selection.
+- Failed and interrupted attempts require review and are not automatically retried.
+ Failed captures do not count as enriched. Ten is an attempt target; upstream
+ restrictions/errors may reduce successful captures. This configuration is not
+ proof of the provider's current licensed allowance.
+- Protected dashboard and CSV export show cohort capture/pending/failure/incomplete
+ counts; the existing Salesforce weekly email includes cohort coverage. Import
+ and capture mutations require an API key, never dashboard credentials alone.
+- Capture progress does not prove Salesforce record updates. Existing household
+ mapping is retained; organizations are captured but not updated by that household
+ mapper. Missing historical net worth cannot drive collection or erase a known value.
+
+## Local validation
+
+Python regression and concurrency tests cover daily limits, duplicate prevention,
+restart behavior, expiry/deletion, account normalization/exclusions, imports and
+API authentication. Browser verified desktop/mobile cohort progress (106/484),
+weekly preview and CSV (484 data rows) against an isolated production-cache copy.
+26 Python tests and Ruff pass. Salesforce validation-only deployment
+`0AfN3000001kBr7KAE` succeeded with six tests and zero component/test errors.
+The first validation exposed recursive mock responses in the schedulable test;
+the scheduler test now returns an exhausted cohort. The final validation client
+timed out, but the server-side deployment report confirms success.
+
+## Deployment and operational limits
+
+No production change has been made. After approval, deploy Railway, import the
+private two-column country/postal-code payload through `/api/cohort/import`, verify
+106 captured / 378 pending, then deploy validated Salesforce classes. Preserve
+existing schedule and credentials. Import/history schema are additive; retain the
+original source and a database backup before rollout. No donor addresses or cohort
+membership should enter GitHub.
+
+Observe the first scheduled run for ten distinct new attempts and actual captures.
+If a Salesforce validation rule rejects an update, that queueable fails visibly;
+its API capture remains saved and will not be scraped again. The remainder of that
+chain stops. A response/network interruption can similarly leave a captured code
+without a Salesforce update. Reconcile such records from stored results without
+repeating upstream collection. Incomplete claims must be reviewed before any retry.
+
+The confirmed recent stall is 70 cache hits and zero upstream attempts August 31–
+September 6; latest cache timestamp is July 3. Full historical causality for every
+day since July 3 has not been independently reconstructed. Database SSH was
+unavailable with the configured local key; no new key was registered.
+
+
+## Late-2025 historical dataset merge (prepared September 6)
+
+Beaux supplied `/Users/beauxwalton/prizm-api/prizm_cache_v2 copy.db` and requires
+current Railway records to take priority over historical rows. Source database
+passes SQLite quick_check; SHA-256 remains unchanged after analysis/import.
+
+- Source: 551 rows (517 success, 33 error, 1 invalid). Four rows have invalid
+ Canadian postal-code formats and are excluded with a separate audit CSV.
+- 547 valid distinct codes: 159 overlap current data and remain entirely untouched.
+ 388 additional records (358 successes, 30 failures) are inserted.
+- Preview has 551 total records: all 163 existing rows verified field-for-field
+ unchanged, plus 388 imports. The preview is built from an API-export fixture;
+ it is not a byte-for-byte production backup and must not replace the live DB.
+- Cohort coverage increases from 106 to 362 of 484 codes; 122 remain uncaptured.
+ At ten successful new captures per day this is approximately 13 collection days.
+
+`historical_import.py` reads the source read-only and inserts only absent normalized
+codes within a transaction. The authenticated `/api/cache/import-history` endpoint
+applies the same rules to the private prepared payload during rollout. It checks
+live overlaps at execution time, so intervening current captures also win. Reruns
+are idempotent. Historical successes populate permanent capture history; failed
+historical lookups do not count as captured.
+
+Preserve original expiry, confirmation and demographic values. The source has no
+cached_at/scrape timestamp: imported cached_at stays NULL and unknown capture dates
+stay blank. Do not infer scrape time from expiry or set it to import time. Historical
+provenance is visible in details/CSV. All original expiry dates are already past;
+imports remain available in full-cache views/exports but do not become fresh cache
+hits. Net-worth amounts from the old scrape are retained as historical data rather
+than overwritten by the segment fallback.
+
+Private artifacts: historical-import-payload.json, historical-merge-summary.json,
+historical-rejected-postal-codes.csv and merged-history-preview.db under
+`private/prizm-major-donors/` outside the repo. Do not commit the donor membership
+list, database or import payload.
+
+Revised rollout: back up production; deploy updated Railway code; import historical
+payload; verify current rows unchanged and historical counts; import cohort; verify
+362 captured / 122 pending; then deploy already validated Salesforce classes.
+No upstream scrape, Salesforce mutation or production database import was performed.
diff --git a/historical_import.py b/historical_import.py
new file mode 100644
index 0000000..8e356d1
--- /dev/null
+++ b/historical_import.py
@@ -0,0 +1,94 @@
+"""Import legacy cache rows without replacing current records or refreshing their age."""
+import argparse
+import hashlib
+import json
+import sqlite3
+from pathlib import Path
+
+from cohort_queue import postal_code
+
+FIELDS = (
+ 'postal_code', 'segment_number', 'segment_name', 'segment_description', 'who_they_are',
+ 'average_household_income', 'education', 'urbanity', 'average_household_net_worth',
+ 'occupation', 'diversity', 'family_life', 'tenure', 'home_type', 'status', 'confirmed',
+ 'expires_at',
+)
+
+
+def read_legacy(path):
+ """Open the source read-only; no source migration or modification."""
+ with sqlite3.connect(Path(path).resolve().as_uri() + '?mode=ro', uri=True) as db:
+ db.row_factory = sqlite3.Row
+ if db.execute('PRAGMA quick_check').fetchone()[0] != 'ok':
+ raise ValueError('Source database integrity check failed')
+ rows = [dict(row) for row in db.execute('SELECT * FROM postal_code_cache ORDER BY postal_code')]
+ return rows
+
+
+def merge_history(db_path, rows, source='late-2025'):
+ if not isinstance(rows, list) or len(rows) > 1000:
+ raise ValueError('Expected at most 1000 historical rows')
+ prepared, skipped, seen = [], 0, set()
+ for row in rows:
+ if not isinstance(row, dict):
+ raise ValueError('Historical rows must be objects')
+ code = postal_code(row.get('postal_code'))
+ if code is None:
+ skipped += 1
+ continue
+ if code in seen:
+ raise ValueError('Duplicate normalized historical postal code')
+ seen.add(code)
+ if row.get('status') not in ('success', 'error', 'invalid') or not row.get('expires_at'):
+ raise ValueError('Historical status and original expiry are required')
+ values = {key: row.get(key) for key in FIELDS}
+ values['postal_code'] = code
+ prepared.append(values)
+ db = sqlite3.connect(db_path, timeout=30)
+ try:
+ with db:
+ db.execute('BEGIN IMMEDIATE')
+ db.execute('''CREATE TABLE IF NOT EXISTS historical_imports (
+ postal_code TEXT PRIMARY KEY, source TEXT NOT NULL,
+ original_expires_at TEXT, imported_at TEXT DEFAULT CURRENT_TIMESTAMP
+ )''')
+ # Match codes across spacing/case variations, preserving every current field.
+ current = {postal_code(row[0]) for row in db.execute('SELECT postal_code FROM postal_code_cache')}
+ inserted = preserved = successes = 0
+ for row in prepared:
+ code = row['postal_code']
+ if row['status'] == 'success':
+ # Empty date explicitly means capture time unknown; never substitute import time.
+ db.execute("INSERT OR IGNORE INTO capture_history VALUES (?, '')", (code,))
+ if code in current:
+ preserved += 1
+ continue
+ columns = FIELDS + ('cached_at', 'average_household_net_worth_amount', 'historical_source')
+ amount = row.get('average_household_net_worth')
+ if not isinstance(amount, (int, float)):
+ amount = None
+ db.execute('INSERT INTO postal_code_cache (' + ','.join(columns) + ') VALUES (' + ','.join('?' for _ in columns) + ')',
+ [row[key] for key in FIELDS] + [None, amount, source])
+ db.execute('INSERT OR IGNORE INTO historical_imports (postal_code,source,original_expires_at) VALUES (?,?,?)',
+ (code, source, row['expires_at']))
+ current.add(code)
+ inserted += 1
+ successes += row['status'] == 'success'
+ return {'source_rows': len(rows), 'inserted': inserted, 'inserted_successes': successes,
+ 'preserved_current': preserved, 'skipped_invalid_format': skipped}
+ finally:
+ db.close()
+
+
+if __name__ == '__main__':
+ from cache_manager_new import CacheManager
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument('--source', required=True)
+ parser.add_argument('--target', required=True)
+ args = parser.parse_args()
+ if Path(args.source).resolve() == Path(args.target).resolve():
+ parser.error('Source and target must differ')
+ rows = read_legacy(args.source)
+ CacheManager(args.target)
+ source_hash = hashlib.sha256(Path(args.source).read_bytes()).hexdigest()
+ print(json.dumps(merge_history(args.target, rows, 'late-2025 sha256:' + source_hash), indent=2))
diff --git a/salesforce/classes/WCPrizmDailyEnrichment.cls b/salesforce/classes/WCPrizmDailyEnrichment.cls
index 975c105..501edc2 100644
--- a/salesforce/classes/WCPrizmDailyEnrichment.cls
+++ b/salesforce/classes/WCPrizmDailyEnrichment.cls
@@ -1,6 +1,4 @@
public without sharing class WCPrizmDailyEnrichment implements Schedulable, Database.AllowsCallouts {
- private static final Integer MAX_POSTAL_CODES_PER_RUN = 10;
- private static final Integer AGGREGATE_SCAN_LIMIT = 500;
private static final Set