gizmoduck/ingestion/guided_imports.py

"""Safe discovery of local source documents for the guided import workflow."""

import ctypes
import os
import stat
from contextlib import contextmanager
from dataclasses import dataclass, replace
from hashlib import sha256
from pathlib import Path

from django.conf import settings
from django.core.exceptions import ValidationError

from finance.models import Account
from finance.services import Actor, DomainError
from finance.services.base import identity

from . import services
from .models import ImportBatch
from .parsers import MAX_BYTES, ParseError, parse_document, partial_context_from_creation_line


@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"),
}

_SUPPORTS_SAFE_DIRECTORY_FDS = (
    os.name == "posix"
    and os.open in os.supports_dir_fd
    and os.stat in os.supports_dir_fd
    and os.listdir in os.supports_fd
    and hasattr(os, "O_DIRECTORY")
    and hasattr(os, "O_NOFOLLOW")
)


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 _unreadable(name: str) -> DiscoveredFile:
    return DiscoveredFile(name, "", None, (), "UNREADABLE_FILE")


def _close(descriptor: int | None) -> None:
    if descriptor is not None:
        try:
            os.close(descriptor)
        except OSError:
            pass


def _open_import_root() -> int | None:
    if not _SUPPORTS_SAFE_DIRECTORY_FDS:
        return None
    try:
        root = os.path.normpath(os.fspath(settings.IMPORT_ROOT))
        if not os.path.isabs(root):
            root = os.path.abspath(root)
        path = Path(root)
        descriptor = os.open(path.anchor, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        for component in path.parts[1:]:
            next_descriptor = os.open(
                component,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                dir_fd=descriptor,
            )
            _close(descriptor)
            descriptor = next_descriptor
        if not stat.S_ISDIR(os.fstat(descriptor).st_mode):
            _close(descriptor)
            return None
        return descriptor
    except (OSError, RuntimeError, TypeError, ValueError):
        _close(locals().get("descriptor"))
        return None


def _read_file_bytes(name: str, root_descriptor: int, expected: os.stat_result) -> bytes:
    descriptor = None
    try:
        flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0)
        if no_follow := getattr(os, "O_NOFOLLOW", 0):
            flags |= no_follow
        descriptor = os.open(name, flags, dir_fd=root_descriptor)
        opened = os.fstat(descriptor)
        if not stat.S_ISREG(opened.st_mode) or not os.path.samestat(expected, opened):
            raise ParseError("UNREADABLE_FILE")
        with os.fdopen(descriptor, "rb", closefd=False) as stream:
            data = stream.read(MAX_BYTES + 1)
    except OSError:
        raise ParseError("UNREADABLE_FILE") from None
    finally:
        _close(descriptor)
    if len(data) > MAX_BYTES:
        raise ParseError("INPUT_TOO_LARGE")
    return data


def _read_discovered_file(name: str, root_descriptor: int, expected: os.stat_result) -> DiscoveredFile:
    try:
        data = _read_file_bytes(name, root_descriptor, expected)
    except ParseError as error:
        return DiscoveredFile(name, "", None, (), error.code)
    document_hash = sha256(data).hexdigest()
    if Path(name).suffix.casefold() not in _PROFILES_BY_SUFFIX:
        return DiscoveredFile(name, document_hash, None, (), "UNSUPPORTED_FILE")
    profile, sections = _profile_and_sections(Path(name), data)
    if profile is None:
        return DiscoveredFile(name, document_hash, None, (), "UNRECOGNIZED_LAYOUT")
    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 _rename_without_replacement(name: str, source_fd: int, destination_fd: int) -> None:
    # The supported Docker runtime is Linux. Fail closed if atomic no-replace
    # rename is unavailable instead of falling back to a check-then-replace race.
    try:
        rename = ctypes.CDLL(None, use_errno=True).renameat2
    except AttributeError:
        raise ParseError("UNREADABLE_FILE") from None
    rename.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
    rename.restype = ctypes.c_int
    encoded_name = os.fsencode(name)
    if rename(source_fd, encoded_name, destination_fd, encoded_name, 1) != 0:  # RENAME_NOREPLACE
        raise OSError(ctypes.get_errno(), "Cannot move import source")


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")
        if overwrite:
            os.replace(name, name, src_dir_fd=source_fd, dst_dir_fd=destination_fd)
        else:
            _rename_without_replacement(name, source_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:
            return ()
        discovered = []
        for name in entries:
            try:
                expected = os.stat(name, dir_fd=root_descriptor, follow_symlinks=False)
            except OSError:
                discovered.append(_unreadable(name))
                continue
            if stat.S_ISREG(expected.st_mode):
                discovered.append(_read_discovered_file(name, root_descriptor, expected))
        return tuple(sorted(discovered, key=lambda row: row.name))
    finally:
        _close(root_descriptor)


@dataclass(frozen=True)
class PreparedImport:
    name: str
    document_hash: str
    profile: str | None
    sections: tuple[str, ...]
    state: str
    batch_id: str | None
    revision: int | None
    counts: dict[str, int]
    code: str | None
    staged: bool = False


_SAFE_CODES = frozenset({
    "UNREADABLE_FILE", "INPUT_TOO_LARGE", "UNSUPPORTED_FILE", "UNRECOGNIZED_LAYOUT",
    "FILE_CHANGED", "INVALID_SECTION_MAPPING", "MISSING_CREATION_DATE", "INVALID_CREATION_DATE",
    "STALE_REVISION", "STALE_PREVIEW", "BATCH_NOT_FOUND", "BATCH_NOT_PREVIEW",
    "REVIEW_REQUIRED", "INVALID_PREPARED_IMPORT",
})


def _verified_bytes(discovered: DiscoveredFile | PreparedImport) -> bytes:
    name = discovered.name
    _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)
    if sha256(data).hexdigest() != discovered.document_hash:
        raise ParseError("FILE_CHANGED")
    profile, sections = _profile_and_sections(Path(name), data)
    if profile != discovered.profile or sections != discovered.sections or profile is None:
        raise ParseError("FILE_CHANGED")
    return data


