feat(migration): import prior periods so a closed year can be shown on its own

Legacy ran a year-end corte: it summed the closing year, wrote that total back
as each customer's Jan-1 BALANCE FORWARD, and started the next year clean. The
platform inherited those opening rows but never the years behind them — only
the current year's charges were ever staged, so every prior year held receipts
and no bills. Rendering one would have shown a customer credits with nothing
owed against them, which is worse than showing nothing.

The archives are whole-database Access snapshots named for the period they
hold, so `2025.accdb` is discovered by filename, staged, and loaded through the
existing DATOS2 branch — a snapshot's `datos2` is the identical shape, one year
older. Only the ledger and DATGRAL come out of a snapshot; everything else in
it is a year-stale copy of a live table. The cash side is deliberately left
behind: EFECTIVO is a lifetime journal, so the snapshot's copy is a subset of
the live one and importing it would double-book every prior-year receipt.

Period travels with the row, in legacySourceTable as `datos2@2025`. That tag,
not the date, is what a year view should filter on: the archives are not
cleanly bounded (2025 carries ten undated rows and two dated into 2026) and
legacy never filtered by date either — its reader is `SELECT ... FROM \`2025\``.
The tag also keeps legacyId safe, since it is a positional ordinal that
restarts at 0 in every archive and would otherwise collide row-for-row.

Two guards, because attaching a prior year by NUMid is the one thing here that
can go quietly wrong:

  - Reissued numbers are skipped, not imported. Comparing each archive's
    DATGRAL against the live one, 13 names moved since 2025 and 40 since 2024;
    most are the same customer re-described, but a few are a different
    household holding a recycled number, and filing their ledger under the new
    owner would show a stranger's charges. Sharing any word of three or more
    characters separates a rename from a reissue. Names are compared
    legacy-to-legacy: `customers.name` has been through blank-name recovery,
    and comparing to it reported 121 drifts where there are 13.
  - Every period is checked against the corte identity it must satisfy —
    SUM(year N) == BALANCE FORWARD(N+1) — and the result is reported per year.
    A truncated export, a file dropped under the wrong year, or a botched
    customer match all fail loudly here. 2025 reconciles 1,159/1,167 (99.3%)
    and 2024 1,144/1,156 (99.0%); the recycle guard raised 2024 from 98.2%.

Balances are untouched: BALANCE_FLOOR_JOIN floors on the newest BALANCE FORWARD
per customer, so rows behind it are already excluded from every balance read.

Uploads go through the existing ingest endpoint, allowlisted by an anchored
`AAAA.accdb` pattern that also keeps a caller-supplied name inside the ingest
directory. The Operaciones page grows an entry point for an archive that has no
row yet, reading the period off the chosen file's own name.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

migration/config.py

