rmancinasandClaude Opus 5
1d689d8f46
The EFECTIVO ledgers have no type column — in Access the transaction type
is implied by which table a row lives in — so unlike DATOS2 there was no
string to map and typeId came out NULL on all 13,496 rows.
That is not just a blank label. handleGetAccountDetails in
my.jorgecuadros.com identifies payments by matching TYPEOFTRX against
('PAYMENT THANK YOU', 'PAYPAL', 'CASH DEPOSIT', 'CHECK DEPOSIT') to reset
the running balance in mode=current. An unlabelled payment is not
recognised, so the balance silently diverges from legacy — 285 rows across
129 customers in the current year alone.
"CASH DEPOSIT" is measured, not chosen: matching the unlabelled rows to the
live site on (NUMid, date, amount) resolves unanimously to that label —
66/66 in the current-year `datosfreak` and 100/100 in the prior-year `2025`
table, the only two periods the site allowlists.
The FM3 fee streams (EFECTIVO FM3 627, CHEQUE FM3 157) have the same
missing-type problem and are deliberately left NULL: every row predates both
exposed periods, so nothing can be matched against a legacy label and none
can reach a customer. Guessing "CHECK DEPOSIT" there would feed the
payment-detection list on no evidence.
type_id_for(None) returns None, so call sites without a label are unchanged.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
339 lines
15 KiB
Python
339 lines
15 KiB
Python
"""
|
|
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
|
|
|
|
def type_id_for(raw) -> str | None:
|
|
"""Resolve a transaction type, minting one when the lookup lacks it.
|
|
|
|
The Access `TYPE OF TRX` table is a stale pick-list, not a constraint —
|
|
staff free-text straight into DATOS2, so 78 values covering 3,939 rows
|
|
(BALANCE FORWARD 1,188, ANNUAL FEE 1,116, IZZI 367, ...) appear in the
|
|
ledger but not the lookup. Leaving those unmapped stored typeId NULL and
|
|
lost the label outright: nothing else on `transactions` carries the type
|
|
text, so the row rendered blank and was unrecoverable after migration.
|
|
Minting from the literal keeps the display string; nameEs stays NULL
|
|
because only the lookup has translations.
|
|
"""
|
|
en = s(raw)
|
|
if not en:
|
|
return None
|
|
key = en.upper()
|
|
tid = type_map.get(key)
|
|
if tid is None:
|
|
tid = str(uuid.uuid4())
|
|
type_rows.append((tid, en, None, 0))
|
|
type_map[key] = tid
|
|
return 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,
|
|
type_label=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.
|
|
|
|
`type_label` names the transaction type for every row. These tables have
|
|
no type column at all — in Access the type is implied by which table the
|
|
row lives in — so unlike DATOS2 there is no string to map and typeId came
|
|
out NULL for all of them.
|
|
|
|
That is not merely a blank label. handleGetAccountDetails in
|
|
my.jorgecuadros.com identifies payments by matching TYPEOFTRX against
|
|
('PAYMENT THANK YOU', 'PAYPAL', 'CASH DEPOSIT', 'CHECK DEPOSIT') to reset
|
|
the running balance in mode=current; an unlabelled payment is not
|
|
recognised and the balance silently diverges from legacy.
|
|
"""
|
|
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"]),
|
|
typeid=type_id_for(type_label),
|
|
src_db=src_db, src_tbl=legacy_tbl, legacy=str(int(r["_row_num"])))
|
|
|
|
def fm3(name, legacy_tbl, check_col=None):
|
|
"""FM3 fee streams. Deliberately left unlabelled, unlike EFECTIVO.
|
|
|
|
These rows (EFECTIVO FM3 627, CHEQUE FM3 157) also have no type column,
|
|
but every one of them predates the two periods the site exposes — it
|
|
allowlists only the current year and the prior year — so none can be
|
|
matched against a legacy label, and none can reach a customer. Inventing
|
|
a plausible name like "CHECK DEPOSIT" would feed the payment-detection
|
|
list in handleGetAccountDetails on nothing but a guess. Leave them NULL
|
|
until a real mapping is available.
|
|
"""
|
|
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_id_for(r["type_of_trx"])
|
|
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()
|
|
# "CASH DEPOSIT" is not a guess: matching these rows to the live site on
|
|
# (NUMid, date, amount) resolves to that label unanimously — 66/66 in the
|
|
# current-year `datosfreak` and 100/100 in the prior-year `2025` table,
|
|
# which are the only two periods the site exposes.
|
|
efectivo_like("stg_utilities", "efectivo", "UTILITY", util_cust, "UTILITIES",
|
|
"EFECTIVO", seen=cash_seen, type_label="CASH DEPOSIT")
|
|
efectivo_like("stg_utilities", "efectivo_backup", "UTILITY", util_cust, "UTILITIES",
|
|
"EFECTIVO_BACKUP", seen=cash_seen, type_label="CASH DEPOSIT")
|
|
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()
|
|
# Same record shape in the seguros DB. Labelled for consistency in the
|
|
# platform's own UI; unverifiable against the site, which only ever reads
|
|
# domain='UTILITY', so no customer-facing behaviour depends on it.
|
|
efectivo_like("stg_seguros", "efectivo", "INSURANCE", ins_cust, "SEGUROS 16_be",
|
|
"EFECTIVO", type_label="CASH DEPOSIT")
|
|
|
|
if sync_mode:
|
|
# Transaction types are rebuilt with fresh uuids each run; resolve them
|
|
# against the rows already in the DB by English name (inserting any that
|
|
# are new) and remap each tx's typeId onto the persisted id so the FK to
|
|
# type_transactions holds. exchange_rates isn't referenced by tx, so it
|
|
# is left untouched in sync.
|
|
c.execute("SELECT id,nameEn FROM type_transactions")
|
|
db_types = {(nm or "").upper(): i for i, nm in c.fetchall()}
|
|
fresh_name = {tid: (en or "").upper() for tid, en, es, active in type_rows}
|
|
new_types = []
|
|
for tid, en, es, active in type_rows:
|
|
if (en or "").upper() not in db_types:
|
|
db_types[(en or "").upper()] = tid
|
|
new_types.append((tid, en, es, active))
|
|
if new_types:
|
|
c.executemany("INSERT INTO type_transactions (id,nameEn,nameEs,isService) VALUES (%s,%s,%s,%s)", new_types)
|
|
tx = [(t[0], t[1], t[2], (db_types.get(fresh_name.get(t[3])) if t[3] else None), *t[4:]) for t in tx]
|
|
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),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 or '(manual)'):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()
|