""" 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 `//.`, while # blob_extract writes `//__.`. # 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()