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>
This commit is contained in:
Claude Opus 4.8
@@ -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()
|
||||
Reference in New Issue
Block a user