rmancinasandClaude Opus 5.5
2cf07ed5ee
DATMEX.MEDIDOR/RUTA were unpivoted onto the WATER row, so every electric service arrived without a meter and water carried CFE's. They are filled on exactly the 1113 rows that carry an RPU (100% overlap), the meters are CFE's alphanumeric form (53MT58), and every legacy query that reads them is a LUZ one. medidor2/ruta2 pair with rpu2 the same way. Takes effect on the next reimport. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
284 lines
13 KiB
Python
284 lines
13 KiB
Python
"""
|
|
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 — the CESPT account only. DATMEX has no water meter column.
|
|
if s(row["agua"]) or flag("water1", False):
|
|
svc("WATER", account=s(row["agua"]), due=s(row["diagua"]),
|
|
active=flag("water1"))
|
|
# ELECTRIC — rpu account + the CFE meter and route beside it. MEDIDOR/
|
|
# RUTA are the electric meter, not water: they are filled on exactly
|
|
# the 1113 rows that carry an RPU (100% overlap), the meters are CFE's
|
|
# alphanumeric `53MT58` form, and every legacy query that reads them is
|
|
# a LUZ one (LUZ, FALTANTES DE LUZ, "luz ruta 325 y 329"). They used to
|
|
# land on the WATER row, leaving every electric service without a meter.
|
|
# Paired by suffix: medidor2/ruta2 belong to rpu2. luz_tipo as note.
|
|
for rc, mc, rtc in (("rpu", "medidor", "ruta"), ("rpu2", "medidor2", "ruta2"),
|
|
("rpu3", "medidor3", "ruta3")):
|
|
if s(row[rc]) or s(row[mc]):
|
|
svc("ELECTRIC", account=s(row[rc]), meter=s(row[mc]), route=s(row[rtc]),
|
|
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()
|