jorgecuadros-platform/migration/transform_customers.py

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