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
289 lines
11 KiB
Python
289 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import date
|
||
|
||
from app.imports.adapters.base import (
|
||
DIRECTION_IN,
|
||
DIRECTION_OUT,
|
||
ROCKETBANK_DEBIT_PDF_VERSION,
|
||
SourceRow,
|
||
StatementDraft,
|
||
UnsupportedStatementLayout,
|
||
account_fingerprint,
|
||
money_to_kopecks,
|
||
)
|
||
from app.imports.readers import open_pdf
|
||
|
||
MONTH_NAMES = {
|
||
"января": 1,
|
||
"февраля": 2,
|
||
"марта": 3,
|
||
"апреля": 4,
|
||
"мая": 5,
|
||
"июня": 6,
|
||
"июля": 7,
|
||
"августа": 8,
|
||
"сентября": 9,
|
||
"октября": 10,
|
||
"ноября": 11,
|
||
"декабря": 12,
|
||
}
|
||
|
||
DAY_HEADING_RE = re.compile(
|
||
r"^(\d{1,2})\s+(января|февраля|марта|апреля|мая|июня|июля|августа|сентября|"
|
||
r"октября|ноября|декабря)\s+(\d{4})\s+баланс на начало дня:\s*(-?[\d\s]+,\d{2})\s*rur$",
|
||
re.IGNORECASE,
|
||
)
|
||
RECORD_RE = re.compile(
|
||
r"^(\d{1,2}:\d{2})\s+(\d{6,})\s+(.*?)([+-]?\s?[\d\s]+,\d{2})\s*rur$",
|
||
re.IGNORECASE,
|
||
)
|
||
HEADER_VALUES_RE = re.compile(
|
||
r"^(-?[\d\s]+,\d{2})\s*rur\s+(-?[\d\s]+,\d{2})\s*rur(?:\s+(\d{2}\.\d{2}\.\d{4}))?$",
|
||
re.IGNORECASE,
|
||
)
|
||
PERIOD_RE = re.compile(r"^с\s+(\d{2}\.\d{2}\.\d{4})\s+по\s+(\d{2}\.\d{2}\.\d{4})$", re.IGNORECASE)
|
||
AMOUNT_RE = re.compile(r"^([+-]?)\s*(\d[\d\s]*,\d{2})$")
|
||
MCC_RE = re.compile(r"\bmcc\s*:?\s*(\d{4})\b", re.IGNORECASE)
|
||
LOCATION_RE = re.compile(r"rur,\s*([^,]+?),\s*mcc\b", re.IGNORECASE)
|
||
|
||
|
||
class RocketbankDebitPdfAdapter:
|
||
"""Rocketbank debit statement contract, evidenced against the real document.
|
||
|
||
Independent text-layer state machine: every page repeats the statement
|
||
header and column captions; day sections start with a Russian day heading
|
||
('DD MONTH YYYY БАЛАНС НА НАЧАЛО ДНЯ: X RUR'); each record is a
|
||
'time + operation code + multiline description + signed RUR amount' line
|
||
whose description column starts slightly above and continues below that
|
||
line, so the surrounding non-structural lines are folded into one
|
||
description. Daily totals and repeated captions are ignored. Statement
|
||
period, account, opening/closing balances and declared inflow/outflow
|
||
totals come from the first-page header block.
|
||
"""
|
||
|
||
adapter_id = "rocketbank_debit_pdf"
|
||
product = "debit"
|
||
parser_version = ROCKETBANK_DEBIT_PDF_VERSION
|
||
bank_code = "rocketbank"
|
||
currency = "RUB"
|
||
|
||
title_marker = "выписка по счету"
|
||
card_re = re.compile(r"дебетовая карта рокет\s*(\d{10,30})\s*rur", re.IGNORECASE)
|
||
column_caption = "время код операции описание сумма"
|
||
issued_prefix = "дата выдачи выписки:"
|
||
page_footer_prefix = "страница"
|
||
daily_total_prefixes = ("всего поступлений:", "всего расходов:")
|
||
opening_label = "входящий остаток"
|
||
closing_label = "исходящий остаток"
|
||
max_fold_lines = 6
|
||
|
||
def detect(self, raw: bytes) -> bool:
|
||
try:
|
||
with open_pdf(raw) as pdf:
|
||
if not pdf.pages:
|
||
return False
|
||
text = pdf.pages[0].extract_text() or ""
|
||
except Exception:
|
||
return False
|
||
lowered = text.lower()
|
||
return self.title_marker in lowered and "дебетовая карта рокет" in lowered
|
||
|
||
def parse(self, raw: bytes) -> StatementDraft:
|
||
lines: list[tuple[int, str]] = []
|
||
with open_pdf(raw) as pdf:
|
||
if not pdf.pages:
|
||
raise UnsupportedStatementLayout("PDF has no pages")
|
||
for page_number, page in enumerate(pdf.pages, start=1):
|
||
text = page.extract_text() or ""
|
||
if page_number == 1 and not text.strip():
|
||
raise UnsupportedStatementLayout(
|
||
"PDF has no extractable text layer; OCR is not supported"
|
||
)
|
||
for line in text.splitlines():
|
||
stripped = line.strip()
|
||
if stripped:
|
||
lines.append((page_number, stripped))
|
||
return self._draft_from_lines(lines)
|
||
|
||
def _draft_from_lines(self, lines: list[tuple[int, str]]) -> StatementDraft:
|
||
blob = "\n".join(line.lower() for _, line in lines)
|
||
if self.title_marker not in blob:
|
||
raise UnsupportedStatementLayout("Rocketbank statement marker not found")
|
||
card_match = self.card_re.search(blob)
|
||
if card_match is None:
|
||
raise UnsupportedStatementLayout("Rocketbank card line not found")
|
||
|
||
period_start: date | None = None
|
||
period_end: date | None = None
|
||
opening: int | None = None
|
||
closing: int | None = None
|
||
declared_inflow: int | None = None
|
||
declared_outflow: int | None = None
|
||
rows: list[SourceRow] = []
|
||
current_day: date | None = None
|
||
pending: _PendingRecord | None = None
|
||
pre_lines: list[tuple[int, str]] = []
|
||
record_ordinals: dict[int, int] = {}
|
||
|
||
index = 0
|
||
while index < len(lines):
|
||
page, line = lines[index]
|
||
low = line.lower()
|
||
if (
|
||
low == self.title_marker
|
||
or low.startswith(self.issued_prefix)
|
||
or low.startswith(self.page_footer_prefix)
|
||
or low == self.column_caption
|
||
or low.startswith(self.daily_total_prefixes)
|
||
):
|
||
index += 1
|
||
continue
|
||
period_match = PERIOD_RE.match(low)
|
||
if period_match is not None and period_start is None:
|
||
period_start = _date_from(period_match.group(1))
|
||
period_end = _date_from(period_match.group(2))
|
||
index += 1
|
||
continue
|
||
if self.opening_label in low or self.closing_label in low:
|
||
if index + 1 >= len(lines):
|
||
raise UnsupportedStatementLayout("Rocketbank balance values missing")
|
||
values_match = HEADER_VALUES_RE.match(lines[index + 1][1].lower())
|
||
if values_match is None:
|
||
raise UnsupportedStatementLayout("Rocketbank balance values missing")
|
||
first = money_to_kopecks(values_match.group(1))
|
||
second = money_to_kopecks(values_match.group(2))
|
||
if self.opening_label in low:
|
||
opening, declared_inflow = first, second
|
||
else:
|
||
closing, declared_outflow = first, second
|
||
index += 2
|
||
continue
|
||
day_match = DAY_HEADING_RE.match(low)
|
||
if day_match is not None:
|
||
if pending is not None:
|
||
rows.append(self._record(pending, current_day))
|
||
pending = None
|
||
pre_lines = []
|
||
current_day = date(
|
||
int(day_match.group(3)),
|
||
MONTH_NAMES[day_match.group(2).lower()],
|
||
int(day_match.group(1)),
|
||
)
|
||
index += 1
|
||
continue
|
||
record_match = RECORD_RE.match(line)
|
||
if record_match is not None:
|
||
if pending is not None:
|
||
rows.append(self._record(pending, current_day))
|
||
ordinal = record_ordinals.get(page, 0) + 1
|
||
record_ordinals[page] = ordinal
|
||
pending = _PendingRecord(
|
||
row_ref=f"page{page}:record{ordinal}",
|
||
code=record_match.group(2),
|
||
description=record_match.group(3).strip(),
|
||
amount_text=record_match.group(4),
|
||
pre_lines=pre_lines,
|
||
)
|
||
pre_lines = []
|
||
index += 1
|
||
continue
|
||
if pending is not None:
|
||
pending.post_lines.append((page, line))
|
||
index += 1
|
||
continue
|
||
pre_lines.append((page, line))
|
||
if len(pre_lines) > self.max_fold_lines:
|
||
raise UnsupportedStatementLayout("unrecognized content between Rocketbank sections")
|
||
index += 1
|
||
if pending is not None:
|
||
rows.append(self._record(pending, current_day))
|
||
pre_lines = []
|
||
if pre_lines:
|
||
raise UnsupportedStatementLayout("unrecognized trailing Rocketbank content")
|
||
if (
|
||
period_start is None
|
||
or opening is None
|
||
or closing is None
|
||
or declared_inflow is None
|
||
or declared_outflow is None
|
||
):
|
||
raise UnsupportedStatementLayout("Rocketbank statement header block not found")
|
||
if not rows:
|
||
raise UnsupportedStatementLayout("no Rocketbank operation rows found")
|
||
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, card_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 _record(
|
||
self,
|
||
pending: _PendingRecord,
|
||
current_day: date | None,
|
||
) -> SourceRow:
|
||
if current_day is None:
|
||
raise UnsupportedStatementLayout("Rocketbank record outside a day section")
|
||
chunks = [text for _, text in pending.pre_lines]
|
||
chunks.append(pending.description)
|
||
chunks.extend(text for _, text in pending.post_lines)
|
||
description = " ".join(" ".join(chunk.split()) for chunk in chunks if chunk.strip())
|
||
amount_match = AMOUNT_RE.match(" ".join(pending.amount_text.split()))
|
||
if amount_match is None:
|
||
raise UnsupportedStatementLayout("Rocketbank record amount missing")
|
||
sign = amount_match.group(1)
|
||
amount = money_to_kopecks(amount_match.group(2))
|
||
direction = DIRECTION_OUT if sign == "-" else DIRECTION_IN
|
||
mcc_match = MCC_RE.search(description)
|
||
location_match = LOCATION_RE.search(description)
|
||
return SourceRow(
|
||
external_id=pending.code,
|
||
occurred_on=current_day,
|
||
posted_on=None,
|
||
amount_kopecks=amount,
|
||
direction=direction,
|
||
description=description or None,
|
||
mcc=int(mcc_match.group(1)) if mcc_match is not None else None,
|
||
bank_category=None,
|
||
location=location_match.group(1).strip() if location_match is not None else None,
|
||
row_ref=pending.row_ref,
|
||
warnings=(),
|
||
)
|
||
|
||
|
||
class _PendingRecord:
|
||
__slots__ = ("row_ref", "code", "description", "amount_text", "pre_lines", "post_lines")
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
row_ref: str,
|
||
code: str,
|
||
description: str,
|
||
amount_text: str,
|
||
pre_lines: list[tuple[int, str]],
|
||
) -> None:
|
||
self.row_ref = row_ref
|
||
self.code = code
|
||
self.description = description
|
||
self.amount_text = amount_text
|
||
self.pre_lines = pre_lines
|
||
self.post_lines: list[tuple[int, str]] = []
|
||
|
||
|
||
def _date_from(text: str) -> date:
|
||
day, month, year = (int(part) for part in text.split("."))
|
||
return date(year, month, day)
|