@@ -18,6 +18,7 @@ the four Access source files.
 """
 
 import os
+import re
 from pathlib import Path
 
 # The folder holding the four Access source files. Overridable via INGEST_DIR so
@@ -95,3 +96,59 @@ SOURCES = {
         },
     },
 }
+
+
+# --- prior-period archives ----------------------------------------------
+#
+# Legacy ran a year-end *corte*: it summed the closing year, wrote that total
+# back as each customer's Jan-1 BALANCE FORWARD, and started the next year
+# clean. Access keeps the closed year as a whole-database snapshot named for
+# the period it holds — `2025.accdb` is UTILITIES as it stood when 2025 was
+# cut — and the office archives one per year.
+#
+# Only the ledger is staged out of a snapshot. Everything else in it (DATMEX,
+# PROFILE, EFECTIVO, ...) is a year-stale copy of a table the live
+# UTILITIES.accdb already provides, and staging all ~50 of them would triple
+# the extract time to import data we would then have to ignore. DATGRAL comes
+# along solely to check that a NUMid still means the same customer it did that
+# year; see the recycle guard in transform_transactions.py.
+#
+# The cash side is deliberately NOT taken from the snapshot: `EFECTIVO` is a
+# lifetime journal, so the snapshot's copy is a subset of the live one and
+# importing it would double-book every prior-year receipt.
+PERIOD_FILE_RE = re.compile(r"^(\d{4})\.accdb$", re.IGNORECASE)
+PERIOD_TABLES = {"datos2", "DATGRAL"}
+
+
+def period_schema(year: int) -> str:
+    return f"stg_period_{year}"
+
+
+def discover_periods(root: Path) -> dict[str, dict]:
+    """Find every `YYYY.accdb` archive sitting in the ingest folder.
+
+    Discovery is by filename because that is the whole upload contract: the
+    operator drops `2025.accdb` on the Operaciones page and the period is 2025.
+    Nothing inside the file names the year — a snapshot's `datos2` looks
+    identical to the live one — so the name is the only declaration of intent
+    we get, and it is what the allowlist on the upload endpoint enforces.
+    """
+    found: dict[str, dict] = {}
+    if not root.is_dir():
+        return found
+    for path in sorted(root.iterdir()):
+        m = PERIOD_FILE_RE.match(path.name)
+        if not m:
+            continue
+        year = int(m.group(1))
+        found[f"period_{year}"] = {
+            "path": path,
+            "schema": period_schema(year),
+            "exclude": set(),
+            "include": set(PERIOD_TABLES),
+            "period_year": year,
+        }
+    return found
+
+
+SOURCES.update(discover_periods(SOURCE_ROOT))

migration/load_staging.py

@@ -40,6 +40,17 @@ def stage_source(source_name: str, source_cfg: dict, sink) -> None:
     tables = extract.list_tables(cnxn)
     excluded = source_cfg["exclude"]
 
+    # A source may name the only tables it is worth staging. Prior-period
+    # archives do: they are whole-database snapshots, but everything in them
+    # except the ledger is a year-stale copy of a live table, so staging the
+    # rest costs minutes per file to produce data nothing reads.
+    include = source_cfg.get("include")
+    if include is not None:
+        missing = include - set(tables)
+        if missing:
+            print(f"  [WARN] {source_name}: missing expected table(s) {sorted(missing)}", file=sys.stderr)
+        tables = [t for t in tables if t in include]
+
     for table_name in tables:
         if table_name in excluded:
             print(f"  [exclude] {table_name}")

migration/transform_transactions.py

@@ -90,6 +90,62 @@ def load(src, name):
     return df
 
 
+def verify_corte(c, year: int) -> None:
+    """Assert legacy's corte identity: SUM(period Y) == BALANCE FORWARD(Y+1).
+
+    This is the whole reason a year can be shown on its own. Legacy closed each
+    year by summing it and writing that total back as every customer's Jan-1
+    opening row for the next one, so if an archive is the right file, complete,
+    and attached to the right customers, its per-customer total lands exactly on
+    the next year's BALANCE FORWARD. A truncated export, a file dropped under
+    the wrong year, or a botched customer match all break the identity loudly
+    here instead of quietly six months from now.
+
+    Reported, never fatal. Legacy publishes on its own schedule, so a handful of
+    customers legitimately drift between the snapshot and the cut — the run that
+    established this reconciled 1,160 of 1,170.
+    """
+    c.execute(
+        """
+        SELECT sums.customerId, sums.total, bf.amount
+        FROM (
+            SELECT customerId, ROUND(SUM(amount), 2) AS total
+            FROM transactions
+            WHERE legacySourceTable = %s AND voidedAt IS NULL
+            GROUP BY customerId
+        ) sums
+        LEFT JOIN (
+            SELECT t.customerId, ROUND(SUM(t.amount), 2) AS amount
+            FROM transactions t
+            JOIN type_transactions tt ON tt.id = t.typeId
+            WHERE tt.nameEn = 'BALANCE FORWARD' AND t.voidedAt IS NULL
+              AND t.transactionDate >= %s AND t.transactionDate < %s
+            GROUP BY t.customerId
+        ) bf ON bf.customerId = sums.customerId
+        """,
+        (f"datos2@{year}", f"{year + 1}-01-01", f"{year + 1}-01-02"),
+    )
+    rows = c.fetchall()
+    matched = mismatched = 0
+    missing = 0
+    drift = Decimal(0)
+    for _cid, total, amount in rows:
+        if amount is None:
+            missing += 1
+            continue
+        if abs(Decimal(str(total)) - Decimal(str(amount))) < Decimal("0.02"):
+            matched += 1
+        else:
+            mismatched += 1
+            drift += abs(Decimal(str(total)) - Decimal(str(amount)))
+    checked = matched + mismatched
+    pct = (100 * matched / checked) if checked else 0
+    print(
+        f"  corte {year} -> BF {year + 1}: {matched}/{checked} match ({pct:.1f}%)"
+        f", {mismatched} off by {drift:,.2f}, {missing} with no BF row"
+    )
+
+
 def main():
     env, sync_mode = parse_mode()
     conn = connect(env)
@@ -230,9 +286,13 @@ def main():
                 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):
+    def billing(name, legacy_tbl, *, src="stg_utilities", skip_numids=None):
         """Load a DATOS2-shaped billing ledger.
 
