From 4720d3053f3d66bc030491412309b42e4b126ed0 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 9 Sep 2026 20:00:58 -0300 Subject: [PATCH] feat: discover local import files --- ingestion/guided_imports.py | 71 +++++++++++++++++++++++++++ ingestion/parsers.py | 42 +++++++++++++--- tests/test_guided_import_discovery.py | 61 +++++++++++++++++++++++ tests/test_import_parsers.py | 29 ++++++++++- 4 files changed, 195 insertions(+), 8 deletions(-) create mode 100644 ingestion/guided_imports.py create mode 100644 tests/test_guided_import_discovery.py diff --git a/ingestion/guided_imports.py b/ingestion/guided_imports.py new file mode 100644 index 0000000..e1a29be --- /dev/null +++ b/ingestion/guided_imports.py @@ -0,0 +1,71 @@ +"""Safe discovery of local source documents for the guided import workflow.""" + +from dataclasses import dataclass +from hashlib import sha256 +from pathlib import Path + +from django.conf import settings + +from .parsers import MAX_BYTES, ParseError, parse_document + + +@dataclass(frozen=True) +class DiscoveredFile: + name: str + document_hash: str + profile: str | None + sections: tuple[str, ...] + code: str | None + + +_PROFILES_BY_SUFFIX = { + ".csv": ("BANK_CSV_V1",), + ".txt": ("OUROCARD_FULL_V1", "OUROCARD_PARTIAL_V1"), +} + + +def _profile_and_sections(path: Path, data: bytes) -> tuple[str | None, tuple[str, ...]]: + for profile in _PROFILES_BY_SUFFIX.get(path.suffix.casefold(), ()): + try: + result = parse_document(data, profile) + except ParseError: + continue + items = tuple(item for item in result.items if not item.errors) + if items: + return profile, tuple(dict.fromkeys(item.section for item in items)) + return None, () + + +def _read_discovered_file(path: Path) -> DiscoveredFile: + try: + with path.open("rb") as stream: + data = stream.read(MAX_BYTES + 1) + except OSError: + return DiscoveredFile(path.name, "", None, (), "UNREADABLE_FILE") + if len(data) > MAX_BYTES: + return DiscoveredFile(path.name, "", None, (), "INPUT_TOO_LARGE") + document_hash = sha256(data).hexdigest() + if path.suffix.casefold() not in _PROFILES_BY_SUFFIX: + return DiscoveredFile(path.name, document_hash, None, (), "UNSUPPORTED_FILE") + profile, sections = _profile_and_sections(path, data) + if profile is None: + return DiscoveredFile(path.name, document_hash, None, (), "UNRECOGNIZED_LAYOUT") + return DiscoveredFile(path.name, document_hash, profile, sections, None) + + +def discover_imports() -> tuple[DiscoveredFile, ...]: + """Return safe metadata for direct regular files under ``IMPORT_ROOT``.""" + root = Path(settings.IMPORT_ROOT).resolve() + try: + entries = root.iterdir() + except OSError: + return () + discovered = [] + for path in entries: + try: + if path.is_symlink() or not path.is_file(): + continue + except OSError: + continue + discovered.append(_read_discovered_file(path)) + return tuple(sorted(discovered, key=lambda row: row.name)) diff --git a/ingestion/parsers.py b/ingestion/parsers.py index ed831e8..a2bc0c3 100644 --- a/ingestion/parsers.py +++ b/ingestion/parsers.py @@ -6,6 +6,8 @@ import re import unicodedata from collections import Counter from dataclasses import dataclass +from datetime import date +from calendar import monthrange from types import MappingProxyType from typing import Mapping @@ -112,22 +114,48 @@ def _txt_items(text, profile, skipped): 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') +def _decode_source(data: bytes) -> tuple[str, str]: 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) + return encoding, data.decode(encoding) except UnicodeDecodeError: - encoding, text = 'latin-1', data.decode('latin-1') + return 'latin-1', data.decode('latin-1') + + +def partial_context_from_creation_line(data: bytes) -> dict[str, str]: + """Build the bounded partial-statement interval from its line-21 issue date.""" + _, text = _decode_source(data) + lines = text.splitlines() + if len(lines) < 21: + raise ParseError('MISSING_CREATION_DATE') + match = re.fullmatch(r'DATA\s+DE\s+EMISS(?:A|Ã)O\s*:\s*(\d{2}/\d{2}/\d{4})', lines[20].strip(), re.I) + if match is None: + raise ParseError('INVALID_CREATION_DATE') + try: + creation = date.fromisoformat('-'.join(reversed(match[1].split('/')))) + except ValueError: + raise ParseError('INVALID_CREATION_DATE') from None + start_year = creation.year if creation.month == 12 else creation.year - 1 + start_month = creation.month % 12 + 1 + start = date(start_year, start_month, 1) + end = date(creation.year, creation.month, monthrange(creation.year, creation.month)[1]) + if (end - start).days > 366: + raise ParseError('INVALID_CREATION_DATE') + return {'start_date': start.isoformat(), 'end_date': end.isoformat()} + + +def parse_document(data: bytes, profile: str, context: dict | None = None) -> ParseResult: + if profile not in PROFILES: + raise ParseError('UNSUPPORTED_PROFILE') + 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') + encoding, text = _decode_source(data) skipped = Counter() source = _csv_items(text, skipped) if profile == 'BANK_CSV_V1' else _txt_items(text, profile, skipped) items = [] diff --git a/tests/test_guided_import_discovery.py b/tests/test_guided_import_discovery.py new file mode 100644 index 0000000..64feb11 --- /dev/null +++ b/tests/test_guided_import_discovery.py @@ -0,0 +1,61 @@ +"""Synthetic checks for source-safe local import discovery.""" + +from hashlib import sha256 + +from django.test import override_settings + +from ingestion.guided_imports import discover_imports + + +HEADER = b"Data,Dependencia,Historico,Documento,Tipo,Valor\n" + + +def test_discovery_returns_only_safe_metadata_for_direct_supported_files(tmp_path): + bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n" + full = b"1 - SYNTHETIC OWNER\n" + b"15.02.2026" + b"SYNTHETIC CARD ROW".ljust(37) + b"BR 12,34 0,00\n" + nested = tmp_path / "nested" + nested.mkdir() + (tmp_path / "bank.csv").write_bytes(bank) + (tmp_path / "card.txt").write_bytes(full) + (nested / "hidden.csv").write_bytes(bank) + + with override_settings(IMPORT_ROOT=tmp_path): + rows = discover_imports() + + assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [ + ("bank.csv", "BANK_CSV_V1", ("1",), None), + ("card.txt", "OUROCARD_FULL_V1", ("1",), None), + ] + assert [row.document_hash for row in rows] == [sha256(bank).hexdigest(), sha256(full).hexdigest()] + assert all(not hasattr(row, "source_text") for row in rows) + + +def test_discovery_reports_safe_codes_for_direct_unsupported_or_invalid_files(tmp_path): + (tmp_path / "unknown.ofx").write_bytes(b"SYNTHETIC OFX") + (tmp_path / "statement.pdf").write_bytes(b"%PDF-SYNTHETIC") + (tmp_path / "invalid.txt").write_bytes(b"SYNTHETIC UNRECOGNIZED TEXT") + (tmp_path / "oversize.csv").write_bytes(b"x" * (10 * 1024 * 1024 + 1)) + + with override_settings(IMPORT_ROOT=tmp_path): + rows = discover_imports() + + assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [ + ("invalid.txt", None, (), "UNRECOGNIZED_LAYOUT"), + ("oversize.csv", None, (), "INPUT_TOO_LARGE"), + ("statement.pdf", None, (), "UNSUPPORTED_FILE"), + ("unknown.ofx", None, (), "UNSUPPORTED_FILE"), + ] + assert all(not hasattr(row, "source_text") for row in rows) + + +def test_discovery_ignores_symlinks_and_non_files(tmp_path): + bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n" + target = tmp_path / "target.csv" + target.write_bytes(bank) + (tmp_path / "linked.csv").symlink_to(target) + (tmp_path / "directory.csv").mkdir() + + with override_settings(IMPORT_ROOT=tmp_path): + rows = discover_imports() + + assert [row.name for row in rows] == ["target.csv"] diff --git a/tests/test_import_parsers.py b/tests/test_import_parsers.py index 3fc3ae7..688a17b 100644 --- a/tests/test_import_parsers.py +++ b/tests/test_import_parsers.py @@ -5,7 +5,7 @@ from pathlib import Path import pytest -from ingestion.parsers import ParseError, parse_document +from ingestion.parsers import ParseError, parse_document, partial_context_from_creation_line from ingestion.normalization import normalize @@ -127,6 +127,33 @@ def test_nonzero_payment_secondary_column_is_ambiguous(): assert card('15.02.2026PGTO DEBITO CONTA 200211 33,00 2,00')[0].errors == ('AMBIGUOUS_AMOUNT',) +def test_partial_context_uses_the_validated_line_21_creation_date(): + data = ('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/2026']) + '\n').encode() + + assert partial_context_from_creation_line(data) == { + 'start_date': '2025-02-01', + 'end_date': '2026-01-31', + } + + +@pytest.mark.parametrize('data,code', [ + (b'SYNTHETIC\n' * 20, 'MISSING_CREATION_DATE'), + (('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/26']) + '\n').encode(), 'INVALID_CREATION_DATE'), +], ids=['absent-line-21', 'malformed-line-21']) +def test_partial_context_rejects_missing_or_malformed_line_21_creation_date(data, code): + with pytest.raises(ParseError, match=code): + partial_context_from_creation_line(data) + + +def test_partial_context_resolves_december_rows_to_the_previous_year_after_january_creation(): + data = ('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/2026']) + '\n').encode() + context = partial_context_from_creation_line(data) + + rows = card(card_line(day='31/12', partial=True), True, context) + + assert rows[0].posting_date == '2025-12-31' + + @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']),