From 1b79b43a54fdcc3d4d1d432728c55cff43376593 Mon Sep 17 00:00:00 2001 From: Ricardo Mancinas Date: Fri, 24 Jul 2026 13:37:22 -0700 Subject: [PATCH] fix(migration): make Phase B additive sync actually work + verify end-to-end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The --sync path had never been run and was broken in several ways. Fixed and verified against the dev DB (two consecutive syncs, both exit 0, 32/32 assertions: stable PKs, manual-row preservation, changed-row updates, legacy-delete, no child duplication, zero FK orphans; idempotent). - policies/properties: reuse each legacy row's existing id (by provenance) BEFORE building child rows, so children no longer point at a discarded fresh uuid; rebuild legacy-owned children via scoped delete + reinsert. - customers: replace zip(customers, refs) (mispaired almost every row) with a ref-grouped id remap; names now restore and no spurious customers appear. - drop the invalid Vehicle @@unique(legacySourceTable, legacyId) — one legacy policy row carries up to 3 vehicles sharing a legacyId; handle via delete+reinsert. - upsert lookup tables (policy_types, insurance_providers, type_transactions, adjusters) by natural name and remap child FKs instead of inserting fresh uuids that nothing points at. - transactions: drop updatedAt=NOW() (no such column); guard report formatting on NULL legacySourceTable (manual rows). Same report guard in bank. - add manual-safe prune (prune_empty_customers.py --sync, in SYNC_STEPS): prune only legacy-owned empties, never manually-added customers. web: customer-detail mini tx list now strikes voided rows with an "(anulado)" tag (was the last void-UI rendering gap; /estado-cuenta already handled it). docs: RESUME.md updated — Phase B sync marked verified end-to-end, void-UI browser pass recorded. Co-Authored-By: Claude Opus 4.8 --- RESUME.md | 60 ++++++++++++------ apps/web/src/app/clientes/[id]/page.tsx | 12 +++- apps/web/src/lib/types.ts | 2 + migration/prune_empty_customers.py | 31 ++++++--- migration/run_all.py | 3 + migration/transform_bank.py | 2 +- migration/transform_customers.py | 45 ++++++++++---- migration/transform_policies.py | 83 ++++++++++++++++++------- migration/transform_properties.py | 25 +++++--- migration/transform_transactions.py | 20 +++++- packages/database/prisma/schema.prisma | 4 +- 11 files changed, 211 insertions(+), 76 deletions(-) diff --git a/RESUME.md b/RESUME.md index 13a3f39..9b3cb4f 100644 --- a/RESUME.md +++ b/RESUME.md @@ -127,8 +127,14 @@ To rerun (from `migration/`, venv at `migration/.venv`): ```bash ./.venv/bin/python load_staging.py --output-dir ./output # re-extract from Access (needs mdbtools + the source files) ./.venv/bin/python run_all.py --env dev # full transform+load; add --stage to re-extract first +./.venv/bin/python run_all.py --env dev --sync # additive sync: upsert legacy by provenance, keep manual rows, prune legacy empties ``` +`--sync` mode (Phase B) upserts legacy-owned rows by their provenance keys and preserves +manual rows (`legacyId IS NULL`); every transform reuses each row's existing PK, rebuilds +legacy-owned children by scoped delete + reinsert, and drops legacy rows gone from source. +Verified end-to-end against dev 2026-07-24 — see §6 item 6. + ## 5. Infrastructure & sync architecture (designed, not yet built) - **Internal server** — on-prem, private IP `192.168.1.xx`, no inbound internet exposure. Runs the platform + canonical MySQL (source of truth). @@ -162,17 +168,30 @@ the reconciliation pass (done, then corrected) are all closed. See §3 and §8. 5. **`TRASPASOS PAYPAL` is a clearing account, not a customer** — carries -7.03M MXN over 309 movements and therefore tops the adeudo worklist. Deliberately not special-cased in code; needs a business decision on how to model it. -6. **DB Operations — Phase B (additive sync) — IMPLEMENTED, verification pending.** Phase A provides - the admin-only `/operaciones` page + `ops` API module (ability `db:manage`, ADMIN), ingest - folder, backup, restore, and destructive re-import. Phase B now enables `SYNC`: `OpsService` - creates a safety backup and runs `run_all.py --sync`; transforms upsert legacy-owned rows by - provenance keys while preserving existing PKs and rows whose `legacyId IS NULL` (manual). - Prisma now enforces provenance uniqueness for properties, policies, transactions, vehicles, - and bank transactions. Sync intentionally skips prune/blob steps so manual customers and - document pointers are not removed. Python compilation plus API/web production builds pass; - still required before production use: push updated Prisma schema and run an end-to-end sync - against a disposable/dev DB proving stable PKs, manual-row preservation, changed-row updates, - and legacy-delete handling. +6. **DB Operations — Phase B (additive sync) — VERIFIED END-TO-END against dev DB 2026-07-24.** + Phase A provides the admin-only `/operaciones` page + `ops` API module (ability `db:manage`, + ADMIN), ingest folder, backup, restore, and destructive re-import. Phase B enables `SYNC`: + `OpsService` creates a safety backup and runs `run_all.py --sync`; transforms upsert + legacy-owned rows by provenance keys while preserving existing PKs and rows whose + `legacyId IS NULL` (manual). Prisma enforces provenance uniqueness for properties, policies, + transactions, and bank transactions (the vehicle unique was **removed** — one legacy policy + row carries up to 3 vehicles that share a `legacyId`, so provenance is not unique per + vehicle; vehicles are rebuilt by scoped delete + reinsert). Sync skips blob extraction, and + runs a **manual-safe prune** (`prune_empty_customers.py --sync` — only prunes empties that + carry a legacy ref, never manually-added customers) because the customer upsert otherwise + re-creates every previously-pruned empty from Parquet. + + **The as-written sync was broken and had never been run; a batch of bugs were fixed on + 2026-07-24 before it passed** (fresh-uuid child FKs in policies/properties, unconditional + child inserts, a `zip(customers, refs)` mispairing in transform_customers, invalid vehicle + unique, lookup tables built with fresh uuids but never upserted, a `updatedAt=NOW()` on a + table with no such column, and report crashes on NULL `legacySourceTable` for manual rows). + Verified with `migration/` `verify_sync.py`-style harness: two consecutive `run_all.py --sync` + runs both exit 0 and pass 32/32 assertions (stable PKs, manual-row preservation, changed-row + updates, legacy-delete, no child duplication, zero FK orphans), idempotent (customers stable + at 1537). Schema pushed to dev, Prisma client regenerated, API build clean. Migration/web + changes uncommitted as of this update. Still open before production: run the same sync from + the `/operaciones` UI (OpsService path) and against a prod-shaped DB. ## 7. Environment notes (current macOS machine) @@ -385,15 +404,20 @@ for what's actually next. --- -- **Sync implementation — DONE, validation pending.** `run_all.py --sync` performs the - non-destructive legacy upsert path for customers, properties, policies, transactions, and - bank rows. It preserves manual rows and stable legacy-owned primary keys; the admin SYNC job - automatically creates a pre-sync backup. Next validation: apply schema changes, then exercise - sync against a disposable DB with added, changed, removed, and manually-created rows. +- **Sync implementation — DONE + VALIDATED end-to-end against dev 2026-07-24.** `run_all.py + --sync` performs the non-destructive legacy upsert path for customers, properties, policies, + transactions, and bank rows, plus a manual-safe empty-customer prune. It preserves manual rows + and stable legacy-owned primary keys; the admin SYNC job auto-creates a pre-sync backup. The + as-written code was broken and had never been run — a batch of bugs was fixed before it passed + (see §6 item 6). Two consecutive syncs both exit 0 and pass 32/32 assertions (added, changed, + removed, and manually-created rows), idempotent. Remaining: exercise the same path from the + `/operaciones` admin UI and against a prod-shaped DB. - **Plan step 9: portal sync worker** remains separate and blocked on VPS provisioning. This Phase B feature synchronizes Access source files into the internal platform; it does not yet poll `utility_dbo` inbox tables or replicate portal-facing data to a VPS. - **Small / open:** (a) `TRASPASOS PAYPAL` clearing account still tops the adeudo worklist (§6.4d) — a business modelling call, not code. (b) Credential rotation on the old repo's - exposed MySQL password. (c) The `/estado-cuenta` browser visual pass — `/banco` was verified - in-browser this session; `/estado-cuenta` still worth a look. + exposed MySQL password. (c) ~~The `/estado-cuenta` browser visual pass.~~ **DONE 2026-07-24** — + verified vs dev: Anular buttons admin-gated, voided rows struck + excluded from totals, + clicking Anular voids end-to-end (note: it uses a blocking `window.confirm`). Customer-detail + mini tx list now also strikes voided rows ("(anulado)" tag) — was the last void-UI gap. diff --git a/apps/web/src/app/clientes/[id]/page.tsx b/apps/web/src/app/clientes/[id]/page.tsx index 3a21d60..d9c4900 100644 --- a/apps/web/src/app/clientes/[id]/page.tsx +++ b/apps/web/src/app/clientes/[id]/page.tsx @@ -781,9 +781,10 @@ function TxRow({ t }: { t: Transaction }) { const tipo = t.type?.nameEs || t.type?.nameEn || "—"; const concept = t.message || t.period || "—"; + const voided = !!t.voidedAt; return ( - + {formatDate(t.transactionDate)} @@ -793,7 +794,14 @@ function TxRow({ t }: { t: Transaction }) { {tipo} {t.reference || "—"} - {concept} + + {concept} + {voided && ( + + (anulado) + + )} + {formatMoney(t.amount, t.currency)} diff --git a/apps/web/src/lib/types.ts b/apps/web/src/lib/types.ts index 3440ae5..a111143 100644 --- a/apps/web/src/lib/types.ts +++ b/apps/web/src/lib/types.ts @@ -667,6 +667,8 @@ export interface Transaction { period?: string | null; message: string | null; checkNumber: string | null; + /** App-voided (`voidedAt` set). UI strikes; totals exclude. */ + voidedAt?: string | null; type: TransactionType | null; } diff --git a/migration/prune_empty_customers.py b/migration/prune_empty_customers.py index ec633d9..06219a2 100644 --- a/migration/prune_empty_customers.py +++ b/migration/prune_empty_customers.py @@ -50,11 +50,21 @@ def main() -> None: ap.add_argument("--env", default="dev", help="target environment (reads deploy/.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}") + 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] @@ -67,7 +77,7 @@ def main() -> None: ORDER BY r.sourceSystem SEPARATOR ' | ') FROM customers c LEFT JOIN customer_legacy_refs r ON r.customerId = c.id - WHERE {EMPTY_WHERE} + WHERE {where} GROUP BY c.id ORDER BY c.nameMissing, c.name """) @@ -86,11 +96,18 @@ def main() -> None: if args.dry_run: print(f" dry run — {len(rows)} would be pruned, nothing deleted") else: - cur.execute(f"DELETE r FROM customer_legacy_refs r JOIN customers c ON c.id = r.customerId " - f"WHERE {EMPTY_WHERE}") - refs_deleted = cur.rowcount - cur.execute(f"DELETE c FROM customers c WHERE {EMPTY_WHERE}") - deleted = cur.rowcount + # 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") diff --git a/migration/run_all.py b/migration/run_all.py index 8028b1d..9ef4084 100644 --- a/migration/run_all.py +++ b/migration/run_all.py @@ -53,6 +53,9 @@ SYNC_STEPS = [ "transform_properties.py", "transform_policies.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", "transform_bank.py", ] diff --git a/migration/transform_bank.py b/migration/transform_bank.py index bfdf957..bfc9bdd 100644 --- a/migration/transform_bank.py +++ b/migration/transform_bank.py @@ -134,7 +134,7 @@ def main(): print(f" skipped (unparseable date): {skip_date}") print(f" -> bank_transactions : {count('bank_transactions')}") for src, n, tot in by_src: - print(f" {src:10} {n:6} sum {tot}") + print(f" {(src or '(manual)'):10} {n:6} sum {tot}") print(f" net balance movement : {net}") print(f" -> business_line_categories: {count('business_line_categories')}") print(" validation: OK") diff --git a/migration/transform_customers.py b/migration/transform_customers.py index 84a4506..3fdb74b 100644 --- a/migration/transform_customers.py +++ b/migration/transform_customers.py @@ -293,18 +293,33 @@ def main() -> None: refs.append((str(uuid.uuid4()), cust_id, "insurance", "DATGRAL", ins_id)) placeholders = ",".join(["%s"] * len(_CUST_COLS)) + remap: dict[str, str] = {} # in-memory customer id -> stable (DB) id if sync_mode: - existing = {} - cur.execute("SELECT id,sourceSystem,sourceTable,legacyId,customerId FROM customer_legacy_refs") - for rid, system, table, legacy, customer_id in cur.fetchall(): - existing[(system, table, legacy)] = (rid, customer_id) - for rec, ref in zip(customers, refs): - key = (ref[2], ref[3], ref[4]) - customer_id = existing.get(key, (None, rec["id"]))[1] - rec["id"] = customer_id + # Resolve each in-memory customer to a stable id: if ANY of its legacy + # refs already exists in the DB, reuse that customer's id (keeps PKs + # stable and preserves manual edits). `customers` and `refs` are + # different-length, differently-ordered lists — merged identities add a + # ref without a customer — so refs are grouped by their owning customer, + # never positionally zipped (the old zip mispaired almost every row). + cur.execute("SELECT sourceSystem,sourceTable,legacyId,customerId FROM customer_legacy_refs") + ref_existing = {(sy, tb, lg): cid for sy, tb, lg, cid in cur.fetchall()} + refs_by_cust: dict[str, list] = {} + for ref in refs: # ref = (refId, custInMemId, system, table, legacyId) + refs_by_cust.setdefault(ref[1], []).append(ref) + for rec in customers: + stable = None + for ref in refs_by_cust.get(rec["id"], []): + cid = ref_existing.get((ref[2], ref[3], ref[4])) + if cid: + stable = cid + break + remap[rec["id"]] = stable or rec["id"] + for rec in customers: + rec["id"] = remap[rec["id"]] cur.execute(f"INSERT INTO customers ({','.join(f'`{c}`' for c in _CUST_COLS)}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE name=VALUES(name),nameSource=VALUES(nameSource),nameMissing=VALUES(nameMissing),addressLine1=VALUES(addressLine1),addressLine2=VALUES(addressLine2),city=VALUES(city),state=VALUES(state),zipCode=VALUES(zipCode),country=VALUES(country),phone=VALUES(phone),mobile=VALUES(mobile),fax=VALUES(fax),email=VALUES(email),notes=VALUES(notes),identificationType=VALUES(identificationType),identificationNumber=VALUES(identificationNumber),identificationExpiration=VALUES(identificationExpiration),customerSince=VALUES(customerSince),status=VALUES(status),feeAmount=VALUES(feeAmount),updatedAt=VALUES(updatedAt)", tuple(rec[c] for c in _CUST_COLS)) - ref = (existing.get(key, (ref[0], customer_id))[0], customer_id, *ref[2:]) - cur.execute("INSERT INTO customer_legacy_refs (id,customerId,sourceSystem,sourceTable,legacyId) VALUES (%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId)", ref) + for ref in refs: + cur.execute("INSERT INTO customer_legacy_refs (id,customerId,sourceSystem,sourceTable,legacyId) VALUES (%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId)", + (ref[0], remap[ref[1]], ref[2], ref[3], ref[4])) else: cur.executemany( f"INSERT INTO customers ({','.join(f'`{c}`' for c in _CUST_COLS)}) VALUES ({placeholders})", @@ -317,7 +332,10 @@ def main() -> None: # Enrich linked customers with insurance-only ID-doc fields, and fill any # contact fields the utilities master left empty (COALESCE keeps master's). + # In sync mode the enrich targets carry in-memory ids, so map them to the + # stable DB ids resolved above (identity map in full mode). for cust_id, ir in enrich: + cust_id = remap.get(cust_id, cust_id) cur.execute( "UPDATE customers SET " "identificationType = COALESCE(identificationType, %s), " @@ -366,8 +384,11 @@ def main() -> None: print(f" from {src:16} : {n}") print(f" of which via the linked insurance record : {from_ins_side}") print(f" still {NO_NAME} : {still_unnamed}") - assert n_cust == len(util) + new_ins, "customer count mismatch" - assert n_refs == len(util) + len(ins), "legacy ref count mismatch" + if not sync_mode: + # Full-load invariants only: sync upserts into an already-loaded (and + # pruned) table, so these exact counts don't hold. + assert n_cust == len(util) + new_ins, "customer count mismatch" + assert n_refs == len(util) + len(ins), "legacy ref count mismatch" print(" validation: OK") conn.close() diff --git a/migration/transform_policies.py b/migration/transform_policies.py index 7a98cde..0df4a65 100644 --- a/migration/transform_policies.py +++ b/migration/transform_policies.py @@ -37,7 +37,7 @@ from pathlib import Path import pandas as pd from dbenv import connect, env_arg -from sync import parse_mode +from sync import parse_mode, existing_ids, delete_missing STG = Path(__file__).parent / "output" / "stg_seguros" LEGACY_DB = "SEGUROS 16_be" @@ -180,6 +180,14 @@ def main(): c.execute("SELECT legacyId, customerId FROM customer_legacy_refs WHERE sourceSystem='insurance'") cust = {r[0]: r[1] for r in c.fetchall()} + # Sync mode reuses each legacy policy's existing id (keyed by provenance) so + # its PK is stable and every child row built below points at the right + # parent. New legacy policies fall through to a fresh uuid. + existing_pol = existing_ids( + c, "policies", ("legacySourceDb", "legacySourceTable", "legacyId"), + "WHERE legacyId IS NOT NULL") if sync_mode else {} + pol_keys: set = set() + policies, insts, vehicles, drivers = [], [], [], [] polno_to_id = {} # policy number -> a policyId (for BENEF/DATOS linking) providers, ptypes = set(), set() @@ -198,7 +206,10 @@ def main(): if not cid: skipped += 1 continue - pid = str(uuid.uuid4()) + legacy_pid = str(int(row["_row_num"])) + pkey = (LEGACY_DB, table, legacy_pid) + pol_keys.add(pkey) + pid = existing_pol.get(pkey) or str(uuid.uuid4()) comp = s(row.get(F.get("comp", ""), None)) if F.get("comp") else None if comp: providers.add(comp) @@ -316,20 +327,60 @@ def main(): dt(r["fecha_cheque"]), s(r["num_cheque"]), 1 if truthy(r["concluido"]) else 0, s(r["resolucion"]))) + pol_cols = ("id,policyNumber,customerId,policyTypeId,insuranceProviderId,agentName,policyDate," + "policyFrom,policyTo,netPremium,policyFee,commission,total,currency,observations," + "coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb," + "legacySourceTable,legacyId,updatedAt") + ph = ",".join(["%s"] * 23) + pol_upsert = ( + f"INSERT INTO policies ({pol_cols}) VALUES ({ph}) ON DUPLICATE KEY UPDATE " + "customerId=VALUES(customerId),policyNumber=VALUES(policyNumber),policyTypeId=VALUES(policyTypeId)," + "insuranceProviderId=VALUES(insuranceProviderId),agentName=VALUES(agentName),policyDate=VALUES(policyDate)," + "policyFrom=VALUES(policyFrom),policyTo=VALUES(policyTo),netPremium=VALUES(netPremium),policyFee=VALUES(policyFee)," + "commission=VALUES(commission),total=VALUES(total),currency=VALUES(currency),observations=VALUES(observations)," + "coveragesJson=VALUES(coveragesJson),liquidated=VALUES(liquidated),liquidationNumber=VALUES(liquidationNumber)," + "liquidationDate=VALUES(liquidationDate),updatedAt=VALUES(updatedAt),archivedAt=NULL") + if sync_mode: - c.execute("SELECT id,name FROM policy_types") + # policy_types / providers: upsert by their name unique, keep ids stable. + c.execute("SELECT name,id FROM policy_types") ptype_ids = dict(c.fetchall()) for n in ptypes: ptype_ids.setdefault(n, str(uuid.uuid4())) c.executemany("INSERT INTO policy_types (id,name) VALUES (%s,%s) ON DUPLICATE KEY UPDATE name=VALUES(name)", [(i, n) for n, i in ptype_ids.items()]) - c.execute("SELECT id,name FROM insurance_providers") + c.execute("SELECT name,id FROM insurance_providers") prov_ids = dict(c.fetchall()) for n in providers: prov_ids.setdefault(n, str(uuid.uuid4())) c.executemany("INSERT INTO insurance_providers (id,name) VALUES (%s,%s) ON DUPLICATE KEY UPDATE name=VALUES(name)", [(i, n) for n, i in prov_ids.items()]) - for p in policies: - p = list(p); p[3] = ptype_ids[p[3]]; p[4] = prov_ids.get(p[4]) - c.execute(f"INSERT INTO policies ({pol_cols}) VALUES ({','.join(['%s'] * 23)}) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),policyNumber=VALUES(policyNumber),policyTypeId=VALUES(policyTypeId),insuranceProviderId=VALUES(insuranceProviderId),agentName=VALUES(agentName),policyDate=VALUES(policyDate),policyFrom=VALUES(policyFrom),policyTo=VALUES(policyTo),netPremium=VALUES(netPremium),policyFee=VALUES(policyFee),commission=VALUES(commission),total=VALUES(total),currency=VALUES(currency),observations=VALUES(observations),coveragesJson=VALUES(coveragesJson),liquidated=VALUES(liquidated),liquidationNumber=VALUES(liquidationNumber),liquidationDate=VALUES(liquidationDate),updatedAt=VALUES(updatedAt),archivedAt=NULL", tuple(p)) + + # Adjusters carry no provenance key, so they can't be upserted by one. + # Resolve claims against the adjusters already in the DB (manual + prior + # loads), inserting only names not present yet — keeps manual adjusters + # and every claim's adjusterId FK valid. + c.execute("SELECT id,name FROM adjusters") + db_adj = {(nm or "").upper(): i for i, nm in c.fetchall()} + adj_id_name = {aid: (nm or "").upper() for aid, comp, city, nm, tel, bp in adj_rows} + new_adj = [] + for aid, comp, city, nm, tel, bp in adj_rows: + if nm and nm.upper() not in db_adj: + db_adj[nm.upper()] = aid + new_adj.append((aid, comp, city, nm, tel, bp)) + if new_adj: + c.executemany("INSERT INTO adjusters (id,company,city,name,phone,beeper) VALUES (%s,%s,%s,%s,%s,%s)", new_adj) + claims = [(cl[0], cl[1], *cl[2:6], db_adj.get(adj_id_name.get(cl[6])) if cl[6] else None, *cl[7:]) for cl in claims] + + # Rebuild every child of a legacy-owned policy before re-inserting the + # children below (manual rows survive: vehicles by their own null + # provenance, the rest by their parent policy's null provenance). + c.execute("DELETE FROM vehicles WHERE legacyId IS NOT NULL") + for tbl in ("policy_payment_installments", "insured_drivers", "policy_beneficiaries", "claims"): + c.execute(f"DELETE ch FROM {tbl} ch JOIN policies p ON p.id=ch.policyId WHERE p.legacyId IS NOT NULL") + + pol_rows = [tuple([p[0], p[1], p[2], ptype_ids.get(p[3]), prov_ids.get(p[4]), *p[5:]]) for p in policies] + c.executemany(pol_upsert, pol_rows) + # Drop legacy policies that vanished from source (children already gone). + delete_missing(c, "policies", ("legacySourceDb", "legacySourceTable", "legacyId"), pol_keys, "WHERE legacyId IS NOT NULL") else: c.execute("SET FOREIGN_KEY_CHECKS=0") for t in ("policy_payment_installments", "vehicles", "insured_drivers", @@ -342,22 +393,8 @@ def main(): prov_ids = {n: str(uuid.uuid4()) for n in providers} c.executemany("INSERT INTO insurance_providers (id,name) VALUES (%s,%s)", [(i, n) for n, i in prov_ids.items()]) c.executemany("INSERT INTO adjusters (id,company,city,name,phone,beeper) VALUES (%s,%s,%s,%s,%s,%s)", adj_rows) - - pol_cols = ("id,policyNumber,customerId,policyTypeId,insuranceProviderId,agentName,policyDate," - "policyFrom,policyTo,netPremium,policyFee,commission,total,currency,observations," - "coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb," - "legacySourceTable,legacyId,updatedAt") - if not sync_mode: - fixed = [] - for p in policies: - p = list(p) - p[3] = ptype_ids.get(p[3]) - p[4] = prov_ids.get(p[4]) - fixed.append(tuple(p)) - ph = ",".join(["%s"] * 23) - c.executemany(f"INSERT INTO policies ({pol_cols}) VALUES ({ph})", fixed) - else: - c.executemany("INSERT INTO policies (id,policyNumber,customerId,policyTypeId,insuranceProviderId,agentName,policyDate,policyFrom,policyTo,netPremium,policyFee,commission,total,currency,observations,coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb,legacySourceTable,legacyId,updatedAt) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),policyNumber=VALUES(policyNumber),policyTypeId=VALUES(policyTypeId),insuranceProviderId=VALUES(insuranceProviderId),agentName=VALUES(agentName),policyDate=VALUES(policyDate),policyFrom=VALUES(policyFrom),policyTo=VALUES(policyTo),netPremium=VALUES(netPremium),policyFee=VALUES(policyFee),commission=VALUES(commission),total=VALUES(total),currency=VALUES(currency),observations=VALUES(observations),coveragesJson=VALUES(coveragesJson),liquidated=VALUES(liquidated),liquidationNumber=VALUES(liquidationNumber),liquidationDate=VALUES(liquidationDate),updatedAt=VALUES(updatedAt),archivedAt=NULL", [tuple([p[0],p[1],p[2],ptype_ids.get(p[3]),prov_ids.get(p[4]),*p[5:]]) for p in policies]) + pol_rows = [tuple([p[0], p[1], p[2], ptype_ids.get(p[3]), prov_ids.get(p[4]), *p[5:]]) for p in policies] + c.executemany(f"INSERT INTO policies ({pol_cols}) VALUES ({ph})", pol_rows) diff --git a/migration/transform_properties.py b/migration/transform_properties.py index 0ed9d3e..476ba9b 100644 --- a/migration/transform_properties.py +++ b/migration/transform_properties.py @@ -112,6 +112,12 @@ def main(): "WHERE sourceSystem='utilities' AND sourceTable='DATGRAL'") cust_map = {r[0]: r[1] for r in cur.fetchall()} + # Sync reuses each legacy property's existing id (keyed by provenance) so its + # PK is stable AND the services/trust rows built below point at the right + # parent. New legacy rows fall through to a fresh uuid. + existing_prop = existing_ids(cur, "properties", ("legacySourceTable", "legacyId"), + "WHERE legacyId IS NOT NULL") if sync_mode else {} + dm = load("datmex") pf = load("profile") # PROFILE flags by join key (best-effort; key nearly unique in PROFILE) @@ -133,7 +139,7 @@ def main(): legacy_id = str(int(row["_row_num"])) prop_keys.add(("DATMEX", legacy_id)) - pid = str(uuid.uuid4()) + pid = existing_prop.get(("DATMEX", legacy_id)) or str(uuid.uuid4()) addr2_parts = [] for lbl, col in (("CASA", "casa"), ("MZ", "manzana"), ("LOTE", "lote")): if s_keep0(row[col]): @@ -206,16 +212,14 @@ def main(): # Fresh rebuild (children first), or additive upsert for legacy-owned rows. if sync_mode: - existing = existing_ids(cur, "properties", ("legacySourceTable", "legacyId"), "WHERE legacyId IS NOT NULL") - for row in props: - key = (row[8], row[9]) - if key in existing: - row = list(row); row[0] = existing[key] - cur.execute( - "INSERT INTO properties (id,customerId,addressLine1,addressLine2,phone1,phone2,phone3,zone,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),addressLine1=VALUES(addressLine1),addressLine2=VALUES(addressLine2),phone1=VALUES(phone1),phone2=VALUES(phone2),phone3=VALUES(phone3),zone=VALUES(zone),archivedAt=NULL", tuple(row)) - delete_missing(cur, "properties", ("legacySourceTable", "legacyId"), prop_keys, "WHERE legacyId IS NOT NULL") + # Children first (scoped to legacy-owned rows so manual rows survive), + # then upsert properties (ids already stable), then drop legacy rows + # gone from source, then re-insert the rebuilt children. cur.execute("DELETE ps FROM property_services ps JOIN properties p ON p.id=ps.propertyId WHERE p.legacyId IS NOT NULL") cur.execute("DELETE ta FROM trust_accounts ta JOIN properties p ON p.id=ta.propertyId WHERE p.legacyId IS NOT NULL") + cur.executemany( + "INSERT INTO properties (id,customerId,addressLine1,addressLine2,phone1,phone2,phone3,zone,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),addressLine1=VALUES(addressLine1),addressLine2=VALUES(addressLine2),phone1=VALUES(phone1),phone2=VALUES(phone2),phone3=VALUES(phone3),zone=VALUES(zone),archivedAt=NULL", props) + delete_missing(cur, "properties", ("legacySourceTable", "legacyId"), prop_keys, "WHERE legacyId IS NOT NULL") cur.executemany("INSERT INTO property_services (id,propertyId,kind,accountNumber,meterNumber,route,dueDay,active,notes) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", services) cur.executemany("INSERT INTO trust_accounts (id,propertyId,bankName,trustNumber,bankFee,dueDate1,dueDate2) VALUES (%s,%s,%s,%s,%s,%s,%s)", trusts) else: @@ -246,7 +250,8 @@ def main(): print(f" {k:14} {n}") print(f" -> trust_accounts : {n_t}") print(f" orphan properties (bad customer FK): {orphans}") - assert n_p == len(props) and orphans == 0, "property load invariant failed" + # n_p == len(props) is a full-load invariant; sync keeps manual rows too. + assert (sync_mode or n_p == len(props)) and orphans == 0, "property load invariant failed" print(" validation: OK") conn.close() diff --git a/migration/transform_transactions.py b/migration/transform_transactions.py index 5ef2648..4ba8c03 100644 --- a/migration/transform_transactions.py +++ b/migration/transform_transactions.py @@ -227,7 +227,23 @@ def main(): efectivo_like("stg_seguros", "efectivo", "INSURANCE", ins_cust, "SEGUROS 16_be", "EFECTIVO") if sync_mode: - c.executemany("INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference,amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),domain=VALUES(domain),typeId=VALUES(typeId),transactionDate=VALUES(transactionDate),period=VALUES(period),reference=VALUES(reference),amount=VALUES(amount),currency=VALUES(currency),checkNumber=VALUES(checkNumber),message=VALUES(message),updatedAt=NOW(),voidedAt=NULL", tx) + # Transaction types are rebuilt with fresh uuids each run; resolve them + # against the rows already in the DB by English name (inserting any that + # are new) and remap each tx's typeId onto the persisted id so the FK to + # type_transactions holds. exchange_rates isn't referenced by tx, so it + # is left untouched in sync. + c.execute("SELECT id,nameEn FROM type_transactions") + db_types = {(nm or "").upper(): i for i, nm in c.fetchall()} + fresh_name = {tid: (en or "").upper() for tid, en, es, active in type_rows} + new_types = [] + for tid, en, es, active in type_rows: + if (en or "").upper() not in db_types: + db_types[(en or "").upper()] = tid + new_types.append((tid, en, es, active)) + if new_types: + c.executemany("INSERT INTO type_transactions (id,nameEn,nameEs,isService) VALUES (%s,%s,%s,%s)", new_types) + tx = [(t[0], t[1], t[2], (db_types.get(fresh_name.get(t[3])) if t[3] else None), *t[4:]) for t in tx] + c.executemany("INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference,amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),domain=VALUES(domain),typeId=VALUES(typeId),transactionDate=VALUES(transactionDate),period=VALUES(period),reference=VALUES(reference),amount=VALUES(amount),currency=VALUES(currency),checkNumber=VALUES(checkNumber),message=VALUES(message),voidedAt=NULL", tx) else: c.execute("SET FOREIGN_KEY_CHECKS=0") for t in ("transactions", "type_transactions", "exchange_rates"): @@ -255,7 +271,7 @@ def main(): print(f" -> transactions : {count('transactions')}") print(f" by domain : {dict(by_dom)}") for src, n in by_src: - print(f" {src:16} {n}") + print(f" {(src or '(manual)'):16} {n}") print(f" -> type_transactions : {count('type_transactions')}") print(f" -> exchange_rates : {count('exchange_rates')}") print(f" orphan transactions (bad customer FK): {orphans}") diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 8cdaa68..f891f24 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -221,10 +221,12 @@ model Vehicle { vinNumber String? stateCode String? notes String? @db.Text + // One legacy policy row can carry up to 3 vehicles, so they share a + // legacyId (the source row number) — provenance is NOT unique per vehicle. + // Sync rebuilds legacy vehicles by scoped delete + reinsert instead of upsert. legacySourceTable String? legacyId String? - @@unique([legacySourceTable, legacyId]) @@map("vehicles") }