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
from sync import delete_missing, existing_ids, parse_mode

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():
    env, sync_mode = parse_mode()
    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()}

    # Sync reuses each legacy property's existing id (keyed by provenance) so its
    # PK is stable AND the services/trust rows built below point at the right
    # parent. New legacy rows fall through to a fresh uuid.
    existing_prop = existing_ids(cur, "properties", ("legacySourceTable", "legacyId"),
                                 "WHERE legacyId IS NOT NULL") if sync_mode else {}

    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 = [], [], []
    prop_keys: set[tuple] = set()
    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

        legacy_id = str(int(row["_row_num"]))
        prop_keys.add(("DATMEX", legacy_id))
        pid = existing_prop.get(("DATMEX", legacy_id)) or 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"]), s(row["clave"]), "DATMEX", legacy_id,

        ))

        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 — the column mixes two things: an account/meter number for 160 of
        # the 334 filled rows, and a tank descriptor ("ESTACIONARIO",
        # "CILINDRO") for the rest. Only the numeric form can be matched
        # against a scanned gas statement, so it is promoted to meterNumber;
        # the descriptor stays a note, as before.
        if s(row["gas"]) or flag("gas1", False):
            gas_val = s(row["gas"])
            gas_meter = gas_val if gas_val and gas_val.isdigit() and len(gas_val) >= 5 else None
            svc("GAS", meter=gas_meter, 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"))
        # TELEPHONE — DATMEX never had a phone *service*, only the contact
        # numbers unpivoted into Property.phone1/2/3 above, even though the
        # legacy ledger billed phone as its own transaction type. A Telnor bill
        # can only be matched against a service row, so the primary number
        # becomes one. Only phone1: of 1518 properties, 534 have phone1, 18
        # phone2 and exactly 1 phone3 — the secondaries are alternate contacts,
        # not additional billed lines. Stored as the bare local number, which is
        # how DATMEX holds it and what a printed bill reduces to once the 664
        # Tijuana LADA is stripped.
        tel = s(row["telefono"])
        if tel:
            svc("TELEPHONE", account="".join(ch for ch in tel if ch.isdigit()) or None,
                notes="from DATMEX.telefono")
        # 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), or additive upsert for legacy-owned rows.
    if sync_mode:
        # Children first (scoped to legacy-owned rows so manual rows survive),
        # then upsert properties (ids already stable), then drop legacy rows
        # gone from source, then re-insert the rebuilt children.
        cur.execute("DELETE ps FROM property_services ps JOIN properties p ON p.id=ps.propertyId WHERE p.legacyId IS NOT NULL")
        cur.execute("DELETE ta FROM trust_accounts ta JOIN properties p ON p.id=ta.propertyId WHERE p.legacyId IS NOT NULL")
        cur.executemany(
            "INSERT INTO properties (id,customerId,addressLine1,addressLine2,phone1,phone2,phone3,zone,cadastralKey,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),addressLine1=VALUES(addressLine1),addressLine2=VALUES(addressLine2),phone1=VALUES(phone1),phone2=VALUES(phone2),phone3=VALUES(phone3),zone=VALUES(zone),cadastralKey=VALUES(cadastralKey),archivedAt=NULL", props)
        delete_missing(cur, "properties", ("legacySourceTable", "legacyId"), prop_keys, "WHERE legacyId IS NOT NULL")
        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)
    else:
        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,cadastralKey,legacySourceTable,legacyId) VALUES (%s,%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}")
    # n_p == len(props) is a full-load invariant; sync keeps manual rows too.
    assert (sync_mode or n_p == len(props)) and orphans == 0, "property load invariant failed"
    print("  validation: OK")
    conn.close()


if __name__ == "__main__":
    main()