jorgecuadros-platform/migration/reconcile.py

"""
Migration plan step 2: reconciliation pass.

The legacy Access files carry several sets of near-duplicate / overlapping
tables (snapshot copies, per-period exports, filtered views). Before the
transform+load (step 3) can union them, we need to know — from the data, not
from a guess — which rows are exact duplicates across those tables and which
are genuinely distinct, so the de-dup / union rule is decided by evidence.

Reads the staged Parquet produced by load_staging.py (`--output-dir`) and
prints a Markdown reconciliation report. Regenerate with:

    ./.venv/bin/python reconcile.py > RECONCILIATION.md

Method note: naive full-row matching across these tables gives ~0 overlap,
which is a trap — it hides *why*. The interesting question is whether two
tables hold the SAME economic records under cosmetic differences, or
genuinely DIFFERENT records. So each group is probed on a deliberate
*business key* (the columns that identify the real-world thing) and the
volatile/identity columns are examined separately.

Method correction (2026-07-22): this file previously assumed that "every table
went through the same mdb-export path, so identical source values serialize
identically". That is false. `mdb-export` formats a numeric column according to
its *Access column type*, so the same amount is emitted as `5000` from one table
and `27000.0000` from another. String-comparing a numeric key column therefore
reports near-zero overlap between tables holding identical records — which is
exactly what produced the original (wrong) "EFECTIVO / EFECTIVO_BACKUP are
near-disjoint ledgers" verdict. Every key column is now canonicalized (numeric
columns parsed and re-formatted to a fixed precision) before comparison.
"""

from __future__ import annotations

from pathlib import Path

import pandas as pd

STG = Path(__file__).parent / "output" / "stg_utilities"
_META = ("_legacy_source_table", "_row_num")
_NULL = "∅"


def load(name: str) -> pd.DataFrame:
    df = pd.read_parquet(STG / f"{name}.parquet")
    df = df[[c for c in df.columns if c not in _META]].copy()
    for c in df.columns:
        df[c] = df[c].astype("string").str.strip().fillna(_NULL)
    return df


def canon(sr: pd.Series) -> pd.Series:
    """Canonicalize one key column so it compares across tables.

    A column that parses as a number in (nearly) every populated cell is
    re-emitted at fixed precision, which erases the per-table formatting
    mdb-export applies from the Access column type. Anything else (dates,
    free text) is left as the already-stripped string.
    """
    populated = sr.ne(_NULL)
    if not populated.any():
        return sr
    num = pd.to_numeric(sr.where(populated), errors="coerce")
    if num.notna().sum() < 0.9 * populated.sum():
        return sr
    return num.map(lambda v: _NULL if pd.isna(v) else f"{v:.4f}").astype("string")


def keyset(df: pd.DataFrame, cols: list[str]) -> set[str]:
    keyed = pd.DataFrame({c: canon(df[c]) for c in cols})
    return set(keyed.agg("\x1f".join, axis=1))


def overlap(a: pd.DataFrame, b: pd.DataFrame, cols: list[str]):
    ka, kb = keyset(a, cols), keyset(b, cols)
    return len(ka & kb), len(ka - kb), len(kb - ka)


def constant_cols(df: pd.DataFrame) -> list[str]:
    return [c for c in df.columns if df[c].nunique(dropna=False) <= 1]


def p(*a):
    print(*a)


