From a680ad2bb02109f46e3b72f786c79681454f43c2 Mon Sep 17 00:00:00 2001 From: Ricardo Mancinas Date: Wed, 22 Jul 2026 18:29:31 -0700 Subject: [PATCH] Transform+load: properties/services/trust + env-parameterize migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit migration/transform_properties.py loads properties, property_services and trust_accounts from staged DATMEX/PROFILE, resolving each property's customer FK through customer_legacy_refs. Services are derived from DATMEX's own account/route/meter fields (the authoritative data); PROFILE flags — merged best-effort on (numer_id,casa,direccion), which matched 1519/1519 — only refine each service's `active`. Trust accounts are 1:1 from DATMEX trust fields; TRUSTVENCE (overlapping) deferred to reconciliation; blobs are step 4. Loaded/validated (dev): 1519 properties (0 orphans, 1 blank id skipped), 3486 services (ELECTRIC 1118 / PROPERTY_TAX 939 / WATER 859 / GAS 335 / OTHER 115 / FEDERAL_ZONE 76 / CABLE 41 / ALARM 3), 553 trust accounts — counts track the PROFILE enrollment flags. Reproducibility (asked: dev must be redoable in prod): - migration/dbenv.py: single DB-target source = deploy/.env.'s DATABASE_URL; connect(env) + env_arg() (--env, default dev). - transform_customers.py / transform_properties.py now take --env instead of hardcoding .env.dev. - migration/run_all.py: runs every step in dependency order against --env (optional --stage re-extracts from Access first). Reproducing dev->prod is `run_all.py --env prod` after deploying the prod stack + prisma db push. All steps are idempotent (truncate+rebuild); re-run yields identical counts. Co-Authored-By: Claude Opus 4.8 --- migration/dbenv.py | 72 +++++++++ migration/run_all.py | 63 ++++++++ migration/transform_customers.py | 16 +- migration/transform_properties.py | 243 ++++++++++++++++++++++++++++++ 4 files changed, 383 insertions(+), 11 deletions(-) create mode 100644 migration/dbenv.py create mode 100644 migration/run_all.py create mode 100644 migration/transform_properties.py diff --git a/migration/dbenv.py b/migration/dbenv.py new file mode 100644 index 0000000..691c9b1 --- /dev/null +++ b/migration/dbenv.py @@ -0,0 +1,72 @@ +""" +Environment selection for the migration scripts. + +Every transform is environment-agnostic: it reads the staged Parquet (same for +all environments) and writes to whichever database `--env` selects. The target +is defined entirely by `deploy/.env.` (the same file Portainer is fed at +deploy time), whose `DATABASE_URL` is the single source of truth for host / +port / credentials / database. + +Reproduce the whole migration in a new environment (e.g. prod) by: + 1. deploy the DB stack for that env (deploy/jorgecuadros-db.stack.yml) + 2. write deploy/.env. with its DATABASE_URL + 3. push the schema: DATABASE_URL=... npx prisma@5 db push --schema=packages/database/prisma/schema.prisma + 4. run: ./.venv/bin/python run_all.py --env + +Usage in a script: + from dbenv import connect, env_arg + env = env_arg() # --env dev|prod (default dev) + conn = connect(env) +""" + +from __future__ import annotations + +import argparse +from pathlib import Path +from urllib.parse import unquote, urlparse + +import pymysql + +REPO = Path(__file__).resolve().parents[1] + + +def load_env(env: str) -> dict: + f = REPO / "deploy" / f".env.{env}" + if not f.exists(): + raise SystemExit( + f"missing {f} — deploy the '{env}' DB stack and write its .env first " + f"(see dbenv.py header)." + ) + out = {} + for line in f.read_text().splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + k, v = line.split("=", 1) + out[k] = v + if "DATABASE_URL" not in out: + raise SystemExit(f"{f} has no DATABASE_URL") + return out + + +def connect(env: str): + url = load_env(env)["DATABASE_URL"] + u = urlparse(url) # mysql://user:pass@host:port/db + return pymysql.connect( + host=u.hostname, + port=u.port or 3306, + user=unquote(u.username or ""), + password=unquote(u.password or ""), + database=(u.path or "/").lstrip("/"), + autocommit=False, + charset="utf8mb4", + ) + + +def env_arg(extra_args=None) -> str: + """Parse --env (default 'dev') and return it. Scripts that need more args + can pass an argparse parser via extra_args(parser).""" + p = argparse.ArgumentParser() + p.add_argument("--env", default="dev", help="target environment (dev|prod|...): reads deploy/.env.") + if extra_args: + extra_args(p) + return p.parse_args().env diff --git a/migration/run_all.py b/migration/run_all.py new file mode 100644 index 0000000..62740b1 --- /dev/null +++ b/migration/run_all.py @@ -0,0 +1,63 @@ +""" +Run the full data migration against one environment, in dependency order. + +Every step is idempotent (truncate + rebuild), so this is safe to re-run. The +target DB is chosen with --env (reads deploy/.env.); the same staged +Parquet feeds every environment. + +Prerequisites (once per environment, NOT done here): + 1. DB stack deployed (deploy/jorgecuadros-db.stack.yml) and deploy/.env. written. + 2. Prisma schema pushed to it: + DATABASE_URL="" \ + npx prisma@5 db push --schema=packages/database/prisma/schema.prisma + +Then: + ./.venv/bin/python run_all.py --env dev # data only (staging already present) + ./.venv/bin/python run_all.py --env prod --stage # re-extract from Access first, then load + +Reproducing dev -> prod is exactly `--env prod` (plus --stage if the staged +Parquet isn't present on the machine running it). +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys +from pathlib import Path + +HERE = Path(__file__).parent +PY = sys.executable # the venv python running this orchestrator + +# Dependency order — extend as later modules land (policies, transactions, bank). +STEPS = [ + "transform_customers.py", # customers + customer_legacy_refs (everything FKs to these) + "transform_properties.py", # properties + services + trust accounts +] + + +def run(cmd: list[str]) -> None: + print("+ " + " ".join(cmd), flush=True) + r = subprocess.run(cmd) + if r.returncode: + sys.exit(r.returncode) + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--env", default="dev", help="target environment (reads deploy/.env.)") + ap.add_argument("--stage", action="store_true", + help="re-run the raw staging load first (needs the Access files + mdbtools)") + args = ap.parse_args() + + if args.stage: + run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")]) + + for step in STEPS: + run([PY, str(HERE / step), "--env", args.env]) + + print(f"\n✓ migration complete for env={args.env}") + + +if __name__ == "__main__": + main() diff --git a/migration/transform_customers.py b/migration/transform_customers.py index c059b77..591a24e 100644 --- a/migration/transform_customers.py +++ b/migration/transform_customers.py @@ -31,9 +31,9 @@ from decimal import Decimal, InvalidOperation from pathlib import Path import pandas as pd -import pymysql -REPO = Path(__file__).resolve().parents[1] +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 @@ -161,15 +161,9 @@ _CUST_COLS = [ def main() -> None: - envf = REPO / "deploy" / ".env.dev" - env = dict( - l.strip().split("=", 1) for l in envf.read_text().splitlines() - if "=" in l and not l.startswith("#") and not l.startswith("DATABASE_URL") - ) - conn = pymysql.connect( - host="192.168.4.212", port=int(env["MYSQL_PORT"]), user=env["MYSQL_USER"], - password=env["MYSQL_PASSWORD"], database=env["MYSQL_DATABASE"], autocommit=False, - ) + env = env_arg() + conn = connect(env) + print(f"[customers] target env: {env}") cur = conn.cursor() # Fresh, idempotent rebuild. diff --git a/migration/transform_properties.py b/migration/transform_properties.py new file mode 100644 index 0000000..46be9da --- /dev/null +++ b/migration/transform_properties.py @@ -0,0 +1,243 @@ +""" +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()