Migration: prune customers with no business records

144 customers owned zero properties, zero policies and zero transactions —
the legacy DATGRAL row exists but nothing in either business line ever
attached to it. They padded the staff customer list with rows that can't be
acted on. 27 were also nameless (dead ID slots); the other 117 have real
names and sometimes contact details, and read as never-activated prospects
or lapsed clients rather than junk. Removing both sets is a deliberate call.

Implemented as a separate step rather than a filter inside
transform_customers.py: emptiness is only knowable after properties, policies
and transactions have loaded, and deciding it there would mean re-deriving
each downstream transform's source-matching logic against the staged Parquet.
Runs after transform_transactions.py in run_all.py.

Safe by construction — a customer with no rows in any of the three tables has
nothing pointing at it, so the delete cannot orphan anything; only its own
customer_legacy_refs go with it. The step asserts zero orphans afterwards.

Every pruned customer is written to output/pruned_customers.csv with its
legacy provenance before the delete, and --dry-run reports without touching
anything. Nothing is unrecoverable: the Access sources are untouched and a
pipeline run without this step brings them all back.

Verified: full run_all.py pass ends at 1538 customers (from 1682), with
1519 properties / 2378 policies / 45861 transactions all intact and zero
orphans. 17 nameless customers remain, all of which carry real records.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

migration/prune_empty_customers.py

@@ -0,0 +1,119 @@
+"""
+Migration step 5: drop customers that carry no business records at all.
+
+A customer is "empty" when it owns zero properties, zero policies and zero
+transactions — the legacy DATGRAL row exists, but nothing in either business
+line ever attached to it. These are dead ID slots and never-activated
+prospects from the Access era, and they pad the staff customer list with
+rows that can't be acted on.
+
+This runs LAST in the customer graph, not inside transform_customers.py,
+because emptiness is only knowable after properties, policies and
+transactions have loaded. Deciding it here also means the rule stays in one
+place instead of being re-derived from the staged Parquet by duplicating each
+downstream transform's source-matching logic.
+
+Safe by construction: a customer with zero rows in all three tables has
+nothing pointing at it, so the delete cannot orphan anything. Only its own
+`customer_legacy_refs` rows go with it.
+
+Nothing is silently destroyed — every deleted customer is written to
+`output/pruned_customers.csv` (with its legacy provenance) before the delete,
+and the whole step is reproducible from the Access sources by re-running the
+pipeline without it.
+
+Run:  ./.venv/bin/python prune_empty_customers.py --env dev [--dry-run]
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+from pathlib import Path
+
+from dbenv import connect
+
+AUDIT = Path(__file__).parent / "output" / "pruned_customers.csv"
+
+# The emptiness test. Kept as one string so the audit dump and the delete can
+# never disagree about what "empty" means.
+EMPTY_WHERE = """
+    NOT EXISTS (SELECT 1 FROM properties   x WHERE x.customerId = c.id)
+AND NOT EXISTS (SELECT 1 FROM policies     x WHERE x.customerId = c.id)
+AND NOT EXISTS (SELECT 1 FROM transactions x WHERE x.customerId = c.id)
+"""
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--env", default="dev", help="target environment (reads deploy/.env.<env>)")
+    ap.add_argument("--dry-run", action="store_true",
+                    help="report and write the audit CSV, but delete nothing")
+    args = ap.parse_args()
+
+    conn = connect(args.env)
+    cur = conn.cursor()
+    print(f"[prune] target env: {args.env}")
+
+    cur.execute("SELECT COUNT(*) FROM customers")
+    before = cur.fetchone()[0]
+
+    # Snapshot what is about to go, provenance included, for the audit trail.
+    cur.execute(f"""
+        SELECT c.id, c.name, c.nameSource, c.nameMissing, c.city, c.state,
+               c.email, c.phone, c.mobile, c.status,
+               GROUP_CONCAT(CONCAT(r.sourceSystem, ':', r.sourceTable, ':', r.legacyId)
+                            ORDER BY r.sourceSystem SEPARATOR ' | ')
+        FROM customers c
+        LEFT JOIN customer_legacy_refs r ON r.customerId = c.id
+        WHERE {EMPTY_WHERE}
+        GROUP BY c.id
+        ORDER BY c.nameMissing, c.name
+    """)
+    rows = cur.fetchall()
+
+    AUDIT.parent.mkdir(parents=True, exist_ok=True)
+    with AUDIT.open("w", newline="", encoding="utf-8") as fh:
+        w = csv.writer(fh)
+        w.writerow(["id", "name", "nameSource", "nameMissing", "city", "state",
+                    "email", "phone", "mobile", "status", "legacyRefs"])
+        w.writerows(rows)
+
+    named = sum(1 for r in rows if not r[3])
+    nameless = len(rows) - named
+
+    if args.dry_run:
+        print(f"  dry run — {len(rows)} would be pruned, nothing deleted")
+    else:
+        cur.execute(f"DELETE r FROM customer_legacy_refs r JOIN customers c ON c.id = r.customerId "
+                    f"WHERE {EMPTY_WHERE}")
+        refs_deleted = cur.rowcount
+        cur.execute(f"DELETE c FROM customers c WHERE {EMPTY_WHERE}")
+        deleted = cur.rowcount
+        conn.commit()
+        print(f"  deleted {deleted} customers, {refs_deleted} legacy refs")
+
+    cur.execute("SELECT COUNT(*) FROM customers")
+    after = cur.fetchone()[0]
+
+    print("=== Empty-customer prune complete ===")
+    print(f"  customers before : {before}")
+    print(f"  empty (no property, policy or transaction) : {len(rows)}")
+    print(f"     named          : {named}")
+    print(f"     without a name : {nameless}")
+    print(f"  customers after  : {after}")
+    print(f"  audit trail      : {AUDIT}")
+
+    # Whatever survived must still have every downstream FK intact.
+    for table in ("properties", "policies", "transactions"):
+        cur.execute(f"SELECT COUNT(*) FROM {table} t "
+                    f"LEFT JOIN customers c ON c.id = t.customerId WHERE c.id IS NULL")
+        orphans = cur.fetchone()[0]
+        assert orphans == 0, f"{table}: {orphans} orphaned rows after prune"
+    print("  validation: OK (0 orphans)")
+    conn.close()
+
+
+if __name__ == "__main__":
+    main()

migration/run_all.py

@@ -43,6 +43,7 @@ STEPS = [
     "transform_properties.py",    # properties + services + trust accounts
     "transform_policies.py",      # policies + installments/vehicles/drivers/benef/claims/adjusters
     "transform_transactions.py",  # shared ledger + type_transactions + exchange_rates
+    "prune_empty_customers.py",   # drop customers with no property/policy/transaction
     "transform_bank.py",          # SCOTHIA bank register (no customer FK; independent)
     "blob_extract.py",            # document pointers; must follow properties + policies
 ]