gizmoduck/ingestion/guided_imports.py

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

import os
import stat
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)


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:
        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


_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
    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:
        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)
    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):
    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"
    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)
        data = _verified_bytes(discovered)
        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")
        _verified_bytes(prepared)
        batch = services.commit_preview(batch_id=batch.pk, revision=prepared.revision, actor=actor)
        return replace(prepared, state="COMMITTED", 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)