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 _normalise_date_cell(value) -> str: """Normalise PDF table line-wrap artefacts without changing the date value.""" text = norm(value) text = re.sub(r"\s*([/-])\s*", r"\1", text) return text def _is_date(value) -> bool: text = _normalise_date_cell(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 any(_is_date(row[index]) for index in range(min(4, len(row)))) 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) # Modern ICICI and similar indexed statements: # serial no, transaction id, transaction date, cheque no, description, # withdrawal, deposit, available balance. if width >= 8 and _is_date(sample[2]): return LayoutTemplate( "indexed_transaction_id_debit_credit_balance", { "transaction_date": 2, "value_date": 2, "reference_no": 1, "narration": 4, "debit": 5, "credit": 6, "balance": 7, }, 0.92, ) # 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 = _normalise_date_cell(cells[mapping["transaction_date"]]) if not _is_date(txn_text): continue value_index = mapping.get("value_date", mapping["transaction_date"]) value_text = _normalise_date_cell(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)