You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
90 lines
3.0 KiB
90 lines
3.0 KiB
"""Safe discovery of local source documents for the guided import workflow.""" |
|
|
|
import os |
|
import stat |
|
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 _unreadable(path: Path) -> DiscoveredFile: |
|
return DiscoveredFile(path.name, "", None, (), "UNREADABLE_FILE") |
|
|
|
|
|
def _read_discovered_file(path: Path, expected: os.stat_result) -> DiscoveredFile: |
|
descriptor = None |
|
try: |
|
flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) |
|
if no_follow := getattr(os, "O_NOFOLLOW", 0): |
|
flags |= no_follow |
|
descriptor = os.open(path, flags) |
|
opened = os.fstat(descriptor) |
|
if not stat.S_ISREG(opened.st_mode) or not os.path.samestat(expected, opened): |
|
return _unreadable(path) |
|
with os.fdopen(descriptor, "rb", closefd=False) as stream: |
|
data = stream.read(MAX_BYTES + 1) |
|
except OSError: |
|
return _unreadable(path) |
|
finally: |
|
if descriptor is not None: |
|
os.close(descriptor) |
|
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``.""" |
|
try: |
|
root = Path(settings.IMPORT_ROOT).resolve() |
|
entries = tuple(root.iterdir()) |
|
except (OSError, RuntimeError): |
|
return () |
|
discovered = [] |
|
for path in entries: |
|
try: |
|
expected = path.lstat() |
|
if not stat.S_ISREG(expected.st_mode): |
|
continue |
|
except OSError: |
|
discovered.append(_unreadable(path)) |
|
continue |
|
discovered.append(_read_discovered_file(path, expected)) |
|
return tuple(sorted(discovered, key=lambda row: row.name))
|
|
|