Transform+load: unified customer master (migration step 3, customers)

migration/transform_customers.py builds `customers` + `customer_legacy_refs`
from staged DATGRAL, implementing the reconciliation rules: utilities DATGRAL
is the customer master; insurance DATGRAL folds in via its num_util
cross-reference (matches enrich the master with the ID-document fields
utilities lacks); COBRO3 excluded as a charge batch. Every legacy row gets a
provenance ref, so the load is auditable and idempotent (truncate+rebuild).

Loaded and validated against the dev DB (192.168.4.212:3307):
  1682 customers (1172 utilities master + 510 insurance-only)
  2242 legacy refs (1172 utilities + 1070 insurance) — 0 orphans
  560 insurance rows linked via num_util, 0 broken cross-refs
  542 merged identities spanning both business lines
Spot-checked a merged customer: single record carrying utilities fee +
insurance passport/ID enriched in, both provenance refs present.

RESUME.md: mark customers done, record dev-DB infra + the pnpm/npm caveat.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.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

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
+4. **Transform + load** (plan step 3) — IN PROGRESS.
-   other module depends on it): DATGRAL (utilities) is the master; join insurance `DATGRAL`
+   - **Customers — DONE** (`migration/transform_customers.py`). Loaded into the dev DB: 1682
-   via its `NUM UTIL` cross-ref + name/address matching; COBRO3 excluded from customers. Then
+     customers (1172 utilities master + 510 insurance-only), 2242 legacy refs (all traceable),
-   the ledger union per the reconciliation rules (both EFECTIVO tables, all three billing tables,
+     560 insurance rows linked via `num_util` with 0 broken refs, **542 merged identities**
-   provenance-keyed; normalize `monedas` currency variants).
+     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

deploy/jorgecuadros-db.stack.yml

@@ -0,0 +1,76 @@
+# Canonical internal MySQL for the Jorge Cuadros platform.
+#
+# Target: Portainer "local" endpoint on cubex (192.168.4.212:9443), which is a
+# 3-node Docker Swarm. This is the source-of-truth database from the plan's
+# "internal server" — the migration transform+load and the NestJS API both
+# point at it, and (later) it becomes the replication SOURCE for the VPS.
+#
+# DEV / PROD as two stacks from this one file. Deploy it twice with different
+# stack names + env_data:
+#   dev : stack jorgecuadros-dev-db  MYSQL_PORT=3307  MYSQL_SERVER_ID=11
+#   prod: stack jorgecuadros-prod-db MYSQL_PORT=3306  MYSQL_SERVER_ID=1
+# Swarm namespaces the named volume by stack name, so the two environments get
+# fully isolated data (jorgecuadros-dev-db_mysql_data vs -prod-db_...) with no
+# extra config. Distinct published ports let both run on the same node at once.
+# server-id must be unique per environment (prod=1 is the replication source).
+#
+# Swarm statefulness rules (see portainer-gitea-deploy skill):
+#  - named volume, never a relative bind mount (the API deploy path won't
+#    create host dirs -> "bind source path does not exist").
+#  - a named volume is LOCAL to whichever node the task lands on. With no
+#    shared storage, the DB MUST be pinned to one node or a reschedule would
+#    start against a fresh empty volume. Pinned here via a node label so it is
+#    not tied to a hostname: label exactly one node with
+#        docker node update --label-add jorgecuadros_db=true <node>
+#    and MySQL will always run there, reusing the same volume.
+#
+# Secrets (MYSQL_ROOT_PASSWORD, MYSQL_PASSWORD) are injected at deploy time via
+# Portainer env_data / stack environment, not committed here.
+
+version: "3.8"
+
+services:
+  mysql:
+    image: mysql:8.4
+    command:
+      # (caching_sha2_password is already the default in 8.4; the old
+      #  --default-authentication-plugin flag was REMOVED in 8.4 and aborts boot.)
+      # binlog + GTID on from day one so this node can be the replication
+      # SOURCE for the VPS replica later without a restart/reconfigure.
+      - --server-id=${MYSQL_SERVER_ID:-1}
+      - --log-bin=mysql-bin
+      - --binlog-format=ROW
+      - --gtid-mode=ON
+      - --enforce-gtid-consistency=ON
+    environment:
+      MYSQL_DATABASE: ${MYSQL_DATABASE:-jorgecuadros}
+      MYSQL_USER: ${MYSQL_USER:-jorgecuadros}
+      MYSQL_PASSWORD: ${MYSQL_PASSWORD:?MYSQL_PASSWORD must be set}
+      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD:?MYSQL_ROOT_PASSWORD must be set}
+    ports:
+      # Swarm ingress publishes on every node, so this is reachable at
+      # 192.168.4.212:${MYSQL_PORT} regardless of which node the task pins to.
+      - target: 3306
+        published: ${MYSQL_PORT:-3306}
+        protocol: tcp
+        mode: ingress
+    volumes:
+      - mysql_data:/var/lib/mysql
+    deploy:
+      replicas: 1
+      placement:
+        constraints:
+          - node.labels.jorgecuadros_db == true
+      restart_policy:
+        condition: any
+      update_config:
+        order: stop-first
+    healthcheck:
+      test: ["CMD", "mysqladmin", "ping", "-h", "localhost", "-u", "root", "-p$$MYSQL_ROOT_PASSWORD"]
+      interval: 10s
+      timeout: 5s
+      retries: 12
+      start_period: 40s
+
+volumes:
+  mysql_data:

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()