diff --git a/app/modules/bank_statement_analyzer/parsers/registry.py b/app/modules/bank_statement_analyzer/parsers/registry.py index d9e71b5..edfe084 100644 --- a/app/modules/bank_statement_analyzer/parsers/registry.py +++ b/app/modules/bank_statement_analyzer/parsers/registry.py @@ -7,7 +7,7 @@ from .hsbc import HSBCParser from .indian_bank import IndianBankModernParser, IndianBankLegacyParser from .indusind import IndusIndParser from .kotak import KotakParser -from .sbi import SBIModernParser, SBIOtherParser +from .sbi import SBIYONORelationshipParser, SBIStandardParser, SBICompactParser, SBIModernParser, SBIOtherParser from .base import extract_text PARSERS = [ @@ -19,8 +19,9 @@ PARSERS = [ IndianBankLegacyParser, IndusIndParser, KotakParser, - SBIOtherParser, - SBIModernParser, + SBIYONORelationshipParser, + SBIStandardParser, + SBICompactParser, ] BANK_OPTIONS = [ @@ -43,7 +44,7 @@ BANK_PARSERS = { "indian_bank": [IndianBankModernParser, IndianBankLegacyParser], "indusind": [IndusIndParser], "kotak": [KotakParser], - "sbi": [SBIOtherParser, SBIModernParser], + "sbi": [SBIYONORelationshipParser, SBIStandardParser, SBICompactParser], } diff --git a/app/modules/bank_statement_analyzer/parsers/sbi.py b/app/modules/bank_statement_analyzer/parsers/sbi.py index dde3f81..fa806c2 100644 --- a/app/modules/bank_statement_analyzer/parsers/sbi.py +++ b/app/modules/bank_statement_analyzer/parsers/sbi.py @@ -1,120 +1,409 @@ from __future__ import annotations -import re, pandas as pd + +import os +import re +import subprocess +import tempfile +from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path -from .base import * -from .common import DATE_TOKEN_PATTERN, date_iso, find -class SBIModernParser(BaseParser): - bank_name='State Bank of India'; parser_name='SBIModernParser' - @classmethod - def detect(cls,text): - u=norm(text).upper() - bank_marker='STATE BANK OF INDIA' in u - digital_layout='REF NO./CHEQUE' in u and 'DETAILS' in u - yono_print_layout=( - 'STATEMENT OF ACCOUNT' in u - and 'VALUE DATE' in u - and 'POST DATE' in u - and 'DEBIT' in u - and 'CREDIT' in u - and 'BALANCE' in u +import pandas as pd +import pdfplumber + +from .base import BaseParser, StatementMeta, amount, extract_text, finalize, norm +from .common import DATE_TOKEN_PATTERN, date_iso, find, parse_flexible_date + +_MONEY_RE = re.compile(r"(? int | None: + counts = re.findall( + r"(?:Brought\s+Forward.*?)(\d{1,3})\s+(\d{1,3})\s+[\d,]+\.\d{2}\s+[\d,]+\.\d{2}", + text, + re.I | re.S, + ) + if not counts: + return None + return sum(int(debit_count) + int(credit_count) for debit_count, credit_count in counts) + + +def _ocr_yono_page(pdf_path: Path, page_number: int, work_dir: Path, dpi: int) -> tuple[int, str]: + prefix = work_dir / f"sbi_yono_{page_number:05d}" + image_path = prefix.with_suffix(".png") + render = subprocess.run( + [ + "pdftoppm", "-f", str(page_number), "-l", str(page_number), + "-r", str(dpi), "-png", "-singlefile", str(pdf_path), str(prefix), + ], + capture_output=True, + text=True, + timeout=180, + check=False, + ) + if render.returncode != 0 or not image_path.exists(): + raise RuntimeError(render.stderr.strip() or f"Unable to render SBI page {page_number}.") + try: + environment = os.environ.copy() + environment.setdefault("OMP_THREAD_LIMIT", "1") + ocr = subprocess.run( + [ + "tesseract", str(image_path), "stdout", "-l", "eng", + "--psm", "6", "-c", "preserve_interword_spaces=1", + ], + capture_output=True, + text=True, + timeout=180, + check=False, + env=environment, ) - return 0.98 if bank_marker and (digital_layout or yono_print_layout) else 0 - def parse(self,path,text=None): - text=text or extract_text(path); meta=StatementMeta(bank_name=self.bank_name,source_file=Path(path).name,parser_name=self.parser_name,confidence='High') - meta.customer_name=find(r'Account Name\s*:?\s*([^\n]+)',text) - if not meta.customer_name: - meta.customer_name=find(r'\bMs\.?\s+([A-Z][A-Z .]{3,})',text) - meta.account_number=find(r'Account Number\s*:?\s*([0-9X*]+)',text) - meta.ifsc=find(r'(?:IFS|IFSC) Code\s*:?\s*([A-Z0-9]+)',text) - m=re.search(rf'(?:Account Statement from|Statement From\s*:?)\s*({DATE_TOKEN_PATTERN})\s+to\s+({DATE_TOKEN_PATTERN})', text, re.I) - if m: - meta.period_from=date_iso(m.group(1)); meta.period_to=date_iso(m.group(2)) - meta.opening_balance=amount(find(r'Balance as on[^\n]*\n\s*([\d,]+\.\d{2})',text)) - # Format A: date details ref debit credit balance - date_re=re.compile(rf'^\s*({DATE_TOKEN_PATTERN})\s+(.*)$', re.I) - rows=[]; cur=None; page=1 - for line in text.splitlines(): - if '\f' in line: page += line.count('\f') - m=date_re.match(line) - if m: - if cur: rows.append(cur) - cur={'transaction_date':m.group(1),'value_date':m.group(1),'lines':[m.group(2)],'source_page':page} - elif cur and line.strip() and not re.match(r'^(Date\s+Details|Txn Date|Account Statement|State Bank)',line.strip(),re.I): - cur['lines'].append(line) - if cur: rows.append(cur) - out=[]; prev=meta.opening_balance - for r in rows: - first=r['lines'][0]; alltxt=norm(' '.join(r['lines'])) - nums=list(re.finditer(r'(?=2: - txn=amount(nums[-2].group(1)); p=nums[-2].start() - if prev is not None and bal is not None: - if abs((prev-txn)-bal)<0.05: debit=txn - elif abs((prev+txn)-bal)<0.05: credit=txn - if debit is None and credit is None: - upper=alltxt.upper() - if '/CR/' in upper or upper.startswith('DEP TFR') or 'INTEREST CREDIT' in upper: - credit=txn - elif '/DR/' in upper or upper.startswith('WDL TFR') or upper.startswith('DEBIT '): - debit=txn - elif p >= 78: - credit=txn - else: - debit=txn - narr=first[:nums[-2].start() if len(nums)>=2 else nums[-1].start()].strip()+' '+' '.join(x.strip() for x in r['lines'][1:]) - ref=''; z=re.search(r'(?:UPI|NEFT|IMPS|RTGS)[/A-Z0-9-]{6,}',alltxt,re.I); ref=z.group(0) if z else '' - if debit is None and credit is None: continue - out.append({**r,'narration':norm(narr),'reference_no':ref,'debit':debit,'credit':credit,'balance':bal}) - if bal is not None: prev=bal - if out: - if meta.opening_balance is None: - f=out[0]; meta.opening_balance=round((f['balance'] or 0)+(f.get('debit') or 0)-(f.get('credit') or 0),2) - meta.closing_balance=out[-1]['balance'] - return meta,finalize(pd.DataFrame(out),meta) + if ocr.returncode != 0: + raise RuntimeError(ocr.stderr.strip() or f"SBI OCR failed for page {page_number}.") + return page_number, ocr.stdout or "" + finally: + image_path.unlink(missing_ok=True) + + +def _extract_yono_table_text(path: str | Path) -> str: + pdf_path = Path(path) + dpi = max(120, min(int(os.getenv("BANK_ANALYZER_SBI_OCR_DPI", "150")), 220)) + workers = max(1, min(int(os.getenv("BANK_ANALYZER_SBI_OCR_WORKERS", "2")), 3)) + with pdfplumber.open(str(pdf_path)) as pdf: + page_count = len(pdf.pages) + pages: dict[int, str] = {} + with tempfile.TemporaryDirectory(prefix="sbi_yono_ocr_") as temporary: + work_dir = Path(temporary) + with ThreadPoolExecutor(max_workers=min(workers, page_count)) as executor: + futures = { + executor.submit(_ocr_yono_page, pdf_path, page, work_dir, dpi): page + for page in range(1, page_count + 1) + } + for future in as_completed(futures): + page_number, page_text = future.result() + pages[page_number] = page_text + return "\n\f\n".join(pages.get(page, "") for page in range(1, page_count + 1)) + +def _page_increment(line: str) -> int: + return line.count("\f") + + +def _extract_common_meta(text: str, path: str | Path, parser_name: str) -> StatementMeta: + meta = StatementMeta( + bank_name="State Bank of India", + source_file=Path(path).name, + parser_name=parser_name, + confidence="High", + ) + meta.customer_name = find(r"Account Name\s*:?\s*([^\n]+)", text) + if not meta.customer_name: + meta.customer_name = find(r"\b(?:Ms|Mr|Mrs)\.?\s+([A-Z][A-Z .]{3,})", text) + meta.account_number = find(r"Account Number\s*:?\s*([0-9Xx*]+)", text) + meta.ifsc = find(r"(?:IFS|IFSC) Code\s*:?\s*([A-Z0-9]+)", text) + + periods = re.findall( + rf"(?:Account Statement from|Statement From\s*:?)\s*({DATE_TOKEN_PATTERN})\s+to\s+({DATE_TOKEN_PATTERN})", + text, + re.I, + ) + if periods: + parsed = [ + (parse_flexible_date(start), parse_flexible_date(end)) + for start, end in periods + ] + parsed = [(start, end) for start, end in parsed if not pd.isna(start) and not pd.isna(end)] + if parsed: + meta.period_from = min(start for start, _ in parsed).strftime("%Y-%m-%d") + meta.period_to = max(end for _, end in parsed).strftime("%Y-%m-%d") + + return meta + + +def _opening_from_summary(text: str) -> float | None: + patterns = ( + r"Brought\s+Forward\s*(?:\([^)]*\))?\s*[:\-]?\s*([\d,]+\.\d{2})\s*(CR|DR)?", + r"Balance\s+as\s+on[^\n]*?([\d,]+\.\d{2})\s*(CR|DR)?", + ) + for pattern in patterns: + match = re.search(pattern, text, re.I | re.S) + if not match: + continue + 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 = -value + if value is not None: + return value + return None + + +def _reference(narration: str) -> str: + match = _REFERENCE_RE.search(narration or "") + return norm(match.group(0)) if match else "" + + +def _semantic_side(narration: str) -> str: + upper = norm(narration).upper() + credit_markers = ( + "/CR/", "BY TRANSFER", "TRANSFER FROM", "DEP TFR", "DEPOSIT", + "INTEREST CREDIT", "CREDIT INTEREST", "CASH DEP", "REVERSAL", "REFUND", + ) + debit_markers = ( + "/DR/", "TO TRANSFER", "TRANSFER TO", "WDL TFR", "WITHDRAWAL", + "ATM WDL", "DEBIT-", "DEBIT ", "CAS PRES", "CHQ", "CHEQUE", + ) + if any(marker in upper for marker in credit_markers): + return "credit" + if any(marker in upper for marker in debit_markers): + return "debit" + return "" + + +def _row_values(first_line: str, narration: str, previous_balance: float | None): + tokens = list(_MONEY_RE.finditer(first_line)) + if not tokens: + return None, None, None, first_line + + balance = amount(tokens[-1].group(1)) + movement_candidates = [amount(match.group(1)) for match in tokens[:-1]] + movement_candidates = [value for value in movement_candidates if value is not None] + debit = credit = None + + if previous_balance is not None and balance is not None: + delta = round(balance - previous_balance, 2) + if abs(delta) >= 0.01: + matching = next( + (value for value in reversed(movement_candidates) if abs(abs(delta) - value) <= 0.05), + None, + ) + movement = matching if matching is not None else abs(delta) + if delta < 0: + debit = movement + else: + credit = movement + elif movement_candidates: + movement = movement_candidates[-1] + side = _semantic_side(narration) + if side == "credit": + credit = movement + else: + debit = movement + + narration_end = tokens[-2].start() if len(tokens) >= 2 else tokens[-1].start() + return debit, credit, balance, first_line[:narration_end].strip() + + +def _parse_rows( + text: str, + meta: StatementMeta, + start_re: re.Pattern[str], + date_groups: int, +) -> pd.DataFrame: + # OCR engines occasionally place two adjacent SBI transactions on one + # physical line. Split before every detected date pair so each transaction + # reaches the normal row collector independently. + text = re.sub( + r"(? 0.01: + credit = delta + if debit is None and credit is None: + continue + + output.append( + { + "transaction_date": row["transaction_date"], + "value_date": row["value_date"], + "narration": narration, + "reference_no": _reference(full_text), + "debit": debit, + "credit": credit, + "balance": balance, + "source_page": row["source_page"], + } + ) + previous_balance = balance + + frame = pd.DataFrame(output) + if not frame.empty: + if meta.opening_balance is None: + first = output[0] + meta.opening_balance = round( + float(first["balance"]) + float(first.get("debit") or 0) - float(first.get("credit") or 0), + 2, + ) + meta.closing_balance = float(output[-1]["balance"]) + meta.total_debit = round(float(frame["debit"].fillna(0).sum()), 2) + meta.total_credit = round(float(frame["credit"].fillna(0).sum()), 2) + return finalize(frame, meta) + + +class SBIYONORelationshipParser(BaseParser): + bank_name = "State Bank of India" + parser_name = "SBIYONORelationshipParser" -class SBIOtherParser(BaseParser): - bank_name='State Bank of India'; parser_name='SBIOtherParser' @classmethod - def detect(cls,text): - u=text.upper(); return 0.97 if 'TXN DATE' in u and 'VALUE' in u and 'REF NO./CHEQUE' in u and 'BY TRANSFER' in u else 0 - def parse(self,path,text=None): - text=text or extract_text(path); meta=StatementMeta(bank_name=self.bank_name,source_file=Path(path).name,parser_name=self.parser_name,confidence='High') - meta.customer_name=find(r'Account Name\s*:\s*([^\n]+)',text) - meta.account_number=find(r'Account Number\s*:?\s*([0-9X*]+)',text) - meta.ifsc=find(r'IFS Code\s*:?\s*([A-Z0-9]+)',text) - m=re.search(rf'Account Statement from\s*({DATE_TOKEN_PATTERN})\s+to\s+({DATE_TOKEN_PATTERN})', text, re.I) - if m: - meta.period_from=date_iso(m.group(1)); meta.period_to=date_iso(m.group(2)) - m=re.search(r'Balance as on\s+[^\n]+\n',text,re.I) - date_re=re.compile(rf'^\s*({DATE_TOKEN_PATTERN})\s+({DATE_TOKEN_PATTERN})\s+(.*)$', re.I) - rows=[]; cur=None; page=1 - for line in text.splitlines(): - if '\f' in line: page += line.count('\f') - mm=date_re.match(line) - if mm: - if cur: rows.append(cur) - cur={'transaction_date':mm.group(1),'value_date':mm.group(2),'lines':[mm.group(3)],'source_page':page} - elif cur and line.strip() and not re.match(r'^(Txn Date|Account Statement|Account Name)',line.strip(),re.I): cur['lines'].append(line) - if cur: rows.append(cur) - out=[] - for r in rows: - first=r['lines'][0]; alltxt=norm(' '.join(r['lines'])) - nums=list(re.finditer(r'(?=2: - txn=amount(nums[-2].group(1)); p=nums[-2].start() - # Based on header layout: Debit starts before Credit - if p < 55: debit=txn - else: credit=txn - narr=first[:nums[-2].start() if len(nums)>=2 else nums[-1].start()].strip()+' '+' '.join(x.strip() for x in r['lines'][1:]) - ref=''; z=re.search(r'(?:UPI|NEFT|IMPS|RTGS)[/A-Z0-9-]{6,}',alltxt,re.I); ref=z.group(0) if z else '' - if debit is None and credit is None: continue - out.append({**r,'narration':norm(narr),'reference_no':ref,'debit':debit,'credit':credit,'balance':bal}) - if out: - f=out[0]; meta.opening_balance=round((f['balance'] or 0)+(f.get('debit') or 0)-(f.get('credit') or 0),2); meta.closing_balance=out[-1]['balance'] - return meta,finalize(pd.DataFrame(out),meta) + def detect(cls, text: str) -> float: + upper = norm(text).upper() + markers = ( + "RELATIONSHIP SUMMARY", + "STATEMENT OF ACCOUNT", + "VALUE DATE", + "POST DATE", + "STATEMENT SUMMARY", + ) + score = sum(marker in upper for marker in markers) + return 0.995 if "STATE BANK OF INDIA" in upper and score >= 3 else 0.0 + + def parse(self, path, text=None): + extracted_text = text or extract_text(path) + start_re = re.compile( + r"^\s*(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\s+" + r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\s+(.*)$", + re.I, + ) + + meta = _extract_common_meta(extracted_text, path, self.parser_name) + meta.opening_balance = _opening_from_summary(extracted_text) + frame = _parse_rows(extracted_text, meta, start_re, 2) + expected = _expected_summary_transactions(extracted_text) + + # The shared OCR path is sufficient for detection, but dense YONO tables + # occasionally need a table-oriented OCR pass. Retry only when the bank's + # printed debit/credit counts prove that rows were missed. + if expected and len(frame) < max(1, int(expected * 0.95)): + try: + table_text = _extract_yono_table_text(path) + retry_meta = _extract_common_meta(table_text, path, self.parser_name) + retry_meta.opening_balance = _opening_from_summary(table_text) + retry_frame = _parse_rows(table_text, retry_meta, start_re, 2) + if len(retry_frame) > len(frame): + meta, frame = retry_meta, retry_frame + except (OSError, RuntimeError, subprocess.SubprocessError): + # Preserve the already extracted result when the optional targeted + # retry is unavailable; the common OCR failure handling remains + # unchanged. + pass + + return meta, frame + + +class SBIStandardParser(BaseParser): + bank_name = "State Bank of India" + parser_name = "SBIStandardParser" + + @classmethod + def detect(cls, text: str) -> float: + upper = norm(text).upper() + if "STATE BANK OF INDIA" not in upper and "ACCOUNT STATEMENT FROM" not in upper: + return 0.0 + has_table = ( + "TXN DATE" in upper + and "DESCRIPTION" in upper + and "REF NO./CHEQUE" in upper + and "DEBIT" in upper + and "CREDIT" in upper + and "BALANCE" in upper + and ("VALUE DATE" in upper or "TXN DATE VALUE DESCRIPTION" in upper) + ) + return 0.99 if has_table else 0.0 + + def parse(self, path, text=None): + text = text or extract_text(path) + meta = _extract_common_meta(text, path, self.parser_name) + meta.opening_balance = _opening_from_summary(text) + start_re = re.compile( + rf"^\s*({DATE_TOKEN_PATTERN})\s+({DATE_TOKEN_PATTERN})\s+(.*)$", + re.I, + ) + return meta, _parse_rows(text, meta, start_re, 2) + + +class SBICompactParser(BaseParser): + bank_name = "State Bank of India" + parser_name = "SBICompactParser" + + @classmethod + def detect(cls, text: str) -> float: + upper = norm(text).upper() + markers = ("STATE BANK OF INDIA", "DATE DETAILS", "REF NO./CHEQUE", "SEARCH FOR") + return 0.985 if all(marker in upper for marker in markers) else 0.0 + + def parse(self, path, text=None): + text = text or extract_text(path) + meta = _extract_common_meta(text, path, self.parser_name) + meta.opening_balance = _opening_from_summary(text) + start_re = re.compile(rf"^\s*({DATE_TOKEN_PATTERN})\s+(.*)$", re.I) + return meta, _parse_rows(text, meta, start_re, 1) + + +# Backward-compatible names retained because the production registry and any +# external imports may still refer to these classes. +SBIModernParser = SBICompactParser +SBIOtherParser = SBIStandardParser