From 3e6d4f8e81c11df129f9a1320badecd40d259f8a Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 8 Sep 2026 16:35:54 -0300 Subject: [PATCH] feat: add reviewed import pipeline --- docs/imports.md | 165 +++++++++ .../plans/2026-09-06-reviewed-imports.md | 48 +++ .../2026-09-06-reviewed-imports-design.md | 210 +++++++++++ ingestion/__init__.py | 0 ingestion/duplicates.py | 25 ++ ingestion/management/__init__.py | 0 ingestion/management/commands/__init__.py | 0 .../management/commands/import_review.py | 208 +++++++++++ ingestion/migrations/0001_initial.py | 58 +++ ingestion/migrations/__init__.py | 0 ingestion/models.py | 38 ++ ingestion/normalization.py | 117 ++++++ ingestion/parsers.py | 136 +++++++ ingestion/services.py | 335 ++++++++++++++++++ ingestion/validation.py | 57 +++ tests/fixtures/synthetic/ingestion/README.md | 12 + tests/fixtures/synthetic/ingestion/bank.csv | 3 + tests/fixtures/synthetic/ingestion/full.txt | 4 + .../fixtures/synthetic/ingestion/partial.txt | 3 + tests/test_import_command.py | 132 +++++++ tests/test_import_concurrency.py | 72 ++++ tests/test_import_parsers.py | 127 +++++++ tests/test_import_workflow.py | 264 ++++++++++++++ 23 files changed, 2014 insertions(+) create mode 100644 docs/imports.md create mode 100644 docs/superpowers/plans/2026-09-06-reviewed-imports.md create mode 100644 docs/superpowers/specs/2026-09-06-reviewed-imports-design.md create mode 100644 ingestion/__init__.py create mode 100644 ingestion/duplicates.py create mode 100644 ingestion/management/__init__.py create mode 100644 ingestion/management/commands/__init__.py create mode 100644 ingestion/management/commands/import_review.py create mode 100644 ingestion/migrations/0001_initial.py create mode 100644 ingestion/migrations/__init__.py create mode 100644 ingestion/models.py create mode 100644 ingestion/normalization.py create mode 100644 ingestion/parsers.py create mode 100644 ingestion/services.py create mode 100644 ingestion/validation.py create mode 100644 tests/fixtures/synthetic/ingestion/README.md create mode 100644 tests/fixtures/synthetic/ingestion/bank.csv create mode 100644 tests/fixtures/synthetic/ingestion/full.txt create mode 100644 tests/fixtures/synthetic/ingestion/partial.txt create mode 100644 tests/test_import_command.py create mode 100644 tests/test_import_concurrency.py create mode 100644 tests/test_import_parsers.py create mode 100644 tests/test_import_workflow.py diff --git a/docs/imports.md b/docs/imports.md new file mode 100644 index 0000000..c20e115 --- /dev/null +++ b/docs/imports.md @@ -0,0 +1,165 @@ +# Reviewed local imports + +Imports require a saved preview and explicit row decisions before financial +records are written. The `import_review` management command supports `preview`, +`export`, `apply`, `refresh`, `commit`, and `cancel`. Every operation requires +`--actor-kind HUMAN|APPLICATION|AGENT` and `--actor-identifier`. + +Keep source documents, mappings and review JSON in the ignored repository-root +`imports/` directory. The management container mounts it at `/app/imports`. +Command paths may be absolute under that root or relative to it; traversal and +symlink escapes are rejected. Output files must not already exist: choose a new +filename for every export. Input documents and JSON files are limited to 10 MiB. +Review files contain financial descriptions and must never enter Git, logs or +external services. See [data policy](data-policy.md). + +## PowerShell workflow + +Run the setup and migrations in [development](development.md) first. Create the +destination bank account or card through the existing financial services; use +its UUID below. These names and UUIDs are illustrative, synthetic examples. + +```powershell +New-Item -ItemType Directory -Force imports | Out-Null +git check-ignore imports/source.csv imports/mapping.json imports/review-1.json +@' +{"1":"11111111-1111-4111-8111-111111111111"} +'@ | Set-Content -Encoding utf8 imports/mapping.json +$actor = @('--actor-kind', 'HUMAN', '--actor-identifier', 'local-owner') +docker compose run --rm manage import_review preview ` + --input source.csv --profile BANK_CSV_V1 --mapping mapping.json @actor +``` + +Place the source CSV in `imports/source.csv` before previewing it. Substitute +the batch UUID printed by `preview`, then export its review: + +```powershell +$batch = '22222222-2222-4222-8222-222222222222' +docker compose run --rm manage import_review export ` + --batch $batch --output review-1.json @actor +notepad imports/review-1.json +``` + +Read `items`, including candidate amounts/dates, validation issues and duplicate +suggestions. Edit only the top-level `decisions` array. Preserve every other +field, including each item's stored `decision`. Alternatively save a minimal +object with exactly `batch_id`, integer `revision`, and `decisions`. Unknown +keys, repeated JSON keys, repeated locators, foreign batch items, stale +revisions and changes to source evidence are rejected. + +```json +{ + "batch_id": "22222222-2222-4222-8222-222222222222", + "revision": 1, + "decisions": [ + {"locator": "line:2", "action": "CREATE"}, + {"locator": "line:3", "action": "SKIP", "reason": "USER_EXCLUDED"} + ] +} +``` + +Use the actual locators in the exported file. Apply the complete decision set, +then export again to inspect the resulting issues and latest revision: + +```powershell +docker compose run --rm manage import_review apply ` + --batch $batch --input review-1.json @actor +docker compose run --rm manage import_review export ` + --batch $batch --output review-2.json @actor +notepad imports/review-2.json +$revision = 2 # Replace with the revision printed by apply/export. +docker compose run --rm manage import_review commit ` + --batch $batch --revision $revision @actor +``` + +Only `commit` writes financial transactions, source provenance and new +installment plans. It posts all accepted rows atomically. Repeating a successful +commit with its latest revision returns the recorded result. If live account, +reference or duplicate information has materially changed, refresh and review +again. Refresh increments the revision and clears all decisions: + +```powershell +docker compose run --rm manage import_review refresh ` + --batch $batch --revision $revision @actor +docker compose run --rm manage import_review export ` + --batch $batch --output review-refreshed.json @actor +``` + +To abandon a preview, use its current revision: + +```powershell +docker compose run --rm manage import_review cancel ` + --batch $batch --revision $revision @actor +``` + +Cancellation creates no financial entries and prevents later commitment. +Terminal output contains batch identifiers, state, revision and counts; inspect +the local review file for financial details. Errors expose only safe codes. + +## Decisions and duplicate handling + +Every candidate needs one explicit decision. Applying a file replaces the +complete decision set. The supported actions are: + +- `CREATE`: create a transaction. Optional annotations are `kind`, `category_id`, + `payee_id`, `statement_id`, `refund_of_id`, `installment_plan_id`, and `new_plan`. + Possible duplicates require `"acknowledge_duplicate": true` after review. +- `MATCH`: attach the source to the existing `transaction_id` selected from + compatible duplicate suggestions. The immutable financial fields and signature + must still agree. Existing annotations are preserved. +- `SKIP`: omit the item with a `reason` code: `USER_EXCLUDED`, `MALFORMED`, + `UNRECOGNIZED`, `ALREADY_IMPORTED`, or `SECTION_EXCLUDED`. Malformed rows and + unknown lines require explicit skipping; file-level corruption cannot be + overridden. Exact already-imported source items cannot be created or matched + again; acknowledge them with `SKIP` / `ALREADY_IMPORTED`. + +For example, an installment decision can include +`"new_plan": "synthetic-plan-a"`. This is a batch-scoped plan name, not an +existing UUID; the account and installment count come from the candidate. +Reuse the same name only for rows belonging to the same plan, or select an +existing `installment_plan_id`. Plan identity is never inferred from merchant +names, and missing installments are never generated. Refunds can remain unlinked +or use an explicit compatible `refund_of_id`. No missing statements or transfer +counterparts are created automatically. + +Section mapping JSON maps ordinal string keys such as `"1"` and `"2"` to account +UUIDs, or `null` to exclude a section. Map every section explicitly; several +sections may share one account. Owner names are never database lookup keys. +Correct mappings or immutable dates/amounts by creating a new preview from a +corrected input. They cannot be rewritten in decisions. + +## Supported compatibility profiles + +All profiles decode UTF-8 (including BOM), then Latin-1 without replacement +characters. NUL bytes, files over 10 MiB and more than 20,000 candidates are +rejected. Initial profiles require BRL destination accounts. Source dates become +both occurrence and posting dates. OFX and PDF are unsupported. + +| Profile | Accepted layout | +| --- | --- | +| `BANK_CSV_V1` | Bank account CSV, comma-delimited with CSV quoting, at least six header columns; date at column 0 as `DD/MM/YYYY`, description at 2, signed decimal-point amount at 5. Each record must match the header width. | +| `OUROCARD_FULL_V1` | Card TXT, `DD.MM.YYYY`, 37-character description/location field, two-character country field, debit and credit in Brazilian decimal notation. A separate `PGTO DEBITO CONTA` layout is recognized. | +| `OUROCARD_PARTIAL_V1` | Card TXT, `DD/MM`, 43-character description/location field, country, debit and credit; requires an explicit date interval. | + +CSV header aliases are case/accent insensitive: date accepts `Data`, +`Data movimento`, `Data do movimento`, or `Data lancamento`; description accepts +`Historico`, `Descricao`, or `Descricao do lancamento`; amount accepts `Valor`, +`Valor lancamento`, or `Valor do lancamento`. Other columns are retained only as +layout positions. Recognized balance rows are excluded and counted. + +Card amounts use credit minus debit. Nonzero amounts in both columns block +creation pending explicit review; known account-debit payment lines propose an +unmatched positive transfer. Card credits propose refunds, bank credits income, +and debits expenses. `PARC n/total` supplies installment metadata. Section +headings establish ordinal mappings; unknown nonblank text remains visible for +acknowledgement, and transaction-like malformed lines are never silently lost. + +Yearless dates require both `--context-start` and `--context-end` as ISO dates +in an inclusive interval of at most 366 days. A date must resolve uniquely; +the importer never guesses the current year. For example: + +```powershell +docker compose run --rm manage import_review preview ` + --input card.txt --profile OUROCARD_PARTIAL_V1 --mapping card-mapping.json ` + --context-start 2025-12-01 --context-end 2026-01-31 @actor +``` diff --git a/docs/superpowers/plans/2026-09-06-reviewed-imports.md b/docs/superpowers/plans/2026-09-06-reviewed-imports.md new file mode 100644 index 0000000..1ce56c8 --- /dev/null +++ b/docs/superpowers/plans/2026-09-06-reviewed-imports.md @@ -0,0 +1,48 @@ +# Reviewed Imports Implementation Plan + +**Spec:** [Approved design](../specs/2026-09-06-reviewed-imports-design.md) +**Goal:** Ship pure format parsing, persisted previews, explicit decisions, and atomic audited imports through local commands. +**Architecture:** ingestion/parsers.py + normalization.py -> validation.py + duplicates.py -> services.py lifecycle -> management command. Finance internal posting helpers are reused inside one outer transaction. + +## Interfaces + +Profiles BANK_CSV_V1, OUROCARD_FULL_V1, OUROCARD_PARTIAL_V1. +parse_document(data: bytes, profile: str, context: dict | None) returns immutable ParseResult(encoding,profile,items,skipped) or raises ParseError(code). +normalize(result) returns tuple of immutable Candidate with to_dict(): locator, section, posting_date ISO or null, occurrence_date ISO or null, description, amount decimal string or null, kind, installment_number, installment_total, errors list of safe codes. + +Public services require actor=finance.services.Actor: +create_preview(data,profile,section_mapping,context=None) -> ImportBatch; +export_review(batch_id) -> dict; apply_decisions(batch_id,revision,decisions) -> ImportBatch; +refresh_preview(batch_id,revision) -> ImportBatch; commit_preview(batch_id,revision) -> ImportBatch; +cancel_preview(batch_id,revision) -> ImportBatch. DomainError.code for failures. +Decisions list entries: locator + action CREATE/MATCH/SKIP, optional acknowledge_duplicate bool, transaction_id for MATCH, reason for SKIP (USER_EXCLUDED, MALFORMED, UNRECOGNIZED, ALREADY_IMPORTED, SECTION_EXCLUDED), kind/category_id/payee_id/statement_id/refund_of_id/installment_plan_id/new_plan for CREATE. new_plan is a batch-scoped string name; count/account derived from candidate. +section_mapping dict keys numeric ordinal strings (default section '1'), values account UUID strings or null (skip section explicitly). Multi-section no implicit mapping. +Export JSON: batch_id, revision, state, profile, encoding, summary, items (locator,candidate,issues,duplicates,decision), decisions list template. Command decision file accepts ONLY batch_id/revision/decisions, or entire exported review with read-only metadata verified against current export. + +- [x] Task 1: Pure parser/normalization, synthetic fixtures and edge tests. Owned parser files and tests/test_import_parsers.py. +- [x] Task 2: Batch/item models, duplicate/validation snapshots and lifecycle services. Persist no finance objects until commit. Revalidate locked snapshots and revision; preserve idempotency and independent failure audits. +- [x] Task 3: Management command and service/concurrency tests; demonstrate explicit review/edit/apply/commit with synthetic inputs. +- [x] Task 4: Docker/settings/migrations/runtime grants, full PostgreSQL verification, fresh/existing DB migration, review and developer docs. + +Ruling: creation rejects file-level parser errors before staging; malformed individual rows remain reviewable but require an explicit SKIP. Section mappings may be corrected by creating a new preview; candidate bytes/context are immutable. +Ruling: refresh increments revision and clears all decisions, requiring renewed consent after live state changes. Applying decisions replaces the complete decision set, never incrementally changes implicit defaults. +Ruling: successful repeated commit returns the recorded batch without a second success audit; latest committed revision must match the request. + +## Completion evidence + +- All 138 tests passed against PostgreSQL. Coverage includes the three parser + profiles, malformed and unknown lines, immutable preview evidence, explicit + decisions, duplicate matching, rollback, stale reviews, and concurrent commits. +- Parser reviewability regressions failed before unknown/malformed source labels + were retained in local candidates, then passed after the fix. Generic issue + codes remain free of source text. +- The ingestion migration applied to both the existing database and a fresh + `gizmoduck-import-verify` database. Django system checks, migration drift, and + pending-migration checks passed. +- Runtime DML and sequence grants passed for both finance and ingestion tables; + DDL and role escalation remained denied. +- The documented command flow ran end to end on the isolated database: preview, + export, edit, apply, commit, and repeated idempotent commit. Exactly one + transaction and one provenance record were created from synthetic input. +- The rehearsal files were removed afterward. The isolated verification + database container is stopped and its synthetic volume is retained. diff --git a/docs/superpowers/specs/2026-09-06-reviewed-imports-design.md b/docs/superpowers/specs/2026-09-06-reviewed-imports-design.md new file mode 100644 index 0000000..c8e48d3 --- /dev/null +++ b/docs/superpowers/specs/2026-09-06-reviewed-imports-design.md @@ -0,0 +1,210 @@ +# Reviewed Imports Design + +**Date:** 2026-09-06 +**Status:** Approved +**Milestone:** Roadmap step 4 + +## Goal and delivery surface + +Import supported bank-account CSV and Banco do Brasil/Ourocard TXT through +separate parsing, normalization, validation, duplicate detection, preview, and +atomic commit stages. Use synthetic fixtures and the existing financial services. + +Recommend a local management-command workflow for this milestone: persist a +preview in PostgreSQL, export its review document into ignored local storage, +then commit explicit row decisions. This delivers a usable review boundary +without introducing authentication, upload endpoints, and product screens before +roadmap step 5. A web wizard is a viable alternative but expands this milestone +into UI and session-security design. An automatic import command would not meet +the required review-before-commit behavior. + +## Source evidence and supported profiles + +Inspected code cells only in Robopato's account and credit-card ETL notebooks. +No notebook output or financial records are copied into this repository. Its +parsers establish these initial compatibility profiles: + +- Account CSV: Latin-1 text, comma delimiter, a header followed by records with + at least six columns. Column 0 is DD/MM/YYYY, column 2 is description, column 5 + is a signed decimal with a decimal point. CSV quoting must work correctly. + Recognized `S A L D O` balance rows are excluded with an explicit summary count. +- Full Ourocard TXT: DD.MM.YYYY followed by fixed-width description/location + fields (37 characters in the regular profile), two-character country field, + and two Brazilian-formatted monetary columns. Installment markers use + `PARC n/total`. The known `PGTO DEBITO CONTA` payment line has a separate layout. +- Partial Ourocard TXT: DD/MM, 43-character description/location field, + two-character country, and two Brazilian-formatted monetary columns. +- Owner/section headers delimit source sections. Store an ordinal section ID + for mapping; do not depend on owner-name database lookups. + +These are explicit versioned compatibility profiles, not a claim to support +all bank exports. Validate expected CSV header width and row shapes rather than +guessing column mappings. Exact accepted header aliases and TXT grammar will +be documented alongside synthetic fixtures during implementation. The parser +must fail closed on date-like or transaction-like lines it cannot classify. +Known section headings, blank lines, and recognized totals may be skipped with +counts; other unrecognized nonblank text is surfaced for acknowledgement. + +Decode UTF-8 (including BOM) first, then Latin-1. Record the selected encoding; +never use replacement-character decoding. Reject NUL bytes and oversized input +(over 10 MiB or 20,000 candidate rows). Preserve physical line/item locators. +No OFX, PDF parsing, network fetches, bank APIs, or pandas dependency. + +## Context, dates, and monetary interpretation + +Caller selects the parser profile and destination account; formats must match +account kind. Bank and Ourocard profiles in this milestone post BRL amounts +only. Reject other account currencies, because neither profile supplies a +reliable conversion contract. The domain still supports other currencies for +manual activity. + +Full dates are used as written. For yearless lines require an explicit inclusive +start/end context of at most 366 days; resolve each DD/MM only if exactly one +valid date lies in that interval. Missing/ambiguous dates block commitment. +Never use the current year. Source transaction dates become occurrence and +posting dates for these initial profiles; statement association does not replace +individual dates. Show this convention in preview. + +CSV amounts retain their signs. In regular TXT rows the first monetary column +is debit and the second credit: normalized amount = credit - debit using exact +Decimal operations independent of caller context. Negative debit represents a +credit adjustment. Nonzero values in both columns require review as an ambiguous +row rather than silently netting. Known payment lines map the payment amount to +a positive unmatched TRANSFER leg; no artificial bank leg is created. Regular +card credits are proposed REFUND, bank credits INCOME, and debits EXPENSE. + +Source labels are proposals: a reviewer may classify permitted rows as unmatched +TRANSFER or REFUND, subject to amount sign and domain validation. Do not infer a +transfer from a merchant-like keyword. Balance/previous-balance totals never +become expenses or opening entries automatically. + +## Pure parser and normalized candidate contracts + +Create an `ingestion` Django application; avoid naming a source package `imports` +because that name is reserved for ignored local financial files. + +Parsers take bytes plus explicitly supplied date context and return immutable +parsed items/issues. They do no database or filesystem access. Normalization +converts parsed items into candidates containing source locator, section ordinal, +posting/occurrence dates, original description, signed decimal string, proposed +kind, optional installment number/count, and parser version. Candidate IDs are +stable per document/locator. Parser issues have code, severity, and locator, with +no raw line in generic diagnostic messages. + +Validation resolves account/reference IDs against existing domain rules and +returns field-level issues. Duplicate detection consumes validated candidates +and database snapshots through a separate boundary. Preview persistence records +results without creating Transaction, Transfer, SourceRecord, or InstallmentPlan +rows. Preview audit events and staging records are allowed. + +## Persisted review lifecycle + +ImportBatch: UUID, document SHA-256, format/profile/version, context, selected +account/section mapping, actor identity, creation timestamp, state (PREVIEW, +COMMITTED, CANCELLED), monotonically increasing revision, and commit result. +ImportItem: batch, unique physical locator, normalized candidate payload, +validation issues, duplicate suggestions, and explicit decision metadata. + +Staged descriptions and review documents are financial data: keep them only in +local PostgreSQL and ignored local files. Do not store uploaded bytes in the +batch or logs. CLI reads source documents from ignored `imports/`; only the +management service mounts that directory, not the web container. Commit uses +the immutable stored candidates, not a re-read of a mutable source file. + +A review document is JSON with batch UUID/revision and decisions keyed by item +locator. It may contain sensitive descriptions for local human inspection. +Reject unknown keys, duplicate locators, foreign-batch items, and stale revisions. +Review decisions may specify kind, category, payee, statement, refund reference, +and installment plan mapping; they cannot rewrite amount, date, or source text. +An incorrect amount/date requires a corrected source/new preview or later manual +correction workflow, not an untraceable edit to imported evidence. + +Updating decisions increments revision and recomputes preview validation. +Every importable row requires an explicit decision. Cancelled batches cannot be +committed. Committing an already committed batch returns the recorded result, +without new transactions or provenance; retry arguments must refer to that batch. + +## Duplicate handling and review decisions + +Exact physical source identity remains (document hash, item locator). An already +attached source item is shown as ALREADY_IMPORTED with its existing transaction; +it cannot be created or attached again, even through a new batch or parser version. +If it belongs to another account, show a blocking account conflict. + +Possible duplicates use the same account, currency, posting date, exact amount, +normalized description (Unicode NFC, casefold, collapsed whitespace), kind, +and installment number/count when present. Compare both existing records and +items within the batch. Do not impose a semantic uniqueness constraint: two +legitimate purchases can share this signature. + +Allowed decisions: + +- CREATE: record a new transaction. Possible duplicates require an explicit + acknowledgement flag; default preview never silently deduplicates them. +- MATCH: attach this source item to a specifically selected existing transaction, + only when its immutable financial fields and duplicate signature still match. + No new transaction is created. Do not mutate the existing record's annotations. +- SKIP: explicitly omit a row. Require a reason code, not a free-form sensitive + diagnostic. A skipped source remains unclaimed and can be reviewed in a future + batch. Already-imported rows are skipped with an explicit existing-source reason. + +Malformed transaction rows block commit until explicitly skipped with a reason; +file-level corruption or unsupported profile cannot be overridden. Unrecognized +nontransaction lines require acknowledgement. Do not suppress skipped/error +counts from the review result. + +## Installments, refunds, and multiple sections + +Do not guess installment identity from merchant and amount. Rows with installment +metadata require an explicit existing plan or a named new-plan decision scoped +to this batch, with account and total count validated. Create new plans only in +the commit transaction, reuse the chosen plan across selected rows, and never +create unobserved installments. Do not use bank periodic-transfer text to infer +an expense installment automatically; show metadata as a review hint. + +Refunds may remain unlinked or explicitly reference a compatible expense. +Statements may be selected from existing records but are not auto-created from +incomplete header data. A source with multiple owner sections must map each +section explicitly to an account or mark it skipped. The same card account can +receive multiple sections. Never use a person's name as an automatic account key. + +## Commit integrity and concurrency + +Use one outermost transaction for all accepted rows, provenance, chosen new +plans, batch status and success audit. Reuse internal domain validation/posting +helpers, not nested public service transactions. A rejected commit rolls back +all financial writes and records a sanitized failure audit after rollback; +PREVIEW remains available for correction. + +Lock the batch, then all affected accounts in UUID order; lock reference rows +consistently before target transactions. Revalidate archive state, relationships, +source identity and duplicate suggestions under locks immediately before writes. +Lock existing candidate-match transactions before comparing annotations. If live +state differs materially from the reviewed snapshot, return STALE_PREVIEW and +require a refreshed preview/review rather than silently changing a decision. +Account locks serialize commits and ordinary postings on the same account. +Database source uniqueness resolves identical-file races across accounts. + +Ensure the multi-row commit uses one consistent lock order across the entire +batch, not each row independently. Plan creation, source attachment and auditing +must leave no partial records if a later row fails. Only transaction UUIDs, +counts, operation and error codes enter generic completion messages. + +## Commands and acceptance + +Provide management commands to create/export a preview, apply a decision file, +refresh a preview, commit a specific batch/revision, and cancel a batch. Commands +require explicit actor kind/identifier and account/section mapping. Document +PowerShell examples using the management container and ignored local import +files. No real financial records are included in the examples. + +Test with synthetic files representing quoting, encodings, invalid dates/money, +full/partial card layouts, year boundaries, payments, refunds, balance lines, +multiple sections, installments, and unknown lines. Verify preview creates no +financial rows, cancellation is inert, skipping is explicit, exact reimport is +idempotent, semantic duplicates remain reviewable, and MATCH preserves history. + +PostgreSQL tests must cover a failing later row rolling back the entire batch, +concurrent identical imports, stale review after other writes, archive changes, +repeated commit, source conflicts, and failure audits. Verify new table grants, +fresh/existing migrations, all financial-domain regressions, and app health. diff --git a/ingestion/__init__.py b/ingestion/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestion/duplicates.py b/ingestion/duplicates.py new file mode 100644 index 0000000..8cb9fdf --- /dev/null +++ b/ingestion/duplicates.py @@ -0,0 +1,25 @@ +"""Compare snapshots without writing or discarding candidate movements.""" +from decimal import Decimal +import unicodedata + + +def signature(*, account_id, currency, posting_date, amount, description, kind, + installment_number=None, installment_total=None): + return (str(account_id), currency, str(posting_date), Decimal(amount), + " ".join(unicodedata.normalize("NFC", description).casefold().split()), + kind, installment_number, installment_total) + + +def candidate_signature(candidate, account, decision): + return signature(account_id=account.pk, currency=account.currency, + posting_date=candidate["posting_date"], amount=candidate["amount"], + description=candidate["description"], kind=decision.get("kind", candidate["kind"]), + installment_number=candidate.get("installment_number"), + installment_total=candidate.get("installment_total")) + + +def transaction_signature(row): + return signature(account_id=row.account_id, currency=row.account.currency, + posting_date=row.posting_date, amount=row.amount, description=row.description, + kind=row.kind, installment_number=row.installment_number, + installment_total=row.installment_plan.total_installments if row.installment_plan_id else None) diff --git a/ingestion/management/__init__.py b/ingestion/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestion/management/commands/__init__.py b/ingestion/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestion/management/commands/import_review.py b/ingestion/management/commands/import_review.py new file mode 100644 index 0000000..746ffb1 --- /dev/null +++ b/ingestion/management/commands/import_review.py @@ -0,0 +1,208 @@ +"""Local, explicit review workflow. Financial details never go to stdout.""" +import json +import re +from datetime import date +from pathlib import Path +from uuid import UUID + +from django.conf import settings +from django.core.management.base import BaseCommand, CommandError + +from finance.services import Actor, DomainError +from ingestion import services + + +MAX_BYTES = 10 * 1024 * 1024 +PROFILES = ("BANK_CSV_V1", "OUROCARD_FULL_V1", "OUROCARD_PARTIAL_V1") + + +def local_path(value): + root = Path(getattr(settings, "IMPORT_ROOT", "/app/imports")).resolve() + requested = Path(value) + if ".." in requested.parts: + raise CommandError("INVALID_PATH") + resolved = (root / requested).resolve() + if not resolved.is_relative_to(root) or resolved == root: + raise CommandError("INVALID_PATH") + return resolved + + +def read_bytes(value): + path = local_path(value) + if not path.is_file(): + raise CommandError("INVALID_INPUT") + with path.open("rb") as stream: + data = stream.read(MAX_BYTES + 1) + if len(data) > MAX_BYTES: + raise CommandError("INPUT_TOO_LARGE") + return data + + +def unique_object(pairs): + result = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate key") + result[key] = value + return result + + +def reject_constant(value): + raise ValueError("nonfinite JSON number") + + +def read_json(value): + data = read_bytes(value) + try: + return json.loads(data.decode("utf-8-sig"), object_pairs_hook=unique_object, + parse_constant=reject_constant) + except (ValueError, RecursionError): + raise CommandError("INVALID_JSON") from None + + +def identifier(value): + try: + return str(UUID(str(value))) + except (ValueError, AttributeError, TypeError): + raise CommandError("INVALID_IDENTIFIER") from None + + +def revision_number(value): + if type(value) is not int or value < 1: + raise CommandError("INVALID_REVISION") + return value + + +def checked_decisions(payload, batch_id, actor): + minimum = {"batch_id", "revision", "decisions"} + if not isinstance(payload, dict) or not minimum.issubset(payload): + raise CommandError("INVALID_REVIEW") + if identifier(payload["batch_id"]) != batch_id: + raise CommandError("INVALID_REVIEW") + revision_number(payload["revision"]) + if not isinstance(payload["decisions"], list): + raise CommandError("INVALID_REVIEW") + locators = set() + for decision in payload["decisions"]: + if not isinstance(decision, dict) or not isinstance(decision.get("locator"), str): + raise CommandError("INVALID_REVIEW") + locator = decision["locator"] + if locator in locators: + raise CommandError("INVALID_REVIEW") + locators.add(locator) + if set(payload) != minimum: + current = services.export_review(batch_id=batch_id) + if set(payload) != set(current): + raise CommandError("INVALID_REVIEW") + # All exported fields except the dedicated decisions array are evidence. + # JSON encoding also distinguishes true/1 and numeric/string edits. + for key in current: + if key != "decisions" and json.dumps(payload[key], sort_keys=True) != json.dumps(current[key], sort_keys=True): + raise CommandError("READ_ONLY_REVIEW_CHANGED") + return payload["revision"], payload["decisions"] + + +class Command(BaseCommand): + help = "Preview, export, apply, refresh, commit or cancel a local reviewed import." + requires_system_checks = [] + + def add_arguments(self, parser): + commands = parser.add_subparsers(dest="operation", required=True) + for operation in ("preview", "export", "apply", "refresh", "commit", "cancel"): + sub = commands.add_parser(operation) + sub.add_argument("--actor-kind", required=True, choices=("HUMAN", "APPLICATION", "AGENT")) + sub.add_argument("--actor-identifier", required=True) + if operation == "preview": + sub.add_argument("--input", required=True) + sub.add_argument("--mapping", required=True) + sub.add_argument("--profile", required=True, choices=PROFILES) + sub.add_argument("--context-start") + sub.add_argument("--context-end") + else: + sub.add_argument("--batch", required=True) + if operation == "export": + sub.add_argument("--output", required=True) + if operation == "apply": + sub.add_argument("--input", required=True) + if operation in {"refresh", "commit", "cancel"}: + sub.add_argument("--revision", required=True, type=int) + + def handle(self, *args, **options): + try: + self.perform(options) + except CommandError: + raise + except DomainError as exc: + code = exc.code + if not isinstance(code, str) or re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,79}", code) is None: + code = "IMPORT_FAILED" + raise CommandError(code) from None + except Exception: + # Source paths, source text and database errors are local financial data. + raise CommandError("IMPORT_FAILED") from None + + def perform(self, options): + actor = Actor(options["actor_kind"], options["actor_identifier"]) + if actor.kind not in {"HUMAN", "APPLICATION", "AGENT"} or not actor.identifier.strip() or len(actor.identifier) > 200: + raise CommandError("INVALID_ACTOR") + operation = options["operation"] + if operation == "preview": + mapping = read_json(options["mapping"]) + if not isinstance(mapping, dict) or not mapping: + raise CommandError("INVALID_MAPPING") + for section, account in mapping.items(): + if re.fullmatch(r"[1-9][0-9]*", section) is None: + raise CommandError("INVALID_MAPPING") + if account is not None: + if not isinstance(account, str): + raise CommandError("INVALID_MAPPING") + identifier(account) + start, end = options.get("context_start"), options.get("context_end") + context = None + if start or end: + try: + first, last = date.fromisoformat(start), date.fromisoformat(end) + if not 0 <= (last - first).days <= 365: + raise ValueError + except (TypeError, ValueError): + raise CommandError("INVALID_CONTEXT") from None + context = {"start_date": first.isoformat(), "end_date": last.isoformat()} + batch = services.create_preview( + data=read_bytes(options["input"]), + profile=options["profile"], + section_mapping=mapping, + context=context, + actor=actor, + ) + else: + batch_id = identifier(options["batch"]) + if operation == "export": + path = local_path(options["output"]) + review = services.export_review(batch_id=batch_id) + encoded = json.dumps(review, ensure_ascii=False, indent=2, allow_nan=False) + "\n" + # Exclusive creation protects source documents and previous reviews. + with path.open("x", encoding="utf-8") as stream: + stream.write(encoded) + self.stdout.write(json.dumps({"batch_id": batch_id, "state": review["state"], + "revision": review["revision"], "item_count": len(review["items"])})) + return + if operation == "apply": + revision, decisions = checked_decisions(read_json(options["input"]), batch_id, actor) + batch = services.apply_decisions( + batch_id=batch_id, + revision=revision, + decisions=decisions, + actor=actor, + ) + else: + revision = revision_number(options["revision"]) + action = {"refresh": services.refresh_preview, "commit": services.commit_preview, + "cancel": services.cancel_preview}[operation] + batch = action(batch_id=batch_id, revision=revision, actor=actor) + counts = getattr(batch, "result", {}) if batch.state == "COMMITTED" else getattr(batch, "summary", {}) + safe_counts = { + key: value for key, value in counts.items() + if key in {"rows", "error_rows", "decided_rows", "created", "matched", "skipped"} + } + self.stdout.write(json.dumps({"batch_id": str(batch.pk), "state": batch.state, + "revision": batch.revision, "counts": safe_counts})) diff --git a/ingestion/migrations/0001_initial.py b/ingestion/migrations/0001_initial.py new file mode 100644 index 0000000..ebc0c0a --- /dev/null +++ b/ingestion/migrations/0001_initial.py @@ -0,0 +1,58 @@ +# Generated by Django 5.2.17 on 2026-09-07 20:58 + +import django.db.models.deletion +import uuid +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + ] + + operations = [ + migrations.CreateModel( + name='ImportBatch', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('document_hash', models.CharField(max_length=64)), + ('source_format', models.CharField(max_length=16)), + ('profile', models.CharField(max_length=32)), + ('parser_version', models.CharField(default='1', max_length=32)), + ('encoding', models.CharField(max_length=32)), + ('context', models.JSONField(default=dict)), + ('section_mapping', models.JSONField(default=dict)), + ('summary', models.JSONField(default=dict)), + ('snapshot', models.JSONField(default=dict)), + ('actor_kind', models.CharField(max_length=16)), + ('actor_identifier', models.CharField(max_length=200)), + ('state', models.CharField(default='PREVIEW', max_length=16)), + ('revision', models.PositiveIntegerField(default=1)), + ('result', models.JSONField(default=dict)), + ], + options={ + 'constraints': [models.CheckConstraint(condition=models.Q(('state__in', ['PREVIEW', 'COMMITTED', 'CANCELLED'])), name='import_batch_state'), models.CheckConstraint(condition=models.Q(('revision__gte', 1)), name='import_batch_revision')], + }, + ), + migrations.CreateModel( + name='ImportItem', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('locator', models.CharField(max_length=255)), + ('candidate', models.JSONField()), + ('issues', models.JSONField(default=list)), + ('duplicates', models.JSONField(default=dict)), + ('decision', models.JSONField(default=dict)), + ('batch', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='items', to='ingestion.importbatch')), + ], + options={ + 'constraints': [models.UniqueConstraint(fields=('batch', 'locator'), name='import_item_locator_unique')], + }, + ), + ] diff --git a/ingestion/migrations/__init__.py b/ingestion/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingestion/models.py b/ingestion/models.py new file mode 100644 index 0000000..726f01a --- /dev/null +++ b/ingestion/models.py @@ -0,0 +1,38 @@ +from django.db import models +from django.db.models import Q +from finance.models.base import TimestampedModel + + +class ImportBatch(TimestampedModel): + document_hash = models.CharField(max_length=64) + source_format = models.CharField(max_length=16) + profile = models.CharField(max_length=32) + parser_version = models.CharField(max_length=32, default="1") + encoding = models.CharField(max_length=32) + context = models.JSONField(default=dict) + section_mapping = models.JSONField(default=dict) + summary = models.JSONField(default=dict) + snapshot = models.JSONField(default=dict) + actor_kind = models.CharField(max_length=16) + actor_identifier = models.CharField(max_length=200) + state = models.CharField(max_length=16, default="PREVIEW") + revision = models.PositiveIntegerField(default=1) + result = models.JSONField(default=dict) + + class Meta: + constraints = [ + models.CheckConstraint(condition=Q(state__in=["PREVIEW", "COMMITTED", "CANCELLED"]), name="import_batch_state"), + models.CheckConstraint(condition=Q(revision__gte=1), name="import_batch_revision"), + ] + + +class ImportItem(TimestampedModel): + batch = models.ForeignKey(ImportBatch, on_delete=models.PROTECT, related_name="items") + locator = models.CharField(max_length=255) + candidate = models.JSONField() + issues = models.JSONField(default=list) + duplicates = models.JSONField(default=dict) + decision = models.JSONField(default=dict) + + class Meta: + constraints = [models.UniqueConstraint(fields=["batch", "locator"], name="import_item_locator_unique")] diff --git a/ingestion/normalization.py b/ingestion/normalization.py new file mode 100644 index 0000000..4a1f49b --- /dev/null +++ b/ingestion/normalization.py @@ -0,0 +1,117 @@ +"""Convert parser evidence into immutable, serializable review candidates.""" + +import re +from dataclasses import dataclass, asdict +from datetime import date, datetime +from decimal import Decimal + +from .parsers import ParseResult + + +@dataclass(frozen=True) +class Candidate: + locator: str + section: str + posting_date: str | None + occurrence_date: str | None + description: str + amount: str | None + kind: str + installment_number: int | None + installment_total: int | None + errors: tuple[str, ...] + + def to_dict(self): + result = asdict(self) + result['errors'] = list(self.errors) + return result + + +def _date(text, profile, context): + if profile != 'OUROCARD_PARTIAL_V1': + pattern = r'\d{2}/\d{2}/\d{4}' if profile == 'BANK_CSV_V1' else r'\d{2}\.\d{2}\.\d{4}' + if not re.fullmatch(pattern, text): + raise ValueError('INVALID_DATE') + try: + return datetime.strptime(text, '%d/%m/%Y' if profile == 'BANK_CSV_V1' else '%d.%m.%Y').date().isoformat() + except ValueError: + raise ValueError('INVALID_DATE') from None + if not context.get('start_date') or not context.get('end_date'): + raise ValueError('MISSING_DATE_CONTEXT') + try: + start, end = date.fromisoformat(context['start_date']), date.fromisoformat(context['end_date']) + if not 0 <= (end - start).days <= 365: + raise ValueError + except (ValueError, TypeError): + raise ValueError('INVALID_DATE_CONTEXT') from None + if not re.fullmatch(r'\d{2}/\d{2}', text): + raise ValueError('INVALID_DATE') + day, month = map(int, text.split('/')) + possible = [] + for year in range(start.year, end.year + 1): + try: + candidate = date(year, month, day) + except ValueError: + continue + if start <= candidate <= end: + possible.append(candidate) + if len(possible) != 1: + raise ValueError('AMBIGUOUS_DATE' if possible else 'INVALID_DATE') + return possible[0].isoformat() + + +def _cents(text, localized=False): + pattern = r'[+-]?(?:\d+|\d{1,3}(?:\.\d{3})+),\d{2}' if localized else r'[+-]?\d+(?:\.\d{1,2})?' + if len(text) > 40 or not re.fullmatch(pattern, text): + raise ValueError('INVALID_AMOUNT') + clean = text.replace('.', '').replace(',', '.') if localized else text + # Integer minor units keep all arithmetic independent of Decimal context. + negative = clean.startswith('-') + clean = clean.lstrip('+-') + whole, _, fraction = clean.partition('.') + return (-1 if negative else 1) * (int(whole) * 100 + int(fraction.ljust(2, '0'))) + + +def _format_cents(value): + return ('-' if value < 0 else '') + f'{abs(value) // 100}.{abs(value) % 100:02d}' + + +def normalize(result: ParseResult) -> tuple[Candidate, ...]: + candidates = [] + for item in result.items: + errors = list(item.errors) + posting_date = amount = installment_number = installment_total = None + kind = 'EXPENSE' + if not item.errors: + try: + posting_date = _date(item.date_text, result.profile, result.context) + except ValueError as exc: + errors.append(str(exc)) + try: + if result.profile == 'BANK_CSV_V1': + cents = _cents(item.debit) + kind = 'INCOME' if cents > 0 else 'EXPENSE' + else: + debit, credit = _cents(item.debit, True), _cents(item.credit, True) + if item.payment: + if debit and credit: + errors.append('AMBIGUOUS_AMOUNT') + # copy_abs is exact even when caller Decimal context traps rounding. + cents = int(Decimal(debit).copy_abs()) + kind = 'TRANSFER' + else: + if debit and credit: + errors.append('AMBIGUOUS_AMOUNT') + cents = credit - debit + kind = 'REFUND' if cents > 0 else 'EXPENSE' + amount = _format_cents(cents) + except ValueError as exc: + errors.append(str(exc)) + if result.profile != 'BANK_CSV_V1' and re.search(r'\bPARC\b', item.description): + match = re.search(r'\bPARC\s+(\d{1,4})/(\d{1,4})\b', item.description) + if match and 1 <= int(match[1]) <= int(match[2]): + installment_number, installment_total = map(int, match.groups()) + else: + errors.append('INVALID_INSTALLMENT') + candidates.append(Candidate(item.locator, item.section, posting_date, posting_date, item.description, amount, kind, installment_number, installment_total, tuple(errors))) + return tuple(candidates) diff --git a/ingestion/parsers.py b/ingestion/parsers.py new file mode 100644 index 0000000..da9df14 --- /dev/null +++ b/ingestion/parsers.py @@ -0,0 +1,136 @@ +"""Pure, versioned source grammars. No filesystem or database access.""" + +import csv +import io +import re +import unicodedata +from collections import Counter +from dataclasses import dataclass +from types import MappingProxyType +from typing import Mapping + + +PARSER_VERSION = '1' +PROFILES = frozenset({'BANK_CSV_V1', 'OUROCARD_FULL_V1', 'OUROCARD_PARTIAL_V1'}) +MAX_BYTES = 10 * 1024 * 1024 +MAX_ITEMS = 20000 + + +class ParseError(ValueError): + def __init__(self, code): + self.code = code + super().__init__(code) + + +@dataclass(frozen=True) +class ParsedItem: + locator: str + section: str + date_text: str = '' + description: str = '' + debit: str = '' + credit: str = '' + payment: bool = False + errors: tuple[str, ...] = () + + +@dataclass(frozen=True) +class ParseResult: + encoding: str + profile: str + items: tuple[ParsedItem, ...] + skipped: Mapping[str, int] + context: Mapping[str, str] + + +def _header_key(value): + return ''.join(c for c in unicodedata.normalize('NFD', value.strip().casefold()) if not unicodedata.combining(c)) + + +def _balance(value): + return bool(re.fullmatch(r'(?:S\s*A\s*L\s*D\s*O|SALDO ANTERIOR)(?:\s+[-\d.,]+)?', value.strip(), re.I)) + + +def _csv_items(text, skipped): + reader = csv.reader(io.StringIO(text, newline=''), strict=True) + try: + header = next(reader) + aliases = ({'data', 'data movimento', 'data do movimento', 'data lancamento'}, {'historico', 'descricao', 'descricao do lancamento'}, {'valor', 'valor lancamento', 'valor do lancamento'}) + if len(header) < 6 or any(_header_key(header[i]) not in allowed for i, allowed in zip((0, 2, 5), aliases)): + raise ParseError('INVALID_HEADER') + while True: + line = reader.line_num + 1 + try: + row = next(reader) + except StopIteration: + return + if not row or all(not cell.strip() for cell in row): + skipped['blank'] += 1 + continue + if len(row) != len(header): + description = row[2] if len(row) > 2 else '' + yield ParsedItem(f'line:{line}', '1', description=description, + errors=('INVALID_ROW',)) + elif _balance(row[2]): + skipped['balance'] += 1 + else: + yield ParsedItem(f'line:{line}', '1', row[0].strip(), row[2], row[5].strip()) + except StopIteration: + raise ParseError('INVALID_HEADER') from None + except csv.Error: + raise ParseError('INVALID_CSV') from None + + +def _txt_items(text, profile, skipped): + partial = profile == 'OUROCARD_PARTIAL_V1' + date_size, body_size = (5, 43) if partial else (10, 37) + date_re = r'\d{2}/\d{2}' if partial else r'\d{2}\.\d{2}\.\d{4}' + regular = re.compile(rf'^({date_re})(.{{{body_size}}})([A-Z ]{{2}})\s+(\S+)\s+(\S+)\s*$') + payment = re.compile(rf'^({date_re})\s*(PGTO DEBITO CONTA.*?)200211\s+(\S+)\s+(\S+)\s*$') + section, section_count = '1', 0 + for number, line in enumerate(text.splitlines(), 1): + locator = f'line:{number}' + if not line.strip(): + skipped['blank'] += 1 + elif re.match(r'^\s*\d\s*-\s*[A-Z]', line): + section_count += 1 + section = str(section_count) + skipped['section'] += 1 + elif _balance(line): + skipped['balance'] += 1 + elif match := payment.fullmatch(line): + yield ParsedItem(locator, section, match[1], match[2].strip(), match[3], match[4], True) + elif match := regular.fullmatch(line): + if _balance(match[2]): + skipped['balance'] += 1 + else: + yield ParsedItem(locator, section, match[1], match[2].rstrip(), match[4], match[5]) + else: + code = 'INVALID_ROW' if re.match(r'^\s*\d{1,2}[./]', line) else 'UNRECOGNIZED_LINE' + yield ParsedItem(locator, section, description=line, errors=(code,)) + + +def parse_document(data: bytes, profile: str, context: dict | None = None) -> ParseResult: + if profile not in PROFILES: + raise ParseError('UNSUPPORTED_PROFILE') + if not isinstance(data, bytes): + raise ParseError('INVALID_INPUT') + if len(data) > MAX_BYTES: + raise ParseError('INPUT_TOO_LARGE') + if b'\x00' in data: + raise ParseError('NUL_BYTE') + if context is not None and (not isinstance(context, dict) or any(not isinstance(v, str) for v in context.values())): + raise ParseError('INVALID_DATE_CONTEXT') + try: + encoding = 'utf-8-sig' if data.startswith(b'\xef\xbb\xbf') else 'utf-8' + text = data.decode(encoding) + except UnicodeDecodeError: + encoding, text = 'latin-1', data.decode('latin-1') + skipped = Counter() + source = _csv_items(text, skipped) if profile == 'BANK_CSV_V1' else _txt_items(text, profile, skipped) + items = [] + for item in source: + items.append(item) + if len(items) > MAX_ITEMS: + raise ParseError('TOO_MANY_ITEMS') + return ParseResult(encoding, profile, tuple(items), MappingProxyType(dict(skipped)), MappingProxyType(dict(context or {}))) diff --git a/ingestion/services.py b/ingestion/services.py new file mode 100644 index 0000000..c88a2e8 --- /dev/null +++ b/ingestion/services.py @@ -0,0 +1,335 @@ +"""Audited review lifecycle. No financial rows exist until an explicit commit.""" +from datetime import date +from functools import wraps +import hashlib +import re +from uuid import UUID, uuid5, NAMESPACE_URL + +from django.core.exceptions import ValidationError +from django.db import IntegrityError, connection, transaction +from django.db.models import Q + +from finance.models import Account, Payee, Category, CardStatement, InstallmentPlan, Transaction, SourceRecord +from finance.services import Actor, DomainError +from finance.services.base import audit, identity, save +from finance.services.posting import record +from ingestion.models import ImportBatch, ImportItem +from ingestion.parsers import parse_document, ParseError +from ingestion.normalization import normalize +from ingestion.duplicates import candidate_signature, transaction_signature +from ingestion.validation import validate + + +REASONS = {"USER_EXCLUDED", "MALFORMED", "UNRECOGNIZED", "ALREADY_IMPORTED", "SECTION_EXCLUDED"} +REFERENCE_FIELDS = {"payee_id": ("payees", Payee), "category_id": ("categories", Category), + "statement_id": ("statements", CardStatement), "installment_plan_id": ("plans", InstallmentPlan)} + + +def operation(name): + def decorate(function): + @wraps(function) + def wrapped(*args, actor, **kwargs): + if not isinstance(actor, Actor) or actor.kind not in {"HUMAN", "APPLICATION", "AGENT"} or not isinstance(actor.identifier, str) or not actor.identifier.strip() or len(actor.identifier) > 200: + raise DomainError("invalid_actor") + if connection.in_atomic_block: + raise DomainError("transaction_boundary") + try: + with transaction.atomic(durable=True): + batch, fields, emit = function(*args, actor=actor, **kwargs) + if emit: + audit(actor, name, "SUCCESS", batch.pk, fields) + return batch + except (DomainError, ParseError, ValidationError, IntegrityError) as error: + code = getattr(error, "code", None) + if not isinstance(code, str): + code = "CONFLICT" if isinstance(error, IntegrityError) else "VALIDATION_ERROR" + with transaction.atomic(durable=True): + audit(actor, name, "FAILURE", error_code=code) + raise DomainError(code) from None + return wrapped + return decorate + + +def _batch(batch_id, revision=None, *, preview=True): + try: + batch = ImportBatch.objects.select_for_update().get(pk=identity(batch_id)) + except ImportBatch.DoesNotExist: + raise DomainError("BATCH_NOT_FOUND") from None + if revision is not None and (type(revision) is not int or revision != batch.revision): + raise DomainError("STALE_REVISION") + if preview and batch.state != "PREVIEW": + raise DomainError("BATCH_NOT_PREVIEW") + return batch + + +def _rows(query, locking): + if locking: + query = query.select_for_update(of=("self",)) + return {str(obj.pk): obj for obj in query.order_by("pk")} + + +def _load(batch, items, locking): + account_ids = [value for value in batch.section_mapping.values() if value] + accounts = _rows(Account.objects.filter(pk__in=account_ids), locking) + refs = {} + for field, (key, model) in REFERENCE_FIELDS.items(): + ids = [item.decision[field] for item in items if item.decision.get(field)] + refs[key] = _rows(model.objects.filter(pk__in=ids), locking) + dates = set() + for item in items: + try: + dates.add(date.fromisoformat(item.candidate.get("posting_date"))) + except (TypeError, ValueError): + pass + target_ids = [item.decision[key] for item in items for key in ("transaction_id", "refund_of_id") if item.decision.get(key)] + query = Transaction.objects.filter(Q(account_id__in=account_ids, posting_date__in=dates) | Q(pk__in=target_ids)) + refs["transactions"] = _rows(query.select_related("account", "installment_plan"), locking) + sources = {row.item_locator: row for row in SourceRecord.objects.filter(document_hash=batch.document_hash).select_related("transaction")} + return accounts, refs, sources + + +def _fingerprint(obj): + # IDs and modification timestamps detect reference annotations/archive changes. + return [str(obj.pk), obj.updated_at.isoformat()] + + +def _evaluate(batch, items, *, locking=False): + accounts, refs, sources = _load(batch, items, locking) + keys = {} + existing = {} + for row in refs["transactions"].values(): + existing.setdefault(transaction_signature(row), []).append(str(row.pk)) + for item in items: + account = accounts.get(batch.section_mapping.get(item.candidate["section"])) + if account and not item.candidate.get("errors"): + try: + keys[item.locator] = candidate_signature(item.candidate, account, item.decision) + except (KeyError, TypeError, ValueError): + pass + results = {} + plan_definitions = {} + plan_memberships = set() + for item in items: + candidate, decision = item.candidate, item.decision + mapped = batch.section_mapping.get(candidate["section"]) + account = accounts.get(mapped) + issues = validate(candidate, decision, account, refs, batch.profile) if mapped else ["SECTION_EXCLUDED"] + key = keys.get(item.locator) + source = sources.get(item.locator) + matches = sorted(existing.get(key, [])) if key is not None else [] + within = sorted(locator for locator, other in keys.items() if locator != item.locator and other == key) if key is not None else [] + duplicate = {"existing": matches, "within": within, + "source": str(source.transaction_id) if source else None, + "source_account": str(source.transaction.account_id) if source else None, + "status": "ALREADY_IMPORTED" if source else None} + if source: + issues.append("ALREADY_IMPORTED" if str(source.transaction.account_id) == mapped else "SOURCE_ACCOUNT_CONFLICT") + if decision.get("new_plan") and decision.get("action") == "CREATE": + name = decision["new_plan"] + definition = (mapped, candidate.get("installment_total")) + if name in plan_definitions and plan_definitions[name] != definition: + issues.append("NEW_PLAN_CONFLICT") + plan_definitions[name] = definition + if decision.get("action") == "CREATE" and candidate.get("installment_number"): + plan_key = decision.get("installment_plan_id") or decision.get("new_plan") + membership = (plan_key, candidate["installment_number"]) + if plan_key and membership in plan_memberships: + issues.append("DUPLICATE_INSTALLMENT") + plan_memberships.add(membership) + results[item.locator] = {"issues": sorted(set(issues)), "duplicates": duplicate} + reference_snapshot = {name: [_fingerprint(obj) for obj in table.values()] for name, table in refs.items() if name != "transactions"} + relevant_ids = {tid for result in results.values() for tid in result["duplicates"]["existing"]} + relevant_ids.update(item.decision[key] for item in items for key in ("transaction_id", "refund_of_id") if item.decision.get(key)) + reference_snapshot["transactions"] = [_fingerprint(obj) for key, obj in refs["transactions"].items() if key in relevant_ids] + return {"accounts": [_fingerprint(obj) for obj in accounts.values()], "references": reference_snapshot, "items": results} + + +def _persist(batch, items, snapshot): + for item in items: + current = snapshot["items"][item.locator] + item.issues, item.duplicates = current["issues"], current["duplicates"] + ImportItem.objects.bulk_update(items, ["issues", "duplicates", "decision"]) + batch.snapshot = snapshot + batch.summary = {**batch.summary, "rows": len(items), "error_rows": sum(bool(item.issues) for item in items), + "decided_rows": sum(bool(item.decision) for item in items)} + batch.save() + + +def _current(batch, items): + snapshot = _evaluate(batch, items, locking=True) + if snapshot != batch.snapshot: + raise DomainError("STALE_PREVIEW") + return snapshot + + +@operation("import_preview") +def create_preview(*, data, profile, section_mapping, context=None, actor): + parsed = parse_document(data, profile, context or {}) + candidates = [candidate.to_dict() for candidate in normalize(parsed)] + if not isinstance(section_mapping, dict) or any(not isinstance(key, str) for key in section_mapping): + raise DomainError("INVALID_SECTION_MAPPING") + sections = {item["section"] for item in candidates} + if set(section_mapping) != sections: + raise DomainError("SECTION_MAPPING_REQUIRED") + mapping = {key: str(identity(value)) if value is not None else None for key, value in section_mapping.items()} + batch = ImportBatch.objects.create(document_hash=hashlib.sha256(data).hexdigest(), + source_format="BANK_CSV" if profile == "BANK_CSV_V1" else "OUROCARD_TXT", + profile=profile, encoding=parsed.encoding, context=context or {}, section_mapping=mapping, + summary={"parser_skipped": dict(parsed.skipped)}, actor_kind=actor.kind, actor_identifier=actor.identifier) + items = [ImportItem(batch=batch, locator=candidate["locator"], candidate=candidate) for candidate in candidates] + ImportItem.objects.bulk_create(items) + _persist(batch, items, _evaluate(batch, items, locking=True)) + return batch, ["preview"], True + + +def export_review(*, batch_id): + # Keep exported revision and item decisions consistent with concurrent editing. + with transaction.atomic(): + batch = _batch(batch_id, preview=False) + items = list(batch.items.order_by("created_at", "locator")) + return {"batch_id": str(batch.pk), "revision": batch.revision, "state": batch.state, + "profile": batch.profile, "encoding": batch.encoding, "summary": batch.summary, + "items": [{"locator": row.locator, "candidate_id": str(uuid5(NAMESPACE_URL, batch.document_hash + ":" + row.locator)), + "candidate": row.candidate, "issues": row.issues, "duplicates": row.duplicates, + "decision": row.decision} for row in items], + "decisions": [{"locator": row.locator, **row.decision} for row in items]} + + +def _decisions(value, locators): + if not isinstance(value, list): + raise DomainError("INVALID_DECISIONS") + result = {} + for supplied in value: + if not isinstance(supplied, dict): + raise DomainError("INVALID_DECISION") + row = dict(supplied) + locator = row.pop("locator", None) + if not isinstance(locator, str) or locator not in locators or locator in result: + raise DomainError("INVALID_LOCATOR") + action = row.get("action") + allowed = {"action"} + if action == "CREATE": + allowed |= {"acknowledge_duplicate", "kind", "payee_id", "category_id", "statement_id", "refund_of_id", "installment_plan_id", "new_plan"} + elif action == "MATCH": + allowed.add("transaction_id") + elif action == "SKIP": + allowed.add("reason") + else: + raise DomainError("INVALID_ACTION") + if set(row) - allowed: + raise DomainError("UNKNOWN_DECISION_FIELD") + if "acknowledge_duplicate" in row and type(row["acknowledge_duplicate"]) is not bool: + raise DomainError("INVALID_ACKNOWLEDGEMENT") + if action == "SKIP" and row.get("reason") not in REASONS: + raise DomainError("SKIP_REASON_REQUIRED") + if action == "MATCH" and not row.get("transaction_id"): + raise DomainError("MATCH_TARGET_REQUIRED") + for field in set(row) & {"transaction_id", "payee_id", "category_id", "statement_id", "refund_of_id", "installment_plan_id"}: + if row[field] is not None: + row[field] = str(identity(row[field])) + if "new_plan" in row and (not isinstance(row["new_plan"], str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", row["new_plan"])): + raise DomainError("INVALID_NEW_PLAN") + if "kind" in row and row["kind"] not in {"INCOME", "EXPENSE", "REFUND", "TRANSFER"}: + raise DomainError("INVALID_KIND") + result[locator] = row + return result + + +@operation("import_decisions") +def apply_decisions(*, batch_id, revision, decisions, actor): + batch = _batch(batch_id, revision) + items = list(batch.items.order_by("locator")) + _current(batch, items) + values = _decisions(decisions, {item.locator for item in items}) + for item in items: + item.decision = values.get(item.locator, {}) + batch.revision += 1 + _persist(batch, items, _evaluate(batch, items, locking=True)) + return batch, ["decisions", "revision"], True + + +@operation("import_refresh") +def refresh_preview(*, batch_id, revision, actor): + batch = _batch(batch_id, revision) + items = list(batch.items.order_by("locator")) + for item in items: + item.decision = {} + batch.revision += 1 + _persist(batch, items, _evaluate(batch, items, locking=True)) + return batch, ["preview", "decisions", "revision"], True + + +@operation("import_cancel") +def cancel_preview(*, batch_id, revision, actor): + batch = _batch(batch_id, revision) + batch.state = "CANCELLED" + batch.save() + return batch, ["state"], True + + +def _check_decision(item): + decision = item.decision + if not decision: + raise DomainError("DECISION_REQUIRED") + if "SOURCE_ACCOUNT_CONFLICT" in item.issues: + raise DomainError("SOURCE_ACCOUNT_CONFLICT") + if decision["action"] == "SKIP": + if item.duplicates.get("source") and decision["reason"] != "ALREADY_IMPORTED": + raise DomainError("ALREADY_IMPORTED") + if "SECTION_EXCLUDED" in item.issues and decision["reason"] != "SECTION_EXCLUDED": + raise DomainError("SECTION_EXCLUDED") + return + if item.duplicates.get("source"): + raise DomainError("ALREADY_IMPORTED") + # MATCH validates the signature rather than requiring a new plan assignment. + issues = [issue for issue in item.issues if not (decision["action"] == "MATCH" and issue == "INSTALLMENT_PLAN_REQUIRED")] + if issues: + raise DomainError("UNRESOLVED_ISSUES") + if decision["action"] == "CREATE" and (item.duplicates["existing"] or item.duplicates["within"]) and not decision.get("acknowledge_duplicate"): + raise DomainError("DUPLICATE_ACKNOWLEDGEMENT_REQUIRED") + if decision["action"] == "MATCH" and decision["transaction_id"] not in item.duplicates["existing"]: + raise DomainError("MATCH_SIGNATURE_MISMATCH") + + +@operation("import_commit") +def commit_preview(*, batch_id, revision, actor): + batch = _batch(batch_id, revision, preview=False) + if batch.state == "COMMITTED": + return batch, [], False + if batch.state != "PREVIEW": + raise DomainError("BATCH_NOT_PREVIEW") + items = list(batch.items.order_by("locator")) + _current(batch, items) + for item in items: + _check_decision(item) + plans = {} + result = {"created": 0, "matched": 0, "skipped": 0, "transaction_ids": [], "parser_skipped": batch.summary.get("parser_skipped", {}), "error_rows": batch.summary.get("error_rows", 0)} + for item in items: + candidate, decision = item.candidate, item.decision + if decision["action"] == "SKIP": + result["skipped"] += 1 + continue + if decision["action"] == "MATCH": + row = Transaction.objects.get(pk=decision["transaction_id"]) + result["matched"] += 1 + else: + account_id = batch.section_mapping[candidate["section"]] + plan_id = decision.get("installment_plan_id") + if decision.get("new_plan"): + name = decision["new_plan"] + if name not in plans: + plans[name] = save(InstallmentPlan(account_id=account_id, description=name, total_installments=candidate["installment_total"])) + plan_id = plans[name].pk + fields = {key: decision.get(key) for key in ("payee_id", "category_id", "statement_id", "refund_of_id")} + row = record(account_id=account_id, amount=candidate["amount"], posting_date=candidate["posting_date"], + occurrence_date=candidate["occurrence_date"], kind=decision.get("kind", candidate["kind"]), + description=candidate["description"], installment_plan_id=plan_id, + installment_number=candidate.get("installment_number"), **fields) + result["created"] += 1 + save(SourceRecord(transaction=row, document_hash=batch.document_hash, source_format=batch.source_format, + parser_version=batch.parser_version, item_locator=item.locator)) + result["transaction_ids"].append(str(row.pk)) + batch.state = "COMMITTED" + batch.result = result + batch.save() + return batch, ["state", "transactions", "source_records", "plans"], True diff --git a/ingestion/validation.py b/ingestion/validation.py new file mode 100644 index 0000000..f6442c7 --- /dev/null +++ b/ingestion/validation.py @@ -0,0 +1,57 @@ +"""Read-only domain validation for review; commit also uses finance posting rules.""" +from datetime import date +from finance.services.base import DomainError, money +from finance.models import Transaction + + +def validate(candidate, decision, account, refs, profile): + issues = list(candidate.get("errors", [])) + if account is None: + return issues + ["ACCOUNT_NOT_FOUND"] + if account.archived: + issues.append("ACCOUNT_ARCHIVED") + if account.currency != "BRL": + issues.append("UNSUPPORTED_CURRENCY") + expected = "BANK" if profile == "BANK_CSV_V1" else "CREDIT_CARD" + if account.kind != expected: + issues.append("ACCOUNT_KIND_MISMATCH") + if issues: + return issues + try: + amount = money(candidate["amount"]) + date.fromisoformat(candidate["posting_date"]) + except (DomainError, ValueError, TypeError, KeyError): + return ["INVALID_MOVEMENT"] + kind = decision.get("kind", candidate["kind"]) + if kind not in {"INCOME", "EXPENSE", "REFUND", "TRANSFER"} or (kind in {"INCOME", "REFUND"} and amount <= 0) or (kind == "EXPENSE" and amount >= 0): + issues.append("INVALID_KIND_OR_SIGN") + for field, table in (("payee_id", "payees"), ("category_id", "categories")): + if decision.get(field): + ref = refs[table].get(decision[field]) + if ref is None or ref.archived: + issues.append("INVALID_" + field.upper()) + category = refs["categories"].get(decision.get("category_id")) + if category and category.kind != {"INCOME": "INCOME", "EXPENSE": "EXPENSE", "REFUND": "EXPENSE"}.get(kind): + issues.append("CATEGORY_KIND_MISMATCH") + statement = refs["statements"].get(decision.get("statement_id")) + if decision.get("statement_id") and (statement is None or statement.account_id != account.pk or account.kind != "CREDIT_CARD"): + issues.append("STATEMENT_ACCOUNT_MISMATCH") + original = refs["transactions"].get(decision.get("refund_of_id")) + if decision.get("refund_of_id") and (original is None or kind != "REFUND" or original.kind != "EXPENSE" or original.account.currency != account.currency): + issues.append("INVALID_REFUND_REFERENCE") + number, total = candidate.get("installment_number"), candidate.get("installment_total") + if number is not None or total is not None: + if not isinstance(number, int) or not isinstance(total, int) or not 1 <= number <= total or kind != "EXPENSE": + issues.append("INVALID_INSTALLMENT") + plan_id, new_plan = decision.get("installment_plan_id"), decision.get("new_plan") + if bool(plan_id) == bool(new_plan): + issues.append("INSTALLMENT_PLAN_REQUIRED") + if plan_id: + plan = refs["plans"].get(plan_id) + if plan is None or plan.account_id != account.pk or plan.total_installments != total: + issues.append("INSTALLMENT_PLAN_MISMATCH") + elif decision.get("action") == "CREATE" and Transaction.objects.filter(installment_plan=plan, installment_number=number).exists(): + issues.append("INSTALLMENT_ALREADY_POSTED") + elif decision.get("installment_plan_id") or decision.get("new_plan"): + issues.append("UNEXPECTED_INSTALLMENT_PLAN") + return issues diff --git a/tests/fixtures/synthetic/ingestion/README.md b/tests/fixtures/synthetic/ingestion/README.md new file mode 100644 index 0000000..4d35068 --- /dev/null +++ b/tests/fixtures/synthetic/ingestion/README.md @@ -0,0 +1,12 @@ +# Synthetic import fixtures + +Every name, amount, date, and identifier in this directory is invented for +tests. The files model the three versioned import profiles and contain no data +copied from Robopato or a financial institution. + +- `bank.csv`: UTF-8, comma-separated columns with date at index 0, + description at index 2, and signed decimal amount at index 5. +- `full.txt`: full `DD.MM.YYYY` Ourocard-style fixed-width rows, including a + credit adjustment and account-payment line. +- `partial.txt`: yearless `DD/MM` fixed-width rows crossing a year boundary; + parsing requires explicit start and end dates. diff --git a/tests/fixtures/synthetic/ingestion/bank.csv b/tests/fixtures/synthetic/ingestion/bank.csv new file mode 100644 index 0000000..77266ff --- /dev/null +++ b/tests/fixtures/synthetic/ingestion/bank.csv @@ -0,0 +1,3 @@ +Data,Dependencia,Historico,Documento,Tipo,Valor +01/02/2026,,Synthetic purchase,,,-12.34 +02/02/2026,,Synthetic income,,,100.00 diff --git a/tests/fixtures/synthetic/ingestion/full.txt b/tests/fixtures/synthetic/ingestion/full.txt new file mode 100644 index 0000000..c71e47b --- /dev/null +++ b/tests/fixtures/synthetic/ingestion/full.txt @@ -0,0 +1,4 @@ +1 - SYNTHETIC OWNER +15.02.2026SYNTHETIC SHOP BR 12,34 0,00 +16.02.2026SYNTHETIC REFUND BR -8,11 0,00 +17.02.2026PGTO DEBITO CONTA 200211 33,00 0,00 diff --git a/tests/fixtures/synthetic/ingestion/partial.txt b/tests/fixtures/synthetic/ingestion/partial.txt new file mode 100644 index 0000000..143b2c1 --- /dev/null +++ b/tests/fixtures/synthetic/ingestion/partial.txt @@ -0,0 +1,3 @@ +1 - SYNTHETIC OWNER +31/12SYNTHETIC PARTIAL BR 12,34 0,00 +01/01SYNTHETIC NEW YEAR BR 12,34 0,00 diff --git a/tests/test_import_command.py b/tests/test_import_command.py new file mode 100644 index 0000000..3f1b044 --- /dev/null +++ b/tests/test_import_command.py @@ -0,0 +1,132 @@ +"""Synthetic command-boundary tests; lifecycle persistence has separate coverage.""" +import copy +import importlib +import json +from io import StringIO +from types import SimpleNamespace +from uuid import UUID + +import pytest +from django.core.management import call_command +from django.core.management.base import CommandError + + +BATCH = "11111111-1111-4111-8111-111111111111" +ACTOR = ["--actor-kind", "HUMAN", "--actor-identifier", "synthetic-owner"] + + +@pytest.fixture +def command(tmp_path, settings, monkeypatch): + settings.IMPORT_ROOT = tmp_path + module = importlib.import_module("ingestion.management.commands.import_review") + review = { + "batch_id": BATCH, "revision": 1, "state": "PREVIEW", + "profile": "BANK_CSV_V1", "encoding": "utf-8", + "summary": {"items": 1}, + "items": [{"locator": "line:2", "candidate": {"description": "SYNTHETIC PRIVATE LABEL"}, + "issues": [], "duplicates": [], "decision": None}], + "decisions": [{"locator": "line:2", "action": "CREATE"}], + } + batch = SimpleNamespace(pk=UUID(BATCH), state="PREVIEW", revision=1) + monkeypatch.setattr(module.services, "export_review", lambda *a, **kw: copy.deepcopy(review)) + monkeypatch.setattr(module.services, "apply_decisions", lambda *a, **kw: batch) + return module, review + + +def run(command, operation, *args): + output = StringIO() + call_command(command[0].Command(), operation, *args, *ACTOR, stdout=output) + return output.getvalue() + + +def test_export_keeps_candidate_text_in_local_file_only(command, tmp_path): + output = run(command, "export", "--batch", BATCH, "--output", "review.json") + assert "SYNTHETIC PRIVATE LABEL" not in output + assert json.loads((tmp_path / "review.json").read_text())["items"][0]["candidate"]["description"] == "SYNTHETIC PRIVATE LABEL" + assert json.loads(output)["batch_id"] == BATCH + + +@pytest.mark.parametrize("path", ["../outside.json", "nested/../../outside.json"]) +def test_output_rejects_traversal(command, path): + with pytest.raises(CommandError, match="INVALID_PATH"): + run(command, "export", "--batch", BATCH, "--output", path) + + +def test_output_never_overwrites_input_or_existing_review(command, tmp_path): + destination = tmp_path / "source.json" + destination.write_text("synthetic original") + with pytest.raises(CommandError): + run(command, "export", "--batch", BATCH, "--output", "source.json") + assert destination.read_text() == "synthetic original" + + +def test_output_rejects_symlink_escape(command, tmp_path): + link = tmp_path / "escape" + try: + link.symlink_to(tmp_path.parent, target_is_directory=True) + except OSError: + pytest.skip("Symlink creation unavailable") + with pytest.raises(CommandError, match="INVALID_PATH"): + run(command, "export", "--batch", BATCH, "--output", "escape/review.json") + + +@pytest.mark.parametrize("payload", [ + '{"batch_id":"' + BATCH + '","revision":1,"revision":2,"decisions":[]}', + '{"batch_id":"' + BATCH + '","revision":1,"decisions":[{"locator":"2","action":"CREATE","action":"SKIP"}]}', + '{"batch_id":"' + BATCH + '","revision":NaN,"decisions":[]}', +]) +def test_apply_rejects_ambiguous_json(command, tmp_path, payload): + (tmp_path / "decisions.json").write_text(payload) + with pytest.raises(CommandError, match="INVALID_JSON"): + run(command, "apply", "--batch", BATCH, "--input", "decisions.json") + + +@pytest.mark.parametrize("change", ["candidate", "metadata", "unknown", "revision", "batch"]) +def test_apply_rejects_modified_evidence_or_wrong_envelope(command, tmp_path, change): + payload = copy.deepcopy(command[1]) + if change == "candidate": + payload["items"][0]["candidate"]["description"] = "tampered" + elif change == "metadata": + payload["encoding"] = "latin-1" + elif change == "unknown": + payload["extra"] = True + elif change == "revision": + payload["revision"] = True + else: + payload["batch_id"] = "22222222-2222-4222-8222-222222222222" + (tmp_path / "decisions.json").write_text(json.dumps(payload)) + with pytest.raises(CommandError): + run(command, "apply", "--batch", BATCH, "--input", "decisions.json") + + +@pytest.mark.parametrize("full", [True, False]) +def test_apply_accepts_decisions_only_or_unchanged_full_export(command, tmp_path, full): + payload = copy.deepcopy(command[1]) + payload["decisions"] = [{"locator": "line:2", "action": "SKIP", "reason": "USER_EXCLUDED"}] + if not full: + payload = {key: payload[key] for key in ("batch_id", "revision", "decisions")} + (tmp_path / "decisions.json").write_text(json.dumps(payload)) + output = run(command, "apply", "--batch", BATCH, "--input", "decisions.json") + assert "SYNTHETIC PRIVATE LABEL" not in output + assert json.loads(output)["state"] == "PREVIEW" + + +def test_file_read_is_bounded(command, tmp_path): + with (tmp_path / "large.json").open("wb") as stream: + stream.truncate(10 * 1024 * 1024 + 1) + with pytest.raises(CommandError, match="INPUT_TOO_LARGE"): + run(command, "apply", "--batch", BATCH, "--input", "large.json") + + +def test_service_exception_does_not_disclose_source(command, monkeypatch): + def fail(*args, **kwargs): + raise RuntimeError("SYNTHETIC PRIVATE LABEL") + monkeypatch.setattr(command[0].services, "export_review", fail) + with pytest.raises(CommandError) as caught: + run(command, "export", "--batch", BATCH, "--output", "review.json") + assert "SYNTHETIC" not in str(caught.value) + + +def test_all_operations_require_explicit_actor(command): + with pytest.raises(CommandError): + call_command(command[0].Command(), "export", "--batch", BATCH, "--output", "review.json") diff --git a/tests/test_import_concurrency.py b/tests/test_import_concurrency.py new file mode 100644 index 0000000..a853a5f --- /dev/null +++ b/tests/test_import_concurrency.py @@ -0,0 +1,72 @@ +"""Competing import commits use independent PostgreSQL connections.""" + +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier + +import pytest +from django.db import close_old_connections, connection, connections + +from finance.models import AuditEvent, SourceRecord, Transaction +from finance.services import Actor, DomainError, create_account +from ingestion import services + +pytestmark = pytest.mark.django_db(transaction=True) +ACTOR = Actor(kind="AGENT", identifier="synthetic-import-concurrency") +DATA = b"Data,Documento,Historico,Agencia,Lote,Valor\n15/01/2026,1,Synthetic purchase,0,0,-12.50\n" + + +def reviewed(bank): + batch = services.create_preview(data=DATA, profile="BANK_CSV_V1", section_mapping={"1": str(bank.pk)}, actor=ACTOR) + return services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR, + decisions=[{"locator": batch.items.get().locator, "action": "CREATE"}]) + + +def race(batches): + assert connection.vendor == "postgresql" + barrier = Barrier(2) + + def attempt(batch): + close_old_connections() + try: + with connection.cursor() as cursor: + cursor.execute("SET statement_timeout = '20s'") + cursor.execute("SELECT pg_backend_pid()") + backend_pid = cursor.fetchone()[0] + barrier.wait(timeout=15) + try: + result = services.commit_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) + return backend_pid, "success", result.result + except DomainError as error: + return backend_pid, "failure", error.code + finally: + connections.close_all() + + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(attempt, batches)) + assert len({item[0] for item in results}) == 2 + return results + + +def test_two_commits_of_same_batch_return_same_result_without_duplicate_audit(): + bank = create_account(name="Synthetic concurrent bank", kind="BANK", actor=ACTOR) + batch = reviewed(bank) + previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) + results = race([batch, batch]) + assert [item[1] for item in results] == ["success", "success"] + assert results[0][2] == results[1][2] + assert Transaction.objects.count() == SourceRecord.objects.count() == 1 + assert list(AuditEvent.objects.exclude(pk__in=previous_audits).values_list("outcome", flat=True)) == ["SUCCESS"] + + +def test_identical_batches_have_one_success_and_one_stale_review(): + bank = create_account(name="Synthetic concurrent bank", kind="BANK", actor=ACTOR) + batches = [reviewed(bank), reviewed(bank)] + previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) + results = race(batches) + assert sorted(item[1] for item in results) == ["failure", "success"] + assert next(item[2] for item in results if item[1] == "failure") == "STALE_PREVIEW" + assert Transaction.objects.count() == SourceRecord.objects.count() == 1 + for batch in batches: + batch.refresh_from_db() + assert sorted(batch.state for batch in batches) == ["COMMITTED", "PREVIEW"] + assert sorted(AuditEvent.objects.exclude(pk__in=previous_audits).values_list("outcome", flat=True)) == ["FAILURE", "SUCCESS"] diff --git a/tests/test_import_parsers.py b/tests/test_import_parsers.py new file mode 100644 index 0000000..d86d090 --- /dev/null +++ b/tests/test_import_parsers.py @@ -0,0 +1,127 @@ +"""All examples in this module are invented synthetic financial records.""" + +from decimal import localcontext, Inexact +from pathlib import Path + +import pytest + +from ingestion.parsers import ParseError, parse_document +from ingestion.normalization import normalize + + +HEADER = 'Data,Dependencia,Historico,Documento,Tipo,Valor\n' + + +def bank(rows, encoding='utf-8'): + return parse_document((HEADER + rows).encode(encoding), 'BANK_CSV_V1') + + +def card_line(day='15.02.2026', body='LOJA INVENTADA', debit='12,34', credit='0,00', partial=False): + return day + body.ljust(43 if partial else 37) + 'BR' + f' {debit} {credit}' + + +def card(text, partial=False, context=None): + return normalize(parse_document(text.encode(), 'OUROCARD_PARTIAL_V1' if partial else 'OUROCARD_FULL_V1', context)) + + +def test_csv_quotes_signs_balances_encoding_and_physical_locators(): + result = bank('01/02/2026,,"Café, inventado",,,-12.34\n02/02/2026,,S A L D O,,,22.00\n\n03/02/2026,,RECEITA,,,23.00\n', 'latin-1') + rows = normalize(result) + assert result.encoding == 'latin-1' + assert sum(result.skipped.values()) == 2 + assert [r.to_dict()['locator'] for r in rows] == ['line:2', 'line:5'] + assert rows[0].to_dict() == dict(locator='line:2', section='1', posting_date='2026-02-01', occurrence_date='2026-02-01', description='Café, inventado', amount='-12.34', kind='EXPENSE', installment_number=None, installment_total=None, errors=[]) + assert rows[1].kind == 'INCOME' + + +@pytest.mark.parametrize('data,profile,code', [(b'\x00','BANK_CSV_V1','NUL_BYTE'), (b'x','OFX','UNSUPPORTED_PROFILE'), (b'a,b,c\n','BANK_CSV_V1','INVALID_HEADER'), (b'x' * (10*1024*1024+1),'BANK_CSV_V1','INPUT_TOO_LARGE')], ids=['nul','unsupported','header','size']) +def test_file_errors(data, profile, code): + with pytest.raises(ParseError) as exc: + parse_document(data, profile) + assert exc.value.code == code + + +def test_bad_csv_rows_are_reviewable_without_raw_diagnostics(): + rows = normalize(bank('31/02/2026,,SECRET,,,NaN\nnot,a,row\n')) + assert set(rows[0].errors) == {'INVALID_DATE', 'INVALID_AMOUNT'} + assert rows[0].description == 'SECRET' + assert rows[1].errors == ('INVALID_ROW',) + assert rows[1].description == 'row' + assert 'SECRET' not in str(rows[0].errors) + + +def test_full_card_sections_installments_credit_and_payment(): + rows = card('1 - SYNTHETIC OWNER\n' + card_line(body='LOJA PARC 2/3') + '\n2 - SECOND INVENTED OWNER\n' + card_line(debit='-8,11') + '\n15.02.2026PGTO DEBITO CONTA 200211 33,00 0,00\nSaldo Anterior 20,00\nunknown text') + assert rows[0].section == '1' + assert rows[0].installment_number == 2 and rows[0].installment_total == 3 + assert rows[1].section == '2' and rows[1].amount == '8.11' and rows[1].kind == 'REFUND' + assert rows[2].amount == '33.00' and rows[2].kind == 'TRANSFER' + assert rows[3].errors == ('UNRECOGNIZED_LINE',) + assert rows[3].description == 'unknown text' + + +def test_card_ambiguity_bad_shapes_and_installments_block(): + assert 'AMBIGUOUS_AMOUNT' in card(card_line(debit='2,00', credit='1,00'))[0].errors + assert 'INVALID_ROW' in card('15.02.2026 MALFORMED')[0].errors + assert 'INVALID_INSTALLMENT' in card(card_line(body='LOJA PARC 3/2'))[0].errors + assert 'INVALID_AMOUNT' in card(card_line(debit='1.23,00'))[0].errors + + +def test_partial_dates_require_explicit_unique_context_and_retain_snapshot(): + text = card_line(day='31/12', partial=True) + '\n' + card_line(day='01/01', partial=True) + context = {'start_date':'2025-12-01','end_date':'2026-01-31'} + parsed = parse_document(text.encode(), 'OUROCARD_PARTIAL_V1', context) + context['start_date'] = '2000-01-01' + assert [r.posting_date for r in normalize(parsed)] == ['2025-12-31','2026-01-01'] + assert 'MISSING_DATE_CONTEXT' in card(text, True)[0].errors + assert 'AMBIGUOUS_DATE' in card(card_line(day='01/01',partial=True), True, {'start_date':'2025-01-01','end_date':'2026-01-01'})[0].errors + assert 'INVALID_DATE_CONTEXT' in card(text, True, {'start_date':'2024-01-01','end_date':'2026-01-01'})[0].errors + + +def test_decimal_context_does_not_change_amounts(): + with localcontext() as ctx: + ctx.prec = 2 + ctx.traps[Inexact] = True + assert card(card_line(debit='12.345,67'))[0].amount == '-12345.67' + assert card('15.02.2026PGTO DEBITO CONTA 200211 -12.345,67 0,00')[0].amount == '12345.67' + + +def test_csv_multiline_locator_and_bom(): + parsed = parse_document(('\ufeff' + HEADER + '01/02/2026,,"two\nlines",,,1.00\n02/02/2026,,next,,,2.00\n').encode(), 'BANK_CSV_V1') + assert parsed.encoding == 'utf-8-sig' + assert [r.locator for r in normalize(parsed)] == ['line:2','line:4'] + + +def test_candidate_row_limit(): + with pytest.raises(ParseError, match='TOO_MANY_ITEMS'): + bank('01/02/2026,,INVENTED,,,1.00\n' * 20001) + + +def test_invalid_quoting_is_file_error_and_header_positions_are_fixed(): + with pytest.raises(ParseError, match='INVALID_CSV'): + bank('01/02/2026,,"unterminated,,,1.00') + with pytest.raises(ParseError, match='INVALID_HEADER'): + parse_document(b'Valor,Unused,Historico,Unused,Unused,Data\n', 'BANK_CSV_V1') + + +def test_partial_leap_date_and_invalid_ranges(): + text = card_line(day='29/02', partial=True) + assert card(text, True, {'start_date':'2024-01-01','end_date':'2024-12-31'})[0].posting_date == '2024-02-29' + assert card(text, True, {'start_date':'2025-01-01','end_date':'2025-12-31'})[0].errors == ('INVALID_DATE',) + assert card(text, True, {'start_date':'2026-01-01','end_date':'2025-12-31'})[0].errors == ('INVALID_DATE_CONTEXT',) + + +def test_nonzero_payment_secondary_column_is_ambiguous(): + assert card('15.02.2026PGTO DEBITO CONTA 200211 33,00 2,00')[0].errors == ('AMBIGUOUS_AMOUNT',) + + +@pytest.mark.parametrize('filename,profile,amounts', [ + ('bank.csv', 'BANK_CSV_V1', ['-12.34','100.00']), + ('full.txt', 'OUROCARD_FULL_V1', ['-12.34','8.11','33.00']), + ('partial.txt', 'OUROCARD_PARTIAL_V1', ['-12.34','-12.34']), +]) +def test_synthetic_fixture_profiles(filename, profile, amounts): + data = (Path(__file__).parent / 'fixtures' / 'synthetic' / 'ingestion' / filename).read_bytes() + rows = normalize(parse_document(data, profile, {'start_date':'2025-12-01','end_date':'2026-01-31'})) + assert [row.amount for row in rows] == amounts + assert all(not row.errors for row in rows) diff --git a/tests/test_import_workflow.py b/tests/test_import_workflow.py new file mode 100644 index 0000000..7a4d608 --- /dev/null +++ b/tests/test_import_workflow.py @@ -0,0 +1,264 @@ +"""Synthetic acceptance cases for explicit review and atomic import persistence.""" + +from datetime import date +from decimal import Decimal +from hashlib import sha256 + +import pytest + +from finance.models import AuditEvent, InstallmentPlan, SourceRecord, Transaction, Transfer +from finance.services import Actor, DomainError, archive, create_account, create_category, create_payee, record_transaction +from ingestion import services + +pytestmark = pytest.mark.django_db(transaction=True) +ACTOR = Actor(kind="AGENT", identifier="synthetic-import-test") +HEADER = "Data,Documento,Historico,Agencia,Lote,Valor\n" +ROW = "15/01/2026,1,Synthetic purchase,0,0,-12.50\n" +DATA = (HEADER + ROW).encode() + + +def account(name="Synthetic import bank"): + return create_account(name=name, kind="BANK", actor=ACTOR) + + +def preview(bank, data=DATA): + return services.create_preview(data=data, profile="BANK_CSV_V1", section_mapping={"1": str(bank.pk)}, actor=ACTOR) + + +def decide(batch, **fields): + return services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR, + decisions=[{"locator": item.locator, "action": "CREATE", **fields} + for item in batch.items.all()]) + + +def commit(batch): + return services.commit_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) + + +def assert_no_financial_rows(): + for model in (Transaction, SourceRecord, InstallmentPlan, Transfer): + assert model.objects.count() == 0 + + +def test_preview_and_decisions_are_inert_then_commit_preserves_source_and_audit(): + batch = preview(account()) + assert batch.state == "PREVIEW" + assert batch.revision == 1 + assert_no_financial_rows() + review = services.export_review(batch_id=batch.pk) + assert review["batch_id"] == str(batch.pk) + assert review["revision"] == 1 + assert len(review["items"]) == 1 + assert review["items"][0]["candidate"]["amount"] == "-12.50" + batch = decide(batch) + assert batch.revision == 2 + assert_no_financial_rows() + previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) + result = commit(batch) + assert result.state == "COMMITTED" + assert result.revision == batch.revision + row = Transaction.objects.get() + assert row.amount == Decimal("-12.50") + assert row.posting_date == row.occurrence_date == date(2026, 1, 15) + source = SourceRecord.objects.get() + assert source.transaction_id == row.pk + assert source.document_hash == sha256(DATA).hexdigest() + assert source.item_locator == batch.items.get().locator + events = list(AuditEvent.objects.exclude(pk__in=previous_audits)) + assert len(events) == 1 + assert events[0].outcome == "SUCCESS" + assert events[0].actor_identifier == ACTOR.identifier + audit_count = AuditEvent.objects.count() + repeated = commit(batch) + assert repeated.result == result.result + assert Transaction.objects.count() == SourceRecord.objects.count() == 1 + assert AuditEvent.objects.count() == audit_count + + +def test_every_row_requires_explicit_decision_and_rejected_commit_is_audited(): + batch = preview(account()) + previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) + with pytest.raises(DomainError) as caught: + commit(batch) + assert caught.value.code + assert_no_financial_rows() + batch.refresh_from_db() + assert batch.state == "PREVIEW" + event = AuditEvent.objects.exclude(pk__in=previous_audits).get() + assert event.outcome == "FAILURE" + assert event.error_code == caught.value.code + assert event.actor_identifier == ACTOR.identifier + + +def test_malformed_row_can_only_be_omitted_with_explicit_skip(): + batch = preview(account(), (HEADER + "31/02/2026,1,Synthetic invalid date,0,0,-12.50\n").encode()) + assert batch.items.get().issues + with pytest.raises(DomainError): + commit(batch) + batch = decide(batch, action="SKIP", reason="MALFORMED") + assert commit(batch).state == "COMMITTED" + assert_no_financial_rows() + + +def test_exact_reimport_requires_already_imported_skip(): + bank = account() + commit(decide(preview(bank))) + batch = preview(bank) + assert "ALREADY_IMPORTED" in str(batch.items.get().duplicates) + with pytest.raises(DomainError): + commit(batch) + batch = decide(batch, action="SKIP", reason="ALREADY_IMPORTED") + assert commit(batch).state == "COMMITTED" + assert Transaction.objects.count() == SourceRecord.objects.count() == 1 + + +def test_semantic_duplicates_require_acknowledgement_but_can_be_legitimate(): + bank = account() + commit(decide(preview(bank))) + # The document ID changes the physical source without changing financial fields. + batch = preview(bank, DATA.replace(b",1,", b",2,")) + assert batch.items.get().duplicates + batch = decide(batch) + with pytest.raises(DomainError): + commit(batch) + batch = decide(batch, acknowledge_duplicate=True) + commit(batch) + assert Transaction.objects.count() == SourceRecord.objects.count() == 2 + + +def test_match_attaches_provenance_without_changing_existing_annotations(): + bank = account() + category = create_category(name="Synthetic category", kind="EXPENSE", actor=ACTOR) + payee = create_payee(name="Synthetic payee", actor=ACTOR) + row = record_transaction(account_id=bank.pk, amount=Decimal("-12.50"), kind="EXPENSE", + posting_date=date(2026, 1, 15), occurrence_date=date(2026, 1, 15), + description="Synthetic purchase", category_id=category.pk, + payee_id=payee.pk, actor=ACTOR) + before = (row.description, row.category_id, row.payee_id, row.updated_at) + batch = decide(preview(bank), action="MATCH", transaction_id=str(row.pk)) + commit(batch) + row.refresh_from_db() + assert (row.description, row.category_id, row.payee_id, row.updated_at) == before + assert Transaction.objects.count() == 1 + assert SourceRecord.objects.get().transaction_id == row.pk + + +def test_cancellation_and_stale_revision_cannot_commit(): + batch = preview(account()) + old_revision = batch.revision + batch = decide(batch) + with pytest.raises(DomainError): + services.commit_preview(batch_id=batch.pk, revision=old_revision, actor=ACTOR) + batch = services.cancel_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) + assert batch.state == "CANCELLED" + with pytest.raises(DomainError): + commit(batch) + assert_no_financial_rows() + + +def test_new_duplicate_makes_review_stale_and_refresh_requires_new_consent(): + bank = account() + batch = decide(preview(bank)) + record_transaction(account_id=bank.pk, amount=Decimal("-12.50"), kind="EXPENSE", + posting_date=date(2026, 1, 15), occurrence_date=date(2026, 1, 15), + description="Synthetic purchase", actor=ACTOR) + with pytest.raises(DomainError) as caught: + commit(batch) + assert caught.value.code == "STALE_PREVIEW" + assert SourceRecord.objects.count() == 0 + refreshed = services.refresh_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR) + assert refreshed.revision == batch.revision + 1 + assert all(not item.decision for item in refreshed.items.all()) + with pytest.raises(DomainError): + commit(refreshed) + commit(decide(refreshed, acknowledge_duplicate=True)) + assert Transaction.objects.count() == 2 + assert SourceRecord.objects.count() == 1 + + +def test_archiving_destination_invalidates_review(): + bank = account() + batch = decide(preview(bank)) + archive(bank, actor=ACTOR) + with pytest.raises(DomainError) as caught: + commit(batch) + assert caught.value.code == "STALE_PREVIEW" + assert_no_financial_rows() + + +def test_later_row_failure_rolls_back_posting_sources_and_success_audit(monkeypatch): + batch = decide(preview(account(), (HEADER + ROW + "16/01/2026,2,Synthetic second purchase,0,0,-7.25\n").encode())) + original = services.record + calls = [] + + def fail_second(**kwargs): + calls.append(kwargs) + if len(calls) == 2: + assert Transaction.objects.count() == SourceRecord.objects.count() == 1 + raise DomainError("synthetic_later_row_failure") + return original(**kwargs) + + monkeypatch.setattr(services, "record", fail_second) + previous_audits = set(AuditEvent.objects.values_list("pk", flat=True)) + with pytest.raises(DomainError) as caught: + commit(batch) + assert len(calls) == 2 + assert caught.value.code == "synthetic_later_row_failure" + assert_no_financial_rows() + batch.refresh_from_db() + assert batch.state == "PREVIEW" + event = AuditEvent.objects.exclude(pk__in=previous_audits).get() + assert event.outcome == "FAILURE" + assert event.error_code == caught.value.code + + +@pytest.mark.parametrize("extra", [{"amount": "-1.00"}, {"description": "Replacement"}, {"unexpected": True}]) +def test_decisions_cannot_change_source_evidence_or_accept_unknown_fields(extra): + batch = preview(account()) + with pytest.raises(DomainError): + decide(batch, **extra) + assert_no_financial_rows() + + +def test_duplicate_locators_are_rejected(): + batch = preview(account()) + decision = {"locator": batch.items.get().locator, "action": "CREATE"} + with pytest.raises(DomainError): + services.apply_decisions(batch_id=batch.pk, revision=batch.revision, decisions=[decision, decision], actor=ACTOR) + assert_no_financial_rows() + + +def test_installments_share_explicit_new_plan_without_creating_unobserved_rows(): + card = create_account(name="Synthetic installment card", kind="CREDIT_CARD", actor=ACTOR) + data = (f"15.01.2026{'Synthetic purchase PARC 2/6':37}BR 12,50 0,00\n" + f"15.02.2026{'Synthetic purchase PARC 3/6':37}BR 12,50 0,00\n").encode() + batch = services.create_preview(data=data, profile="OUROCARD_FULL_V1", + section_mapping={"1": str(card.pk)}, actor=ACTOR) + batch = decide(batch, new_plan="synthetic-purchase") + assert_no_financial_rows() + commit(batch) + plan = InstallmentPlan.objects.get() + assert plan.account_id == card.pk + assert plan.total_installments == 6 + assert set(Transaction.objects.values_list("installment_number", flat=True)) == {2, 3} + assert set(Transaction.objects.values_list("installment_plan_id", flat=True)) == {plan.pk} + assert SourceRecord.objects.count() == 2 + + +def test_applying_decisions_replaces_complete_set_and_missing_rows_block_commit(): + batch = decide(preview(account(), (HEADER + ROW + "16/01/2026,2,Synthetic second purchase,0,0,-7.25\n").encode())) + first = batch.items.order_by("locator").first() + batch = services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR, + decisions=[{"locator": first.locator, "action": "CREATE"}]) + assert not batch.items.exclude(pk=first.pk).get().decision + with pytest.raises(DomainError): + commit(batch) + assert_no_financial_rows() + + +def test_source_already_attached_to_another_account_cannot_be_claimed(): + commit(decide(preview(account()))) + batch = decide(preview(account("Synthetic second import bank"))) + with pytest.raises(DomainError): + commit(batch) + assert Transaction.objects.count() == SourceRecord.objects.count() == 1