Add SBI Account Summary layout template parser

2026-08-04 10:29:49 +05:30
parent 7db38c7b04
commit 1dbde5e0b9
2 changed files with 107 additions and 7 deletions
@@ -26,7 +26,39 @@ _HEADER_RE = re.compile(



+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,
@@ -37,6 +69,16 @@ def _expected_summary_transactions(text: str) -> int | 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")
@@ -211,7 +253,7 @@ def _parse_rows(
    # physical line. Split before every detected date pair so each transaction
    # reaches the normal row collector independently.
    text = re.sub(
-        r"(?<!\n)(?=(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\s+"
+        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,
@@ -226,9 +268,10 @@ def _parse_rows(

    rows: list[dict] = []
    current: dict | None = None
+    pending_prefix: list[str] = []
    page = 1

-    for raw_line in text.splitlines():
+    for raw_line in text.split("\n"):
        page += _page_increment(raw_line)
        line = raw_line.replace("\f", "").rstrip()
        match = start_re.match(line)
@@ -242,11 +285,24 @@ def _parse_rows(
                "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)

@@ -257,7 +313,7 @@ def _parse_rows(
    previous_balance = meta.opening_balance
    for row in rows:
        first_line = row["lines"][0]
-        full_text = norm(" ".join(row["lines"]))
+        full_text = norm(" ".join([*row.get("prefix_lines", []), *row["lines"]]))
        debit, credit, balance, first_narration = _row_values(
            first_line,
            full_text,
@@ -266,7 +322,7 @@ def _parse_rows(
        if balance is None:
            continue

-        narration = norm(" ".join([first_narration, *row["lines"][1:]]))
+        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:
@@ -354,6 +410,49 @@ class SBIYONORelationshipParser(BaseParser):
        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"