jorgecuadros-platform/migration/transform_properties.py

"""
Migration plan step 3 (properties): properties + services + trust accounts.

Reads staged DATMEX (one row per property) / PROFILE (enrollment flags) and
loads `properties`, `property_services`, `trust_accounts` in the dev MySQL,
resolving each property's customer FK through `customer_legacy_refs` (so this
must run AFTER transform_customers.py).

Design decisions (validated against the staged data):
  - DATMEX.numer_id -> the utilities customer number; ALL 1171 distinct ids
    match a loaded customer (0 orphans). numer_id is NOT unique in DATMEX
    (customers can own several properties) -> one Property per DATMEX row.
  - DATMEX and PROFILE are NOT row-aligned (only 2.5% match by position), so
    PROFILE flags are merged best-effort on (numer_id, casa, direccion).
    Services are derived from the presence of DATMEX's own account/route/meter
    fields (the authoritative data); the PROFILE flag, when matched, only
    refines the service's `active` value. This keeps correctness independent
    of the fragile join.
  - TrustAccount is 1:1 with a property; built from DATMEX's own trust fields
    when a trust is indicated. TRUSTVENCE (549) overlaps these per-property
    fields and is deferred to a later reconciliation rather than loaded as a
    conflicting second source.
  - doc_1/doc_2 (LONGBINARY) are skipped here — document extraction is step 4.

Run:  ./.venv/bin/python transform_properties.py
"""

from __future__ import annotations

import uuid
from decimal import Decimal, InvalidOperation
from pathlib import Path

import pandas as pd

from dbenv import connect, env_arg

STG = Path(__file__).parent / "output" / "stg_utilities"
NULL = "∅"
_TRUE = {"1", "-1", "true", "si", "sí", "yes"}


def load(name: str) -> pd.DataFrame:
    df = pd.read_parquet(STG / f"{name}.parquet").sort_values("_row_num").reset_index(drop=True)
    keep = [c for c in df.columns if c != "_legacy_source_table"]
    df = df[keep].copy()
    for c in df.columns:
        if c != "_row_num":
            df[c] = df[c].astype("string").str.strip()
    return df


def s(v):
    if v is None or pd.isna(v):
        return None
    v = str(v).strip()
    return None if v in ("", NULL, "0", "0000-00-00") else v


def s_keep0(v):
    """Like s() but keeps '0' (used for join keys / house numbers)."""
    if v is None or pd.isna(v):
        return None
    v = str(v).strip()
    return None if v in ("", NULL) else v


def norm_id(v):
    v = s_keep0(v)
    if v is None:
        return None
    if v.endswith(".0"):
        v = v[:-2]
    return None if v == "0" else v


def as_dec(v):
    v = s(v)
    if v is None:
        return None
    try:
        return Decimal(v.replace(",", ""))
    except (InvalidOperation, ValueError):
        return None


def as_date(v):
    v = s(v)
    if v is None:
        return None
    dt = pd.to_datetime(v, errors="coerce")
    return None if pd.isna(dt) else dt.to_pydatetime()


def jkey(numer, casa, direc):
    return "|".join([
        norm_id(numer) or "",
        (s_keep0(casa) or ""),
        (s(direc) or "").upper(),
    ])


