diff --git a/.gitignore b/.gitignore index ac95f9f..47176c5 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,6 @@ __pycache__/ *.pyc packages/database/generated/ .DS_Store + +# local package-manager shim (machine npm is pnpm-aliased); npm is canonical +pnpm-lock.yaml diff --git a/RESUME.md b/RESUME.md index fba8b9a..833bc3e 100644 --- a/RESUME.md +++ b/RESUME.md @@ -169,11 +169,24 @@ Homebrew. No Access ODBC driver, `node_modules` not installed, staging Parquet n (folio collides; migrate both), the billing tables are disjoint period runs (union all, no de-dup), and COBRO3 is a charge batch not a customer snapshot (DATGRAL is sole master). The decided union/de-dup rules are in `PLAN.md` migration step 2. -4. **Transform + load** (plan step 3) — NEXT. Start with `Customer`/`CustomerLegacyRef` (every - other module depends on it): DATGRAL (utilities) is the master; join insurance `DATGRAL` - via its `NUM UTIL` cross-ref + name/address matching; COBRO3 excluded from customers. Then - the ledger union per the reconciliation rules (both EFECTIVO tables, all three billing tables, - provenance-keyed; normalize `monedas` currency variants). +4. **Transform + load** (plan step 3) — IN PROGRESS. + - **Customers — DONE** (`migration/transform_customers.py`). Loaded into the dev DB: 1682 + customers (1172 utilities master + 510 insurance-only), 2242 legacy refs (all traceable), + 560 insurance rows linked via `num_util` with 0 broken refs, **542 merged identities** + spanning both business lines; linked customers enriched with insurance-only ID-doc fields. + COBRO3 excluded. Re-runnable (truncate+rebuild); needs staged Parquet present + (`load_staging.py --output-dir ./output` first). + - **NEXT:** properties+services (DATMEX/PROFILE), policies (+installments/vehicles/drivers/ + beneficiaries/claims), then the ledger union per the reconciliation rules (both EFECTIVO + tables, all three billing tables, provenance-keyed; normalize `monedas` currency variants), + SCOTHIA bank register. Each resolves its customer FK through `customer_legacy_refs`. +5b. **Infra done:** dev MySQL deployed to the cubex Swarm via Portainer API as stack + `jorgecuadros-dev-db` (MySQL 8.4, `192.168.4.212:3307`, node `cubex` labeled + `jorgecuadros_db=true`); Prisma schema pushed (26 tables). Stack file: + `deploy/jorgecuadros-db.stack.yml` (same file deploys prod as `jorgecuadros-prod-db` :3306). + Creds in gitignored `deploy/.env.dev`. NOTE: machine `npm` is pnpm-aliased and pnpm ignores + the `workspaces` field — full workspace install needs `pnpm-workspace.yaml` or real npm; for + now Prisma CLI is run via `npx prisma@5`. 5. **Customer module** in `apps/api`/`apps/web` (list/search/detail) — first real feature, Spanish-first UI. Run `npm install` at repo root first (node_modules absent here). 6. **Sync design finalization** — now unblocked: map the internal→VPS replicated subset and the diff --git a/migration/transform_customers.py b/migration/transform_customers.py new file mode 100644 index 0000000..c059b77 --- /dev/null +++ b/migration/transform_customers.py @@ -0,0 +1,278 @@ +""" +Migration plan step 3 (customers): build the unified customer master. + +Reads staged Parquet and loads `customers` + `customer_legacy_refs` in the +Prisma-managed MySQL. This is the core of the whole project — one customer +record shared by both business lines — so every later module (policies, +properties, transactions) resolves its customer FK through the legacy refs +written here. + +Rules come from the reconciliation pass (RECONCILIATION.md): + - Utilities `DATGRAL` (1172) is the customer master; one Customer each. + - Insurance `DATGRAL` (1070) links to a utilities customer via its + `num_util` cross-reference. Matches fold into the existing customer (and + enrich it with the ID-document fields the utilities master lacks); + non-matches become new insurance-only customers. + - `COBRO3` is a charge batch, NOT a customer source -> excluded here. + +Every legacy row folded in gets a `customer_legacy_refs` row +(sourceSystem, sourceTable=DATGRAL, legacyId) so the merge is auditable and +the load is idempotent (re-run = truncate + rebuild). + +Run: ./.venv/bin/python transform_customers.py +(reads deploy/.env.dev for the dev DATABASE credentials) +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from decimal import Decimal, InvalidOperation +from pathlib import Path + +import pandas as pd +import pymysql + +REPO = Path(__file__).resolve().parents[1] +STG = Path(__file__).parent / "output" +NULL = "∅" +NOW = datetime.now(timezone.utc).replace(tzinfo=None) # naive UTC for MySQL DATETIME + +_TRUE = {"1", "-1", "true", "verdadero", "si", "sí", "activo", "yes", "y", "t"} +_FALSE = {"0", "false", "falso", "no", "inactivo", "n", "f"} + + +# --------------------------- source normalization --------------------------- # +def load(source: str, table: str) -> pd.DataFrame: + df = pd.read_parquet(STG / source / f"{table}.parquet") + df = df[[c for c in df.columns if c not in ("_legacy_source_table", "_row_num")]].copy() + for c in df.columns: + df[c] = df[c].astype("string").str.strip() + return df + + +def s(v) -> str | None: + """Cell -> clean string or None (empty/sentinel -> None).""" + if v is None or pd.isna(v): + return None + v = str(v).strip() + return None if v in ("", NULL, "0000-00-00") else v + + +def norm_id(v) -> str | None: + v = s(v) + if v is None: + return None + if v.endswith(".0"): # some numeric ids serialize as "521.0" + v = v[:-2] + return v if v not in ("0",) else None + + +def as_bool(v, default=True) -> int: + v = s(v) + if v is None: + return 1 if default else 0 + lv = v.lower() + if lv in _TRUE: + return 1 + if lv in _FALSE: + return 0 + return 1 if default else 0 + + +def as_date(v): + v = s(v) + if v is None: + return None + dt = pd.to_datetime(v, errors="coerce") + if pd.isna(dt): + return None + return dt.to_pydatetime() + + +def as_decimal(v): + v = s(v) + if v is None: + return None + v = v.replace(",", "") + try: + return Decimal(v) + except (InvalidOperation, ValueError): + return None + + +# ------------------------------- record builders ---------------------------- # +def customer_from_utilities(row) -> dict: + return dict( + id=str(uuid.uuid4()), + name=s(row["nombre"]) or "(SIN NOMBRE)", + addressLine1=s(row["direccion"]), + addressLine2=s(row["colonia"]), + city=s(row["ciudad"]), + state=s(row["estado"]), + zipCode=s(row["codigo"]), + country=s(row["pais"]), + phone=s(row["telusa"]), + mobile=s(row["cel"]), + fax=s(row["fax"]), + email=s(row["email"]), + notes=s(row["observaciones"]), + identificationType=None, + identificationNumber=None, + identificationExpiration=None, + customerSince=as_date(row["cliente_desde"]), + status=as_bool(row["status"]), + feeAmount=as_decimal(row["fee"]), + updatedAt=NOW, + ) + + +def customer_from_insurance(row) -> dict: + return dict( + id=str(uuid.uuid4()), + name=s(row["nombre"]) or "(SIN NOMBRE)", + addressLine1=s(row["direccion_1"]), + addressLine2=s(row["direccion_2"]), + city=s(row["ciudad"]), + state=s(row["estado"]), + zipCode=s(row["codigo"]), + country=s(row["pais"]), + phone=s(row["telusa"]), + mobile=s(row["tel"]), + fax=s(row["fax"]), + email=s(row["emailaddress"]), + notes=s(row["observaciones"]), + identificationType=s(row["tipo_identificacion"]), + identificationNumber=s(row["no_identificacion"]), + identificationExpiration=as_date(row["expira_identificacion"]), + customerSince=None, + status=1, + feeAmount=None, + updatedAt=NOW, + ) + + +_CUST_COLS = [ + "id", "name", "addressLine1", "addressLine2", "city", "state", "zipCode", + "country", "phone", "mobile", "fax", "email", "notes", "identificationType", + "identificationNumber", "identificationExpiration", "customerSince", + "status", "feeAmount", "updatedAt", +] + + +def main() -> None: + envf = REPO / "deploy" / ".env.dev" + env = dict( + l.strip().split("=", 1) for l in envf.read_text().splitlines() + if "=" in l and not l.startswith("#") and not l.startswith("DATABASE_URL") + ) + conn = pymysql.connect( + host="192.168.4.212", port=int(env["MYSQL_PORT"]), user=env["MYSQL_USER"], + password=env["MYSQL_PASSWORD"], database=env["MYSQL_DATABASE"], autocommit=False, + ) + cur = conn.cursor() + + # Fresh, idempotent rebuild. + cur.execute("SET FOREIGN_KEY_CHECKS=0") + cur.execute("TRUNCATE TABLE customer_legacy_refs") + cur.execute("TRUNCATE TABLE customers") + cur.execute("SET FOREIGN_KEY_CHECKS=1") + + util = load("stg_utilities", "datgral") + ins = load("stg_seguros", "datgral") + + customers: list[dict] = [] + refs: list[tuple] = [] # (id, customerId, sourceSystem, sourceTable, legacyId) + util_map: dict[str, str] = {} # utilities num_id -> customer_id + + # Phase A: utilities DATGRAL = the master. + for _, row in util.iterrows(): + rec = customer_from_utilities(row) + customers.append(rec) + nid = norm_id(row["num_id"]) + legacy = nid or f"rownum_{len(customers)}" + refs.append((str(uuid.uuid4()), rec["id"], "utilities", "DATGRAL", legacy)) + if nid: + util_map[nid] = rec["id"] + + # Phase B: insurance DATGRAL links via num_util, else new customer. + linked = new_ins = unmatched_numutil = 0 + enrich: list[tuple] = [] # (customerId, insurance record) for fill-in + for _, row in ins.iterrows(): + ins_id = norm_id(row["num_id"]) or f"insrow_{new_ins+linked}" + nutil = norm_id(row["num_util"]) + if nutil and nutil in util_map: + cust_id = util_map[nutil] + linked += 1 + enrich.append((cust_id, customer_from_insurance(row))) + else: + if nutil and nutil not in util_map: + unmatched_numutil += 1 + rec = customer_from_insurance(row) + customers.append(rec) + cust_id = rec["id"] + new_ins += 1 + refs.append((str(uuid.uuid4()), cust_id, "insurance", "DATGRAL", ins_id)) + + # Insert customers. + placeholders = ",".join(["%s"] * len(_CUST_COLS)) + cur.executemany( + f"INSERT INTO customers ({','.join(f'`{c}`' for c in _CUST_COLS)}) VALUES ({placeholders})", + [tuple(rec[c] for c in _CUST_COLS) for rec in customers], + ) + # Insert legacy refs. + cur.executemany( + "INSERT INTO customer_legacy_refs (id, customerId, sourceSystem, sourceTable, legacyId) " + "VALUES (%s,%s,%s,%s,%s)", + refs, + ) + + # Enrich linked customers with insurance-only ID-doc fields, and fill any + # contact fields the utilities master left empty (COALESCE keeps master's). + for cust_id, ir in enrich: + cur.execute( + "UPDATE customers SET " + "identificationType = COALESCE(identificationType, %s), " + "identificationNumber = COALESCE(identificationNumber, %s), " + "identificationExpiration = COALESCE(identificationExpiration, %s), " + "email = COALESCE(email, %s), " + "phone = COALESCE(phone, %s), " + "mobile = COALESCE(mobile, %s), " + "updatedAt = %s " + "WHERE id = %s", + (ir["identificationType"], ir["identificationNumber"], + ir["identificationExpiration"], ir["email"], ir["phone"], + ir["mobile"], NOW, cust_id), + ) + + conn.commit() + + # ---- validation / report ---- + cur.execute("SELECT COUNT(*) FROM customers") + n_cust = cur.fetchone()[0] + cur.execute("SELECT COUNT(*) FROM customer_legacy_refs") + n_refs = cur.fetchone()[0] + cur.execute("SELECT sourceSystem, COUNT(*) FROM customer_legacy_refs GROUP BY sourceSystem") + by_sys = dict(cur.fetchall()) + cur.execute("SELECT COUNT(*) FROM customer_legacy_refs " + "GROUP BY customerId HAVING COUNT(*) > 1") + merged = len(cur.fetchall()) + + print("=== Customer load complete ===") + print(f" utilities DATGRAL rows : {len(util)}") + print(f" insurance DATGRAL rows : {len(ins)}") + print(f" linked to a utilities customer : {linked}") + print(f" new insurance-only customers : {new_ins}") + print(f" num_util set but not in utilities master (data-quality) : {unmatched_numutil}") + print(f" -> customers : {n_cust} (expected {len(util)} + {new_ins} = {len(util)+new_ins})") + print(f" -> customer_legacy_refs : {n_refs} (expected {len(util)+len(ins)} = {len(util)+len(ins)})") + print(f" refs by system : {by_sys}") + print(f" customers with >1 ref (merged identities) : {merged}") + 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() + + +if __name__ == "__main__": + main()