def _validated_mapping(discovered, section_mapping):
    if (not isinstance(section_mapping, dict) or set(section_mapping) != set(discovered.sections)
            or any(not isinstance(value, str) for value in section_mapping.values())):
        raise DomainError("INVALID_SECTION_MAPPING")
    try:
        mapping = {key: str(identity(value)) for key, value in section_mapping.items()}
    except DomainError:
        raise DomainError("INVALID_SECTION_MAPPING") from None
    active = set(Account.objects.filter(pk__in=mapping.values(), archived=False, currency="BRL")
                 .values_list("pk", flat=True))
    if {str(pk) for pk in active} != set(mapping.values()):
        raise DomainError("INVALID_SECTION_MAPPING")
    return mapping


def _cancel_unfinished(batch_id, actor):
    # Cancellation must use the latest revision after a stale-review failure.
    for _ in range(3):
        batch = ImportBatch.objects.filter(pk=batch_id, state="PREVIEW").first()
        if batch is None:
            return
        try:
            services.cancel_preview(batch_id=batch.pk, revision=batch.revision, actor=actor)
            return
        except DomainError as error:
            if error.code != "STALE_REVISION":
                raise
    raise DomainError("CANCELLATION_FAILED")


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:
        try:
            _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)


def prepare_import(*, discovered: DiscoveredFile, section_mapping: dict[str, str], actor: Actor) -> PreparedImport:
    """Prepare complete deterministic decisions through the audited lifecycle."""
    prepared = PreparedImport(discovered.name, discovered.document_hash, discovered.profile,
                              discovered.sections, "NEEDS_ATTENTION", None, None,
                              {"created": 0, "matched": 0, "skipped": 0}, None)
    try:
        if discovered.code:
            raise ParseError(discovered.code)
        _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,
                                        section_mapping=mapping, context=context, actor=actor)
        prepared = replace(prepared, batch_id=str(batch.pk), revision=batch.revision)
        decisions = []
        counts = {"created": 0, "matched": 0, "skipped": 0}
        for item in batch.items.order_by("locator"):
            if item.issues == ["ALREADY_IMPORTED"]:
                decisions.append({"locator": item.locator, "action": "SKIP", "reason": "ALREADY_IMPORTED"})
                counts["skipped"] += 1
            elif not item.issues and not item.duplicates["existing"] and not item.duplicates["within"]:
                decisions.append({"locator": item.locator, "action": "CREATE"})
                counts["created"] += 1
            else:
                raise DomainError("REVIEW_REQUIRED")
        if not decisions:
            raise DomainError("REVIEW_REQUIRED")
        batch = services.apply_decisions(batch_id=batch.pk, revision=batch.revision,
                                         decisions=decisions, actor=actor)
        return replace(prepared, state="READY" if counts["created"] else "ALREADY_IMPORTED",
                       revision=batch.revision, counts=counts)
    except (DomainError, ParseError, ValidationError) as error:
        return _attention(prepared, error, actor)


def commit_prepared_import(*, prepared: PreparedImport, actor: Actor) -> PreparedImport:
    """Revalidate source identity and delegate the atomic commit to persistence."""
    try:
        batch_id = identity(prepared.batch_id)
    except DomainError:
        return _attention(prepared, DomainError("INVALID_PREPARED_IMPORT"), actor, cancel=False)
    try:
        batch = ImportBatch.objects.filter(pk=batch_id).first()
        if batch is None:
            raise DomainError("BATCH_NOT_FOUND")
        if batch.document_hash != prepared.document_hash or batch.profile != prepared.profile:
            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")
        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)
        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)