From 47cd5c2cf3c9b5d469ed46d1a01ed9a3f9fd91a2 Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Wed, 5 Aug 2026 17:49:44 +0530 Subject: [PATCH] Fix SBI multiline table parsing and restore template-first bank analysis --- .../parsers/bank_identity.py | 53 ++++-- .../parsers/registry.py | 161 ++++++++++++++---- 2 files changed, 165 insertions(+), 49 deletions(-) diff --git a/app/modules/bank_statement_analyzer/parsers/bank_identity.py b/app/modules/bank_statement_analyzer/parsers/bank_identity.py index 7a69715..339aafc 100644 --- a/app/modules/bank_statement_analyzer/parsers/bank_identity.py +++ b/app/modules/bank_statement_analyzer/parsers/bank_identity.py @@ -151,6 +151,10 @@ def detect_bank_identity( ) -> BankIdentityMatch | None: raw_text = str(text or "") text_upper = _normalise(raw_text) + # Identity headings and account metadata are normally in the first page. + # Restrict strong name evidence to the beginning so counterparty bank names + # inside hundreds of transaction narrations cannot relabel the statement. + header_upper = _normalise(raw_text[:800]) ifsc_upper = re.sub(r"[^A-Z0-9]", "", str(ifsc or "").upper()) file_upper = _normalise(Path(source_file or "").stem) @@ -159,25 +163,36 @@ def detect_bank_identity( score = 0.0 evidence: list[str] = [] - # Longest/special IFSC prefixes carry the strongest identity signal. - matching_prefixes = [prefix for prefix in bank.ifsc_prefixes if ifsc_upper.startswith(prefix.upper())] + matching_prefixes = [ + prefix for prefix in bank.ifsc_prefixes + if ifsc_upper.startswith(prefix.upper()) + ] if matching_prefixes: prefix = max(matching_prefixes, key=len) - score += 0.72 if len(prefix) > 4 else 0.58 + # The account IFSC belongs to the issuing bank and therefore outranks + # bank names appearing merely as transaction counterparties. + score += 0.96 if len(prefix) > 4 else 0.90 evidence.append(f"IFSC prefix {prefix}") for domain in bank.domains: if domain.lower() in raw_text.lower(): - score += 0.72 + score += 0.92 evidence.append(f"domain {domain}") break - alias_hits = [alias for alias in bank.aliases if _alias_present(text_upper, alias)] - if alias_hits: - longest = max(alias_hits, key=len) - # Long formal headings are stronger than abbreviations such as SBI. - score += 0.72 if len(_normalise(longest)) >= 12 else 0.36 - evidence.append(f"name {longest}") + header_alias_hits = [alias for alias in bank.aliases if _alias_present(header_upper, alias)] + if header_alias_hits: + longest = max(header_alias_hits, key=len) + score += 0.88 if len(_normalise(longest)) >= 12 else 0.55 + evidence.append(f"header name {longest}") + else: + # Full-document matches are weak because narrations commonly mention + # beneficiary and remitter banks. They may support another signal but + # cannot identify the issuer by themselves. + body_hits = [alias for alias in bank.aliases if _alias_present(text_upper, alias)] + if body_hits: + score += 0.12 + evidence.append("body mention") if any(_alias_present(file_upper, alias) for alias in bank.aliases): score += 0.10 @@ -186,7 +201,9 @@ def detect_bank_identity( confidence = min(score, 0.99) if confidence < 0.50: continue - candidate = BankIdentityMatch(bank.name, bank.category, confidence, tuple(dict.fromkeys(evidence))) + candidate = BankIdentityMatch( + bank.name, bank.category, confidence, tuple(dict.fromkeys(evidence)) + ) if best is None or candidate.confidence > best.confidence: best = candidate @@ -194,9 +211,21 @@ def detect_bank_identity( def apply_detected_bank_identity(meta, df, text: str): + meta_ifsc = str(getattr(meta, "ifsc", "") or "") + if not meta_ifsc: + # Some PDFs place the IFSC value far from its label in text-reading + # order. Recover the first header-area IFSC token directly; limiting + # the search to the beginning avoids beneficiary-bank IFSC codes. + header_match = re.search(r"\b[A-Z]{4}0[A-Z0-9]{6}\b", str(text or "")[:2500], re.I) + if header_match: + meta_ifsc = header_match.group(0).upper() + try: + meta.ifsc = meta_ifsc + except Exception: + pass match = detect_bank_identity( text, - ifsc=str(getattr(meta, "ifsc", "") or ""), + ifsc=meta_ifsc, source_file=str(getattr(meta, "source_file", "") or ""), ) if match is None: diff --git a/app/modules/bank_statement_analyzer/parsers/registry.py b/app/modules/bank_statement_analyzer/parsers/registry.py index 568c4d8..0703145 100644 --- a/app/modules/bank_statement_analyzer/parsers/registry.py +++ b/app/modules/bank_statement_analyzer/parsers/registry.py @@ -1,5 +1,7 @@ from __future__ import annotations +import logging + from .idfc import IDFCFirstParser from .axis import AxisParser from .hdfc import HDFCParser @@ -17,7 +19,9 @@ from .rbl_bank import RBLBankParser from .andhra_pradesh_grameena_bank import AndhraPradeshGrameenaBankParser from .bank_identity import apply_detected_bank_identity from .base import extract_text +from .template_engine import parse_with_template +logger = logging.getLogger(__name__) PARSERS = [ IDFCFirstParser, @@ -88,54 +92,137 @@ def detect_parser(text): return scored[0][1](), scored[0][0] -def _selected_parser(bank_key: str, text: str): - candidates = BANK_PARSERS.get(bank_key, []) - if not candidates: - return None, 0 +def _usable_result(result) -> bool: + if result is None: + return False + try: + _meta, frame = result + return frame is not None and not frame.empty + except Exception: + return False + + +def _identified(result, text: str): + if not _usable_result(result): + return None + meta, frame = result + return apply_detected_bank_identity(meta, frame, text) + + +def _attempt(parser, path, text: str, route: str): + try: + result = parser.parse(path, text) + result = _identified(result, text) + if result is not None: + meta, frame = result + logger.info( + "Bank analyzer route accepted: route=%s parser=%s rows=%s file=%s", + route, + getattr(meta, "parser_name", parser.__class__.__name__), + len(frame), + path, + ) + return result + logger.warning( + "Bank analyzer route returned no rows: route=%s parser=%s file=%s", + route, + parser.__class__.__name__, + path, + ) + except Exception as exc: + logger.warning( + "Bank analyzer route failed: route=%s parser=%s file=%s error=%s", + route, + parser.__class__.__name__, + path, + exc, + exc_info=True, + ) + return None + + +def _template_first(path, text: str, hint: str): + try: + result = parse_with_template(path, text=text, bank_hint=hint) + result = _identified(result, text) + if result is not None: + meta, frame = result + logger.info( + "Bank analyzer template accepted: parser=%s rows=%s file=%s", + getattr(meta, "parser_name", "TemplateBasedStatementParser"), + len(frame), + path, + ) + return result + logger.warning("Bank analyzer template returned no rows: file=%s", path) + except Exception as exc: + logger.warning( + "Bank analyzer template rejected: file=%s error=%s", + path, + exc, + exc_info=True, + ) + return None + + +def _ordered_bank_candidates(text: str): scored = sorted( - ((parser.detect(text), parser) for parser in candidates), + ((parser.detect(text), parser) for parser in PARSERS), key=lambda item: item[0], reverse=True, ) - if scored and scored[0][0] > 0: - return scored[0][1](), scored[0][0] - return None, 0 - - -def _parse_and_identify(parser, path, text: str): - meta, df = parser.parse(path, text) - return apply_detected_bank_identity(meta, df, text) + return [parser for score, parser in scored if score > 0] def parse_pdf(path, bank_hint: str | None = None): + """Parse a statement using a validated template first, then bank parsers. + + Bank identity is resolved independently after extraction, so using a shared + layout never labels a statement as the bank whose parser happens to have a + similar table structure. + """ hint = (bank_hint or "auto").strip().lower() text = extract_text(path) - if hint == "hsbc": - meta, df = HSBCParser().parse(path, text) - return apply_detected_bank_identity(meta, df, text) + template_result = _template_first(path, text, hint) + if template_result is not None: + return template_result 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 _parse_and_identify(parser, 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: - meta, df = HSBCParser().parse(path, text) - return apply_detected_bank_identity(meta, df, text) - except ValueError: - pass - if parser is None: + candidates = BANK_PARSERS.get(hint, []) + for parser_class in sorted(candidates, key=lambda cls: cls.detect(text), reverse=True): + result = _attempt(parser_class(), path, text, f"selected:{hint}") + if result is not None: + return result + label = dict(BANK_OPTIONS).get(hint, "the selected bank") raise ValueError( - "Unsupported statement format. Select the bank manually or add a bank-specific parser for this statement layout." + f"The uploaded statement could not be parsed by a validated layout template or the " + f"selected {label} parser. Please verify that the PDF is readable and supported." ) - return _parse_and_identify(parser, path, text) + + attempted: set[type] = set() + for parser_class in _ordered_bank_candidates(text): + attempted.add(parser_class) + result = _attempt(parser_class(), path, text, "auto-detected") + if result is not None: + return result + + if not (text or "").strip(): + result = _attempt(HSBCParser(), path, text, "image-only-ocr") + if result is not None: + return result + + # A detector may miss a new bank layout even though an existing dedicated + # parser can still extract and reconcile it. Keep this compatibility pass. + for parser_class in PARSERS: + if parser_class in attempted: + continue + result = _attempt(parser_class(), path, text, "compatibility-fallback") + if result is not None: + return result + + raise ValueError( + "The bank statement format was identified, but no transaction rows could be extracted. " + "The validated template engine and all bank-specific parsers were attempted. " + "Please verify that the PDF is text-readable and that this statement layout is supported." + )