Add support for multiple SBI statement layouts
This commit is contained in:
A R R R Associates
@@ -7,7 +7,7 @@ from .hsbc import HSBCParser
|
|||||||
from .indian_bank import IndianBankModernParser, IndianBankLegacyParser
|
from .indian_bank import IndianBankModernParser, IndianBankLegacyParser
|
||||||
from .indusind import IndusIndParser
|
from .indusind import IndusIndParser
|
||||||
from .kotak import KotakParser
|
from .kotak import KotakParser
|
||||||
from .sbi import SBIModernParser, SBIOtherParser
|
from .sbi import SBIYONORelationshipParser, SBIStandardParser, SBICompactParser, SBIModernParser, SBIOtherParser
|
||||||
from .base import extract_text
|
from .base import extract_text
|
||||||
|
|
||||||
PARSERS = [
|
PARSERS = [
|
||||||
@@ -19,8 +19,9 @@ PARSERS = [
|
|||||||
IndianBankLegacyParser,
|
IndianBankLegacyParser,
|
||||||
IndusIndParser,
|
IndusIndParser,
|
||||||
KotakParser,
|
KotakParser,
|
||||||
SBIOtherParser,
|
SBIYONORelationshipParser,
|
||||||
SBIModernParser,
|
SBIStandardParser,
|
||||||
|
SBICompactParser,
|
||||||
]
|
]
|
||||||
|
|
||||||
BANK_OPTIONS = [
|
BANK_OPTIONS = [
|
||||||
@@ -43,7 +44,7 @@ BANK_PARSERS = {
|
|||||||
"indian_bank": [IndianBankModernParser, IndianBankLegacyParser],
|
"indian_bank": [IndianBankModernParser, IndianBankLegacyParser],
|
||||||
"indusind": [IndusIndParser],
|
"indusind": [IndusIndParser],
|
||||||
"kotak": [KotakParser],
|
"kotak": [KotakParser],
|
||||||
"sbi": [SBIOtherParser, SBIModernParser],
|
"sbi": [SBIYONORelationshipParser, SBIStandardParser, SBICompactParser],
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,120 +1,409 @@
|
|||||||
from __future__ import annotations
|
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 pathlib import Path
|
||||||
from .base import *
|
|
||||||
from .common import DATE_TOKEN_PATTERN, date_iso, find
|
|
||||||
|
|
||||||
class SBIModernParser(BaseParser):
|
import pandas as pd
|
||||||
bank_name='State Bank of India'; parser_name='SBIModernParser'
|
import pdfplumber
|
||||||
@classmethod
|
|
||||||
def detect(cls,text):
|
from .base import BaseParser, StatementMeta, amount, extract_text, finalize, norm
|
||||||
u=norm(text).upper()
|
from .common import DATE_TOKEN_PATTERN, date_iso, find, parse_flexible_date
|
||||||
bank_marker='STATE BANK OF INDIA' in u
|
|
||||||
digital_layout='REF NO./CHEQUE' in u and 'DETAILS' in u
|
_MONEY_RE = re.compile(r"(?<![\d.])([\d,]+\.\d{2})(?!\d)")
|
||||||
yono_print_layout=(
|
_REFERENCE_RE = re.compile(
|
||||||
'STATEMENT OF ACCOUNT' in u
|
r"(?:UPI|NEFT|IMPS|RTGS|UTR|CHEQUE|CHQ)[/ :A-Z0-9._-]{5,}", re.I
|
||||||
and 'VALUE DATE' in u
|
)
|
||||||
and 'POST DATE' in u
|
_HEADER_RE = re.compile(
|
||||||
and 'DEBIT' in u
|
r"^(?:REF\s*NO|VALUE\s*DATE|POST\s*DATE|TXN\s*DATE|DATE\s+DETAILS|"
|
||||||
and 'CREDIT' in u
|
r"ACCOUNT\s+STATEMENT|STATEMENT\s+OF\s+ACCOUNT|STATE\s+BANK\s+OF\s+INDIA|"
|
||||||
and 'BALANCE' in u
|
r"PAGE\s+NO|STATEMENT\s+SUMMARY|BROUGHT\s+FORWARD|DR\s+COUNT|CR\s+COUNT)",
|
||||||
|
re.I,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
def _expected_summary_transactions(text: str) -> 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
|
if ocr.returncode != 0:
|
||||||
def parse(self,path,text=None):
|
raise RuntimeError(ocr.stderr.strip() or f"SBI OCR failed for page {page_number}.")
|
||||||
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')
|
return page_number, ocr.stdout or ""
|
||||||
meta.customer_name=find(r'Account Name\s*:?\s*([^\n]+)',text)
|
finally:
|
||||||
if not meta.customer_name:
|
image_path.unlink(missing_ok=True)
|
||||||
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)
|
def _extract_yono_table_text(path: str | Path) -> str:
|
||||||
m=re.search(rf'(?:Account Statement from|Statement From\s*:?)\s*({DATE_TOKEN_PATTERN})\s+to\s+({DATE_TOKEN_PATTERN})', text, re.I)
|
pdf_path = Path(path)
|
||||||
if m:
|
dpi = max(120, min(int(os.getenv("BANK_ANALYZER_SBI_OCR_DPI", "150")), 220))
|
||||||
meta.period_from=date_iso(m.group(1)); meta.period_to=date_iso(m.group(2))
|
workers = max(1, min(int(os.getenv("BANK_ANALYZER_SBI_OCR_WORKERS", "2")), 3))
|
||||||
meta.opening_balance=amount(find(r'Balance as on[^\n]*\n\s*([\d,]+\.\d{2})',text))
|
with pdfplumber.open(str(pdf_path)) as pdf:
|
||||||
# Format A: date details ref debit credit balance
|
page_count = len(pdf.pages)
|
||||||
date_re=re.compile(rf'^\s*({DATE_TOKEN_PATTERN})\s+(.*)$', re.I)
|
pages: dict[int, str] = {}
|
||||||
rows=[]; cur=None; page=1
|
with tempfile.TemporaryDirectory(prefix="sbi_yono_ocr_") as temporary:
|
||||||
for line in text.splitlines():
|
work_dir = Path(temporary)
|
||||||
if '\f' in line: page += line.count('\f')
|
with ThreadPoolExecutor(max_workers=min(workers, page_count)) as executor:
|
||||||
m=date_re.match(line)
|
futures = {
|
||||||
if m:
|
executor.submit(_ocr_yono_page, pdf_path, page, work_dir, dpi): page
|
||||||
if cur: rows.append(cur)
|
for page in range(1, page_count + 1)
|
||||||
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):
|
for future in as_completed(futures):
|
||||||
cur['lines'].append(line)
|
page_number, page_text = future.result()
|
||||||
if cur: rows.append(cur)
|
pages[page_number] = page_text
|
||||||
out=[]; prev=meta.opening_balance
|
return "\n\f\n".join(pages.get(page, "") for page in range(1, page_count + 1))
|
||||||
for r in rows:
|
|
||||||
first=r['lines'][0]; alltxt=norm(' '.join(r['lines']))
|
def _page_increment(line: str) -> int:
|
||||||
nums=list(re.finditer(r'(?<!\d)([\d,]+\.\d{2})(?!\d)',first))
|
return line.count("\f")
|
||||||
if not nums: continue
|
|
||||||
bal=amount(nums[-1].group(1)); debit=credit=None
|
|
||||||
# use dashes and positions from SBI layout: debit then credit then balance
|
def _extract_common_meta(text: str, path: str | Path, parser_name: str) -> StatementMeta:
|
||||||
if len(nums)>=2:
|
meta = StatementMeta(
|
||||||
txn=amount(nums[-2].group(1)); p=nums[-2].start()
|
bank_name="State Bank of India",
|
||||||
if prev is not None and bal is not None:
|
source_file=Path(path).name,
|
||||||
if abs((prev-txn)-bal)<0.05: debit=txn
|
parser_name=parser_name,
|
||||||
elif abs((prev+txn)-bal)<0.05: credit=txn
|
confidence="High",
|
||||||
if debit is None and credit is None:
|
)
|
||||||
upper=alltxt.upper()
|
meta.customer_name = find(r"Account Name\s*:?\s*([^\n]+)", text)
|
||||||
if '/CR/' in upper or upper.startswith('DEP TFR') or 'INTEREST CREDIT' in upper:
|
if not meta.customer_name:
|
||||||
credit=txn
|
meta.customer_name = find(r"\b(?:Ms|Mr|Mrs)\.?\s+([A-Z][A-Z .]{3,})", text)
|
||||||
elif '/DR/' in upper or upper.startswith('WDL TFR') or upper.startswith('DEBIT '):
|
meta.account_number = find(r"Account Number\s*:?\s*([0-9Xx*]+)", text)
|
||||||
debit=txn
|
meta.ifsc = find(r"(?:IFS|IFSC) Code\s*:?\s*([A-Z0-9]+)", text)
|
||||||
elif p >= 78:
|
|
||||||
credit=txn
|
periods = re.findall(
|
||||||
else:
|
rf"(?:Account Statement from|Statement From\s*:?)\s*({DATE_TOKEN_PATTERN})\s+to\s+({DATE_TOKEN_PATTERN})",
|
||||||
debit=txn
|
text,
|
||||||
narr=first[:nums[-2].start() if len(nums)>=2 else nums[-1].start()].strip()+' '+' '.join(x.strip() for x in r['lines'][1:])
|
re.I,
|
||||||
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
|
if periods:
|
||||||
out.append({**r,'narration':norm(narr),'reference_no':ref,'debit':debit,'credit':credit,'balance':bal})
|
parsed = [
|
||||||
if bal is not None: prev=bal
|
(parse_flexible_date(start), parse_flexible_date(end))
|
||||||
if out:
|
for start, end in periods
|
||||||
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)
|
parsed = [(start, end) for start, end in parsed if not pd.isna(start) and not pd.isna(end)]
|
||||||
meta.closing_balance=out[-1]['balance']
|
if parsed:
|
||||||
return meta,finalize(pd.DataFrame(out),meta)
|
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{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
|
||||||
|
page = 1
|
||||||
|
|
||||||
|
for raw_line in text.splitlines():
|
||||||
|
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],
|
||||||
|
"source_page": page,
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
|
||||||
|
stripped = line.strip()
|
||||||
|
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["lines"]))
|
||||||
|
debit, credit, balance, first_narration = _row_values(
|
||||||
|
first_line,
|
||||||
|
full_text,
|
||||||
|
previous_balance,
|
||||||
|
)
|
||||||
|
if balance is None:
|
||||||
|
continue
|
||||||
|
|
||||||
|
narration = norm(" ".join([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"
|
||||||
|
|
||||||
class SBIOtherParser(BaseParser):
|
|
||||||
bank_name='State Bank of India'; parser_name='SBIOtherParser'
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def detect(cls,text):
|
def detect(cls, text: str) -> float:
|
||||||
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
|
upper = norm(text).upper()
|
||||||
def parse(self,path,text=None):
|
markers = (
|
||||||
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')
|
"RELATIONSHIP SUMMARY",
|
||||||
meta.customer_name=find(r'Account Name\s*:\s*([^\n]+)',text)
|
"STATEMENT OF ACCOUNT",
|
||||||
meta.account_number=find(r'Account Number\s*:?\s*([0-9X*]+)',text)
|
"VALUE DATE",
|
||||||
meta.ifsc=find(r'IFS Code\s*:?\s*([A-Z0-9]+)',text)
|
"POST DATE",
|
||||||
m=re.search(rf'Account Statement from\s*({DATE_TOKEN_PATTERN})\s+to\s+({DATE_TOKEN_PATTERN})', text, re.I)
|
"STATEMENT SUMMARY",
|
||||||
if m:
|
)
|
||||||
meta.period_from=date_iso(m.group(1)); meta.period_to=date_iso(m.group(2))
|
score = sum(marker in upper for marker in markers)
|
||||||
m=re.search(r'Balance as on\s+[^\n]+\n',text,re.I)
|
return 0.995 if "STATE BANK OF INDIA" in upper and score >= 3 else 0.0
|
||||||
date_re=re.compile(rf'^\s*({DATE_TOKEN_PATTERN})\s+({DATE_TOKEN_PATTERN})\s+(.*)$', re.I)
|
|
||||||
rows=[]; cur=None; page=1
|
def parse(self, path, text=None):
|
||||||
for line in text.splitlines():
|
extracted_text = text or extract_text(path)
|
||||||
if '\f' in line: page += line.count('\f')
|
start_re = re.compile(
|
||||||
mm=date_re.match(line)
|
r"^\s*(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\s+"
|
||||||
if mm:
|
r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\s+(.*)$",
|
||||||
if cur: rows.append(cur)
|
re.I,
|
||||||
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)
|
meta = _extract_common_meta(extracted_text, path, self.parser_name)
|
||||||
out=[]
|
meta.opening_balance = _opening_from_summary(extracted_text)
|
||||||
for r in rows:
|
frame = _parse_rows(extracted_text, meta, start_re, 2)
|
||||||
first=r['lines'][0]; alltxt=norm(' '.join(r['lines']))
|
expected = _expected_summary_transactions(extracted_text)
|
||||||
nums=list(re.finditer(r'(?<!\d)([\d,]+\.\d{2})(?!\d)',first))
|
|
||||||
if not nums: continue
|
# The shared OCR path is sufficient for detection, but dense YONO tables
|
||||||
bal=amount(nums[-1].group(1)); debit=credit=None
|
# occasionally need a table-oriented OCR pass. Retry only when the bank's
|
||||||
if len(nums)>=2:
|
# printed debit/credit counts prove that rows were missed.
|
||||||
txn=amount(nums[-2].group(1)); p=nums[-2].start()
|
if expected and len(frame) < max(1, int(expected * 0.95)):
|
||||||
# Based on header layout: Debit starts before Credit
|
try:
|
||||||
if p < 55: debit=txn
|
table_text = _extract_yono_table_text(path)
|
||||||
else: credit=txn
|
retry_meta = _extract_common_meta(table_text, path, self.parser_name)
|
||||||
narr=first[:nums[-2].start() if len(nums)>=2 else nums[-1].start()].strip()+' '+' '.join(x.strip() for x in r['lines'][1:])
|
retry_meta.opening_balance = _opening_from_summary(table_text)
|
||||||
ref=''; z=re.search(r'(?:UPI|NEFT|IMPS|RTGS)[/A-Z0-9-]{6,}',alltxt,re.I); ref=z.group(0) if z else ''
|
retry_frame = _parse_rows(table_text, retry_meta, start_re, 2)
|
||||||
if debit is None and credit is None: continue
|
if len(retry_frame) > len(frame):
|
||||||
out.append({**r,'narration':norm(narr),'reference_no':ref,'debit':debit,'credit':credit,'balance':bal})
|
meta, frame = retry_meta, retry_frame
|
||||||
if out:
|
except (OSError, RuntimeError, subprocess.SubprocessError):
|
||||||
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']
|
# Preserve the already extracted result when the optional targeted
|
||||||
return meta,finalize(pd.DataFrame(out),meta)
|
# 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
|
||||||
|
|||||||
Reference in New Issue
Block a user