def main() -> None:
    env = env_arg()
    conn = connect(env)
    print(f"[properties] target env: {env}")
    cur = conn.cursor()

    # customer map: utilities num_id -> customerId
    cur.execute("SELECT legacyId, customerId FROM customer_legacy_refs "
                "WHERE sourceSystem='utilities' AND sourceTable='DATGRAL'")
    cust_map = {r[0]: r[1] for r in cur.fetchall()}

    dm = load("datmex")
    pf = load("profile")
    # PROFILE flags by join key (best-effort; key nearly unique in PROFILE)
    flags: dict[str, dict] = {}
    for _, r in pf.iterrows():
        flags[jkey(r["numerid"], r["casa"], r["direccion"])] = r

    props, services, trusts = [], [], []
    skipped_no_customer = 0
    matched_profile = 0

    for _, row in dm.iterrows():
        nid = norm_id(row["numer_id"])
        cust_id = cust_map.get(nid) if nid else None
        if not cust_id:
            skipped_no_customer += 1
            continue

        pid = str(uuid.uuid4())
        addr2_parts = []
        for lbl, col in (("CASA", "casa"), ("MZ", "manzana"), ("LOTE", "lote")):
            if s_keep0(row[col]):
                addr2_parts.append(f"{lbl} {s_keep0(row[col])}")
        props.append((
            pid, cust_id, s(row["direccion"]), ", ".join(addr2_parts) or None,
            s(row["telefono"]), s(row["telefono2"]), s(row["telefono3"]),
            s(row["zona"]), "DATMEX", nid,
        ))

        pfrow = flags.get(jkey(row["numer_id"], row["casa"], row["direccion"]))
        if pfrow is not None:
            matched_profile += 1

        def flag(name, default=True):
            if pfrow is None:
                return default
            return (s_keep0(pfrow[name]) or "0").lower() in _TRUE

        def svc(kind, account=None, meter=None, route=None, due=None, notes=None, active=True):
            services.append((str(uuid.uuid4()), pid, kind, account, meter, route,
                             due, 1 if active else 0, notes))

        # WATER (primary + extra meters)
        if s(row["agua"]) or flag("water1", False):
            svc("WATER", account=s(row["agua"]), meter=s(row["medidor"]),
                route=s(row["ruta"]), due=s(row["diagua"]), active=flag("water1"))
        for mcol, rcol in (("medidor2", "ruta2"), ("medidor3", "ruta3")):
            if s(row[mcol]) or s(row[rcol]):
                svc("WATER", meter=s(row[mcol]), route=s(row[rcol]),
                    notes="secondary meter", active=flag("water1"))
        # ELECTRIC (rpu account + extras); luz_tipo as note
        for rc in ("rpu", "rpu2", "rpu3"):
            if s(row[rc]):
                svc("ELECTRIC", account=s(row[rc]),
                    notes=s(row["luz_tipo"]) if rc == "rpu" else None,
                    active=flag("electric"))
        # GAS
        if s(row["gas"]) or flag("gas1", False):
            svc("GAS", due=s(row["gas_vence"]), notes=s(row["gas"]), active=flag("gas1"))
        # CABLE
        if s(row["cable_num"]) or (s_keep0(row["cable_sky"]) or "0") in _TRUE:
            svc("CABLE", account=s(row["cable_num"]), route=s(row["cia_cable"]),
                due=s(row["dia_cable"]), notes=s(row["cable_sky"]),
                active=flag("cable_sky"))
        # PROPERTY TAX
        if s(row["predial"]) or flag("propertytaxes", False):
            svc("PROPERTY_TAX", account=s(row["predial"]), notes=s(row["przn"]),
                active=flag("propertytaxes"))
        # FEDERAL ZONE
        if s(row["zfed"]) or flag("federalzone", False):
            svc("FEDERAL_ZONE", account=s(row["zfed"]), notes=s(row["zfed_t"]),
                active=flag("federalzone"))
        # ALARM
        if s(row["alarm_system"]):
            svc("ALARM", notes=s(row["alarm_system"]))
        # OTHER — free-text service notes from PROFILE.services (telnor/cable history)
        if pfrow is not None and s(pfrow["services"]):
            svc("OTHER", notes=s(pfrow["services"]))

        # TRUST (1:1) — from DATMEX trust fields when indicated
        trust_indicated = s(row["trust_num"]) or flag("trust", False) or s(row["banco"])
        if trust_indicated:
            bank = s(row["banco"])
            if not bank and pfrow is not None:
                bank = s_keep0(pfrow["bank"])
            trusts.append((str(uuid.uuid4()), pid, bank, s(row["trust_num"]),
                           as_dec(row["bfee"]), as_date(row["vence1"]), as_date(row["vence2"])))

    # Fresh rebuild (children first).
    cur.execute("SET FOREIGN_KEY_CHECKS=0")
    for t in ("property_services", "service_documents", "trust_accounts", "properties"):
        cur.execute(f"TRUNCATE TABLE {t}")
    cur.execute("SET FOREIGN_KEY_CHECKS=1")

    cur.executemany(
        "INSERT INTO properties (id,customerId,addressLine1,addressLine2,phone1,phone2,"
        "phone3,zone,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", props)
    cur.executemany(
        "INSERT INTO property_services (id,propertyId,kind,accountNumber,meterNumber,route,"
        "dueDay,active,notes) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", services)
    cur.executemany(
        "INSERT INTO trust_accounts (id,propertyId,bankName,trustNumber,bankFee,dueDate1,dueDate2)"
        " VALUES (%s,%s,%s,%s,%s,%s,%s)", trusts)
    conn.commit()

    cur.execute("SELECT COUNT(*) FROM properties"); n_p = cur.fetchone()[0]
    cur.execute("SELECT COUNT(*) FROM property_services"); n_s = cur.fetchone()[0]
    cur.execute("SELECT COUNT(*) FROM trust_accounts"); n_t = cur.fetchone()[0]
    cur.execute("SELECT kind, COUNT(*) FROM property_services GROUP BY kind ORDER BY 2 DESC")
    by_kind = cur.fetchall()
    cur.execute("SELECT COUNT(*) FROM properties p LEFT JOIN customers c ON p.customerId=c.id "
                "WHERE c.id IS NULL"); orphans = cur.fetchone()[0]

    print("=== Properties load complete ===")
    print(f"  DATMEX rows            : {len(dm)}")
    print(f"    skipped (blank/no customer) : {skipped_no_customer}")
    print(f"    PROFILE flags matched       : {matched_profile}/{len(props)}")
    print(f"  -> properties          : {n_p}")
    print(f"  -> property_services   : {n_s}")
    for k, n in by_kind:
        print(f"       {k:14} {n}")
    print(f"  -> trust_accounts      : {n_t}")
    print(f"  orphan properties (bad customer FK): {orphans}")
    assert n_p == len(props) and orphans == 0, "property load invariant failed"
    print("  validation: OK")
    conn.close()


if __name__ == "__main__":
    main()