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.
71 lines
2.3 KiB
71 lines
2.3 KiB
"""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))
|
|
|