""" Migration plan step 3 (shared ledger): unify every cash/billing ledger into `transactions`, plus the `type_transactions` and `exchange_rates` lookups. Runs AFTER transform_customers.py (customer FK is required). Union rules come from the reconciliation pass (RECONCILIATION.md): - EFECTIVO + EFECTIVO_BACKUP -> EFECTIVO in full, plus only the BACKUP rows whose business key is absent from EFECTIVO (BACKUP is a stale copy: 12386 of its 12387 rows are verbatim duplicates). De-dup on the business key, never on folio (folio collides across the two tables). - datos2 + FEE ANUAL + fee15 -> all three (disjoint period runs; no de-dup) - EFECTIVO FM3 / CHEQUE FM3 -> distinct fee stream (amount = fee+tax+multa) - IVA 2015 -> its own snapshot (no date column -> nominal) - insurance EFECTIVO -> domain INSURANCE (customer via insurance refs) Every row keeps (legacySourceDb, legacySourceTable, legacyId) provenance, so the union is traceable and idempotent (truncate + rebuild). Customer FK is resolved through customer_legacy_refs: utilities ledgers by the utilities num_id (cl/numid/num_id), insurance EFECTIVO by the insurance num_id. Rows whose customer id or transaction date can't be resolved are skipped and counted (both are required columns). Run: ./.venv/bin/python transform_transactions.py --env dev """ from __future__ import annotations import uuid from datetime import datetime from decimal import Decimal, InvalidOperation from pathlib import Path import pandas as pd from dbenv import connect, env_arg from sync import parse_mode STG = Path(__file__).parent / "output" NULL = "∅" IVA_NOMINAL_DATE = datetime(2015, 12, 31) def s(v): if v is None or pd.isna(v): return None v = str(v).strip() return None if v in ("", NULL, "0000-00-00") else v def norm_id(v): v = s(v) if v is None: return None return v[:-2] if v.endswith(".0") else v def dec(v, default=None): v = s(v) if v is None: return default try: return Decimal(v.replace(",", "")) except (InvalidOperation, ValueError): return default def dt(v): v = s(v) if v is None: return None d = pd.to_datetime(v, errors="coerce") return None if pd.isna(d) else d.to_pydatetime() def cur(v): v = (s(v) or "").upper() if v.startswith("DOL") or "DOLLAR" in v or v.startswith("USD"): return "USD" return "MXN" def load(src, name): df = pd.read_parquet(STG / src / f"{name}.parquet").sort_values("_row_num").reset_index(drop=True) df = df[[c for c in df.columns if c != "_legacy_source_table"]].copy() for c in df.columns: if c != "_row_num": df[c] = df[c].astype("string").str.strip() return df def main(): env, sync_mode = parse_mode() conn = connect(env) print(f"[transactions] target env: {env}") c = conn.cursor() c.execute("SELECT sourceSystem, legacyId, customerId FROM customer_legacy_refs") util_cust, ins_cust = {}, {} for sys_, lid, cid in c.fetchall(): (util_cust if sys_ == "utilities" else ins_cust)[lid] = cid # --- lookups: type_transactions + exchange_rates --- tt = load("stg_utilities", "type_of_trx") type_rows, type_map = [], {} for _, r in tt.iterrows(): en = s(r["type_of_trx"]) if not en: continue tid = str(uuid.uuid4()) type_rows.append((tid, en, s(r["espa_ol"]), 0)) type_map[en.upper()] = tid xr = load("stg_utilities", "tipo_hist") xr_rows = [] for _, r in xr.iterrows(): rate = dec(r["tipo_de_cambio"]) d = dt(r["date"]) if rate is None or d is None: continue xr_rows.append((str(uuid.uuid4()), rate, d, dt(r["hour"]))) # --- transactions --- tx = [] skip_cust = skip_date = skip_dupe = 0 def add(cid, domain, tdate, amount, currency, *, period=None, reference=None, typeid=None, check=None, message=None, src_db=None, src_tbl=None, legacy=None): tx.append((str(uuid.uuid4()), cid, domain, typeid, tdate, period, reference, amount if amount is not None else Decimal(0), currency, None, check, message, 0, src_db, src_tbl, legacy)) # Business key of a real cash payment. `folio` is deliberately excluded: it # is a per-table sequential number that collides between EFECTIVO and # EFECTIVO_BACKUP (12363 shared numbers, 12204 of them on different # payments), so it identifies nothing across tables. def biz_key(r): return ( norm_id(r["cl"]), s(r["fecha"]), dec(r["monto"], Decimal(0)), s(r["conepto"]), ) def efectivo_like(src, name, domain, custmap, src_db, legacy_tbl, *, seen=None): """Load an EFECTIVO-shaped cash ledger. `seen` (a set) makes the load de-duplicating: keys are added to it as rows load, and a row whose key is already present is skipped. That is how EFECTIVO_BACKUP contributes only its genuinely-new rows. """ nonlocal skip_cust, skip_date, skip_dupe df = load(src, name) for _, r in df.iterrows(): if seen is not None: key = biz_key(r) if key in seen: skip_dupe += 1; continue seen.add(key) cid = custmap.get(norm_id(r["cl"])) if not cid: skip_cust += 1; continue td = dt(r["fecha"]) if td is None: skip_date += 1; continue add(cid, domain, td, dec(r["monto"], Decimal(0)), cur(r["monedas"]), reference=s(r["folio"]), message=s(r["conepto"]), src_db=src_db, src_tbl=legacy_tbl, legacy=str(int(r["_row_num"]))) def fm3(name, legacy_tbl, check_col=None): nonlocal skip_cust, skip_date df = load("stg_utilities", name) for _, r in df.iterrows(): cid = util_cust.get(norm_id(r["cl"])) if not cid: skip_cust += 1; continue td = dt(r["fecha"]) if td is None: skip_date += 1; continue amt = sum((dec(r[k], Decimal(0)) for k in ("fee", "tax", "multa")), Decimal(0)) add(cid, "UTILITY", td, amt, cur(r["monedas"]), reference=s(r["folio"]), message=s(r["conepto"]), check=s(r[check_col]) if check_col else None, src_db="UTILITIES", src_tbl=legacy_tbl, legacy=str(int(r["_row_num"]))) def billing(name, legacy_tbl): nonlocal skip_cust, skip_date df = load("stg_utilities", name) for _, r in df.iterrows(): cid = util_cust.get(norm_id(r["numid"])) if not cid: skip_cust += 1; continue td = dt(r["date"]) if td is None: skip_date += 1; continue tid = type_map.get((s(r["type_of_trx"]) or "").upper()) add(cid, "UTILITY", td, dec(r["chargecredit"], Decimal(0)), "MXN", period=s(r["period"]), reference=s(r["refer"]), typeid=tid, check=s(r["cheque"]), src_db="UTILITIES", src_tbl=legacy_tbl, legacy=str(int(r["_row_num"]))) def iva(): nonlocal skip_cust df = load("stg_utilities", "iva_2015") for _, r in df.iterrows(): cid = util_cust.get(norm_id(r["num_id"])) if not cid: skip_cust += 1; continue add(cid, "UTILITY", IVA_NOMINAL_DATE, dec(r["fee"], Decimal(0)), "MXN", reference=s(r["recibo"]), message="IVA 2015", src_db="UTILITIES", src_tbl="IVA 2015", legacy=str(int(r["_row_num"]))) # Shared across both calls so BACKUP is de-duplicated against EFECTIVO — # order matters: EFECTIVO is the live table and loads first, so a collision # always resolves in its favour. cash_seen: set = set() efectivo_like("stg_utilities", "efectivo", "UTILITY", util_cust, "UTILITIES", "EFECTIVO", seen=cash_seen) efectivo_like("stg_utilities", "efectivo_backup", "UTILITY", util_cust, "UTILITIES", "EFECTIVO_BACKUP", seen=cash_seen) fm3("efectivo_fm3", "EFECTIVO FM3") fm3("cheque_fm3", "CHEQUE FM3", check_col="num_cheque") billing("datos2", "datos2") billing("fee_anual", "FEE ANUAL") billing("fee15", "fee15") iva() efectivo_like("stg_seguros", "efectivo", "INSURANCE", ins_cust, "SEGUROS 16_be", "EFECTIVO") if sync_mode: c.executemany("INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference,amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),domain=VALUES(domain),typeId=VALUES(typeId),transactionDate=VALUES(transactionDate),period=VALUES(period),reference=VALUES(reference),amount=VALUES(amount),currency=VALUES(currency),checkNumber=VALUES(checkNumber),message=VALUES(message),updatedAt=NOW(),voidedAt=NULL", tx) else: c.execute("SET FOREIGN_KEY_CHECKS=0") for t in ("transactions", "type_transactions", "exchange_rates"): c.execute(f"TRUNCATE TABLE {t}") c.execute("SET FOREIGN_KEY_CHECKS=1") c.executemany("INSERT INTO type_transactions (id,nameEn,nameEs,isService) VALUES (%s,%s,%s,%s)", type_rows) c.executemany("INSERT INTO exchange_rates (id,rate,effectiveDate,effectiveHour) VALUES (%s,%s,%s,%s)", xr_rows) c.executemany( "INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference,amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", tx) conn.commit() def count(t): c.execute(f"SELECT COUNT(*) FROM {t}"); return c.fetchone()[0] c.execute("SELECT domain, COUNT(*) FROM transactions GROUP BY domain") by_dom = c.fetchall() c.execute("SELECT legacySourceTable, COUNT(*) FROM transactions GROUP BY legacySourceTable ORDER BY 2 DESC") by_src = c.fetchall() c.execute("SELECT COUNT(*) FROM transactions t LEFT JOIN customers c ON t.customerId=c.id WHERE c.id IS NULL") orphans = c.fetchone()[0] print("=== Transactions load complete ===") print(f" skipped (unresolved customer): {skip_cust}") print(f" skipped (unparseable date) : {skip_date}") print(f" skipped (EFECTIVO_BACKUP dup): {skip_dupe}") print(f" -> transactions : {count('transactions')}") print(f" by domain : {dict(by_dom)}") for src, n in by_src: print(f" {src:16} {n}") print(f" -> type_transactions : {count('type_transactions')}") print(f" -> exchange_rates : {count('exchange_rates')}") print(f" orphan transactions (bad customer FK): {orphans}") assert orphans == 0, "transaction customer FK invariant failed" print(" validation: OK") conn.close() if __name__ == "__main__": main()