Transform+load: properties/services/trust + env-parameterize migration

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.<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 <noreply@anthropic.com>

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.<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.<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 <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.<env>")
+    if extra_args:
+        extra_args(p)
+    return p.parse_args().env

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.<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.<env> written.
+  2. Prisma schema pushed to it:
+       DATABASE_URL="<that env's 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.<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()

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 = env_arg()
-    env = dict(
+    conn = connect(env)
-        l.strip().split("=", 1) for l in envf.read_text().splitlines()
+    print(f"[customers] target env: {env}")
-        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,
-    )
     cur = conn.cursor()
 
     # Fresh, idempotent rebuild.

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()