jorgecuadros-platform/migration/native_guard.py

"""
Refuse a full re-import that would delete platform-native data.

A full `run_all.py` pass truncates and rebuilds every table it owns from the
Access extract:

    transform_customers.py    customers, customer_legacy_refs
    transform_properties.py   properties, property_services, service_documents,
                              trust_accounts
    transform_policies.py     policies + installments, vehicles, drivers,
                              beneficiaries, claims, adjusters, policy_types,
                              insurance_providers
    transform_transactions.py transactions, type_transactions, exchange_rates
    transform_bank.py         bank tables
    blob_extract.py           service_documents, policy_documents

That was harmless while the platform was a read-only mirror of Access: every
row came from the extract, so wiping and rebuilding lost nothing. It stopped
being harmless when the platform started minting rows Access has never heard
of — portal NUMids from the allocator (apps/api/src/customers/numid.service.ts),
customers created in the staff UI, OCR-captured policies, app-booked ledger
rows, uploaded documents. None of those come back.

`--sync` already avoids all of it: it upserts legacy rows against the existing
refs and leaves everything else alone. So this guard does not try to teach the
full path to preserve anything — it stops the full path when there is something
to preserve, and points at the additive one.

    python native_guard.py --env prod          # report only, exit 3 if blocking
    python run_all.py --env prod --stage       # runs this first, refuses on 3
    python run_all.py --env prod --force-full  # ignore the guard (deletes them)
"""

from __future__ import annotations

import argparse
import re
from pathlib import Path

import pandas as pd

from dbenv import connect

STG = Path(__file__).parent / "output"

# Exit code the orchestrator looks for. Distinct from 1 so a connection failure
# or a bad query is not silently read as "native rows found".
BLOCKED = 3

# transform_customers.py mints these when a legacy row carries no id of its own.
# They are regenerated by every full pass, so they are legacy-owned, not native.
SYNTHETIC_REF_PREFIXES = ("rownum_", "insrow_")

# App-uploaded documents are stored as `<prefix>/<parent>/<uuid>.<ext>`, while
# blob_extract writes `<prefix>/<parent>/<stagedtable>_<row>_<col>.<ext>`.
# service_documents carries no provenance column, so the key shape is the only
# signal available — approximate, and reported as such. Matched in MySQL rather
# than in Python so the whole scan stays one round trip per table.
UUID_KEY_SQL = "/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\\."


def staged_legacy_ids() -> dict[str, set[str]] | None:
    """The ids Access will re-create, per source system.

    None when the staged Parquet is absent — which is not the same as "no legacy
    ids". Returning an empty set there would mark all 1,171 refs as native and
    block every run; returning None lets the caller say "cannot verify" instead.
    """
    sources = {"utilities": "stg_utilities", "insurance": "stg_seguros"}
    out: dict[str, set[str]] = {}
    for system, folder in sources.items():
        path = STG / folder / "datgral.parquet"
        if not path.exists():
            return None
        df = pd.read_parquet(path, columns=["num_id"])
        ids = set()
        for v in df["num_id"].astype("string"):
            if v is None or pd.isna(v):
                continue
            v = str(v).strip()
            if v.endswith(".0"):  # some numeric ids serialize as "521.0"
                v = v[:-2]
            if v and v != "0":
                ids.add(v)
        out[system] = ids
    return out


def native_refs(cur, staged: dict[str, set[str]] | None) -> tuple[int, list[str]]:
    """Legacy refs with no counterpart in the Access extract.

    This is the check that catches an allocated portal NUMid: the customer holds
    a perfectly ordinary-looking (utilities, DATGRAL, '1172') ref, so "customer
    has no refs" does not see it. Only comparing against staging does.
    """
    cur.execute(
        "SELECT sourceSystem, legacyId FROM customer_legacy_refs ORDER BY sourceSystem, legacyId"
    )
    rows = cur.fetchall()
    if staged is None:
        return 0, []

    found = []
    for system, legacy_id in rows:
        if legacy_id.startswith(SYNTHETIC_REF_PREFIXES):
            continue
        known = staged.get(system)
        # An unknown source system has no extract to compare against, so it
        # cannot be re-created either — treat it as native rather than ignoring.
        if known is None or legacy_id not in known:
            found.append(f"{system}/{legacy_id}")
    return len(found), found


