import pdfplumber import re from typing import List, Optional from datetime import datetime from app.parsers.base_parser import BaseBankStatementParser from app.models.transaction import Transaction class MonobankPdfParser(BaseBankStatementParser): """ Парсер PDF-выписок Монобанка (АТ "УНІВЕРСАЛ БАНК") через extract_tables(). Ожидается, что каждая страница содержит таблицу с 9 колонками: [0] Дата/время операции (многострочно) [1] Деталі операції (многострочно) [2] Реквізити контрагента (многострочно) [3] Сумма операции (+/-) [4] Валюта (обычно UAH) [5] Сумма во «внутренней» валюте (игнор) [6] Курс (игнор) [7] Сумма комиссии (число или «—») [8] Остаток после операции (можно игнорировать или использовать) Если строка в первой колонке пустая, считаем, что это продолжение предыдущей строки (для колонок [1] и [2]). """ RE_COMMISSION = re.compile(r'Комісія\s+банку\s+([\d\.,]+)\s*грн') DATETIME_FORMATS = ["%d.%m.%Y %H:%M:%S", "%d.%m.%Y %H:%M", "%d.%m.%Y"] def __init__(self): super().__init__() self.our_company_name: Optional[str] = None self.our_company_inn: Optional[str] = None self.our_company_account: Optional[str] = None self.our_bank_name: Optional[str] = None def parse(self, file_path: str) -> List[Transaction]: transactions: List[Transaction] = [] with pdfplumber.open(file_path) as pdf: # Извлекаем заголовок с первой страницы if pdf.pages: self._extract_header_data(pdf.pages[0]) all_rows = [] for page_index, page in enumerate(pdf.pages): raw_tables = page.extract_tables() if not raw_tables: continue for table in raw_tables: if len(table) < 2 or len(table[0]) < 9: continue data_rows = table[1:] merged = self._merge_rows(data_rows) # --- Отладка: вывод объединённых строк (закомментируйте при релизе) # print(f"--- Page {page_index + 1} merged rows ---") # for r in merged: # print(r) # print("---------------") all_rows.extend(merged) filtered_rows = [row for row in all_rows if self._looks_like_date(row[0])] # --- Отладка: вывод отфильтрованных строк # print("=== Отфильтрованные строки ===") # for r in filtered_rows: # print(r) # print("===================================") for row_i, row in enumerate(filtered_rows, start=1): date_str = row[0].strip() # Дата/время details_str = row[1] # Деталі операції (многострочно) contragent_str = row[2] # Реквізити контрагента (многострочно) amount_str = (row[3] or "").replace(" ", "") currency_str = (row[4] or "").strip() commission_str = (row[7] or "").strip() op_date = self._parse_datetime(date_str) try: amount = float(amount_str.replace(",", ".")) except ValueError: amount = 0.0 doc_number = f"MNB_{row_i}" # Обработка реквизитов контрагента – оставляем всё как есть contragent_inn = self._find_inn(contragent_str) contragent_iban = self._find_iban(contragent_str) contragent_name = self._keep_all_lines(contragent_str) # Преобразуем детали операции в одно строковое представление details_one_line = self._to_single_line(details_str) # Распределяем стороны (приход/расход) if amount < 0: payer_inn = "1" payer_name = self.our_company_name or "OUR_FOP" payer_account = self.our_company_account or "" recipient_inn = contragent_inn recipient_name = contragent_name recipient_account = contragent_iban date_outcome = op_date.date() if op_date else None date_income = None else: payer_inn = contragent_inn payer_name = contragent_name payer_account = contragent_iban recipient_inn = "1" recipient_name = self.our_company_name or "OUR_FOP" recipient_account = self.our_company_account or "" date_outcome = None date_income = op_date.date() if op_date else None # --- Отладка: вывод текущей строки (закомментируйте при релизе) # print(f"Row {row_i}:") # print(" Date:", date_str) # print(" Details:", details_str) # print(" Contragent:", contragent_str) # print(" ContragentName (final):", contragent_name) # print(" Amount:", amount) # print(" Commission field:", commission_str) main_tx = Transaction( number=doc_number, date=(op_date.date() if op_date else None), amount=amount, payer_inn=payer_inn, payer_name=payer_name, payer_account=payer_account, recipient_inn=recipient_inn, recipient_name=recipient_name, recipient_account=recipient_account, payment_details=details_one_line, # Используем однострочное представление date_income=date_income, date_outcome=date_outcome, ) # Если возможно, сохраняем однострочную версию для поля НазначениеПлатежа1 main_tx.payment_details_one_line = details_one_line # Это доп. поле, используйте при генерации 1C transactions.append(main_tx) if amount > 0: found_comm = self._extract_commission(details_str) if found_comm is None: found_comm = self._extract_commission(commission_str) if found_comm and found_comm > 0: doc_number_comm = doc_number + "_COM" comm_tx = Transaction( number=doc_number_comm, date=(op_date.date() if op_date else None), amount=-abs(found_comm), payer_inn="1", payer_name=self.our_company_name or "OUR_FOP", payer_account=self.our_company_account or "", recipient_inn=contragent_inn, recipient_name=contragent_name, recipient_account=contragent_iban, payment_details=f"Комісія банку (из операции: {details_str})", date_income=None, date_outcome=(op_date.date() if op_date else None), ) # Тоже сохраняем однострочную версию для комиссионной операции comm_tx.payment_details_one_line = self._to_single_line( f"Комісія банку (из операции: {details_str})" ) transactions.append(comm_tx) return transactions # ------------------- Вспомогательные методы ------------------- def _extract_header_data(self, page) -> None: text = page.extract_text() or "" for line in text.split("\n"): line = line.strip() m_inn = re.search(r'ІПН:\s*(\d+)', line) if m_inn: self.our_company_inn = m_inn.group(1) m_acc = re.search(r'Рух\s+коштів\s+по\s+рахунку:\s+(UA\d+)', line) if m_acc: self.our_company_account = m_acc.group(1) m_client = re.search(r'Клієнт:\s+(.+)', line) if m_client: self.our_company_name = m_client.group(1).strip() self.our_bank_name = 'АТ "УНІВЕРСАЛ БАНК"' def _merge_rows(self, data_rows: List[List[str]]) -> List[List[str]]: merged = [] current = None for row in data_rows: row = row + [""] * (9 - len(row)) first_col = row[0].strip() if first_col: if self._looks_like_date(first_col): if current: merged.append(current) current = row[:] else: if current: for col in [1, 2]: part = row[col].strip() if part: current[col] += "\n" + part else: current = row[:] else: if current: for col in [1, 2]: part = row[col].strip() if part: current[col] += "\n" + part else: current = row[:] if current: merged.append(current) # --- Отладка: вывод склеенных строк (раскомментируйте при необходимости) # print("=== Итоговые склеенные строки ===") # for r in merged: # print(r) # print("==================================") return merged def _looks_like_date(self, text: str) -> bool: text = text.strip() for fmt in self.DATETIME_FORMATS: try: datetime.strptime(text, fmt) return True except ValueError: continue return False def _parse_datetime(self, date_str: str) -> Optional[datetime]: date_str = date_str.strip() for fmt in self.DATETIME_FORMATS: try: return datetime.strptime(date_str, fmt) except ValueError: pass return None def _extract_commission(self, text: str) -> Optional[float]: match = self.RE_COMMISSION.search(text) if match: val_str = match.group(1).replace(",", ".") try: return float(val_str) except ValueError: return None return None def _find_inn(self, text: str) -> str: m = re.search(r'ЄДРПОУ:\s*(\d+)', text) if m: return m.group(1) m2 = re.search(r'\b\d{8,10}\b', text) if m2: return m2.group(0) return "" def _find_iban(self, text: str) -> str: m = re.search(r'(UA\d{2}\s*\d+)', text) if m: return m.group(1).replace(" ", "") return "" def _keep_all_lines(self, text: str) -> str: # Оставляем всё, что есть в тексте (не удаляем никаких строк) return text.strip() def _to_single_line(self, text: str) -> str: """ Преобразует многострочный текст в одну строку, удаляя переводы строк и лишние пробелы. """ return " ".join(text.split())