jorgecuadros-platform/migration/prune_empty_customers.py

"""
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")
    ap.add_argument("--sync", action="store_true",
                    help="only prune legacy-owned empties; never touch manually-added "
                         "customers (those with no customer_legacy_refs row)")
    args = ap.parse_args()

    conn = connect(args.env)
    cur = conn.cursor()
    print(f"[prune] target env: {args.env}{' (sync: legacy-owned only)' if args.sync else ''}")

    # In sync mode a manually-added customer (no legacy ref) with no records yet
    # is a legitimate new row, not Access-era dead weight — so restrict the prune
    # to customers that carry a legacy ref.
    where = EMPTY_WHERE + (
        "\nAND EXISTS (SELECT 1 FROM customer_legacy_refs lr WHERE lr.customerId = c.id)"
        if args.sync else "")

    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 {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:
        # Delete by the exact id set selected above (which already carries the
        # sync guard). Deleting via the id list avoids referencing the delete
        # target table inside its own WHERE (MySQL error 1093) and keeps refs +
        # customers on the same set regardless of delete order.
        ids = [r[0] for r in rows]
        refs_deleted = deleted = 0
        if ids:
            fmt = ",".join(["%s"] * len(ids))
            cur.execute(f"DELETE FROM customer_legacy_refs WHERE customerId IN ({fmt})", ids)
            refs_deleted = cur.rowcount
            cur.execute(f"DELETE FROM customers WHERE id IN ({fmt})", ids)
            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()