Add validated template fallback for bank statement layouts

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)
+        # Preserve HSBC's specialised OCR parser as the first route.
-        if parser is None:
+        candidates = BANK_PARSERS.get(hint, [])
-            label = dict(BANK_OPTIONS).get(hint, "the selected bank")
+        scored = sorted(
-            raise ValueError(
+            ((parser.detect(text), parser) for parser in candidates),
-                f"The uploaded statement does not match the selected {label} format. "
+            key=lambda item: item[0],
-                "Please verify the selected bank or choose Auto Detect."
+            reverse=True,
-            )
-        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."
         )
-    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

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)