rmancinasandClaude Opus 4.8
594ee7cfca
DATGRAL.NOMBRE is blank on 266 legacy rows (140 utilities, 126 insurance), which surfaced in the UI as 257 customers literally named "(SIN NOMBRE)". The blank is real — those cells are empty in the Access files, not lost in extraction — but the rows mostly are not junk: 176 of the 257 carry a property, a policy, or transactions. The old PHP importer handled this by skipping blank-name rows outright (jorgecuadros-intra-webapp/src/tools/customerAdapter.php:47,81). That was worse than it looks: every other adapter resolved its customer FK through the customer_mapping table those skipped rows never entered, so their properties and policies were silently dropped (customerServiceAdapter.php:45) and their transactions were written against customer_id 0 (customerBalanceAdapter.php:52). So: recover the name instead of skipping. Names come from the secondary tables that still carry them, most trustworthy first — UTILSEG (the office's own hand-maintained name <-> id cross-reference spanning both lines), then the billing runs (IVA 2015, COBRO3) and the policy rows' NOMBRE ASEG (MULT, M EMPR, INCENDIO). A linked customer can also borrow the name its insurance record resolved to. Result: 213 of 257 recovered, 44 still genuinely nameless anywhere in the source. customers.nameSource records which table each recovered name came from, so a reconstructed name is never mistaken for one that was really on the record — the list tags it "nombre recuperado", the detail header names the source, and a still-unnamed customer renders muted italic instead of as a normal name. Also fixes run_all.py: transform_properties and transform_policies truncate service_documents/policy_documents, but blob_extract.py was not in the step list, so a full re-run left the uploaded MinIO objects with no rows pointing at them. Hit exactly that while reloading for this change. Verified end-to-end: full pipeline re-run against dev reproduces every prior count (1682 customers, 1519 properties, 2378 policies, 45861 transactions, 22354 bank rows, 70 documents) with zero orphans, and both apps build clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
76 lines
3.0 KiB
Python
76 lines
3.0 KiB
Python
"""
|
|
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
|
|
|
|
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", # customers + customer_legacy_refs (everything FKs to these)
|
|
"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
|
|
"transform_bank.py", # SCOTHIA bank register (no customer FK; independent)
|
|
"blob_extract.py", # document pointers; must follow properties + policies
|
|
]
|
|
|
|
|
|
def run(cmd: list[str]) -> None:
|
|
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)")
|
|
args = ap.parse_args()
|
|
|
|
if args.stage:
|
|
run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")])
|
|
|
|
for step in STEPS:
|
|
run([PY, str(HERE / step), "--env", args.env])
|
|
|
|
print(f"\n✓ migration complete for env={args.env}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|