diff --git a/app/modules/bank_statement_analyzer/parsers/registry.py b/app/modules/bank_statement_analyzer/parsers/registry.py index 9ee59a0..68de980 100644 --- a/app/modules/bank_statement_analyzer/parsers/registry.py +++ b/app/modules/bank_statement_analyzer/parsers/registry.py @@ -8,13 +8,14 @@ from .hsbc import HSBCParser from .indian_bank import IndianBankModernParser, IndianBankLegacyParser from .indusind import IndusIndParser from .kotak import KotakParser -from .sbi import SBIAccountSummaryParser, SBIModernParser, SBIOtherParser +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 +from .template_engine import TemplateBasedStatementParser, parse_with_template PARSERS = [ IDFCFirstParser, @@ -31,7 +32,6 @@ PARSERS = [ IndianBankLegacyParser, IndusIndParser, KotakParser, - SBIAccountSummaryParser, SBIOtherParser, SBIModernParser, ] @@ -68,7 +68,7 @@ BANK_PARSERS = { "indian_bank": [IndianBankModernParser, IndianBankLegacyParser], "indusind": [IndusIndParser], "kotak": [KotakParser], - "sbi": [SBIAccountSummaryParser, SBIOtherParser, SBIModernParser], + "sbi": [SBIOtherParser, SBIModernParser], } @@ -89,32 +89,71 @@ def _selected_parser(bank_key: str, text: str): return None, 0 +def _valid_result(result) -> bool: + if not result or len(result) != 2: + return False + _meta, frame = result + return frame is not None and not frame.empty + + def parse_pdf(path, bank_hint: str | None = None): + """Parse a bank statement without removing any existing parser behaviour. + + Existing bank-specific parsers remain first priority. The template engine + is an additive fallback when a bank parser does not recognise a new layout + or returns no transactions. A template result is accepted only after + transaction-level running-balance validation. + """ hint = (bank_hint or "auto").strip().lower() - if hint == "hsbc": - return HSBCParser().parse(path, "") text = extract_text(path) + primary_error: Exception | None = None + if hint and hint != "auto": - parser, score = _selected_parser(hint, text) - if parser is None: - label = dict(BANK_OPTIONS).get(hint, "the selected bank") - raise ValueError( - f"The uploaded statement does not match the selected {label} format. " - "Please verify the selected bank or choose Auto Detect." - ) - return parser.parse(path, text) - parser, score = detect_parser(text) - if parser is None and not (text or "").strip(): - # Image-only statements cannot be identified by pdftotext. HSBCParser - # performs its own OCR identification and raises a precise mismatch error. - try: - return HSBCParser().parse(path, text) - except ValueError: - pass - if parser is None: - raise ValueError( - "Unsupported statement format. Select the bank manually or add a bank-specific parser for this statement layout." + # Preserve HSBC's specialised OCR parser as the first route. + candidates = BANK_PARSERS.get(hint, []) + scored = sorted( + ((parser.detect(text), parser) for parser in candidates), + key=lambda item: item[0], + reverse=True, ) - return parser.parse(path, text) + for score, parser_class in scored: + if score <= 0: + continue + try: + result = parser_class().parse(path, text) + if _valid_result(result): + return result + except (ValueError, RuntimeError) as exc: + primary_error = exc + # New layouts from a known bank are routed by table structure rather + # than rejected merely because the bank-specific detector changed. + try: + return parse_with_template(path, text, hint) + except (ValueError, RuntimeError) as template_error: + label = dict(BANK_OPTIONS).get(hint, "the selected bank") + detail = str(primary_error or template_error) + raise ValueError( + f"The uploaded statement could not be reliably parsed as {label}. " + f"{detail}" + ) from template_error + parser, _score = detect_parser(text) + if parser is not None: + try: + result = parser.parse(path, text) + if _valid_result(result): + return result + except (ValueError, RuntimeError) as exc: + primary_error = exc + + # Auto Detect now falls back to structural templates. Bank detection remains + # a hint for metadata and naming, never a hard gate for transaction parsing. + try: + return parse_with_template(path, text, "auto") + except (ValueError, RuntimeError) as template_error: + detail = str(primary_error or template_error) + raise ValueError( + "Unsupported or unreconciled statement format. " + f"{detail}" + ) from template_error diff --git a/app/modules/bank_statement_analyzer/parsers/template_engine.py b/app/modules/bank_statement_analyzer/parsers/template_engine.py new file mode 100644 index 0000000..c7e7e38 --- /dev/null +++ b/app/modules/bank_statement_analyzer/parsers/template_engine.py @@ -0,0 +1,404 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +import pandas as pd +import pdfplumber + +from .base import BaseParser, StatementMeta, amount, extract_text, finalize, norm +from .common import date_iso, find, infer_mode, parse_flexible_date + + +# The template engine is deliberately additive. Existing bank-specific parsers +# remain the primary route. This parser is used only when a selected parser does +# not recognise a layout or when a recognised parser cannot extract rows. + +_DATE_CELL_RE = re.compile( + r"^\s*(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{1,2}-[A-Za-z]{3}-\d{2,4}|\d{1,2}\s+[A-Za-z]{3}\s+\d{2,4})\s*$", + re.I, +) + +_BANK_MARKERS: tuple[tuple[str, tuple[str, ...]], ...] = ( + ("State Bank of India", ("STATE BANK OF INDIA", "SBIN0")), + ("HDFC Bank", ("HDFC BANK", "HDFC0")), + ("Axis Bank", ("AXIS BANK", "UTIB0")), + ("ICICI Bank", ("ICICI BANK", "ICIC0")), + ("HSBC Bank", ("HSBC",)), + ("IDFC FIRST Bank", ("IDFC FIRST BANK", "IDFB0")), + ("Indian Bank", ("INDIAN BANK", "IDIB0")), + ("IndusInd Bank", ("INDUSIND BANK", "INDB0")), + ("Kotak Mahindra Bank", ("KOTAK MAHINDRA BANK", "KKBK0")), + ("Central Bank of India", ("CENTRAL BANK OF INDIA", "CBIN0")), + ("YES Bank", ("YES BANK", "YESB0")), + ("City Union Bank", ("CITY UNION BANK", "CIUB0")), + ("Bank of Baroda", ("BANK OF BARODA", "BARB0")), + ("RBL Bank", ("RBL BANK", "RATN0")), +) + +_BANK_KEY_LABELS = { + "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", + "idfc": "IDFC FIRST Bank", + "indian_bank": "Indian Bank", + "indusind": "IndusInd Bank", + "kotak": "Kotak Mahindra Bank", + "sbi": "State Bank of India", +} + +_HEADER_ALIASES = { + "transaction_date": ("TRANSACTION DATE", "TXN DATE", "POST DATE", "DATE"), + "value_date": ("VALUE DATE",), + "narration": ("TRANSACTION DETAILS", "PARTICULARS", "DESCRIPTION", "DETAILS", "NARRATION"), + "reference_no": ("REF NO./CHEQUE NO.", "REF NO/CHEQUE NO", "CHEQUE NO/REFERENCE NO", "CHQ.NO.", "CHQ.NO", "CHEQUE NO", "REFERENCE NO"), + "debit": ("WITHDRAWAL AMT", "WITHDRAWALS", "WITHDRAWAL", "DEBIT AMT", "DEBIT"), + "credit": ("DEPOSIT AMT", "DEPOSITS", "DEPOSIT", "CREDIT AMT", "CREDIT"), + "balance": ("RUNNING BALANCE", "BALANCE AMT", "BALANCE"), +} + + +@dataclass(frozen=True) +class LayoutTemplate: + name: str + mapping: dict[str, int] + confidence: float + + +def _upper(value) -> str: + return norm(value).upper().replace("\n", " ") + + +def _is_date(value) -> bool: + text = norm(value) + return bool(_DATE_CELL_RE.fullmatch(text)) and not pd.isna(parse_flexible_date(text)) + + +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) + + +def _find_header_mapping(row: list) -> dict[str, int]: + cells = [_upper(cell) for cell in row] + mapping: dict[str, int] = {} + for field, aliases in _HEADER_ALIASES.items(): + for index, cell in enumerate(cells): + if any(alias == cell or alias in cell for alias in aliases): + mapping[field] = index + break + return mapping + + +def _looks_like_data_row(row: list) -> bool: + return bool(row) and (_is_date(row[0]) or (len(row) > 1 and _is_date(row[1]))) + + +def _infer_structural_template(table: list[list]) -> LayoutTemplate | None: + sample = next((list(row) for row in table if _looks_like_data_row(list(row))), None) + if sample is None: + return None + width = len(sample) + # SBI Account Summary / SBI YONO family: + # value date, post date, details, reference, debit, credit, balance. + if width >= 7 and _is_date(sample[0]) and _is_date(sample[1]): + return LayoutTemplate( + "dual_date_debit_credit_balance", + { + "value_date": 0, + "transaction_date": 1, + "narration": 2, + "reference_no": 3, + "debit": 4, + "credit": 5, + "balance": 6, + }, + 0.90, + ) + # RBL-style: + # transaction date, details, cheque no, value date, withdrawal, deposit, balance. + if width >= 7 and _is_date(sample[0]) and _is_date(sample[3]): + return LayoutTemplate( + "transaction_and_value_date", + { + "transaction_date": 0, + "narration": 1, + "reference_no": 2, + "value_date": 3, + "debit": 4, + "credit": 5, + "balance": 6, + }, + 0.88, + ) + # BOB/CUB and many conventional statements. + if width >= 6 and _is_date(sample[0]): + return LayoutTemplate( + "single_date_reference_debit_credit_balance", + { + "transaction_date": 0, + "narration": 1, + "reference_no": 2, + "debit": width - 3, + "credit": width - 2, + "balance": width - 1, + }, + 0.82, + ) + if width >= 5 and _is_date(sample[0]): + return LayoutTemplate( + "single_date_debit_credit_balance", + { + "transaction_date": 0, + "narration": 1, + "debit": width - 3, + "credit": width - 2, + "balance": width - 1, + }, + 0.78, + ) + return None + + +def _template_for_table(table: list[list]) -> tuple[LayoutTemplate | None, int]: + for row_index, raw_row in enumerate(table[:4]): + row = list(raw_row or []) + mapping = _find_header_mapping(row) + required = {"transaction_date", "narration", "debit", "credit", "balance"} + if required.issubset(mapping): + if "value_date" not in mapping: + mapping["value_date"] = mapping["transaction_date"] + return LayoutTemplate("header_alias_template", mapping, 0.96), row_index + 1 + + inferred = _infer_structural_template(table) + return inferred, 0 + + +def _bank_name(text: str, bank_hint: str | None) -> str: + if bank_hint and bank_hint != "auto": + return _BANK_KEY_LABELS.get(bank_hint, bank_hint.replace("_", " ").title()) + upper = (text or "").upper() + best_name = "Unknown Bank" + best_score = 0 + for name, markers in _BANK_MARKERS: + score = sum(marker in upper for marker in markers) + if score > best_score: + best_name, best_score = name, score + return best_name + + +def _extract_reference(narration: str, explicit: str) -> str: + explicit = norm(explicit) + if explicit and explicit != "-": + return explicit + patterns = ( + r"\bUPI/(?:DR|CR|DRC)?/?([0-9]{10,18})\b", + r"\bIMPS[/ ]([A-Z0-9]{10,30})\b", + r"\b(?:NEFT|RTGS)[*/:/ -]([A-Z0-9]{10,35})\b", + r"\bUTR[:/ ]([A-Z0-9]{8,35})\b", + r"\bCHQ(?:UE)?\s*NO\.?\s*([0-9]{2,12})\b", + ) + for pattern in patterns: + match = re.search(pattern, narration, re.I) + if match: + return match.group(1) + return "" + + +def _parse_summary(text: str, meta: StatementMeta) -> None: + opening_patterns = ( + r"Opening\s+Bal(?:ance)?\s*[:\-]?\s*([\d,]+\.\d{2})\s*(CR|DR)?", + r"Brought\s+Forward(?:\([^)]*\))?\s*[:\-]?\s*([\d,]+\.\d{2})\s*(CR|DR)?", + r"\bB/F\s+([\d,]+\.\d{2})\s*(CR|DR)?", + ) + closing_patterns = ( + r"Closing\s+Bal(?:ance)?\s*[:\-]?\s*([\d,]+\.\d{2})\s*(CR|DR)?", + r"Grand\s+Total:[^\n]*?([\d,]+\.\d{2})\s*(CR|DR)?", + ) + + def signed(match) -> float | None: + if not match: + return None + value = amount(match.group(1)) + suffix = (match.group(2) or "").upper() if match.lastindex and match.lastindex >= 2 else "" + if value is not None and suffix == "DR": + value = -abs(value) + return value + + for pattern in opening_patterns: + value = signed(re.search(pattern, text, re.I | re.S)) + if value is not None: + meta.opening_balance = value + break + for pattern in closing_patterns: + value = signed(re.search(pattern, text, re.I | re.S)) + if value is not None: + meta.closing_balance = value + break + + summary = re.search( + r"(?:Total\s+Debits?|Grand\s+Total)\D*([\d,]+\.\d{2})\D+(?:Total\s+Credits?|)([\d,]+\.\d{2})", + text, + re.I | re.S, + ) + if summary and "TOTAL DEBIT" in text.upper(): + meta.total_debit = amount(summary.group(1)) + meta.total_credit = amount(summary.group(2)) + + period = re.search( + r"(?:Statement\s+(?:From|of account for the period of)|Period\s*:?)\s*" + r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{1,2}-[A-Za-z]{3}-\d{2,4})\s+" + r"(?:to|To)\s+" + r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{1,2}-[A-Za-z]{3}-\d{2,4})", + text, + re.I, + ) + if period: + meta.period_from = date_iso(period.group(1)) + meta.period_to = date_iso(period.group(2)) + + +def _extract_metadata(text: str, meta: StatementMeta) -> None: + meta.customer_name = norm( + find(r"(?:A/C Name|Account Name|Accountholder Name|Primary Holder)\s*:?[ \t]*([^\n]+)", text, flags=re.I) + ) + meta.account_number = find( + r"(?:A/C Number|Account Number|Account No\.?|Savings Account No\.)\s*:?[ \t]*([0-9Xx* -]{6,30})", + text, + flags=re.I, + ).replace(" ", "") + meta.customer_id = find(r"(?:CIF ID|CIF Number|Customer ID|Cust Id)\s*:?[ \t]*([0-9Xx*]+)", text, flags=re.I) + meta.ifsc = find(r"(?:IFSC(?:/RTGS/NEFT)?|IFS Code)\s*:?[ \t]*([A-Z]{4}0[A-Z0-9]{6})", text, flags=re.I) + _parse_summary(text, meta) + + +def _continuity_score(rows: list[dict], opening_balance: float | None) -> float: + if not rows: + return 0.0 + checks = matches = 0 + previous = opening_balance + for row in rows: + balance = row.get("balance") + debit = row.get("debit") or 0.0 + credit = row.get("credit") or 0.0 + if balance is None: + continue + if previous is None: + previous = float(balance) + float(debit) - float(credit) + expected = round(float(previous) + float(credit) - float(debit), 2) + checks += 1 + if abs(expected - float(balance)) <= 0.05: + matches += 1 + previous = float(balance) + return matches / checks if checks else 0.0 + + +class TemplateBasedStatementParser(BaseParser): + bank_name = "Template Matched Bank" + parser_name = "TemplateBasedStatementParser" + + @classmethod + def detect(cls, text: str) -> float: + upper = (text or "").upper() + structural = sum( + token in upper + for token in ( + "VALUE DATE", "TRANSACTION DATE", "POST DATE", "WITHDRAWAL", + "DEPOSIT", "DEBIT", "CREDIT", "BALANCE", "PARTICULARS", + "TRANSACTION DETAILS", + ) + ) + return min(0.85, structural * 0.08) if structural >= 4 else 0.0 + + def __init__(self, bank_hint: str | None = None): + self.bank_hint = (bank_hint or "auto").strip().lower() + + def parse(self, path, text=None): + pdf_path = Path(path) + text = text or extract_text(pdf_path) + detected_bank = _bank_name(text, self.bank_hint) + meta = StatementMeta( + bank_name=detected_bank, + source_file=pdf_path.name, + parser_name=self.parser_name, + confidence="Template Matched", + ) + _extract_metadata(text, meta) + + rows: list[dict] = [] + template_names: list[str] = [] + with pdfplumber.open(str(pdf_path)) as pdf: + for page_no, page in enumerate(pdf.pages, start=1): + tables = page.extract_tables() or [] + for table in tables: + if not table: + continue + template, data_start = _template_for_table(table) + if template is None: + continue + page_rows = 0 + for raw in table[data_start:]: + cells = list(raw or []) + mapping = template.mapping + max_index = max(mapping.values()) + if len(cells) <= max_index: + cells += [None] * (max_index + 1 - len(cells)) + txn_text = norm(cells[mapping["transaction_date"]]) + if not _is_date(txn_text): + continue + value_index = mapping.get("value_date", mapping["transaction_date"]) + value_text = norm(cells[value_index]) or txn_text + narration = norm(cells[mapping["narration"]]) + reference = norm(cells[mapping["reference_no"]]) if "reference_no" in mapping else "" + debit = amount(cells[mapping["debit"]]) + credit = amount(cells[mapping["credit"]]) + balance = _signed_balance(cells[mapping["balance"]]) + if balance is None or (debit is None and credit is None): + continue + rows.append( + { + "transaction_date": txn_text, + "value_date": value_text, + "narration": narration, + "reference_no": _extract_reference(narration, reference), + "debit": debit, + "credit": credit, + "balance": balance, + "source_page": page_no, + "mode": infer_mode(narration), + } + ) + page_rows += 1 + if page_rows: + template_names.append(template.name) + + if not rows: + raise ValueError("No transaction table matched a supported statement template.") + + continuity = _continuity_score(rows, meta.opening_balance) + dated_ratio = sum(not pd.isna(parse_flexible_date(row["transaction_date"])) for row in rows) / len(rows) + if len(rows) < 2 or dated_ratio < 0.90 or continuity < 0.80: + raise ValueError( + "A similar table layout was found, but transaction balances did not reconcile reliably. " + "The statement was not accepted to prevent an incorrect analysis." + ) + + meta.parser_name = f"TemplateBasedStatementParser[{','.join(dict.fromkeys(template_names))}]" + meta.confidence = f"High ({continuity:.1%} balance continuity)" + frame = pd.DataFrame(rows) + return meta, finalize(frame, meta) + + +def parse_with_template(path, text: str | None = None, bank_hint: str | None = None): + return TemplateBasedStatementParser(bank_hint=bank_hint).parse(path, text)