jorgecuadros-platform/migration/transform_bank.py

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