""" 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 from dbenv import connect, env_arg 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: env = env_arg() conn = connect(env) print(f"[customers] target env: {env}") 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()