From 6224b721c9fb6427a34ecdf1c78f6fc9d759179c Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Mon, 3 Aug 2026 17:05:41 +0530 Subject: [PATCH] Add City Union Bank Bank of Baroda and RBL statement parsers --- .../parsers/bank_of_baroda.py | 152 ++++++++++++++++++ .../parsers/city_union_bank.py | 107 ++++++++++++ .../parsers/rbl_bank.py | 119 ++++++++++++++ .../parsers/registry.py | 13 ++ 4 files changed, 391 insertions(+) create mode 100644 app/modules/bank_statement_analyzer/parsers/bank_of_baroda.py create mode 100644 app/modules/bank_statement_analyzer/parsers/city_union_bank.py create mode 100644 app/modules/bank_statement_analyzer/parsers/rbl_bank.py diff --git a/app/modules/bank_statement_analyzer/parsers/bank_of_baroda.py b/app/modules/bank_statement_analyzer/parsers/bank_of_baroda.py new file mode 100644 index 0000000..8f4febb --- /dev/null +++ b/app/modules/bank_statement_analyzer/parsers/bank_of_baroda.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import re +from pathlib import Path + +import pandas as pd + +from .base import BaseParser, StatementMeta, amount, extract_text, finalize, norm +from .common import date_iso, find, infer_mode + + +class BankOfBarodaParser(BaseParser): + bank_name = "Bank of Baroda" + parser_name = "BankOfBarodaParser" + + _DATE = re.compile(r"^(\d{2}-\d{2}-\d{2})\s+(.*)$") + _BALANCE = re.compile(r"(-?[0-9,]+\.\d{2})(Cr|Dr)\s*$", re.I) + _MONEY = re.compile(r"(? float: + upper = (text or "").upper() + score = 0.0 + if "BANK OF BARODA" in upper: + score += 0.62 + if "BARB0" in upper or "IFSC CODE: BARB" in upper: + score += 0.20 + if all(token in upper for token in ("DATE", "PARTICULARS", "CHQ.NO.", "WITHDRAWALS", "DEPOSITS", "BALANCE")): + score += 0.16 + if "BOB ADVANTAGE" in upper: + score += 0.02 + return min(score, 0.99) + + @staticmethod + def _signed_balance(value: str, marker: str) -> float | None: + parsed = amount(value) + if parsed is None: + return None + return -abs(parsed) if marker.upper() == "DR" else abs(parsed) + + @staticmethod + def _reference_number(narration: str, cheque_no: str) -> str: + if norm(cheque_no): + return norm(cheque_no) + for pattern in ( + r"\bNEFT[-/]([A-Z0-9]{10,30})\b", + r"\bRTGS[-/]([A-Z0-9]{10,30})\b", + r"\b([0-9]{12,18})\b", + ): + match = re.search(pattern, narration, re.I) + if match: + return match.group(1) + return "" + + def parse(self, path, text=None): + pdf_path = Path(path) + text = text or extract_text(pdf_path) + meta = StatementMeta( + bank_name=self.bank_name, + source_file=pdf_path.name, + parser_name=self.parser_name, + confidence="High", + ) + meta.customer_name = norm(find(r"A/C Name\s*:\s*([^\n]+)", text, flags=re.I)) + meta.account_number = find(r"A/C Number\s*:\s*([0-9]+)", text, flags=re.I) + meta.ifsc = find(r"IFSC CODE:\s*([A-Z0-9]+)", text, flags=re.I) + period = re.search(r"Statement of account for the period of\s+(\d{2}-\d{2}-\d{4})\s+to\s+(\d{2}-\d{2}-\d{4})", text, re.I) + if period: + meta.period_from = date_iso(period.group(1)) + meta.period_to = date_iso(period.group(2)) + + rows: list[dict] = [] + current: dict | None = None + previous_balance: float | None = None + opening_balance: float | None = None + + pages = re.split(r"\f", text) + for page_no, page_text in enumerate(pages, start=1): + for raw_line in page_text.splitlines(): + line = raw_line.rstrip() + match = self._DATE.match(line.strip()) + if match: + if current is not None: + rows.append(current) + previous_balance = current["balance"] + current = None + txn_date, remainder = match.groups() + balance_match = self._BALANCE.search(remainder) + if not balance_match: + continue + balance_value = self._signed_balance(balance_match.group(1), balance_match.group(2)) + before_balance = remainder[:balance_match.start()].rstrip() + if norm(before_balance).upper() == "B/F": + opening_balance = balance_value + previous_balance = balance_value + continue + money_matches = list(self._MONEY.finditer(before_balance)) + narration_part = before_balance + cheque_no = "" + if money_matches: + narration_part = before_balance[:money_matches[0].start()].rstrip() + cheque_match = re.search(r"\s([0-9]{1,8})\s*$", narration_part) + if cheque_match: + cheque_no = cheque_match.group(1) + narration_part = narration_part[:cheque_match.start()].rstrip() + debit_value = credit_value = None + if previous_balance is not None and balance_value is not None: + delta = round(balance_value - previous_balance, 2) + if delta < -0.009: + debit_value = abs(delta) + elif delta > 0.009: + credit_value = delta + elif money_matches: + debit_value = amount(money_matches[-1].group(1)) + current = { + "transaction_date": txn_date, + "value_date": txn_date, + "narration": norm(narration_part), + "reference_no": self._reference_number(norm(narration_part), cheque_no), + "debit": debit_value, + "credit": credit_value, + "balance": balance_value, + "source_page": page_no, + "mode": infer_mode(norm(narration_part)), + } + continue + + if current is not None: + continuation = norm(line) + upper = continuation.upper() + if not continuation or upper.startswith(("PAGE TOTAL", "GRAND TOTAL", "BANK OF BARODA", "A/C NUMBER", "STATEMENT OF ACCOUNT", "DATE PARTICULARS", "NOTE:", "UNLESS ")): + continue + if set(continuation) <= {"-"}: + continue + current["narration"] = norm(f"{current['narration']} {continuation}") + current["reference_no"] = self._reference_number(current["narration"], current["reference_no"]) + current["mode"] = infer_mode(current["narration"]) + + if current is not None: + rows.append(current) + + data = pd.DataFrame(rows) + if not data.empty: + first = data.iloc[0] + first_balance = float(first["balance"]) + first_debit = float(first["debit"]) if pd.notna(first["debit"]) else 0.0 + first_credit = float(first["credit"]) if pd.notna(first["credit"]) else 0.0 + meta.opening_balance = round(opening_balance if opening_balance is not None else first_balance + first_debit - first_credit, 2) + meta.total_debit = round(float(pd.to_numeric(data["debit"], errors="coerce").fillna(0).sum()), 2) + meta.total_credit = round(float(pd.to_numeric(data["credit"], errors="coerce").fillna(0).sum()), 2) + meta.closing_balance = round(float(data.iloc[-1]["balance"]), 2) + return meta, finalize(data, meta) diff --git a/app/modules/bank_statement_analyzer/parsers/city_union_bank.py b/app/modules/bank_statement_analyzer/parsers/city_union_bank.py new file mode 100644 index 0000000..a5f3e20 --- /dev/null +++ b/app/modules/bank_statement_analyzer/parsers/city_union_bank.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import re +from pathlib import Path + +import pandas as pd +import pdfplumber + +from .base import BaseParser, StatementMeta, amount, extract_text, finalize, norm +from .common import date_iso, find, infer_mode + + +class CityUnionBankParser(BaseParser): + bank_name = "City Union Bank" + parser_name = "CityUnionBankParser" + + @classmethod + def detect(cls, text: str) -> float: + upper = (text or "").upper() + score = 0.0 + if "CITY UNION BANK" in upper: + score += 0.62 + if "CIUB0" in upper or "IFSC :CIUB" in upper or "IFSC:CIUB" in upper: + score += 0.20 + if all(token in upper for token in ("DATE", "DESCRIPTION", "CHEQUE NO", "DEBIT", "CREDIT", "BALANCE")): + score += 0.16 + if "END OF STATEMENT - FROM INTERNET BANKING" in upper: + score += 0.02 + return min(score, 0.99) + + @staticmethod + def _reference_number(narration: str, cheque_no: str) -> str: + if norm(cheque_no): + return norm(cheque_no) + for pattern in ( + r"\bUTR[:/\s-]*([A-Z0-9]{10,30})\b", + r"\bCHQ\s*NO\s*([0-9]{1,12})\b", + r"\b([A-Z]{4}H\d{10,})\b", + ): + match = re.search(pattern, narration, re.I) + if match: + return match.group(1) + return "" + + def parse(self, path, text=None): + pdf_path = Path(path) + text = text or extract_text(pdf_path) + meta = StatementMeta( + bank_name=self.bank_name, + source_file=pdf_path.name, + parser_name=self.parser_name, + confidence="High", + ) + meta.customer_name = norm(find(r"CUSTOMER DETAILS\s*:\s*([^\n]+)", text, flags=re.I)) + meta.account_number = find(r"ACCOUNT NO\s*\(15 DIGIT\)\s*:\s*([0-9]+)", text, flags=re.I) + if not meta.account_number: + meta.account_number = find(r"ACCOUNT NO\s*:\s*([A-Z0-9-]+)", text, flags=re.I) + meta.ifsc = find(r"IFSC\s*:\s*([A-Z0-9]+)", text, flags=re.I) + period = re.search(r"STATEMENT OF ACCOUNT\s+from\s+(\d{2}/\d{2}/\d{4})\s+to\s+(\d{2}/\d{2}/\d{4})", text, re.I) + if period: + meta.period_from = date_iso(period.group(1)) + meta.period_to = date_iso(period.group(2)) + + rows: list[dict] = [] + with pdfplumber.open(str(pdf_path)) as pdf: + for page_no, page in enumerate(pdf.pages, start=1): + for table in page.extract_tables() or []: + if not table: + continue + header = [norm(cell).upper() for cell in table[0]] + if not all(token in header for token in ("DATE", "DESCRIPTION", "DEBIT", "CREDIT", "BALANCE")): + continue + for raw in table[1:]: + cells = list(raw) + [None] * (6 - len(raw)) + txn_date, description, cheque_no, debit, credit, balance = cells[:6] + txn_date = norm(txn_date) + if not re.fullmatch(r"\d{2}/\d{2}/\d{4}", txn_date): + continue + debit_value = amount(debit) + credit_value = amount(credit) + balance_value = amount(balance) + if debit_value is None and credit_value is None: + continue + narration = norm(description) + rows.append({ + "transaction_date": txn_date, + "value_date": txn_date, + "narration": narration, + "reference_no": self._reference_number(narration, norm(cheque_no)), + "debit": debit_value, + "credit": credit_value, + "balance": balance_value, + "source_page": page_no, + "mode": infer_mode(narration), + }) + + data = pd.DataFrame(rows) + if not data.empty: + first = data.iloc[0] + first_balance = float(first["balance"]) + first_debit = float(first["debit"]) if pd.notna(first["debit"]) else 0.0 + first_credit = float(first["credit"]) if pd.notna(first["credit"]) else 0.0 + meta.opening_balance = round(first_balance + first_debit - first_credit, 2) + meta.total_debit = round(float(pd.to_numeric(data["debit"], errors="coerce").fillna(0).sum()), 2) + meta.total_credit = round(float(pd.to_numeric(data["credit"], errors="coerce").fillna(0).sum()), 2) + meta.closing_balance = round(float(data.iloc[-1]["balance"]), 2) + return meta, finalize(data, meta) diff --git a/app/modules/bank_statement_analyzer/parsers/rbl_bank.py b/app/modules/bank_statement_analyzer/parsers/rbl_bank.py new file mode 100644 index 0000000..7c7fafc --- /dev/null +++ b/app/modules/bank_statement_analyzer/parsers/rbl_bank.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import re +from pathlib import Path + +import pandas as pd +import pdfplumber + +from .base import BaseParser, StatementMeta, amount, extract_text, finalize, norm +from .common import date_iso, find, infer_mode + + +class RBLBankParser(BaseParser): + bank_name = "RBL Bank" + parser_name = "RBLBankParser" + + @classmethod + def detect(cls, text: str) -> float: + upper = (text or "").upper() + score = 0.0 + if "RBL BANK" in upper or "RBLBANK" in upper: + score += 0.58 + if "RATN000" in upper or "IFSC/RTGS/NEFT" in upper and "RATN" in upper: + score += 0.20 + if "STATEMENT OF TRANSACTIONS IN SAVINGS ACCOUNT" in upper: + score += 0.12 + if all(token in upper for token in ("TRANSACTION DETAILS", "VALUE DATE", "WITHDRAWAL AMT", "DEPOSIT AMT", "BALANCE AMT")): + score += 0.10 + return min(score, 0.99) + + @staticmethod + def _signed_balance(value) -> float | None: + text = norm(value).upper() + parsed = amount(text) + if parsed is None: + return None + return -abs(parsed) if text.endswith("DR") else abs(parsed) + + @staticmethod + def _reference_number(narration: str, cheque_no: str) -> str: + if norm(cheque_no): + return norm(cheque_no) + for pattern in ( + r"\bUPI/([0-9]{10,18})\b", + r"\bIMPS\s+([0-9]{10,18})\b", + r"\b(?:NEFT|RTGS)/([A-Z0-9]{10,30})\b", + r"\b([0-9]{12,18})-IMPS\b", + ): + match = re.search(pattern, narration, re.I) + if match: + return match.group(1) + return "" + + def parse(self, path, text=None): + pdf_path = Path(path) + text = text or extract_text(pdf_path) + meta = StatementMeta( + bank_name=self.bank_name, + source_file=pdf_path.name, + parser_name=self.parser_name, + confidence="High", + ) + meta.customer_name = norm(find(r"Accountholder Name\s*:?\s*([^\n]+)", text, flags=re.I)) + meta.customer_id = find(r"CIF ID\s*:?\s*([0-9]+)", text, flags=re.I) + meta.account_number = find(r"Statement Of Transactions In Savings Account No\.\s*([0-9]+)", text, flags=re.I) + meta.ifsc = find(r"(?:IFSC/RTGS/NEFT|IFSC)\s*:?\s*([A-Z0-9]+)", text, flags=re.I) + period = re.search(r"Period\s*:\s*(\d{2}-\d{2}-\d{4})\s+to\s+(\d{2}-\d{2}-\d{4})", text, re.I) + if period: + meta.period_from = date_iso(period.group(1)) + meta.period_to = date_iso(period.group(2)) + + rows: list[dict] = [] + printed_opening: float | None = None + with pdfplumber.open(str(pdf_path)) as pdf: + for page_no, page in enumerate(pdf.pages, start=1): + for table in page.extract_tables() or []: + if not table: + continue + header = [norm(cell).upper() for cell in table[0]] + if not all(token in header for token in ("DATE", "TRANSACTION DETAILS", "VALUE DATE", "WITHDRAWAL AMT", "DEPOSIT AMT", "BALANCE AMT")): + continue + for raw in table[1:]: + cells = list(raw) + [None] * (7 - len(raw)) + txn_date, description, cheque_no, value_date, withdrawal, deposit, balance = cells[:7] + txn_date = norm(txn_date) + balance_value = self._signed_balance(balance) + if not txn_date and balance_value is not None and printed_opening is None: + printed_opening = balance_value + continue + if not re.fullmatch(r"\d{2}-[A-Za-z]{3}-\d{4}", txn_date): + continue + debit_value = amount(withdrawal) + credit_value = amount(deposit) + if debit_value is None and credit_value is None: + continue + narration = norm(description) + rows.append({ + "transaction_date": txn_date, + "value_date": norm(value_date) or txn_date, + "narration": narration, + "reference_no": self._reference_number(narration, norm(cheque_no)), + "debit": debit_value, + "credit": credit_value, + "balance": balance_value, + "source_page": page_no, + "mode": infer_mode(narration), + }) + + data = pd.DataFrame(rows) + if not data.empty: + first = data.iloc[0] + first_balance = float(first["balance"]) + first_debit = float(first["debit"]) if pd.notna(first["debit"]) else 0.0 + first_credit = float(first["credit"]) if pd.notna(first["credit"]) else 0.0 + meta.opening_balance = round(printed_opening if printed_opening is not None else first_balance + first_debit - first_credit, 2) + meta.total_debit = round(float(pd.to_numeric(data["debit"], errors="coerce").fillna(0).sum()), 2) + meta.total_credit = round(float(pd.to_numeric(data["credit"], errors="coerce").fillna(0).sum()), 2) + meta.closing_balance = round(float(data.iloc[-1]["balance"]), 2) + return meta, finalize(data, meta) diff --git a/app/modules/bank_statement_analyzer/parsers/registry.py b/app/modules/bank_statement_analyzer/parsers/registry.py index f7b63d0..c0151d0 100644 --- a/app/modules/bank_statement_analyzer/parsers/registry.py +++ b/app/modules/bank_statement_analyzer/parsers/registry.py @@ -11,6 +11,9 @@ from .kotak import KotakParser from .sbi import SBIModernParser, SBIOtherParser from .central_bank_of_india import CentralBankOfIndiaParser from .yes_bank import YesBankParser +from .city_union_bank import CityUnionBankParser +from .bank_of_baroda import BankOfBarodaParser +from .rbl_bank import RBLBankParser from .base import extract_text PARSERS = [ @@ -18,6 +21,9 @@ PARSERS = [ AxisParser, CentralBankOfIndiaParser, YesBankParser, + CityUnionBankParser, + BankOfBarodaParser, + RBLBankParser, HDFCParser, ICICIParser, HSBCParser, @@ -34,6 +40,9 @@ BANK_OPTIONS = [ ("axis", "Axis Bank"), ("central_bank_of_india", "Central Bank of India"), ("yes_bank", "YES Bank"), + ("city_union_bank", "City Union Bank"), + ("bank_of_baroda", "Bank of Baroda"), + ("rbl_bank", "RBL Bank"), ("hdfc", "HDFC Bank"), ("icici", "ICICI Bank"), ("hsbc", "HSBC Bank"), @@ -48,6 +57,9 @@ BANK_PARSERS = { "axis": [AxisParser], "central_bank_of_india": [CentralBankOfIndiaParser], "yes_bank": [YesBankParser], + "city_union_bank": [CityUnionBankParser], + "bank_of_baroda": [BankOfBarodaParser], + "rbl_bank": [RBLBankParser], "hdfc": [HDFCParser], "icici": [ICICIParser], "hsbc": [HSBCParser], @@ -104,3 +116,4 @@ def parse_pdf(path, bank_hint: str | None = None): ) return parser.parse(path, text) +