diff --git a/app/modules/bank_statement_analyzer/parsers/central_bank_of_india.py b/app/modules/bank_statement_analyzer/parsers/central_bank_of_india.py new file mode 100644 index 0000000..8564f47 --- /dev/null +++ b/app/modules/bank_statement_analyzer/parsers/central_bank_of_india.py @@ -0,0 +1,116 @@ +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 CentralBankOfIndiaParser(BaseParser): + bank_name = "Central Bank of India" + parser_name = "CentralBankOfIndiaParser" + + @classmethod + def detect(cls, text: str) -> float: + upper = (text or "").upper() + score = 0.0 + if "CENTRAL BANK OF INDIA" in upper: + score += 0.60 + if "CBIN0" in upper or "IFSC CODE: CBIN" in upper: + score += 0.20 + if all(token in upper for token in ("POST DATE", "VALUE DATE", "ACCOUNT DESCRIPTION", "DEBIT", "CREDIT", "BALANCE")): + score += 0.18 + if "END OF STATEMENT - FROM INTERNET BANKING" in upper: + score += 0.02 + return min(score, 0.99) + + @staticmethod + def _signed_balance(value) -> float | None: + if value is None: + return 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"\bRRN\s*[:/]?\s*(\d{10,18})\b", + r"\b(CBINN\d{10,})\b", + r"\b([A-Z]{4,8}R\d{10,})\b", + r"\b(\d{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"Account Number:\s*\d+.*?\n(?:Product Type:.*?\n)?([A-Z][A-Z .]+)", text, flags=re.I | re.S)) + meta.account_number = find(r"Account Number:\s*([0-9]+)", text) + meta.ifsc = find(r"IFSC Code:\s*([A-Z0-9]+)", text) + 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 ("POST DATE" in header and "ACCOUNT DESCRIPTION" in header and "BALANCE" in header): + continue + for raw in table[1:]: + cells = list(raw) + [None] * (8 - len(raw)) + post_date, value_date, branch_code, cheque_no, description, debit, credit, balance = cells[:8] + post_date = norm(post_date) + if not re.fullmatch(r"\d{2}/\d{2}/\d{4}", post_date): + continue + narration = norm(description) + debit_value = amount(debit) + credit_value = amount(credit) + balance_value = self._signed_balance(balance) + if debit_value is None and credit_value is None: + continue + rows.append({ + "transaction_date": post_date, + "value_date": norm(value_date) or post_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/registry.py b/app/modules/bank_statement_analyzer/parsers/registry.py index 77b43ed..f7b63d0 100644 --- a/app/modules/bank_statement_analyzer/parsers/registry.py +++ b/app/modules/bank_statement_analyzer/parsers/registry.py @@ -1,4 +1,4 @@ -from __future__ import annotations +from __future__ import annotations from .idfc import IDFCFirstParser from .axis import AxisParser @@ -9,11 +9,15 @@ from .indian_bank import IndianBankModernParser, IndianBankLegacyParser from .indusind import IndusIndParser from .kotak import KotakParser from .sbi import SBIModernParser, SBIOtherParser +from .central_bank_of_india import CentralBankOfIndiaParser +from .yes_bank import YesBankParser from .base import extract_text PARSERS = [ IDFCFirstParser, AxisParser, + CentralBankOfIndiaParser, + YesBankParser, HDFCParser, ICICIParser, HSBCParser, @@ -28,6 +32,8 @@ PARSERS = [ BANK_OPTIONS = [ ("auto", "Auto Detect"), ("axis", "Axis Bank"), + ("central_bank_of_india", "Central Bank of India"), + ("yes_bank", "YES Bank"), ("hdfc", "HDFC Bank"), ("icici", "ICICI Bank"), ("hsbc", "HSBC Bank"), @@ -40,6 +46,8 @@ BANK_OPTIONS = [ BANK_PARSERS = { "axis": [AxisParser], + "central_bank_of_india": [CentralBankOfIndiaParser], + "yes_bank": [YesBankParser], "hdfc": [HDFCParser], "icici": [ICICIParser], "hsbc": [HSBCParser], @@ -95,3 +103,4 @@ def parse_pdf(path, bank_hint: str | None = None): "Unsupported statement format. Select the bank manually or add a bank-specific parser for this statement layout." ) return parser.parse(path, text) + diff --git a/app/modules/bank_statement_analyzer/parsers/yes_bank.py b/app/modules/bank_statement_analyzer/parsers/yes_bank.py new file mode 100644 index 0000000..df9381a --- /dev/null +++ b/app/modules/bank_statement_analyzer/parsers/yes_bank.py @@ -0,0 +1,96 @@ +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 YesBankParser(BaseParser): + bank_name = "YES Bank" + parser_name = "YesBankParser" + + @classmethod + def detect(cls, text: str) -> float: + upper = (text or "").upper() + score = 0.0 + if "YES BANK" in upper: + score += 0.55 + if "YESB000" in upper or "IFSC CODE: YESB" in upper: + score += 0.20 + if "TRANSACTION DETAILS FOR YOUR ACCOUNT NUMBER" in upper: + score += 0.15 + if all(token in upper for token in ("TRANSACTION DATE", "VALUE DATE", "WITHDRAWALS", "DEPOSITS", "RUNNING")): + score += 0.10 + return min(score, 0.99) + + 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"Primary Holder:\s*([^\n]+?)(?:\s+A/C Opening Date:|\n)", text, flags=re.I)) + if not meta.customer_name: + meta.customer_name = norm(find(r"Primary Account Holder Name:\s*([^\n]+)", text, flags=re.I)) + meta.customer_id = find(r"(?:Cust Id|Customer Id):\s*([0-9]+)", text, flags=re.I) + meta.account_number = find(r"Statement of account:\s*([0-9]+)", text, flags=re.I) + if not meta.account_number: + meta.account_number = find(r"account number\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"Period:\s*From\s+(.+?)\s+To\s+([^\n]+)", text, re.I) + if period: + meta.period_from = date_iso(period.group(1).strip()) + meta.period_to = date_iso(period.group(2).strip()) + + 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 ("TRANSACTION DATE" in header and "VALUE DATE" in header and "DESCRIPTION" in header): + continue + for raw in table[1:]: + cells = list(raw) + [None] * (7 - len(raw)) + txn_date, value_date, reference_no, description, withdrawal, deposit, balance = cells[:7] + txn_date = norm(txn_date) + if not re.fullmatch(r"\d{2}-[A-Za-z]{3}-\d{4}", txn_date): + continue + debit_value = amount(withdrawal) + credit_value = amount(deposit) + 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": norm(value_date) or txn_date, + "narration": narration, + "reference_no": norm(reference_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)