diff --git a/ingestion/guided_imports.py b/ingestion/guided_imports.py index 18ef358..88b1fc8 100644 --- a/ingestion/guided_imports.py +++ b/ingestion/guided_imports.py @@ -2,6 +2,7 @@ import os import stat +from contextlib import contextmanager from dataclasses import dataclass, replace from hashlib import sha256 from pathlib import Path @@ -127,12 +128,94 @@ def _read_discovered_file(name: str, root_descriptor: int, expected: os.stat_res return DiscoveredFile(name, document_hash, profile, sections, None) +@contextmanager +def _directory(name: str | None = None): + root = _open_import_root() + descriptor = None + try: + if root is None: + raise ParseError("UNREADABLE_FILE") + if name is None: + yield root + else: + try: + os.mkdir(name, mode=0o700, dir_fd=root) + except FileExistsError: + pass + descriptor = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root) + yield descriptor + except OSError: + raise ParseError("UNREADABLE_FILE") from None + finally: + _close(descriptor) + _close(root) + + +def _validate_name(name: str) -> None: + if (not isinstance(name, str) or not name or name in {".", ".."} + or any(char in name for char in ("/", "\\", "\x00", ":"))): + raise ParseError("UNREADABLE_FILE") + + +def _move_file(name: str, source: str | None, destination: str | None, *, overwrite=False) -> None: + _validate_name(name) + with _directory(source) as source_fd, _directory(destination) as destination_fd: + try: + expected = os.stat(name, dir_fd=source_fd, follow_symlinks=False) + except FileNotFoundError: + if source == "staging": + return + raise + if not stat.S_ISREG(expected.st_mode): + raise ParseError("UNREADABLE_FILE") + try: + target = os.stat(name, dir_fd=destination_fd, follow_symlinks=False) + except FileNotFoundError: + pass + else: + if not overwrite or not stat.S_ISREG(target.st_mode): + raise ParseError("UNREADABLE_FILE") + os.replace(name, name, src_dir_fd=source_fd, dst_dir_fd=destination_fd) + + +def _recover_staging() -> None: + """Move direct regular staging files back to incoming before discovery.""" + try: + with _directory("staging") as descriptor: + for name in os.listdir(descriptor): + try: + _return_to_incoming(name) + except ParseError: + # Leave unsafe entries or filename collisions for local attention. + continue + except ParseError: + return + + +def _stage(discovered: DiscoveredFile) -> DiscoveredFile: + """Atomically rename one verified incoming file into staging.""" + _verified_bytes(discovered) + _move_file(discovered.name, None, "staging") + return discovered + + +def _return_to_incoming(name: str) -> None: + """Move a staged file back when it was not committed.""" + _move_file(name, "staging", None) + + +def _move_to_processed(name: str) -> None: + """Replace the processed filename after a successful commit.""" + _move_file(name, "staging", "processed", overwrite=True) + + def discover_imports() -> tuple[DiscoveredFile, ...]: """Return safe metadata for direct regular files under ``IMPORT_ROOT``.""" root_descriptor = _open_import_root() if root_descriptor is None: return () try: + _recover_staging() try: entries = os.listdir(root_descriptor) except OSError: @@ -162,6 +245,7 @@ class PreparedImport: revision: int | None counts: dict[str, int] code: str | None + staged: bool = False _SAFE_CODES = frozenset({ @@ -174,21 +258,13 @@ _SAFE_CODES = frozenset({ def _verified_bytes(discovered: DiscoveredFile | PreparedImport) -> bytes: name = discovered.name - if (not isinstance(name, str) or not name or name in {".", ".."} - or any(char in name for char in ("/", "\\", "\x00", ":"))): - raise ParseError("UNREADABLE_FILE") - root_descriptor = _open_import_root() - if root_descriptor is None: - raise ParseError("UNREADABLE_FILE") - try: + _validate_name(name) + directory = "staging" if isinstance(discovered, PreparedImport) and discovered.staged else None + with _directory(directory) as root_descriptor: expected = os.stat(name, dir_fd=root_descriptor, follow_symlinks=False) if not stat.S_ISREG(expected.st_mode): raise ParseError("UNREADABLE_FILE") data = _read_file_bytes(name, root_descriptor, expected) - except OSError: - raise ParseError("UNREADABLE_FILE") from None - finally: - _close(root_descriptor) if sha256(data).hexdigest() != discovered.document_hash: raise ParseError("FILE_CHANGED") profile, sections = _profile_and_sections(Path(name), data) @@ -227,7 +303,7 @@ def _cancel_unfinished(batch_id, actor): raise DomainError("CANCELLATION_FAILED") -def _attention(prepared, error, actor, *, cancel=True): +def _attention(prepared, error, actor, *, cancel=True, return_source=True): code = getattr(error, "code", None) code = code if isinstance(code, str) and code in _SAFE_CODES else "IMPORT_FAILED" if cancel and prepared.batch_id: @@ -235,6 +311,12 @@ def _attention(prepared, error, actor, *, cancel=True): _cancel_unfinished(prepared.batch_id, actor) except DomainError: code = "CANCELLATION_FAILED" + if return_source and prepared.staged: + try: + _return_to_incoming(prepared.name) + prepared = replace(prepared, staged=False) + except ParseError: + code = "UNREADABLE_FILE" return replace(prepared, state="NEEDS_ATTENTION", counts={"created": 0, "matched": 0, "skipped": 0}, code=code) @@ -246,7 +328,9 @@ def prepare_import(*, discovered: DiscoveredFile, section_mapping: dict[str, str try: if discovered.code: raise ParseError(discovered.code) - data = _verified_bytes(discovered) + _stage(discovered) + prepared = replace(prepared, staged=True) + data = _verified_bytes(prepared) mapping = _validated_mapping(discovered, section_mapping) context = partial_context_from_creation_line(data) if discovered.profile == "OUROCARD_PARTIAL_V1" else {} batch = services.create_preview(data=data, profile=discovered.profile, @@ -287,9 +371,17 @@ def commit_prepared_import(*, prepared: PreparedImport, actor: Actor) -> Prepare return _attention(prepared, DomainError("INVALID_PREPARED_IMPORT"), actor, cancel=False) if prepared.state not in {"READY", "ALREADY_IMPORTED", "COMMITTED"} or type(prepared.revision) is not int: raise DomainError("INVALID_PREPARED_IMPORT") - _verified_bytes(prepared) + already_committed = batch.state == "COMMITTED" + if not already_committed: + _verified_bytes(prepared) batch = services.commit_preview(batch_id=batch.pk, revision=prepared.revision, actor=actor) - return replace(prepared, state="COMMITTED", revision=batch.revision, + if not already_committed: + try: + _move_to_processed(prepared.name) + except ParseError as error: + # Persistence succeeded; keep staging for discovery and provenance-based retry. + return _attention(prepared, error, actor, cancel=False, return_source=False) + return replace(prepared, state="COMMITTED", staged=False, revision=batch.revision, counts={key: batch.result[key] for key in ("created", "matched", "skipped")}, code=None) except (DomainError, ParseError, ValidationError) as error: return _attention(prepared, error, actor) diff --git a/tests/test_guided_import_workflow.py b/tests/test_guided_import_workflow.py index 63c49ea..e345718 100644 --- a/tests/test_guided_import_workflow.py +++ b/tests/test_guided_import_workflow.py @@ -73,7 +73,7 @@ def test_exact_second_import_only_skips_and_commit_is_idempotent(source, bank): first = prepare(discovered, bank) assert commit(first).counts["created"] == 2 assert commit(first).counts["created"] == 2 - second = prepare(discovered, bank) + second = prepare(source(), bank) assert second.state == "ALREADY_IMPORTED" assert second.counts == {"created": 0, "matched": 0, "skipped": 2} assert all(item.decision == {"action": "SKIP", "reason": "ALREADY_IMPORTED"} @@ -148,7 +148,8 @@ def test_wrong_account_kind_requires_attention_and_cancels(source): def test_changed_file_cannot_use_scan_hash(source, bank, tmp_path, stage): discovered = source() prepared = prepare(discovered, bank) if stage == "commit" else None - (tmp_path / discovered.name).write_bytes(DATA.replace(b"-12.50", b"-13.50")) + directory = tmp_path / "staging" if stage == "commit" else tmp_path + (directory / discovered.name).write_bytes(DATA.replace(b"-12.50", b"-13.50")) result = commit(prepared) if prepared else prepare(discovered, bank) assert result.state == "NEEDS_ATTENTION" assert result.code == "FILE_CHANGED" @@ -167,7 +168,7 @@ def test_tampered_filename_is_rejected(source, bank, name): def test_symlink_replacement_is_rejected_before_commit(source, bank, tmp_path): discovered = source() prepared = prepare(discovered, bank) - original = tmp_path / discovered.name + original = tmp_path / "staging" / discovered.name other = tmp_path / "other.csv" other.write_bytes(DATA) original.unlink() @@ -325,3 +326,33 @@ def test_partially_imported_source_skips_exact_rows_and_creates_only_new_rows(so assert commit(prepared).counts == {"created": 1, "matched": 0, "skipped": 1} assert Transaction.objects.count() == SourceRecord.objects.count() == 2 assert_no_open_preview() + + +def test_guided_import_stages_then_processes_committed_source(source, bank, tmp_path): + discovered = source() + prepared = prepare(discovered, bank) + assert prepared.state == "READY" + assert not (tmp_path / "bank.csv").exists() + assert (tmp_path / "staging" / "bank.csv").read_bytes() == DATA + processed = tmp_path / "processed" + processed.mkdir(exist_ok=True) + (processed / "bank.csv").write_bytes(b"synthetic previous source") + + committed = commit(prepared) + + assert committed.state == "COMMITTED" + assert not (tmp_path / "bank.csv").exists() + assert not (tmp_path / "staging" / "bank.csv").exists() + assert (processed / "bank.csv").read_bytes() == DATA + + +def test_guided_import_returns_failure_and_recovers_leftover_staging(source, bank, tmp_path): + discovered = source(b"unsupported", "bad.csv") + result = prepare(discovered, bank) + + assert result.state == "NEEDS_ATTENTION" + assert (tmp_path / "bad.csv").exists() + staging = tmp_path / "staging" + staging.mkdir(exist_ok=True) + (staging / "interrupted.csv").write_bytes(DATA) + assert [row.name for row in guided_imports.discover_imports()] == ["bad.csv", "interrupted.csv"]