def scan(conn) -> tuple[list[tuple[str, int, str]], bool]:
    """(label, count, detail) per source of native rows, plus whether staging
    was available to verify the refs."""
    cur = conn.cursor()
    staged = staged_legacy_ids()
    findings: list[tuple[str, int, str]] = []

    ref_count, ref_examples = native_refs(cur, staged)
    if ref_count:
        shown = ", ".join(ref_examples[:8])
        more = f" (+{ref_count - 8} more)" if ref_count > 8 else ""
        findings.append(("customer_legacy_refs", ref_count, f"{shown}{more}"))

    cur.execute(
        "SELECT COUNT(*) FROM customers c"
        " WHERE NOT EXISTS (SELECT 1 FROM customer_legacy_refs r WHERE r.customerId = c.id)"
    )
    n = cur.fetchone()[0]
    if n:
        findings.append(("customers", n, "created in the staff UI, no legacy ref"))

    # Every transform writes legacyId on what it loads, so a NULL is the app's.
    for table, detail in (
        ("transactions", "booked in the app (captura, OCR, manual)"),
        ("policies", "created in the app or captured by policy OCR"),
        ("properties", "created in the app"),
        ("vehicles", "created in the app"),
        ("bank_transactions", "booked in the chequera"),
    ):
        cur.execute(f"SELECT COUNT(*) FROM {table} WHERE legacyId IS NULL")
        n = cur.fetchone()[0]
        if n:
            findings.append((table, n, detail))

    # blob_extract always writes originalColumn; the app never does.
    cur.execute("SELECT COUNT(*) FROM policy_documents WHERE originalColumn IS NULL")
    n = cur.fetchone()[0]
    if n:
        findings.append(("policy_documents", n, "uploaded in the app"))

    cur.execute("SELECT COUNT(*) FROM service_documents WHERE storageKey REGEXP %s",
                (UUID_KEY_SQL,))
    n = cur.fetchone()[0]
    if n:
        findings.append(("service_documents", n, "uploaded in the app (key shape, approximate)"))

    return findings, staged is not None


def report(findings, staged_ok: bool, env: str) -> int:
    print(f"=== Verificación de datos nativos (env={env}) ===", flush=True)

    if not staged_ok:
        print(
            "  ! No hay Parquet en migration/output, así que no se pueden verificar\n"
            "    los refs contra el extracto de Access. Ejecute con --stage.",
            flush=True,
        )

    if not findings:
        print("  Sin filas nativas. Una reimportación completa no destruye nada.", flush=True)
        return 0 if staged_ok else BLOCKED

    total = sum(n for _, n, _ in findings)
    print(f"  {total} filas existen SÓLO en la plataforma y se perderían:", flush=True)
    for label, n, detail in findings:
        print(f"    {n:>7}  {label:<22} {detail}", flush=True)
    print(
        "\n  Una reimportación completa vacía estas tablas y las reconstruye desde\n"
        "  Access, que no conoce ninguna de estas filas.\n"
        "\n  Use la sincronización aditiva (run_all.py --sync), que respeta lo\n"
        "  capturado en la plataforma. Para reimportar de todos modos y BORRARLAS,\n"
        "  ejecute run_all.py --force-full.",
        flush=True,
    )
    return BLOCKED


def main() -> None:
    ap = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    ap.add_argument("--env", default="dev")
    args = ap.parse_args()

    conn = connect(args.env)
    try:
        findings, staged_ok = scan(conn)
    finally:
        conn.close()

    raise SystemExit(report(findings, staged_ok, args.env))


if __name__ == "__main__":
    main()