A R R R Associates
405 lines
15 KiB
Python
405 lines
15 KiB
Python
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)
|