rmancinasandClaude Opus 5.5
b016e53893
The office files every customer in a numbered paper folder and reuses the number once an account closes. That number was the legacy NUMid, which is also an import key and the portal login, so reusing it meant re-pointing an identity. It is now customers.folderNumber: a plain label nothing joins on. - Unique only among non-archived customers, checked in CustomersService (create, update, restore). Archived customers keep theirs as history. - GET /customers/folder/next suggests the lowest free number; the new customer form pre-fills it. - Backfilled from the utilities NUMid (dev: 1,171 folders, 0 shared); the importer copies it on load. Insurance-only customers start without one. - Reports, the printed statement and the customer file show "Carpeta" instead of NUMid. The list searches by it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
418 lines
18 KiB
Python
418 lines
18 KiB
Python
"""
|
|
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
|
|
(it is still read as a name-recovery source, see below).
|
|
|
|
`DATGRAL.NOMBRE` is blank on 266 legacy rows (140 utilities, 126 insurance).
|
|
Names for most of them are recovered from secondary tables — see
|
|
`_NAME_SOURCES` — and `customers.nameSource` records which table each
|
|
recovered name came from.
|
|
|
|
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
|
|
from sync import parse_mode
|
|
|
|
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
|
|
|
|
|
|
# -------------------------------- name recovery ----------------------------- #
|
|
NO_NAME = "(SIN NOMBRE)"
|
|
|
|
# The old PHP importer skipped blank-name DATGRAL rows outright
|
|
# (jorgecuadros-intra-webapp/src/tools/customerAdapter.php:47,81). That also
|
|
# silently dropped those rows' properties, policies and transactions, because
|
|
# every other adapter resolved its customer FK through the customer_mapping
|
|
# table those skipped rows never got into (customerServiceAdapter.php:45), and
|
|
# customerBalanceAdapter.php:52 defaulted the unmapped ones to customer_id 0.
|
|
# Most blank-name rows are real accounts, so recover the name instead of
|
|
# skipping: 176 of the 257 carry a property, policy or transaction.
|
|
#
|
|
# Per side, most trustworthy source first; a later source only fills ids the
|
|
# earlier ones left unresolved. UTILSEG is the office's own hand-maintained
|
|
# name <-> id cross-reference spanning both lines; the rest are billing runs
|
|
# and policy rows that happen to repeat the customer's name.
|
|
# (label, staged source, table, id column, name column)
|
|
_NAME_SOURCES: dict[str, list[tuple[str, str, str, str, str]]] = {
|
|
"utilities": [
|
|
("UTILSEG", "stg_seguros", "utilseg", "util", "nombre"),
|
|
("IVA 2015", "stg_utilities", "iva_2015", "num_id", "nombre"),
|
|
("COBRO3", "stg_utilities", "cobro3", "num_id", "nombre"),
|
|
],
|
|
"insurance": [
|
|
("UTILSEG", "stg_seguros", "utilseg", "seguros", "nombre"),
|
|
("MULT", "stg_seguros", "mult", "num_id", "nombre_aseg"),
|
|
("M EMPR", "stg_seguros", "m_empr", "num_id", "nombre_aseg"),
|
|
("INCENDIO", "stg_seguros", "incendio", "num_id", "nombre_aseg"),
|
|
],
|
|
}
|
|
|
|
|
|
def build_name_index(system: str) -> dict[str, tuple[str, str]]:
|
|
"""legacy num_id -> (recovered name, source label) for one business line."""
|
|
index: dict[str, tuple[str, str]] = {}
|
|
for label, source, table, id_col, name_col in _NAME_SOURCES[system]:
|
|
df = load(source, table)
|
|
if id_col not in df.columns or name_col not in df.columns:
|
|
raise KeyError(f"{table}: expected columns {id_col}/{name_col}, got {list(df.columns)}")
|
|
for _, row in df.iterrows():
|
|
nid, name = norm_id(row[id_col]), s(row[name_col])
|
|
if nid and name and nid not in index:
|
|
index[nid] = (name, label)
|
|
return index
|
|
|
|
|
|
def resolve_name(raw, nid, index) -> tuple[str, str | None]:
|
|
"""(name, nameSource). nameSource stays None when DATGRAL had the name."""
|
|
name = s(raw)
|
|
if name:
|
|
return name, None
|
|
if nid and nid in index:
|
|
return index[nid]
|
|
return NO_NAME, None
|
|
|
|
|
|
# ------------------------------- record builders ---------------------------- #
|
|
def customer_from_utilities(row, name_index) -> dict:
|
|
nid = norm_id(row["num_id"])
|
|
name, name_source = resolve_name(row["nombre"], nid, name_index)
|
|
return dict(
|
|
id=str(uuid.uuid4()),
|
|
name=name,
|
|
nameSource=name_source,
|
|
nameMissing=int(name == NO_NAME),
|
|
# The NUMid is the number on the customer's shelf folder. It is copied
|
|
# into a plain label so the office can reuse it without it ever being
|
|
# an identity; customer_legacy_refs keeps the NUMid itself.
|
|
folderNumber=int(nid) if nid and nid.isdigit() else None,
|
|
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"]),
|
|
# DATGRAL.TIPO is the minimum-balance threshold (100/200/300/500 —
|
|
# 1,017 of 1,172 customers carry one), NOT an identification or account
|
|
# type as the column name suggests. It reaches the website as
|
|
# datosfreak.TIPO and is returned to the customer app as `minBalance`.
|
|
minimumBalance=as_decimal(row["tipo"]),
|
|
updatedAt=NOW,
|
|
)
|
|
|
|
|
|
def customer_from_insurance(row, name_index) -> dict:
|
|
name, name_source = resolve_name(row["nombre"], norm_id(row["num_id"]), name_index)
|
|
return dict(
|
|
id=str(uuid.uuid4()),
|
|
name=name,
|
|
nameSource=name_source,
|
|
nameMissing=int(name == NO_NAME),
|
|
folderNumber=None, # insurance ids are a separate space, not a shelf folder
|
|
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,
|
|
minimumBalance=None,
|
|
updatedAt=NOW,
|
|
)
|
|
|
|
|
|
_CUST_COLS = [
|
|
"id", "name", "nameSource", "nameMissing", "folderNumber", "addressLine1", "addressLine2", "city", "state", "zipCode",
|
|
"country", "phone", "mobile", "fax", "email", "notes", "identificationType",
|
|
"identificationNumber", "identificationExpiration", "customerSince",
|
|
"status", "feeAmount", "minimumBalance", "updatedAt",
|
|
]
|
|
|
|
|
|
def main() -> None:
|
|
env, sync_mode = parse_mode()
|
|
conn = connect(env)
|
|
print(f"[customers] target env: {env}")
|
|
cur = conn.cursor()
|
|
|
|
if not sync_mode:
|
|
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")
|
|
util_names = build_name_index("utilities")
|
|
ins_names = build_name_index("insurance")
|
|
|
|
customers: list[dict] = []
|
|
refs: list[tuple] = [] # (id, customerId, sourceSystem, sourceTable, legacyId)
|
|
util_map: dict[str, str] = {} # utilities num_id -> customer_id
|
|
rec_by_id: dict[str, dict] = {}
|
|
|
|
# Phase A: utilities DATGRAL = the master.
|
|
for _, row in util.iterrows():
|
|
rec = customer_from_utilities(row, util_names)
|
|
customers.append(rec)
|
|
rec_by_id[rec["id"]] = 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 = from_ins_side = 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
|
|
ins_rec = customer_from_insurance(row, ins_names)
|
|
# Last name-recovery path: a master whose own line had no name and
|
|
# no utilities-side fallback can still borrow the name its linked
|
|
# insurance record resolved to.
|
|
master = rec_by_id[cust_id]
|
|
if master["name"] == NO_NAME and ins_rec["name"] != NO_NAME:
|
|
master["name"] = ins_rec["name"]
|
|
master["nameSource"] = ins_rec["nameSource"] or "DATGRAL (seguros)"
|
|
master["nameMissing"] = 0
|
|
from_ins_side += 1
|
|
enrich.append((cust_id, ins_rec))
|
|
else:
|
|
if nutil and nutil not in util_map:
|
|
unmatched_numutil += 1
|
|
rec = customer_from_insurance(row, ins_names)
|
|
customers.append(rec)
|
|
rec_by_id[rec["id"]] = rec
|
|
cust_id = rec["id"]
|
|
new_ins += 1
|
|
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:
|
|
# 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),folderNumber=VALUES(folderNumber),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),minimumBalance=VALUES(minimumBalance),updatedAt=VALUES(updatedAt)", tuple(rec[c] for c in _CUST_COLS))
|
|
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})",
|
|
[tuple(rec[c] for c in _CUST_COLS) for rec in customers],
|
|
)
|
|
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).
|
|
# 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), "
|
|
"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())
|
|
cur.execute("SELECT nameSource, COUNT(*) FROM customers "
|
|
"WHERE nameSource IS NOT NULL GROUP BY nameSource ORDER BY 2 DESC")
|
|
recovered = cur.fetchall()
|
|
cur.execute("SELECT COUNT(*) FROM customers WHERE name = %s", (NO_NAME,))
|
|
still_unnamed = cur.fetchone()[0]
|
|
cur.execute("SELECT COUNT(folderNumber) FROM customers")
|
|
with_folder = cur.fetchone()[0]
|
|
cur.execute("SELECT COUNT(*) FROM (SELECT folderNumber FROM customers "
|
|
"WHERE archivedAt IS NULL AND folderNumber IS NOT NULL "
|
|
"GROUP BY folderNumber HAVING COUNT(*) > 1) d")
|
|
shared_folders = cur.fetchone()[0]
|
|
|
|
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}")
|
|
print(" name recovery (DATGRAL.NOMBRE was blank):")
|
|
for src, n in recovered:
|
|
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}")
|
|
print(f" folder numbers : {with_folder} (one per numeric utilities NUMid)")
|
|
print(f" held by >1 open customer (must be 0) : {shared_folders}")
|
|
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()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|