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  -> both (near-disjoint ledgers; no folio de-dup)
  - 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

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 = env_arg()
    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

    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 = 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))

    def efectivo_like(src, name, domain, custmap, src_db, legacy_tbl):
        nonlocal skip_cust, skip_date
        df = load(src, name)
        for _, r in df.iterrows():
            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_map.get((s(r["type_of_trx"]) or "").upper())
            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"])))

    efectivo_like("stg_utilities", "efectivo", "UTILITY", util_cust, "UTILITIES", "EFECTIVO")
    efectivo_like("stg_utilities", "efectivo_backup", "UTILITY", util_cust, "UTILITIES", "EFECTIVO_BACKUP")
    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")

    # --- write ---
    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"  -> transactions        : {count('transactions')}")
    print(f"     by domain           : {dict(by_dom)}")
    for src, n in by_src:
        print(f"       {src: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()