# -*- coding: utf-8 -*- """ PDF‑парсер АТ «Райффайзен Банк» / форма «Виписка за рахунком». version 1.6  (2025‑07‑01) Отличия от v1.5 ──────────────── * исправлена критическая ошибка ValueError: could not convert string to float: '' (возникала на строках, где регулярка находила «комісія …» без цифр — или цифры содержали лишь пробелы/неразрывные пробелы, которые затем очищались до пустой строки).  Теперь: • регулярка _RE_FEE гарантирует наличие **минимум одной** цифры • конвертация в float выполняется «безопасно» (_safe_to_float) и возвращает None вместо исключения * добавлено подробное логирование ситуаций, когда комиссия не распознана или не может быть преобразована в число * незначительные косметические правки (PEP 8, type hints) """ import logging import re from datetime import datetime from typing import List, Optional import pdfplumber from app.models.transaction import Transaction from app.parsers.base_parser import BaseBankStatementParser logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) logger.addHandler(logging.StreamHandler()) # ───────────────────────── regexes ───────────────────────── _RE_NBSP = re.compile(r"[ \u00A0\u202F]") _RE_FLOAT = re.compile(r"^-?[\d \u00A0\u202F]+[,.]\d{2}$") _RE_IBAN = re.compile(r"\bUA\d{25}\b") _RE_DATE = re.compile(r"\d{2}\.\d{2}\.\d{4}(?:\s+\d{2}:\d{2})?") # Гарантируем хотя бы одну цифру в группе _RE_FEE = re.compile( r"(?:коміс\w*|ком[\s\-]*бан)\s*" r"([0-9][0-9 \u00A0\u202F]*(?:[,.]\d{1,2})?)\s*(?:uah|грн)?", re.IGNORECASE, ) # ───────────────────────── helpers ───────────────────────── def _clean_num(num: str) -> str: """Удаляет обычные/неразрывные пробелы, заменяет запятую на точку.""" return _RE_NBSP.sub("", num).replace(",", ".") def _to_float(num: str) -> float: """Прямое преобразование; бросает ValueError при невалидном числе.""" return float(_clean_num(num)) def _safe_to_float(num: str | None) -> Optional[float]: """Более безопасная версия: возвращает None вместо исключения.""" if not num: return None cleaned = _clean_num(num) if not cleaned: return None try: return float(cleaned) except ValueError: logger.debug("Не удалось преобразовать %r в float (после очистки: %r)", num, cleaned) return None # ───────────────────────── parser ────────────────────────── class RaiffeisenBankPdfParser(BaseBankStatementParser): """Главный публичный метод — parse(file_path) → List[Transaction]""" # ───────────────────── public interface ─────────────────── def parse(self, file_path: str) -> List[Transaction]: logger.debug("Старт парсинга PDF: %s", file_path) txs: List[Transaction] = [] with pdfplumber.open(file_path) as pdf: if not pdf.pages: return txs self._extract_header(pdf.pages[0]) raw_rows = self._collect_all_rows(pdf) filtered = [r for r in raw_rows if not self._is_trash_row(r)] merged: List[List[str]] = self._merge_continuations(filtered) seen: set[tuple] = set() for row in merged: base_tx = self._row_to_transaction(row) if not base_tx: continue key = (base_tx.number, base_tx.date, base_tx.amount) if key in seen: continue # дубликат seen.add(key) txs.append(base_tx) # ── отделяем комиссию, если она есть в назначении ── fee_amt = self._extract_fee(base_tx.payment_details) if fee_amt is None: continue # пропускаем, если в исходном платеже уже списана точно такая сумма if base_tx.amount < 0 and abs(abs(base_tx.amount) - fee_amt) < 0.01: continue txs.append(self._make_fee_tx(base_tx, fee_amt)) logger.debug("Всего Transaction (с комиссиями): %d", len(txs)) return txs # ───────────────────── header helpers ──────────────────── def _extract_header(self, page) -> None: txt = page.extract_text() or "" g = lambda p: (re.search(p, txt) or [None, ""])[1].strip() self.our_company_name = g(r"Назва\s+клієнта:\s*(.+)") self.our_company_inn = g(r"Код\s+за\s+ЄДРПОУ/ІНН:\s*(\d{8,10})") self.our_company_account = g(r"(UA\d{25})").replace(" ", "") logger.debug("Header → name=%s | inn=%s | acc=%s", self.our_company_name, self.our_company_inn, self.our_company_account) # ───────────────────── table helpers ───────────────────── def _collect_all_rows(self, pdf) -> List[List[str]]: rows: List[List[str]] = [] for page in pdf.pages: tbl = ( page.extract_table( { "vertical_strategy": "lines", "horizontal_strategy": "lines", "intersection_tolerance": 2, "snap_tolerance": 3, } ) or page.extract_table( { "vertical_strategy": "text", "horizontal_strategy": "text", "keep_blank_chars": True, } ) ) if tbl: rows += [[(c or "").strip() for c in r] for r in tbl] return rows def _is_trash_row(self, row: List[str]) -> bool: txt = " ".join(row).lower() return not txt.strip() or txt.startswith( ( "номер документ", "оборот за", "вихідний залишок", "вхідний залишок", "сторінка", ) ) def _merge_continuations(self, rows: List[List[str]]) -> List[List[str]]: merged: List[List[str]] = [] cur: Optional[List[str]] = None for r in rows: r += [""] * (9 - len(r)) # нормализуем количество колонок до 9 if r[0]: # новая строка‑операция if cur: merged.append(cur) cur = r.copy() else: # продолжение предыдущей if cur: for idx in (3, 5, 6): # IBAN/Банк | Назва | Призначення if r[idx]: cur[idx] += "\n" + r[idx] if cur: merged.append(cur) return merged # ───────────────── row → Transaction ───────────────────── def _row_to_transaction(self, r: List[str]) -> Optional[Transaction]: doc_num = r[0] if not doc_num: return None dt = self._parse_date((r[2] or r[1]).strip()) debit_raw, credit_raw = (r[7] or "").strip(), (r[8] or "").strip() if debit_raw and _RE_FLOAT.match(debit_raw): amount = -_to_float(debit_raw) elif credit_raw and _RE_FLOAT.match(credit_raw): amount = _to_float(credit_raw) else: return None counter_acc = ( _RE_IBAN.search(r[3]).group().replace(" ", "") if _RE_IBAN.search(r[3]) else "" ) counter_inn = (r[4] or "").strip() counter_name = (r[5] or "").strip() purpose = " ".join(r[6].split()) if amount < 0: # расход payer_name, payer_inn, payer_acc = ( self.our_company_name, self.our_company_inn, self.our_company_account, ) recip_name, recip_inn, recip_acc = counter_name, counter_inn, counter_acc d_out, d_in = dt, None else: # приход payer_name, payer_inn, payer_acc = counter_name, counter_inn, counter_acc recip_name, recip_inn, recip_acc = ( self.our_company_name, self.our_company_inn, self.our_company_account, ) d_out, d_in = None, dt tx = Transaction( number=doc_num, date=dt, amount=amount, payment_details=purpose, payer_name=payer_name, payer_inn=payer_inn, payer_account=payer_acc, recipient_name=recip_name, recipient_inn=recip_inn, recipient_account=recip_acc, date_outcome=d_out, date_income=d_in, ) tx.payment_details_one_line = self._to_single_line(purpose) return tx # ───────────────── commissions ─────────────────────────── @staticmethod def _to_single_line(text: str) -> str: return " ".join(text.split()) @staticmethod def _extract_fee(text: str) -> Optional[float]: """Извлекает сумму комиссии (UAH) или None, если не найдена/непарсится.""" m = _RE_FEE.search(text.replace("\n", " ")) if not m: return None fee = _safe_to_float(m.group(1)) if fee is None: logger.debug("Комиссия найдена, но не распознана как число: %r", m.group(0)) return fee def _make_fee_tx(self, base: Transaction, fee: float) -> Transaction: tx = Transaction( number=f"{base.number}-FEE", date=base.date, amount=-abs(fee), payer_name=self.our_company_name, payer_inn=self.our_company_inn, payer_account=self.our_company_account, recipient_name='АТ "Райффайзен Банк"', recipient_inn='14305909', recipient_account='', payment_details=f"Банківська комісія за док. {base.number}", date_outcome=base.date, date_income=None, ) tx.payment_details_one_line = self._to_single_line(tx.payment_details) return tx # ───────────────── misc utils ──────────────────────────── @staticmethod def _parse_date(s: str) -> datetime: s = s.strip().replace("\n", " ") for fmt in ("%d.%m.%Y %H:%M", "%d.%m.%Y"): try: return datetime.strptime(s, fmt) except ValueError: continue m = _RE_DATE.search(s) return datetime.strptime(m.group(), "%d.%m.%Y") if m else datetime.now()