""" Migration plan step 3 (bank register): SCOTHIA.mdb -> bank_transactions + business_line_categories. This is the office's OWN operating checking account ("chequera"), deliberately separate from customer-facing `transactions` and carrying no customer FK — so it can load independently of the other steps. Sources: - DATOS I (ingresos) -> amount = +ingreso, transferred flag, cleared=operado - DATOS E (egresos) -> amount = -egreso (expenses negative), amountInWords from the spelled-out "cantidad en letra" - TABLA RAMODOS -> business_line_categories (line-of-business lookup) Bank account: SCOTHIA is the Utilities MXN chequera and nothing else — DATOS E/I carry no bank or currency column — so every row loads against the single account seeded by `backfill_bank_accounts.py`, which must have run first. `banks` / `bank_accounts` are NOT truncated here; only the movements are. (In full-rebuild mode that still clears app-captured rows on every account, the same whole-database truncate every transform in this pipeline does — use `--sync` to upsert instead.) Category link: DATOS E/I have no explicit FK to TABLA RAMODOS — the ramo is inferred from the CONCEPTO text, which is a fuzzy classification, not a stored key. So the categories are loaded but bank_transactions.categoryId is left NULL for now; a concept->ramo classifier is a later enhancement. Idempotent (rebuild the legacy rows). Run: ./.venv/bin/python transform_bank.py --env dev """ from __future__ import annotations import uuid from decimal import Decimal, InvalidOperation from pathlib import Path import pandas as pd from backfill_bank_accounts import UTILITIES_ACCOUNT from dbenv import connect, env_arg from sync import parse_mode STG = Path(__file__).parent / "output" / "stg_scothia" NULL = "∅" 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 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 truthy(v): return (s(v) or "0").lower() in {"1", "-1", "true", "si", "sí", "yes"} def load(name): df = pd.read_parquet(STG / 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"[bank] target env: {env}") c = conn.cursor() # Every SCOTHIA row belongs to the one Utilities MXN chequera. Resolved by # label rather than created here, so this script can't silently open a # second copy of the account if the backfill hasn't run. c.execute("SELECT id FROM bank_accounts WHERE label = %s", (UTILITIES_ACCOUNT,)) row = c.fetchone() if not row: raise SystemExit( f"missing bank account {UTILITIES_ACCOUNT!r} — run " f"backfill_bank_accounts.py --env {env} first" ) account_id = row[0] print(f"[bank] account: {UTILITIES_ACCOUNT} ({account_id})") # business_line_categories (dedup TABLA RAMODOS) cats, seen = [], set() for _, r in load("tabla_ramodos").iterrows(): name = s(r["ramo2"]) if name and name.upper() not in seen: seen.add(name.upper()) cats.append((str(uuid.uuid4()), name)) rows = [] skip_date = 0 def add(r, amount, income: bool): nonlocal skip_date td = dt(r["fecha"]) if td is None: skip_date += 1 return rows.append(( str(uuid.uuid4()), account_id, td, s(r["tipo"]), s(r["num"]), s(r["concepto"]), amount, None, # categoryId left NULL (see header) 1 if truthy(r["operado"]) else 0, 1 if (income and truthy(r["transferido"])) else 0, s(r["notas"]), None if income else s(r["cantidad_en_letra"]), "DATOS I" if income else "DATOS E", str(int(r["_row_num"])), )) for _, r in load("datos_i").iterrows(): add(r, dec(r["ingreso"], Decimal(0)), income=True) for _, r in load("datos_e").iterrows(): add(r, -(dec(r["egreso"], Decimal(0))), income=False) COLS = ( "id,bankAccountId,transactionDate,transactionType,reference,concept,amount," "categoryId,cleared,transferred,notes,amountInWords,legacySourceTable,legacyId" ) PLACEHOLDERS = ",".join(["%s"] * 14) if sync_mode: # bankAccountId is deliberately absent from the UPDATE clause: an # account moved by hand in the app must not be dragged back. for row in rows: c.execute(f"INSERT INTO bank_transactions ({COLS}) VALUES ({PLACEHOLDERS}) ON DUPLICATE KEY UPDATE transactionDate=VALUES(transactionDate),transactionType=VALUES(transactionType),reference=VALUES(reference),concept=VALUES(concept),amount=VALUES(amount),cleared=VALUES(cleared),transferred=VALUES(transferred),notes=VALUES(notes),amountInWords=VALUES(amountInWords),voidedAt=NULL", row) else: c.execute("SET FOREIGN_KEY_CHECKS=0") for t in ("bank_transactions", "business_line_categories"): c.execute(f"TRUNCATE TABLE {t}") c.execute("SET FOREIGN_KEY_CHECKS=1") c.executemany("INSERT INTO business_line_categories (id,name) VALUES (%s,%s)", cats) c.executemany( f"INSERT INTO bank_transactions ({COLS}) VALUES ({PLACEHOLDERS})", rows) conn.commit() def count(t): c.execute(f"SELECT COUNT(*) FROM {t}"); return c.fetchone()[0] c.execute("SELECT legacySourceTable, COUNT(*), SUM(amount) FROM bank_transactions GROUP BY legacySourceTable") by_src = c.fetchall() c.execute("SELECT SUM(amount) FROM bank_transactions") net = c.fetchone()[0] print("=== Bank register load complete ===") print(f" skipped (unparseable date): {skip_date}") print(f" -> bank_transactions : {count('bank_transactions')}") for src, n, tot in by_src: print(f" {(src or '(manual)'):10} {n:6} sum {tot}") print(f" net balance movement : {net}") # Per account, never a cross-account total: the registers are in different # currencies and summing them produces a figure that never existed. c.execute( "SELECT a.label, a.currency, COUNT(t.id), COALESCE(SUM(t.amount), 0) " "FROM bank_accounts a LEFT JOIN bank_transactions t ON t.bankAccountId = a.id " "GROUP BY a.id, a.label, a.currency ORDER BY a.label" ) for label, currency, n, total in c.fetchall(): print(f" {label:34} {currency} {n:6} neto {total}") print(f" -> business_line_categories: {count('business_line_categories')}") print(" validation: OK") conn.close() if __name__ == "__main__": main()