+        `src` names the staging schema, so a prior-period archive
+        (stg_period_2025) loads through this same path: the snapshot's `datos2`
+        is the identical eleven-column shape, one year older.
+
         NOPAGO is the legacy "still owed" flag. The website reads it directly —
         `account.statement.php` splits the statement on `NOPAGO = 0` vs
         `NOPAGO = 1` and renders the latter as the "Outstanding Bills Requiring
@@ -241,10 +301,13 @@ def main():
         Only these three tables carry it (76 rows set in DATOS2 today); the
         EFECTIVO/FM3 cash streams have no such column and stay 0.
         """
-        nonlocal skip_cust, skip_date
-        df = load("stg_utilities", name)
+        nonlocal skip_cust, skip_date, skip_recycled
+        df = load(src, name)
         for _, r in df.iterrows():
-            cid = util_cust.get(norm_id(r["numid"]))
+            numid = norm_id(r["numid"])
+            if skip_numids and numid in skip_numids:
+                skip_recycled += 1; continue
+            cid = util_cust.get(numid)
             if not cid:
                 skip_cust += 1; continue
             td = dt(r["date"])
@@ -257,6 +320,53 @@ def main():
                 legacy=str(int(r["_row_num"])),
                 outstanding=1 if s(r["nopago"]) == "1" else 0)
 
+    skip_recycled = 0
+    recycle_report: list[tuple[int, str, str, str]] = []
+
+    def period_numid_guard(year: int) -> set[str]:
+        """NUMids whose prior-period owner is not today's customer.
+
+        Prior-period rows attach by NUMid and nothing else, so a number the
+        office retired and reissued would file one customer's ledger under
+        another's name — the one error this feature must never make, because it
+        shows a stranger's charges to whoever holds the number now.
+
+        Reuse is real but rare: comparing each archive's DATGRAL against the
+        live one, 13 names moved since 2025 and 40 since 2024. Most are the same
+        customer re-described — a typo fixed (VIKIE -> VICKIE), a spouse added
+        or dropped (STEWART, ALAN R. -> STEWART, ALAN & JENNIFER). A few are
+        genuinely a different household (STRONKS, BOB -> SWEET, DONALD E.).
+
+        Sharing any word of three or more characters separates the two cleanly:
+        a rename keeps the surname, a reissue keeps nothing. Names are compared
+        legacy-to-legacy, archive DATGRAL against live DATGRAL, deliberately not
+        against `customers.name` — that column has been through the blank-name
+        recovery pass, and comparing to it reported 121 drifts where there are
+        13, every extra one a false positive that would have discarded good
+        history.
+        """
+        try:
+            arch = load(f"stg_period_{year}", "datgral")
+            live = load("stg_utilities", "datgral")
+        except (FileNotFoundError, OSError):
+            return set()
+
+        def toks(v) -> set[str]:
+            return {w for w in "".join(ch if ch.isalnum() else " " for ch in (s(v) or "").upper()).split() if len(w) >= 3}
+
+        live_names = {norm_id(r["num_id"]): s(r["nombre"]) for _, r in live.iterrows()}
+        blocked: set[str] = set()
+        for _, r in arch.iterrows():
+            numid = norm_id(r["num_id"])
+            was, now = s(r["nombre"]), live_names.get(numid)
+            if not numid or not was or not now:
+                continue
+            if toks(was) & toks(now):
+                continue
+            blocked.add(numid)
+            recycle_report.append((year, numid, was, now))
+        return blocked
+
     def iva():
         nonlocal skip_cust
         df = load("stg_utilities", "iva_2015")
@@ -286,6 +396,31 @@ def main():
     billing("fee_anual", "FEE ANUAL")
     billing("fee15", "fee15")
     iva()
+
+    # --- prior periods -----------------------------------------------------
+    #
+    # Legacy kept each closed year in its own table and opened the next one with
+    # a Jan-1 BALANCE FORWARD carrying the closing total. The platform has one
+    # `transactions` table, so the period a row belongs to has to travel with
+    # the row: it rides in legacySourceTable as `datos2@2025`.
+    #
+    # That tag, not the date, is what a year view should filter on. The archives
+    # are not cleanly bounded — 2025's ledger carries ten undated rows and two
+    # dated into 2026 — and legacy itself never filtered by date either: its
+    # reader is `SELECT ... FROM \`2025\``. Keying on provenance reproduces the
+    # legacy period exactly and strands nothing.
+    #
+    # The tag also keeps the unique key safe. legacyId is a positional row
+    # ordinal, so every archive restarts it at 0 and would collide with the live
+    # `datos2` row-for-row if they shared a source-table name.
+    periods = sorted(
+        int(d.name.rsplit("_", 1)[1])
+        for d in STG.glob("stg_period_*")
+        if d.is_dir() and d.name.rsplit("_", 1)[1].isdigit()
+    )
+    for year in periods:
+        billing("datos2", f"datos2@{year}", src=f"stg_period_{year}",
+                skip_numids=period_numid_guard(year))
     # Same record shape in the seguros DB. Labelled for consistency in the
     # platform's own UI; unverifiable against the site, which only ever reads
     # domain='UTILITY', so no customer-facing behaviour depends on it.
@@ -334,6 +469,7 @@ def main():
     print(f"  skipped (unresolved customer): {skip_cust}")
     print(f"  skipped (unparseable date)   : {skip_date}")
     print(f"  skipped (EFECTIVO_BACKUP dup): {skip_dupe}")
+    print(f"  skipped (reissued NUMid)     : {skip_recycled}")
     print(f"  -> transactions        : {count('transactions')}")
     print(f"     by domain           : {dict(by_dom)}")
     for src, n in by_src:
@@ -342,6 +478,17 @@ def main():
     print(f"  -> exchange_rates      : {count('exchange_rates')}")
     print(f"  orphan transactions (bad customer FK): {orphans}")
     assert orphans == 0, "transaction customer FK invariant failed"
+
+    if recycle_report:
+        print(f"  ! reissued NUMids, prior-period rows NOT imported: {len(recycle_report)}")
+        for year, numid, was, now in recycle_report[:8]:
+            print(f"      {year} NUMid {numid}: '{was}' -> '{now}'")
+        if len(recycle_report) > 8:
+            print(f"      ... {len(recycle_report) - 8} more")
+
+    for year in periods:
+        verify_corte(c, year)
+
     print("  validation: OK")
     conn.close()