jorgecuadros-platform/migration/run_all.py

"""
Run the full data migration against one environment, in dependency order.

Every step is idempotent (truncate + rebuild), so this is safe to re-run. The
target DB is chosen with --env (reads deploy/.env.<env>); the same staged
Parquet feeds every environment.

Prerequisites (once per environment, NOT done here):
  1. DB stack deployed (deploy/jorgecuadros-db.stack.yml) and deploy/.env.<env> written.
  2. Prisma schema pushed to it:
       DATABASE_URL="<that env's url>" \
         npx prisma@5 db push --schema=packages/database/prisma/schema.prisma

Then:
  ./.venv/bin/python run_all.py --env dev            # data only (staging already present)
  ./.venv/bin/python run_all.py --env prod --stage   # re-extract from Access first, then load

A full pass truncates and rebuilds every table it owns from the Access extract,
so anything the platform minted itself — allocated portal NUMids, customers
created in the staff UI, OCR-captured policies, app-booked ledger rows, uploaded
documents — is destroyed. native_guard.py runs first and refuses when the target
database holds any of it; --force-full overrides and deletes them.

--sync swaps the truncate+rebuild steps for the additive upsert ones. It reads
the same staged Parquet, so it needs --stage too unless a previous run left
migration/output populated on this machine — which is never true in a
container, where that directory is part of the image and dies with it.

Reproducing dev -> prod is exactly `--env prod` (plus --stage if the staged
Parquet isn't present on the machine running it).

Note: the blob_extract step re-reads the original Access files directly (the
blobs are not in the staged Parquet), so the machine running this needs
SOURCE_ROOT + mdbtools + MinIO credentials even without --stage.
"""

from __future__ import annotations

import argparse
import subprocess
import sys
from pathlib import Path

HERE = Path(__file__).parent
PY = sys.executable  # the venv python running this orchestrator

# Dependency order. Every step truncates what it owns, so anything downstream
# of a truncated table has to be rebuilt in the same pass — blob_extract is in
# this list because transform_properties and transform_policies truncate
# service_documents / policy_documents, which would otherwise leave the
# uploaded MinIO objects with no rows pointing at them.
STEPS = [
    "transform_customers.py",
    "transform_properties.py",
    # Statement-OCR match fields. transform_properties.py now produces these
    # directly, so on a full rebuild this is a no-op that re-asserts they are
    # there; on a database predating the OCR module it is what fills them in.
    # Must follow transform_properties.py, which truncates both tables it
    # touches.
    "backfill_statement_match_fields.py",
    "transform_policies.py",
    # Premium breakdown (recargo, per-payment figures, forma de pago).
    # transform_policies.py now writes these directly, so on a full rebuild
    # this is a no-op that re-asserts them; on a database migrated before the
    # breakdown existed it is what recovers them out of coveragesJson.
    # Must follow transform_policies.py, which truncates the installments.
    "backfill_policy_premium_breakdown.py",
    "transform_transactions.py",
    "prune_empty_customers.py",
    # Seeds the Scotiabank chequera that every SCOTHIA movement is booked into;
    # transform_bank.py fails fast without it.
    "backfill_bank_accounts.py",
    "transform_bank.py",
    "blob_extract.py",
]

SYNC_STEPS = [
    "transform_customers.py",
    "transform_properties.py",
    # Statement-OCR match fields. transform_properties.py now produces these
    # directly, so on a full rebuild this is a no-op that re-asserts they are
    # there; on a database predating the OCR module it is what fills them in.
    # Must follow transform_properties.py, which truncates both tables it
    # touches.
    "backfill_statement_match_fields.py",
    "transform_policies.py",
    # Premium breakdown (recargo, per-payment figures, forma de pago).
    # transform_policies.py now writes these directly, so on a full rebuild
    # this is a no-op that re-asserts them; on a database migrated before the
    # breakdown existed it is what recovers them out of coveragesJson.
    # Must follow transform_policies.py, which truncates the installments.
    "backfill_policy_premium_breakdown.py",
    "transform_transactions.py",
    # Manual-safe prune: drops legacy-owned empties that the customer upsert
    # re-creates from Parquet, but leaves manually-added customers alone.
    "prune_empty_customers.py",
    # Seeds the Scotiabank chequera that every SCOTHIA movement is booked into;
    # transform_bank.py fails fast without it.
    "backfill_bank_accounts.py",
    "transform_bank.py",
]


# native_guard.py exits with this when the target database holds rows that only
# exist in the platform. Kept in step with the constant there.
GUARD_BLOCKED = 3


def guard(env: str, force: bool) -> None:
    """Stop a full pass that would delete platform-native rows.

    Only the full path needs this: --sync upserts legacy rows against the
    existing refs and leaves everything else alone, so it cannot lose anything.
    Run after staging, because the guard verifies legacy refs against the staged
    Parquet and cannot tell an allocated NUMid from an Access one without it.
    """
    cmd = [PY, str(HERE / "native_guard.py"), "--env", env]
    print("+ " + " ".join(cmd), flush=True)
    r = subprocess.run(cmd)
    if r.returncode == GUARD_BLOCKED and not force:
        sys.exit(r.returncode)
    if r.returncode == GUARD_BLOCKED and force:
        print(
            "\n! --force-full: continuando y BORRANDO las filas nativas listadas.\n",
            flush=True,
        )
    elif r.returncode:
        sys.exit(r.returncode)


def run(cmd: list[str], step: int | None = None, total: int | None = None) -> None:
    # The "[paso i/N] name" marker is a contract with the Operaciones screen,
    # which parses the last one to show progress. Emitting it here rather than
    # letting the UI count STEPS itself keeps the two from drifting when a step
    # is added — the number of steps is only ever stated in this file.
    if step is not None and total is not None:
        print(f"[paso {step}/{total}] {Path(cmd[1]).name}", flush=True)
    print("+ " + " ".join(cmd), flush=True)
    r = subprocess.run(cmd)
    if r.returncode:
        sys.exit(r.returncode)


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("--stage", action="store_true",
                    help="re-run the raw staging load first (needs the Access files + mdbtools)")
    ap.add_argument("--sync", action="store_true",
                    help="upsert legacy rows and archive removed legacy rows; preserve manual rows")
    ap.add_argument("--force-full", action="store_true",
                    help="run the full truncate+rebuild even when it deletes platform-native rows")
    args = ap.parse_args()

    steps = SYNC_STEPS if args.sync else STEPS
    # Staging counts as a step when it runs: it is the slowest part of the pass
    # (mdbtools re-reads every Access file), so leaving it outside the numbering
    # would park the Operaciones progress bar at "nothing yet" for minutes.
    total = len(steps) + (1 if args.stage else 0)
    offset = 1 if args.stage else 0

    if args.stage:
        run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")],
            step=1, total=total)

    # Deliberately not counted as a step: it is a precondition, it takes a
    # second, and the Operaciones progress bar parses those numbers.
    if not args.sync:
        guard(args.env, args.force_full)

    for i, step in enumerate(steps, start=1 + offset):
        cmd = [PY, str(HERE / step), "--env", args.env]
        if args.sync:
            cmd.append("--sync")
        run(cmd, step=i, total=total)

    print(f"\n✓ migration complete for env={args.env}")


if __name__ == "__main__":
    main()