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