From 29ae9fa5bc41c6d07fac1dc6332e6c9c3c617a4e Mon Sep 17 00:00:00 2001 From: Ricardo Mancinas Date: Wed, 19 Aug 2026 15:26:58 -0700 Subject: [PATCH] feat(migration): import prior periods so a closed year can be shown on its own MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- apps/api/src/ops/ops.service.ts | 88 +++++++++++++-- apps/api/src/ops/period-file.spec.ts | 48 ++++++++ apps/web/src/app/operaciones/page.tsx | 62 ++++++++++- apps/web/src/lib/types.ts | 2 + migration/config.py | 57 ++++++++++ migration/load_staging.py | 11 ++ migration/transform_transactions.py | 155 +++++++++++++++++++++++++- 7 files changed, 405 insertions(+), 18 deletions(-) create mode 100644 apps/api/src/ops/period-file.spec.ts diff --git a/apps/api/src/ops/ops.service.ts b/apps/api/src/ops/ops.service.ts index a8407c8..4f2055d 100644 --- a/apps/api/src/ops/ops.service.ts +++ b/apps/api/src/ops/ops.service.ts @@ -31,6 +31,29 @@ export const INGEST_FILES = [ ] as const; export type IngestName = (typeof INGEST_FILES)[number]; +/** + * A prior-period archive: one Access snapshot per closed year, named for the + * period it holds. `2025.accdb` is UTILITIES as it stood when 2025 was cut. + * + * The filename is the entire declaration of the period — nothing inside the + * file names its year, because a snapshot's `datos2` is indistinguishable from + * the live one — so this pattern is both the allowlist and the contract. It is + * anchored and allows no separator, which is what keeps an upload from + * escaping the ingest directory. + */ +const PERIOD_FILE_RE = /^(\d{4})\.accdb$/i; + +/** Earliest period we will accept, so a typo'd year cannot mint a bogus one. */ +const PERIOD_MIN_YEAR = 1990; + +export function periodYearOf(name: string): number | null { + const m = PERIOD_FILE_RE.exec(name); + if (!m) return null; + const year = Number(m[1]); + if (year < PERIOD_MIN_YEAR || year > new Date().getUTCFullYear()) return null; + return year; +} + /** * Prefix for every command containing a pipe. Without it the exit status of * `mysqldump | gzip` is gzip's, so a dump that failed immediately still looks @@ -120,19 +143,30 @@ export class OpsService implements OnModuleInit { /* -------------------------------------------------------------- ingest */ - private assertIngestName(name: string): IngestName { - if (!INGEST_FILES.includes(name as IngestName)) { - throw new BadRequestException( - `Archivo no permitido. Debe ser uno de: ${INGEST_FILES.join(", ")}`, - ); - } - return name as IngestName; + private assertIngestName(name: string): string { + if (INGEST_FILES.includes(name as IngestName)) return name; + if (periodYearOf(name) !== null) return name; + throw new BadRequestException( + `Archivo no permitido. Debe ser uno de: ${INGEST_FILES.join(", ")}` + + `, o un archivo de periodo anterior con nombre AAAA.accdb (por ejemplo 2025.accdb).`, + ); } async listIngest(): Promise< - { name: string; present: boolean; size: number | null; modifiedAt: string | null }[] + { + name: string; + present: boolean; + size: number | null; + modifiedAt: string | null; + /** Set only on a prior-period archive; null on the four fixed sources. */ + periodYear: number | null; + }[] > { - return Promise.all( + // The four fixed sources are listed whether present or not — they are + // required, so "missing" is the useful state to show. Period archives are + // optional and unbounded, so they are listed only once uploaded, newest + // year first. + const fixed = await Promise.all( INGEST_FILES.map(async (name) => { try { const st = await fs.stat(path.join(this.ingestDir, name)); @@ -141,12 +175,46 @@ export class OpsService implements OnModuleInit { present: true, size: st.size, modifiedAt: st.mtime.toISOString(), + periodYear: null as number | null, }; } catch { - return { name, present: false, size: null, modifiedAt: null }; + return { + name, + present: false, + size: null, + modifiedAt: null, + periodYear: null as number | null, + }; } }), ); + + let entries: string[] = []; + try { + entries = await fs.readdir(this.ingestDir); + } catch { + entries = []; + } + const periods = ( + await Promise.all( + entries + .map((name) => ({ name, year: periodYearOf(name) })) + .filter((e): e is { name: string; year: number } => e.year !== null) + .sort((a, b) => b.year - a.year) + .map(async ({ name, year }) => { + const st = await fs.stat(path.join(this.ingestDir, name)); + return { + name, + present: true, + size: st.size, + modifiedAt: st.mtime.toISOString(), + periodYear: year, + }; + }), + ) + ).filter(Boolean); + + return [...fixed, ...periods]; } async saveIngest(name: string, data: Buffer): Promise { diff --git a/apps/api/src/ops/period-file.spec.ts b/apps/api/src/ops/period-file.spec.ts new file mode 100644 index 0000000..9037f7c --- /dev/null +++ b/apps/api/src/ops/period-file.spec.ts @@ -0,0 +1,48 @@ +import { periodYearOf } from "./ops.service"; + +/** + * `periodYearOf` is the upload allowlist for prior-period archives, so it is + * doing two jobs at once: deciding what counts as a period file, and keeping a + * caller-supplied name from escaping the ingest directory. Both are pinned here. + */ +describe("periodYearOf", () => { + it("accepts a four-digit year archive", () => { + expect(periodYearOf("2025.accdb")).toBe(2025); + expect(periodYearOf("1999.accdb")).toBe(1999); + }); + + it("is case-insensitive on the extension", () => { + expect(periodYearOf("2025.ACCDB")).toBe(2025); + }); + + it("rejects a path that would escape the ingest directory", () => { + // The name is joined onto the ingest path, so anything with a separator or + // a parent reference has to fail before it reaches the filesystem. + expect(periodYearOf("../2025.accdb")).toBeNull(); + expect(periodYearOf("../../etc/passwd")).toBeNull(); + expect(periodYearOf("sub/2025.accdb")).toBeNull(); + expect(periodYearOf("2025.accdb/../../x")).toBeNull(); + }); + + it("rejects names that only look like a period", () => { + expect(periodYearOf("202.accdb")).toBeNull(); + expect(periodYearOf("20255.accdb")).toBeNull(); + expect(periodYearOf("2025.mdb")).toBeNull(); + expect(periodYearOf("copia 2025.accdb")).toBeNull(); + expect(periodYearOf("2025.accdb.bak")).toBeNull(); + expect(periodYearOf("UTILITIES.accdb")).toBeNull(); + }); + + it("rejects years outside the plausible range", () => { + // A typo'd year would otherwise mint a period nobody can ever reconcile: + // there is no BALANCE FORWARD for the year after it to check against. + expect(periodYearOf("1889.accdb")).toBeNull(); + expect(periodYearOf(`${new Date().getUTCFullYear() + 1}.accdb`)).toBeNull(); + }); + + it("accepts the current year, which is the earliest a period can be cut", () => { + expect(periodYearOf(`${new Date().getUTCFullYear()}.accdb`)).toBe( + new Date().getUTCFullYear(), + ); + }); +}); diff --git a/apps/web/src/app/operaciones/page.tsx b/apps/web/src/app/operaciones/page.tsx index 3d4a90a..7aa4963 100644 --- a/apps/web/src/app/operaciones/page.tsx +++ b/apps/web/src/app/operaciones/page.tsx @@ -68,6 +68,7 @@ function Operaciones() { const [starting, setStarting] = useState(false); const fileInputs = useRef>({}); + const periodInput = useRef(null); const refreshLists = useCallback(() => { listIngest().then(setIngest).catch(() => setIngest([])); @@ -143,6 +144,20 @@ function Operaciones() { } } + /** Upload a prior-period archive under its own filename. */ + async function handleUploadPeriod(file: File | undefined) { + if (!file) return; + if (!/^\d{4}\.accdb$/i.test(file.name)) { + setError( + `"${file.name}" no es un archivo de periodo. Debe llamarse AAAA.accdb, por ejemplo 2025.accdb.`, + ); + if (periodInput.current) periodInput.current.value = ""; + return; + } + await handleUpload(file.name, file); + if (periodInput.current) periodInput.current.value = ""; + } + async function handleDeleteIngest(name: string) { setError(null); try { @@ -197,7 +212,11 @@ function Operaciones() { else await start("RESTORE", c.file); } - const ingestReady = (ingest ?? []).every((f) => f.present); + // Only the four fixed Access sources gate a run. Period archives are + // optional extras — having none simply means no prior years are available. + const ingestReady = (ingest ?? []) + .filter((f) => f.periodYear === null) + .every((f) => f.present); return ( <> @@ -244,8 +263,9 @@ function Operaciones() {

Carpeta de ingesta

- Los cuatro archivos originales de Access. La reimportación y la - sincronización leen de aquí. Tamaño máximo por archivo: {formatBytes(INGEST_MAX_BYTES)}. + Los cuatro archivos originales de Access, más los archivos de periodos + anteriores. La reimportación y la sincronización leen de aquí. Tamaño + máximo por archivo: {formatBytes(INGEST_MAX_BYTES)}.

@@ -262,7 +282,14 @@ function Operaciones() { {(ingest ?? []).map((f) => ( - +
{f.name} + {f.name} + {f.periodYear !== null && ( + + periodo {f.periodYear} + + )} + {f.present ? "Presente" : "Falta"} @@ -313,6 +340,33 @@ function Operaciones() {
+ + {/* A period archive that has never been uploaded has no row to click, + so it needs its own entry point. The file names itself: the archive + IS `2025.accdb`, and that name is what declares the period, so the + control reads it off the chosen file rather than asking twice. */} +
+ handleUploadPeriod(e.target.files?.[0])} + /> + + + Un archivo de Access por año cerrado, nombrado con su periodo:{" "} + 2025.accdb. Aporta el año anterior al + estado de cuenta; no altera el saldo actual. + +
{/* Operations */} diff --git a/apps/web/src/lib/types.ts b/apps/web/src/lib/types.ts index e9f0b67..f757f1b 100644 --- a/apps/web/src/lib/types.ts +++ b/apps/web/src/lib/types.ts @@ -192,6 +192,8 @@ export interface IngestFile { present: boolean; size: number | null; modifiedAt: string | null; + /** Year of a prior-period archive (`2025.accdb`); null on the four fixed sources. */ + periodYear: number | null; } export interface BackupFile { diff --git a/migration/config.py b/migration/config.py index d960459..810c174 100644 --- a/migration/config.py +++ b/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)) diff --git a/migration/load_staging.py b/migration/load_staging.py index d3fcecd..03fbeaf 100644 --- a/migration/load_staging.py +++ b/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}") diff --git a/migration/transform_transactions.py b/migration/transform_transactions.py index 4eb262b..866d28e 100644 --- a/migration/transform_transactions.py +++ b/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()