jorgecuadros-platform/migration/transform_transactions.py

"""
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):
        """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_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()
    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:
        # 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()