k1nq
e43428c1a2
Gitea Actions CI / detect (push) Successful in 3s
Gitea Actions CI / Compose & nginx validation (push) Failing after 20s
Gitea Actions CI / Frontend (lint / type / test / build) (push) Failing after 24s
Gitea Actions CI / Backend (lint / type / test / build) (push) Failing after 1m34s
43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
from app.imports.adapters.alfa_xlsx import (
|
|
AlfaCreditXlsxAdapter as AlfaCreditXlsxAdapter,
|
|
)
|
|
from app.imports.adapters.alfa_xlsx import AlfaDebitXlsxAdapter
|
|
from app.imports.adapters.base import StatementAdapter, UnsupportedStatementLayout
|
|
from app.imports.adapters.ozon_pdf import OzonCreditPdfAdapter as OzonCreditPdfAdapter
|
|
from app.imports.adapters.ozon_pdf import OzonDebitPdfAdapter
|
|
from app.imports.adapters.rocketbank_pdf import RocketbankDebitPdfAdapter
|
|
|
|
# Only the three evidenced-format development contracts are active. The credit
|
|
# adapters stay present but unregistered until real fixtures exist for them.
|
|
ADAPTERS: tuple[StatementAdapter, ...] = (
|
|
AlfaDebitXlsxAdapter(),
|
|
OzonDebitPdfAdapter(),
|
|
RocketbankDebitPdfAdapter(),
|
|
)
|
|
|
|
XLSX_MAGIC = b"PK\x03\x04"
|
|
PDF_MAGIC = b"%PDF-"
|
|
|
|
|
|
def magic_kind(raw: bytes) -> str | None:
|
|
if raw.startswith(XLSX_MAGIC):
|
|
return "xlsx"
|
|
if raw.startswith(PDF_MAGIC):
|
|
return "pdf"
|
|
return None
|
|
|
|
|
|
def detect_adapter(raw: bytes, magic: str) -> StatementAdapter:
|
|
attempted: list[str] = []
|
|
for adapter in ADAPTERS:
|
|
expected = "xlsx" if adapter.adapter_id.endswith("xlsx") else "pdf"
|
|
if expected != magic:
|
|
continue
|
|
attempted.append(adapter.adapter_id)
|
|
if adapter.detect(raw):
|
|
return adapter
|
|
raise UnsupportedStatementLayout(
|
|
"no registered adapter recognizes this document layout; "
|
|
f"attempted: {', '.join(attempted) or 'none for this file type'}"
|
|
)
|