def main() -> None:
    p("# Reconciliation Report — Overlapping Legacy Tables")
    p("")
    p("_Migration plan step 2. Generated by `reconcile.py` from staged Parquet "
      "(`output/stg_utilities`). Regenerate: `./.venv/bin/python reconcile.py "
      "> RECONCILIATION.md`._")
    p("")
    p("**Headline:** `EFECTIVO_BACKUP` *is* a duplicate of `EFECTIVO` and must "
      "not be double-loaded; the billing tables are genuinely disjoint period "
      "runs; `COBRO3` is a mislabeled charge batch, not a customer master. See "
      "each group's decided rule.")
    p("")

    # ------------------------------------------------------------------ #
    # Group 1: EFECTIVO vs EFECTIVO_BACKUP
    # ------------------------------------------------------------------ #
    p("## 1. Cash ledger — `EFECTIVO` vs `EFECTIVO_BACKUP`")
    p("")
    ef, efb = load("efectivo"), load("efectivo_backup")
    biz = ['cl', 'fecha', 'monto', 'conepto']  # what identifies a real payment
    p(f"- `EFECTIVO`: {len(ef)} rows. `EFECTIVO_BACKUP`: {len(efb)} rows.")
    # folio behaviour
    fe = set(ef['folio']) - {_NULL}
    fb = set(efb['folio']) - {_NULL}
    p(f"- `folio` is per-table sequential: {ef['folio'].nunique()} distinct in "
      f"EFECTIVO (= row count), {efb['folio'].nunique()} in BACKUP. "
      f"{len(fe & fb)} folio *numbers* appear in both.")
    both_folio = fe & fb
    m = ef[ef['folio'].isin(both_folio)].drop_duplicates('folio').set_index('folio')
    n = efb[efb['folio'].isin(both_folio)].drop_duplicates('folio').set_index('folio')
    ci = m.index.intersection(n.index)
    mk = pd.DataFrame({c: canon(m.loc[ci, c]) for c in biz})
    nk = pd.DataFrame({c: canon(n.loc[ci, c]) for c in biz})
    folio_conflict = int((mk != nk).any(axis=1).sum())
    p(f"- **But of those {len(ci)} shared folio numbers, {folio_conflict} carry "
      f"a *different* transaction** (differ on {biz}). → `folio` collides; it is "
      "NOT a stable cross-table id, and it cannot be the de-dup key.")
    b, oa, ob = overlap(ef, efb, biz)
    p(f"- On the real business key `{biz}`, canonicalized: **{b} rows in both**, "
      f"{oa} only in EFECTIVO, {ob} only in BACKUP — i.e. all but {ob} of "
      f"BACKUP's {len(efb)} rows already exist verbatim in EFECTIVO, same "
      "customer, same timestamp to the second, same amount, same concept text.")
    p("- Yearly row counts track each other almost exactly from 2006 to 2023 "
      "(e.g. 2017: 1036 vs 994), which is what a stale copy looks like — not "
      "two ledgers covering different eras.")
    p("")
    p("**Decided rule (corrected):** `EFECTIVO_BACKUP` is a **stale backup copy "
      "of `EFECTIVO`**, not an independent ledger. Load `EFECTIVO` in full, and "
      "load from `EFECTIVO_BACKUP` only the rows whose canonicalized business "
      f"key is absent from `EFECTIVO` ({ob} row(s)). Loading both in full "
      f"double-counts {b} payments and doubles nearly every customer's "
      "historical receipt total, which makes any statement or balance view "
      "wrong. De-dup on the business key, **not** on `folio` (it collides).")
    p("")
    p("> This reverses the original verdict in this report, which put that "
      "overlap at 2. That number came from string-comparing `monto`, which `mdb-export` "
      "serializes with a different precision per table (`5000` vs "
      "`27000.0000`) — see the module docstring.")
    p("")
    p("> `EFECTIVO FM3` (627) and `CHEQUE FM3` (157) are a separate stream — "
      "`fee`/`tax`/`multa` columns instead of `monto` — and migrate as distinct "
      "transactions, not reconciled against EFECTIVO.")
    p("")
    p("> Data-quality note for the transform: `monedas` has case/spelling "
      "variants (`PESOS`/`Pesos`, `DOLARES`/`Dolares`/`DOLLARS`) — normalize "
      "currency on load.")
    p("")

    # ------------------------------------------------------------------ #
    # Group 2: datos2 vs FEE ANUAL vs fee15
    # ------------------------------------------------------------------ #
    p("## 2. Billing/fee logs — `datos2` vs `FEE ANUAL` vs `fee15`")
    p("")
    d2, fa, f15 = load("datos2"), load("fee_anual"), load("fee15")
    p(f"- Rows: `datos2` {len(d2)}, `FEE ANUAL` {len(fa)}, `fee15` {len(f15)}.")

    def date_span(df):
        d = sorted(x for x in df['date'].unique() if x != _NULL)
        return (d[0], d[-1]) if d else ("∅", "∅")
    p(f"- `date` spans: `datos2` {date_span(d2)}, `FEE ANUAL` {date_span(fa)}, "
      f"`fee15` {date_span(f15)} — **three different periods**.")
    p(f"- `FEE ANUAL.refer` and `fee15.refer` are each a single constant value "
      f"({fa['refer'].nunique()}/{f15['refer'].nunique()} distinct) — these are "
      "one-shot per-period fee runs, not general logs.")
    idcols = ['numid', 'date', 'chargecredit']
    _, _, _ = overlap(fa, d2, idcols)
    b1 = overlap(fa, d2, idcols)[0]
    b2 = overlap(f15, d2, idcols)[0]
    b3 = overlap(fa, f15, idcols)[0]
    p(f"- Real-identity overlap `{idcols}`: FEE ANUAL∩datos2 = {b1}, "
      f"fee15∩datos2 = {b2}, FEE ANUAL∩fee15 = {b3}.")
    p("")
    p("**Decided rule:** `datos2`, `FEE ANUAL`, and `fee15` are **disjoint "
      "historical billing runs from different periods** (≈2025–26, 2018, 2017 "
      "respectively), not copies of one another. Migrate **all three** into the "
      "billing side of `transactions`, provenance-tagged; no de-dup is needed "
      "(zero real-identity overlap). Preserve `datos2.due_date` (the other two "
      "lack it — leave null for their rows).")
    p("")

    # ------------------------------------------------------------------ #
    # Group 3: DATGRAL vs COBRO3
    # ------------------------------------------------------------------ #
    p("## 3. Customer master — `DATGRAL` vs `COBRO3`")
    p("")
    dg, cb = load("datgral"), load("cobro3")
    dg_ids, cb_ids = set(dg['num_id']), set(cb['num_id'])
    p(f"- `DATGRAL` {len(dg)} rows / `COBRO3` {len(cb)} rows; both unique on "
      f"`num_id`.")
    p(f"- Every `COBRO3.num_id` is in `DATGRAL` ({len(cb_ids & dg_ids)}/"
      f"{len(cb_ids)}); it adds **no new customer id**.")
    fee_const = cb['fee'].nunique()
    fee_vals = list(cb['fee'].unique())[:3]
    p(f"- **`COBRO3.fee` is constant** ({fee_const} distinct value: {fee_vals}) "
      "while `DATGRAL.fee` varies per customer — so COBRO3 is not a snapshot of "
      "the master's fee field.")
    p("- Sampling shows COBRO3 rows carry the flat charge with names/addresses "
      "denormalized (some blank), i.e. a saved *charge worklist*, not the "
      "authoritative customer record.")
    p("")
    p("**Decided rule:** `COBRO3` is a **billing/charge batch** ('cobro' = "
      "collection), not a customer table. `DATGRAL` is the sole utilities "
      "customer master. **Do not** merge COBRO3 into `customers` or let it "
      "overwrite any master field. If the flat charge has value as history, "
      "model those 181 rows as charge `transactions` (amount = the constant "
      "fee) keyed to the existing DATGRAL customers — otherwise exclude COBRO3 "
      "from the migration entirely. Either way it contributes zero new "
      "customers.")
    p("")


if __name__ == "__main__":
    main()