""" 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"]), "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 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), 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,legacySourceTable,legacyId) VALUES (%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),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,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}") # 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()