k1nq
023612ce0a
- Updated timestamps and AST hashes for multiple files in the manifest.json. - Added new test file backend/tests/test_expected_income.py. - Added documentation files: перево́д_между_своими.md and 0002-fixes.md. - Introduced pytest.ini to configure pytest settings, including test paths and warning filters.
306 lines
13 KiB
Python
306 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import date, datetime
|
|
|
|
from app.imports.adapters.base import (
|
|
DIRECTION_IN,
|
|
DIRECTION_OUT,
|
|
OZON_DEBIT_PDF_VERSION,
|
|
PARSER_VERSION,
|
|
SourceRow,
|
|
StatementDraft,
|
|
UnsupportedStatementLayout,
|
|
account_fingerprint,
|
|
amount_to_kopecks,
|
|
money_to_kopecks,
|
|
)
|
|
from app.imports.readers import open_pdf
|
|
|
|
PERIOD_RE = re.compile(
|
|
r"период:?\s*(\d{2}\.\d{2}\.\d{4})\s*[-–—]\s*(\d{2}\.\d{2}\.\d{4})", re.IGNORECASE
|
|
)
|
|
ROW_RE = re.compile(
|
|
r"^(?P<date>\d{2}\.\d{2}\.\d{4})\s+(?P<desc>.+?)\s+(?P<amount>-?\d[\d\s]*,\d{2})"
|
|
r"(?:\s*(?:₽|руб\.?|RUB))?$"
|
|
)
|
|
MASK_RE = re.compile(r"(?:\*{2,}|•{2,})\s*(\d{4})")
|
|
|
|
|
|
class OzonPdfAdapterBase:
|
|
"""Ozon statement contract: text-layer PDF, per-page line scanning."""
|
|
|
|
adapter_id: str = ""
|
|
product: str = ""
|
|
bank_markers: tuple[str, ...] = ("ozon банк", "озон банк", "ozon bank")
|
|
product_markers: tuple[str, ...] = ()
|
|
|
|
def detect(self, raw: bytes) -> bool:
|
|
try:
|
|
text = _first_page_text(raw)
|
|
except Exception:
|
|
return False
|
|
lowered = text.lower()
|
|
return any(m in lowered for m in self.bank_markers) and any(
|
|
m in lowered for m in self.product_markers
|
|
)
|
|
|
|
def parse(self, raw: bytes) -> StatementDraft:
|
|
with open_pdf(raw) as pdf:
|
|
lines: list[str] = []
|
|
period_start: date | None = None
|
|
period_end: date | None = None
|
|
masks: list[str] = []
|
|
for page_index, page in enumerate(pdf.pages):
|
|
text = page.extract_text() or ""
|
|
if page_index == 0 and not text.strip():
|
|
raise UnsupportedStatementLayout(
|
|
"PDF has no extractable text layer; OCR is not supported"
|
|
)
|
|
lowered = text.lower()
|
|
if not any(m in lowered for m in self.bank_markers):
|
|
raise UnsupportedStatementLayout("bank marker not found in PDF text")
|
|
if not any(m in lowered for m in self.product_markers):
|
|
raise UnsupportedStatementLayout(
|
|
f"expected {self.product} statement markers not found"
|
|
)
|
|
match = PERIOD_RE.search(text)
|
|
if match and period_start is None:
|
|
period_start = datetime.strptime(match.group(1), "%d.%m.%Y").date()
|
|
period_end = datetime.strptime(match.group(2), "%d.%m.%Y").date()
|
|
masks.extend(MASK_RE.findall(text))
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if ROW_RE.match(line):
|
|
lines.append(line)
|
|
if not lines:
|
|
raise UnsupportedStatementLayout("no statement rows matched the layout contract")
|
|
fingerprint = (
|
|
account_fingerprint(self.bank_markers[0].split()[0], f"**{masks[0]}")
|
|
if masks
|
|
else None
|
|
)
|
|
parsed = [_parse_row(line, f"page:line{n + 1}") for n, line in enumerate(lines)]
|
|
return StatementDraft(
|
|
adapter_id=self.adapter_id,
|
|
parser_version=PARSER_VERSION,
|
|
bank_code=self.bank_markers[0].split()[0],
|
|
product=self.product,
|
|
account_fingerprint=fingerprint,
|
|
currency="RUB",
|
|
period_start=period_start,
|
|
period_end=period_end,
|
|
opening_balance_kopecks=None,
|
|
closing_balance_kopecks=None,
|
|
rows=tuple(parsed),
|
|
)
|
|
|
|
|
|
def _parse_row(line: str, row_ref: str) -> SourceRow:
|
|
match = ROW_RE.match(line)
|
|
assert match is not None
|
|
occurred = datetime.strptime(match.group("date"), "%d.%m.%Y").date()
|
|
raw_amount = match.group("amount").replace(" ", "")
|
|
amount = amount_to_kopecks(raw_amount)
|
|
return SourceRow(
|
|
external_id=None,
|
|
occurred_on=occurred,
|
|
posted_on=None,
|
|
amount_kopecks=amount,
|
|
direction=DIRECTION_IN if amount >= 0 else DIRECTION_OUT,
|
|
description=match.group("desc").strip(),
|
|
mcc=None,
|
|
bank_category=None,
|
|
location=None,
|
|
row_ref=row_ref,
|
|
warnings=() if occurred else ("occurred_date_missing",),
|
|
)
|
|
|
|
|
|
def _first_page_text(raw: bytes) -> str:
|
|
with open_pdf(raw) as pdf:
|
|
if not pdf.pages:
|
|
raise UnsupportedStatementLayout("PDF has no pages")
|
|
return pdf.pages[0].extract_text() or ""
|
|
|
|
|
|
class OzonDebitPdfAdapter:
|
|
"""Ozon debit statement contract, evidenced against the real document.
|
|
|
|
A 'Справка о движении средств' text-layer PDF: a legal-bank header block
|
|
with account/period/opening balance, per-page tables whose rows keep
|
|
multiline document IDs and descriptions attached to one transaction, and a
|
|
closing block with declared inflow/outflow totals and the closing balance.
|
|
Amounts are signed strings ('- 500.00 ₽'); the sign maps to direction and
|
|
the stored amount is always positive.
|
|
"""
|
|
|
|
adapter_id = "ozon_debit_pdf"
|
|
product = "debit"
|
|
parser_version = OZON_DEBIT_PDF_VERSION
|
|
bank_code = "ozon"
|
|
currency = "RUB"
|
|
|
|
bank_marker = "озон банк"
|
|
title_marker = "справка о движении средств"
|
|
account_marker_re = re.compile(r"лицевого\s+сч[её]та:?\s*№\s*([\d*•]+)", re.IGNORECASE)
|
|
period_re = re.compile(
|
|
r"период\s+выписки:?\s*(\d{2}\.\d{2}\.\d{4})\s*[–—-]\s*(\d{2}\.\d{2}\.\d{4})",
|
|
re.IGNORECASE,
|
|
)
|
|
opening_re = re.compile(r"входящий\s+остаток:?\s*(.+)", re.IGNORECASE)
|
|
inflow_re = re.compile(r"итого\s+зачислений\s+за\s+период:?\s*(.+)", re.IGNORECASE)
|
|
outflow_re = re.compile(r"итого\s+списаний\s+за\s+период:?\s*(.+)", re.IGNORECASE)
|
|
closing_re = re.compile(r"исходящий\s+остаток:?\s*(.+)", re.IGNORECASE)
|
|
amount_re = re.compile(r"^([+-])\s*(\d[\d\s]*(?:[.,]\d{2})?)\s*(?:₽|rur|rub)?$", re.IGNORECASE)
|
|
table_header_labels = ("дата операции", "документ", "назначение платежа")
|
|
table_subheader_tokens = ("рубли", "валюта")
|
|
|
|
def detect(self, raw: bytes) -> bool:
|
|
try:
|
|
text = _first_page_text(raw)
|
|
except Exception:
|
|
return False
|
|
lowered = _ozon_normalize(text)
|
|
return (
|
|
self.bank_marker in lowered
|
|
and self.title_marker in lowered
|
|
and "лицевого сч" in lowered
|
|
)
|
|
|
|
def parse(self, raw: bytes) -> StatementDraft:
|
|
pages_text: list[str] = []
|
|
pages_tables: list[list[list[list[str | None]]]] = []
|
|
with open_pdf(raw) as pdf:
|
|
if not pdf.pages:
|
|
raise UnsupportedStatementLayout("PDF has no pages")
|
|
for page_index, page in enumerate(pdf.pages):
|
|
text = page.extract_text() or ""
|
|
if page_index == 0 and not text.strip():
|
|
raise UnsupportedStatementLayout(
|
|
"PDF has no extractable text layer; OCR is not supported"
|
|
)
|
|
pages_text.append(text)
|
|
extracted = page.extract_tables()
|
|
pages_tables.append(extracted if extracted else [])
|
|
return self._draft_from_extracted(pages_text, pages_tables)
|
|
|
|
def _draft_from_extracted(
|
|
self, pages_text: list[str], pages_tables: list[list[list[list[str | None]]]]
|
|
) -> StatementDraft:
|
|
page_zero = _ozon_normalize(pages_text[0])
|
|
if self.bank_marker not in page_zero or self.title_marker not in page_zero:
|
|
raise UnsupportedStatementLayout("Ozon statement markers not found")
|
|
account_match = self.account_marker_re.search(page_zero)
|
|
if account_match is None:
|
|
raise UnsupportedStatementLayout("Ozon account line not found")
|
|
period_match = self.period_re.search(page_zero)
|
|
if period_match is None:
|
|
raise UnsupportedStatementLayout("Ozon period line not found")
|
|
period_start = datetime.strptime(period_match.group(1), "%d.%m.%Y").date()
|
|
period_end = datetime.strptime(period_match.group(2), "%d.%m.%Y").date()
|
|
opening = money_to_kopecks(self._required_line(self.opening_re, page_zero, "opening"))
|
|
full_text = _ozon_normalize("\n".join(pages_text))
|
|
declared_inflow = money_to_kopecks(self._required_line(self.inflow_re, full_text, "inflow"))
|
|
declared_outflow = money_to_kopecks(
|
|
self._required_line(self.outflow_re, full_text, "outflow")
|
|
)
|
|
closing = money_to_kopecks(self._required_line(self.closing_re, full_text, "closing"))
|
|
rows: list[SourceRow] = []
|
|
for page_index, tables in enumerate(pages_tables, start=1):
|
|
for table_index, table in enumerate(tables):
|
|
for row_index, cells in enumerate(table):
|
|
row = self._table_row(
|
|
cells, f"page{page_index}:table{table_index}:row{row_index}"
|
|
)
|
|
if row is not None:
|
|
rows.append(row)
|
|
if not rows:
|
|
raise UnsupportedStatementLayout("no statement rows matched the Ozon contract")
|
|
return StatementDraft(
|
|
adapter_id=self.adapter_id,
|
|
parser_version=self.parser_version,
|
|
bank_code=self.bank_code,
|
|
product=self.product,
|
|
account_fingerprint=account_fingerprint(self.bank_code, account_match.group(1)),
|
|
currency=self.currency,
|
|
period_start=period_start,
|
|
period_end=period_end,
|
|
opening_balance_kopecks=opening,
|
|
closing_balance_kopecks=closing,
|
|
rows=tuple(rows),
|
|
declared_inflow_kopecks=declared_inflow,
|
|
declared_outflow_kopecks=declared_outflow,
|
|
declared_hold_kopecks=None,
|
|
)
|
|
|
|
def _required_line(self, pattern: re.Pattern[str], text: str, what: str) -> str:
|
|
match = pattern.search(text)
|
|
if match is None:
|
|
raise UnsupportedStatementLayout(f"Ozon {what} total line not found")
|
|
return match.group(1)
|
|
|
|
def _table_row(self, cells: list[str | None], row_ref: str) -> SourceRow | None:
|
|
if cells is None:
|
|
raise UnsupportedStatementLayout("unexpected Ozon table row shape")
|
|
texts = [_ozon_cell(cell) for cell in cells]
|
|
if not any(texts):
|
|
return None
|
|
lowered = [text.lower() for text in texts]
|
|
joined = " ".join(text for text in lowered if text)
|
|
if all(label in joined for label in self.table_header_labels):
|
|
return None
|
|
if lowered[0] == "" and any(token in joined for token in self.table_subheader_tokens):
|
|
return None
|
|
if lowered[0] == "" or len(texts) < 4:
|
|
raise UnsupportedStatementLayout("unexpected Ozon table row shape")
|
|
occurred = None
|
|
date_match = re.match(r"^(\d{2}\.\d{2}\.\d{4})", texts[0])
|
|
if date_match is not None:
|
|
occurred = datetime.strptime(date_match.group(1), "%d.%m.%Y").date()
|
|
external_id = re.sub(r"\s+", "", texts[1]) if len(texts) > 1 else ""
|
|
description = " ".join(texts[2].split()) if len(texts) > 2 else ""
|
|
warnings: list[str] = []
|
|
if occurred is None:
|
|
warnings.append("occurred_date_missing")
|
|
if not external_id:
|
|
warnings.append("external_id_missing")
|
|
amount_match = self.amount_re.match(texts[3]) if len(texts) > 3 else None
|
|
if amount_match is None:
|
|
amount = 0
|
|
warnings.append("amount_unparseable")
|
|
direction = DIRECTION_OUT
|
|
else:
|
|
sign, body = amount_match.group(1), amount_match.group(2)
|
|
amount = amount_to_kopecks(body.replace(" ", "").replace(",", "."))
|
|
direction = DIRECTION_IN if sign == "+" else DIRECTION_OUT
|
|
return SourceRow(
|
|
external_id=external_id or None,
|
|
occurred_on=occurred,
|
|
posted_on=None,
|
|
amount_kopecks=amount,
|
|
direction=direction,
|
|
description=description or None,
|
|
mcc=None,
|
|
bank_category=None,
|
|
location=None,
|
|
row_ref=row_ref,
|
|
warnings=tuple(warnings),
|
|
)
|
|
|
|
|
|
def _ozon_normalize(text: str) -> str:
|
|
return text.replace("\u00a0", " ").lower()
|
|
|
|
|
|
def _ozon_cell(cell: object) -> str:
|
|
if cell is None:
|
|
return ""
|
|
return str(cell).replace("\u00a0", " ").strip()
|
|
|
|
|
|
class OzonCreditPdfAdapter(OzonPdfAdapterBase):
|
|
adapter_id = "ozon_credit_pdf"
|
|
product = "credit"
|
|
product_markers = ("кредитн", "кредитная")
|