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)

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 (truncate + rebuild). 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 dbenv import connect, env_arg

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 = env_arg()
    conn = connect(env)
    print(f"[bank] target env: {env}")
    c = conn.cursor()

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

    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(
        "INSERT INTO bank_transactions (id,transactionDate,transactionType,reference,concept,"
        "amount,categoryId,cleared,transferred,notes,amountInWords,legacySourceTable,legacyId) "
        "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", 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:10} {n:6}  sum {tot}")
    print(f"     net balance movement    : {net}")
    print(f"  -> business_line_categories: {count('business_line_categories')}")
    print("  validation: OK")
    conn.close()


if __name__ == "__main__":
    main()