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. Every table went through
the same mdb-export path, so identical source values serialize identically —
string comparison on a chosen key is a valid identity test.
"""

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 keyset(df: pd.DataFrame, cols: list[str]) -> set[str]:
    return set(df[cols].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:** none of the three suspected \"duplicate\" groups are what "
      "the plan assumed. They are disjoint historical/period data or a "
      "mislabeled batch — 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)
    folio_conflict = int((m.loc[ci, biz] != n.loc[ci, biz]).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.")
    b, oa, ob = overlap(ef, efb, biz)
    p(f"- On the real business key `{biz}`: **{b} rows in both**, {oa} only in "
      f"EFECTIVO, {ob} only in BACKUP.")
    p("- Date ranges are different eras: BACKUP is dominated by 2017–2022 "
      "records, EFECTIVO by recent ones.")
    p("")
    p("**Decided rule:** the two tables are **near-disjoint ledgers**, not a "
      "live/backup duplicate pair (only " + str(b) + " shared payments out of "
      "~13k+12k). Migrate **both** into `transactions`, each row keyed "
      "internally by `(legacy_source_table, folio)` provenance — do **not** "
      "de-dup on `folio` (it collides) and do **not** drop BACKUP (it holds "
      "~12k older payments absent from EFECTIVO). The " + str(b) + " business-"
      "key matches are the only possible double-counts and should be spot-"
      "checked, but at that volume they don't threaten balance integrity.")
    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()