A R R R Associates
509 lines
18 KiB
Python
509 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
|
|
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"(?<![\d.])([\d,]+\.\d{2})(?!\d)")
|
|
_REFERENCE_RE = re.compile(
|
|
r"(?:UPI|NEFT|IMPS|RTGS|UTR|CHEQUE|CHQ)[/ :A-Z0-9._-]{5,}", re.I
|
|
)
|
|
_HEADER_RE = re.compile(
|
|
r"^(?:REF\s*NO|VALUE\s*DATE|POST\s*DATE|TXN\s*DATE|DATE\s+DETAILS|"
|
|
r"ACCOUNT\s+STATEMENT|STATEMENT\s+OF\s+ACCOUNT|STATE\s+BANK\s+OF\s+INDIA|"
|
|
r"PAGE\s+NO|STATEMENT\s+SUMMARY|BROUGHT\s+FORWARD|DR\s+COUNT|CR\s+COUNT)",
|
|
re.I,
|
|
)
|
|
|
|
|
|
|
|
def _summary_metrics(text: str) -> dict[str, float | int] | None:
|
|
"""Extract SBI's printed statement summary without depending on line wrapping."""
|
|
match = re.search(
|
|
r"Brought\s+Forward.*?Dr\s+Count\s+Cr\s+Count.*?"
|
|
r"Total\s+Debits.*?Total\s+Credits.*?Closing\s+Balance.*?"
|
|
r"([\d,]+\.\d{2})\s*(CR|DR)?\s+"
|
|
r"(\d{1,6})\s+(\d{1,6})\s+"
|
|
r"([\d,]+\.\d{2})\s+([\d,]+\.\d{2})\s+"
|
|
r"([\d,]+\.\d{2})\s*(CR|DR)?",
|
|
text,
|
|
re.I | re.S,
|
|
)
|
|
if not match:
|
|
return None
|
|
|
|
def signed(value: str, suffix: str | None) -> float:
|
|
parsed = float(value.replace(",", ""))
|
|
return -parsed if (suffix or "").upper() == "DR" else parsed
|
|
|
|
return {
|
|
"opening_balance": signed(match.group(1), match.group(2)),
|
|
"debit_count": int(match.group(3)),
|
|
"credit_count": int(match.group(4)),
|
|
"total_debit": float(match.group(5).replace(",", "")),
|
|
"total_credit": float(match.group(6).replace(",", "")),
|
|
"closing_balance": signed(match.group(7), match.group(8)),
|
|
}
|
|
|
|
|
|
def _expected_summary_transactions(text: str) -> int | None:
|
|
summary = _summary_metrics(text)
|
|
if summary:
|
|
return int(summary["debit_count"]) + int(summary["credit_count"])
|
|
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 _apply_printed_summary(meta: StatementMeta, text: str) -> None:
|
|
summary = _summary_metrics(text)
|
|
if not summary:
|
|
return
|
|
meta.opening_balance = float(summary["opening_balance"])
|
|
meta.total_debit = float(summary["total_debit"])
|
|
meta.total_credit = float(summary["total_credit"])
|
|
meta.closing_balance = float(summary["closing_balance"])
|
|
|
|
|
|
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,
|
|
)
|
|
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"(?<=[^\n\d])(?=(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\s+"
|
|
r"(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\s+)",
|
|
"\n",
|
|
text,
|
|
)
|
|
# Typical OCR noise turns 12/05/2025 into 12105/2025 by reading the first
|
|
# slash as the digit 1. Correct only at a transaction-line boundary.
|
|
text = re.sub(
|
|
r"(?m)^(\s*)(\d{1,2})[1|](\d{2})/(\d{4})(?=\s)",
|
|
r"\1\2/\3/\4",
|
|
text,
|
|
)
|
|
|
|
rows: list[dict] = []
|
|
current: dict | None = None
|
|
pending_prefix: list[str] = []
|
|
page = 1
|
|
|
|
for raw_line in text.split("\n"):
|
|
page += _page_increment(raw_line)
|
|
line = raw_line.replace("\f", "").rstrip()
|
|
match = start_re.match(line)
|
|
if match:
|
|
if current:
|
|
rows.append(current)
|
|
transaction_date = match.group(1)
|
|
value_date = match.group(2) if date_groups == 2 else transaction_date
|
|
remainder = match.group(date_groups + 1)
|
|
current = {
|
|
"transaction_date": transaction_date,
|
|
"value_date": value_date,
|
|
"lines": [remainder],
|
|
"prefix_lines": list(pending_prefix),
|
|
"source_page": page,
|
|
}
|
|
pending_prefix = []
|
|
continue
|
|
|
|
stripped = line.strip()
|
|
is_mode_prefix = bool(
|
|
stripped
|
|
and re.fullmatch(
|
|
r"(?:DEP|WDL)\s+TFR(?:\s+INB.*)?|INTEREST\s+CREDIT|CEMTEX\s+DEP",
|
|
stripped,
|
|
re.I,
|
|
)
|
|
)
|
|
if is_mode_prefix:
|
|
pending_prefix = [stripped]
|
|
continue
|
|
if current and stripped and not _HEADER_RE.match(stripped):
|
|
current["lines"].append(stripped)
|
|
|
|
if current:
|
|
rows.append(current)
|
|
|
|
output: list[dict] = []
|
|
previous_balance = meta.opening_balance
|
|
for row in rows:
|
|
first_line = row["lines"][0]
|
|
full_text = norm(" ".join([*row.get("prefix_lines", []), *row["lines"]]))
|
|
debit, credit, balance, first_narration = _row_values(
|
|
first_line,
|
|
full_text,
|
|
previous_balance,
|
|
)
|
|
if balance is None:
|
|
continue
|
|
|
|
narration = norm(" ".join([*row.get("prefix_lines", []), first_narration, *row["lines"][1:]]))
|
|
if debit is None and credit is None and previous_balance is not None:
|
|
delta = round(balance - previous_balance, 2)
|
|
if delta < -0.01:
|
|
debit = abs(delta)
|
|
elif delta > 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"
|
|
|
|
@classmethod
|
|
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 SBIAccountSummaryParser(BaseParser):
|
|
"""SBI Account Summary layout with dual dates and a printed summary page."""
|
|
|
|
bank_name = "State Bank of India"
|
|
parser_name = "SBIAccountSummaryParser"
|
|
|
|
@classmethod
|
|
def detect(cls, text: str) -> float:
|
|
upper = norm(text).upper()
|
|
required = (
|
|
"STATE BANK OF INDIA",
|
|
"ACCOUNT SUMMARY",
|
|
"STATEMENT OF ACCOUNT",
|
|
"STATEMENT SUMMARY",
|
|
"BROUGHT FORWARD",
|
|
"DR COUNT",
|
|
"CR COUNT",
|
|
)
|
|
if all(marker in upper for marker in required):
|
|
return 0.998
|
|
return 0.0
|
|
|
|
def parse(self, path, text=None):
|
|
extracted_text = text or extract_text(path)
|
|
meta = _extract_common_meta(extracted_text, path, self.parser_name)
|
|
_apply_printed_summary(meta, extracted_text)
|
|
if meta.opening_balance is None:
|
|
meta.opening_balance = _opening_from_summary(extracted_text)
|
|
start_re = re.compile(
|
|
rf"^\s*({DATE_TOKEN_PATTERN})\s+({DATE_TOKEN_PATTERN})\s+(.*)$",
|
|
re.I,
|
|
)
|
|
frame = _parse_rows(extracted_text, meta, start_re, 2)
|
|
|
|
summary = _summary_metrics(extracted_text)
|
|
if summary and len(frame) != int(summary["debit_count"]) + int(summary["credit_count"]):
|
|
raise ValueError(
|
|
"SBI Account Summary rows could not be fully reconciled with the printed "
|
|
"debit and credit counts. Please retain the PDF and contact support."
|
|
)
|
|
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
|