Shares one module with the CLI rather than reimplementing the insert: frollo-ingest.ts holds parsing, scoping, de-duplication and the write, and both callers pass in their own SQL executor (Prisma in the route, a pg client in the script). The alternative is two implementations of the same insert, which is how the pantry healthcheck came to be fixed in one repo and left broken in the other. The route refuses rather than guesses. findAnomalies() returns every reason an unattended run should stop - a configured account contributing no rows, an unrecognised account, a near-consecutive-id collapse that might be a real repeat, a batch over ~200 rows, or an export taken with pending included - and the route answers 409 having written nothing. Two defects the wiring surfaced. Deliberately excluded credit cards were reported as unknown accounts, which would have raised the new-account anomaly on every single run and left the automatic path permanently refusing; EXCLUDED_ACCOUNTS now distinguishes excluded from unknown. And pending was tested after account scope, so pending rows on cards - which is all of them so far - classified as out-of-scope and the wrong-export-option signal could never fire; pending is now tested first.
This commit is contained in:
+75
-218
@@ -5,14 +5,18 @@
|
||||
* node --experimental-strip-types scripts/import-frollo.mts --file <csv> --apply
|
||||
*
|
||||
* Dry run by default: it prints exactly what it would do and touches nothing.
|
||||
* Rehearse first — every irreversible step in this repo that skipped that has
|
||||
* cost something, and the failure mode here is doubling reported income.
|
||||
* Rehearse first — the failure mode here is doubling reported income, and the
|
||||
* de-duplication rule that shipped second only survived because a dry run was
|
||||
* read against the salary rows.
|
||||
*
|
||||
* Export the file with PENDING TRANSACTIONS EXCLUDED. A pending row changes both
|
||||
* its id and its description when it settles, so importing one guarantees a
|
||||
* duplicate on the next run; and every pending row observed so far has been on a
|
||||
* credit card, which this importer does not cover anyway. See DECISIONS.md
|
||||
* ING-11 in the smarthome repo.
|
||||
* duplicate on the next run; every pending row observed has been on a credit
|
||||
* card, which this importer does not cover anyway. See DECISIONS.md ING-11.
|
||||
*
|
||||
* All parsing, scoping, de-duplication and insertion live in
|
||||
* `src/lib/frollo-ingest.ts`, shared with POST /api/frollo/ingest so the manual
|
||||
* and automatic paths cannot drift.
|
||||
*
|
||||
* Needs DATABASE_URL. postgres-personal publishes no host port, so from the host:
|
||||
* export DATABASE_URL="postgresql://personal:<pw>@$(docker inspect postgres-personal \
|
||||
@@ -21,239 +25,92 @@
|
||||
*/
|
||||
import { readFileSync } from "node:fs";
|
||||
import pg from "pg";
|
||||
import { parseCSVRows } from "../src/lib/csv-parser.ts";
|
||||
import {
|
||||
accountFor,
|
||||
accountReport,
|
||||
dedupe,
|
||||
readRows,
|
||||
skipReason,
|
||||
toLedgerRow,
|
||||
unknownAccounts,
|
||||
type FrolloRow,
|
||||
type SkipReason,
|
||||
} from "../src/lib/frollo-csv.ts";
|
||||
|
||||
const SOURCE = "frollo";
|
||||
import { ingestFrolloCsv, type SqlExecutor } from "../src/lib/frollo-ingest.ts";
|
||||
|
||||
function arg(name: string, fallback?: string): string | undefined {
|
||||
const i = process.argv.indexOf(`--${name}`);
|
||||
if (i >= 0 && process.argv[i + 1] && !process.argv[i + 1].startsWith("--")) {
|
||||
return process.argv[i + 1];
|
||||
}
|
||||
if (i >= 0 && process.argv[i + 1] && !process.argv[i + 1].startsWith("--")) return process.argv[i + 1];
|
||||
return fallback;
|
||||
}
|
||||
const has = (name: string) => process.argv.includes(`--${name}`);
|
||||
|
||||
const file = arg("file");
|
||||
const apply = has("apply");
|
||||
const ownerId = Number(arg("owner", "1"));
|
||||
const allowMissing = has("allow-missing-accounts");
|
||||
|
||||
if (!file) {
|
||||
console.error("usage: --file <csv> [--apply] [--owner <id>] [--allow-missing-accounts]");
|
||||
console.error("usage: --file <csv> [--apply] [--force] [--owner <id>]");
|
||||
process.exit(2);
|
||||
}
|
||||
if (!process.env.DATABASE_URL) {
|
||||
console.error("DATABASE_URL is not set — cannot compare against the ledger.");
|
||||
process.exit(2);
|
||||
}
|
||||
|
||||
const money = (n: number) => n.toLocaleString("en-AU", { minimumFractionDigits: 2, maximumFractionDigits: 2 });
|
||||
|
||||
const rows = readRows(parseCSVRows(readFileSync(file, "utf8")));
|
||||
console.log(`\n${file}`);
|
||||
console.log(` ${rows.length} rows in file\n`);
|
||||
|
||||
// ---- 1. scope ------------------------------------------------------------
|
||||
const skipped = new Map<SkipReason, FrolloRow[]>();
|
||||
const inScope: FrolloRow[] = [];
|
||||
for (const r of rows) {
|
||||
const why = skipReason(r);
|
||||
if (why) {
|
||||
const list = skipped.get(why);
|
||||
if (list) list.push(r);
|
||||
else skipped.set(why, [r]);
|
||||
} else inScope.push(r);
|
||||
}
|
||||
console.log("SCOPE");
|
||||
for (const [why, list] of [...skipped.entries()].sort((a, b) => b[1].length - a[1].length)) {
|
||||
console.log(` skipped ${String(list.length).padStart(5)} ${why}`);
|
||||
}
|
||||
console.log(` in scope ${String(inScope.length).padStart(4)}\n`);
|
||||
|
||||
if (skipped.has("pending")) {
|
||||
console.log(
|
||||
` NOTE: ${skipped.get("pending")!.length} pending row(s) present — the export was taken\n` +
|
||||
` with pending INCLUDED. They are skipped, but re-export with pending\n` +
|
||||
` excluded so successive files are comparable.\n`
|
||||
);
|
||||
}
|
||||
|
||||
// ---- 2. accounts ---------------------------------------------------------
|
||||
// A lapsed CDR consent removes an account from the export silently; the row
|
||||
// count just drops. Assert the roster rather than trusting it.
|
||||
console.log("ACCOUNTS");
|
||||
const report = accountReport(inScope);
|
||||
for (const a of report) {
|
||||
const mark = a.present ? " " : "!";
|
||||
console.log(` ${mark} ${a.spec.last4} ${a.spec.label.padEnd(24)} ${String(a.rows).padStart(5)} rows`);
|
||||
}
|
||||
const missing = report.filter((a) => !a.present);
|
||||
const unknown = unknownAccounts(rows);
|
||||
if (unknown.length > 0) {
|
||||
console.log("\n accounts in the file this importer does not know (skipped):");
|
||||
for (const u of unknown) console.log(` ${u.accountNumber} ${u.accountName} ${u.rows} rows`);
|
||||
}
|
||||
if (missing.length > 0) {
|
||||
console.log(`\n ${missing.length} configured account(s) contributed NO rows:`);
|
||||
for (const m of missing) console.log(` ${m.spec.last4} ${m.spec.label} (${m.spec.provider})`);
|
||||
console.log(" A CDR consent may have lapsed — check before treating this file as complete.");
|
||||
if (!allowMissing) {
|
||||
console.log(" Refusing to continue. Re-run with --allow-missing-accounts if this is expected.\n");
|
||||
process.exit(1);
|
||||
}
|
||||
}
|
||||
console.log();
|
||||
|
||||
// ---- 3. dedupe -----------------------------------------------------------
|
||||
const { kept, dropped } = dedupe(inScope);
|
||||
console.log("DE-DUPLICATION (re-ingest twins from CDR re-consent)");
|
||||
console.log(` ${dropped.length} dropped, ${kept.length} kept`);
|
||||
if (dropped.length > 0) {
|
||||
const byAccount = new Map<string, number>();
|
||||
for (const d of dropped) {
|
||||
const label = accountFor(d.row.accountNumber)?.label ?? "?";
|
||||
byAccount.set(label, (byAccount.get(label) ?? 0) + 1);
|
||||
}
|
||||
for (const [label, n] of [...byAccount.entries()].sort((a, b) => b[1] - a[1])) {
|
||||
console.log(` ${String(n).padStart(4)} ${label}`);
|
||||
}
|
||||
console.log(" examples:");
|
||||
for (const d of dropped.slice(0, 3)) {
|
||||
console.log(
|
||||
` ${d.row.transactionDate} ${money(d.row.amount).padStart(11)} ${d.row.description.slice(0, 34)}` +
|
||||
` id ${d.row.transactionId} duplicates ${d.duplicateOf.transactionId} (gap ${d.idGap})`
|
||||
);
|
||||
}
|
||||
// Ids this close look like a genuine same-day repeat rather than a twin.
|
||||
// Nothing in scope has ever tripped it; if something does, look before
|
||||
// trusting the collapse.
|
||||
const suspect = dropped.filter((d) => d.suspectGenuineRepeat);
|
||||
if (suspect.length > 0) {
|
||||
console.log(`\n REVIEW: ${suspect.length} collapsed row(s) had near-consecutive ids and may be`);
|
||||
console.log(" real repeats rather than duplicates:");
|
||||
for (const d of suspect.slice(0, 10)) {
|
||||
console.log(
|
||||
` ${d.row.transactionDate} ${money(d.row.amount).padStart(11)} ${d.row.description.slice(0, 40)}` +
|
||||
` ids ${d.duplicateOf.transactionId}/${d.row.transactionId} gap ${d.idGap}`
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
console.log();
|
||||
|
||||
// ---- 4. what would change ------------------------------------------------
|
||||
const ledger = kept.map(toLedgerRow);
|
||||
if (!process.env.DATABASE_URL) {
|
||||
console.error("DATABASE_URL is not set — cannot compare against the ledger.");
|
||||
process.exit(2);
|
||||
}
|
||||
const client = new pg.Client({ connectionString: process.env.DATABASE_URL });
|
||||
await client.connect();
|
||||
const exec: SqlExecutor = async <T,>(sql: string, params: unknown[]) =>
|
||||
(await client.query(sql, params)).rows as T[];
|
||||
|
||||
const existing = await client.query<{ source_ref: string }>(
|
||||
"SELECT source_ref FROM transactions WHERE source = $1",
|
||||
[SOURCE]
|
||||
);
|
||||
const seen = new Set(existing.rows.map((r) => r.source_ref));
|
||||
const fresh = ledger.filter((r) => !seen.has(r.sourceRef));
|
||||
const already = ledger.length - fresh.length;
|
||||
let report;
|
||||
try {
|
||||
report = await ingestFrolloCsv(readFileSync(file, "utf8"), exec, {
|
||||
ownerId: Number(arg("owner", "1")),
|
||||
apply: has("apply"),
|
||||
force: has("force"),
|
||||
});
|
||||
} finally {
|
||||
// Left open, the process hangs after an exception and looks like a slow query.
|
||||
if (!has("keep-open")) await client.end().catch(() => {});
|
||||
}
|
||||
|
||||
console.log(`\n${file}`);
|
||||
console.log(` ${report.totalRows} rows in file\n`);
|
||||
|
||||
console.log("SCOPE");
|
||||
for (const [why, n] of Object.entries(report.skipped).sort((a, b) => b[1]! - a[1]!)) {
|
||||
console.log(` skipped ${String(n).padStart(5)} ${why}`);
|
||||
}
|
||||
console.log(` in scope ${String(report.inScope).padStart(4)}\n`);
|
||||
|
||||
console.log("ACCOUNTS");
|
||||
for (const a of report.accounts) {
|
||||
console.log(
|
||||
` ${a.present ? " " : "!"} ${a.spec.last4} ${a.spec.label.padEnd(24)} ${String(a.rows).padStart(5)} rows`
|
||||
);
|
||||
}
|
||||
if (report.unknownAccounts.length > 0) {
|
||||
console.log("\n accounts in the file this importer does not know (skipped):");
|
||||
for (const u of report.unknownAccounts) {
|
||||
console.log(` ${u.accountNumber} ${u.accountName} ${u.rows} rows`);
|
||||
}
|
||||
}
|
||||
console.log();
|
||||
|
||||
console.log("DE-DUPLICATION (re-ingest twins from CDR re-consent)");
|
||||
console.log(` ${report.duplicatesDropped} dropped`);
|
||||
if (report.suspectRepeats.length > 0) {
|
||||
console.log(`\n REVIEW: ${report.suspectRepeats.length} collapsed row(s) had near-consecutive ids`);
|
||||
console.log(" and may be real repeats rather than duplicates:");
|
||||
for (const s of report.suspectRepeats.slice(0, 10)) {
|
||||
console.log(` ${s.transactionDate} ${money(s.amount).padStart(11)} ${s.description.slice(0, 40)} (gap ${s.idGap})`);
|
||||
}
|
||||
}
|
||||
console.log();
|
||||
|
||||
console.log("LEDGER");
|
||||
console.log(` already imported : ${String(already).padStart(5)}`);
|
||||
console.log(` new to insert : ${String(fresh.length).padStart(5)}`);
|
||||
const debits = fresh.filter((r) => r.transactionType === "debit");
|
||||
const credits = fresh.filter((r) => r.transactionType === "credit");
|
||||
const aud = (rs: typeof fresh) => rs.filter((r) => !r.foreignCurrencyCode);
|
||||
console.log(
|
||||
` of which AUD : ${aud(debits).length} debits ${money(aud(debits).reduce((s, r) => s + r.amount, 0))}` +
|
||||
` / ${aud(credits).length} credits ${money(aud(credits).reduce((s, r) => s + r.amount, 0))}`
|
||||
);
|
||||
const fx = fresh.filter((r) => r.foreignCurrencyCode);
|
||||
if (fx.length > 0) {
|
||||
const byCcy = new Map<string, number>();
|
||||
for (const r of fx) byCcy.set(r.foreignCurrencyCode!, (byCcy.get(r.foreignCurrencyCode!) ?? 0) + 1);
|
||||
console.log(` foreign : ${[...byCcy].map(([c, n]) => `${n} ${c}`).join(", ")} (amount_aud left NULL)`);
|
||||
}
|
||||
if (fresh.length > 0) {
|
||||
console.log("\n first rows to insert:");
|
||||
for (const r of fresh.slice(0, 8)) {
|
||||
console.log(
|
||||
` ${r.transactionDate} ${r.transactionType.padEnd(6)} ${money(r.amount).padStart(11)}` +
|
||||
` ${(r.foreignCurrencyCode ?? "AUD").padEnd(4)} ${r.description.slice(0, 40)}`
|
||||
);
|
||||
}
|
||||
console.log(` already imported : ${String(report.alreadyImported).padStart(5)}`);
|
||||
console.log(` new to insert : ${String(report.toInsert).padStart(5)}`);
|
||||
|
||||
if (report.anomalies.length > 0) {
|
||||
console.log("\nANOMALIES");
|
||||
for (const a of report.anomalies) console.log(` ! ${a}`);
|
||||
}
|
||||
|
||||
// ---- 5. apply ------------------------------------------------------------
|
||||
if (!apply) {
|
||||
console.log("\nDRY RUN — nothing written. Re-run with --apply to insert.\n");
|
||||
await client.end();
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
if (fresh.length === 0) {
|
||||
console.log("\nNothing to insert.\n");
|
||||
await client.end();
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
// row_index is left NULL: it exists to order statement lines, and the
|
||||
// uniqueness this import relies on is uq_transaction_source_ref, not
|
||||
// uq_transaction_identity.
|
||||
//
|
||||
// category is not set, because Frollo's own categories are not trustworthy. Note
|
||||
// what actually lands: trg_transactions_normalize_category rewrites NULL to
|
||||
// 'other' on insert, so these rows arrive categorised 'other' rather than
|
||||
// uncategorised. That is equivalent for display and for the rules engine, which
|
||||
// writes transaction_overrides.category_override and is read ahead of
|
||||
// t.category by EFFECTIVE_CATEGORY — so a rule still wins.
|
||||
let inserted = 0;
|
||||
try {
|
||||
await client.query("BEGIN");
|
||||
for (const r of fresh) {
|
||||
const res = await client.query(
|
||||
`INSERT INTO transactions
|
||||
(statement_id, owner_id, transaction_date, description, amount, transaction_type,
|
||||
merchant_name, foreign_currency_amount, foreign_currency_code,
|
||||
source, source_ref, source_account)
|
||||
VALUES (NULL, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
|
||||
ON CONFLICT (source, source_ref) WHERE source IS NOT NULL AND source_ref IS NOT NULL
|
||||
DO NOTHING`,
|
||||
[
|
||||
ownerId,
|
||||
r.transactionDate,
|
||||
r.description,
|
||||
r.amount,
|
||||
r.transactionType,
|
||||
r.merchantName,
|
||||
r.foreignCurrencyAmount,
|
||||
r.foreignCurrencyCode,
|
||||
SOURCE,
|
||||
r.sourceRef,
|
||||
r.sourceAccount,
|
||||
]
|
||||
);
|
||||
inserted += res.rowCount ?? 0;
|
||||
}
|
||||
await client.query("COMMIT");
|
||||
} catch (e) {
|
||||
await client.query("ROLLBACK");
|
||||
console.error("\nROLLED BACK:", e);
|
||||
await client.end();
|
||||
if (report.applied) {
|
||||
console.log(`\nInserted ${report.inserted} row(s).\n`);
|
||||
} else if (report.anomalies.length > 0 && has("apply")) {
|
||||
console.log("\nNOT APPLIED — anomalies above. Read them, then re-run with --force.\n");
|
||||
process.exit(1);
|
||||
} else {
|
||||
console.log("\nDRY RUN — nothing written. Re-run with --apply to insert.\n");
|
||||
}
|
||||
|
||||
const after = await client.query<{ n: string }>(
|
||||
"SELECT count(*) AS n FROM transactions WHERE source = $1",
|
||||
[SOURCE]
|
||||
);
|
||||
console.log(`\nInserted ${inserted} row(s). ${after.rows[0].n} Frollo rows now in the ledger.\n`);
|
||||
await client.end();
|
||||
|
||||
Reference in New Issue
Block a user