A R R R Associates
377 lines
14 KiB
Python
377 lines
14 KiB
Python
from __future__ import annotations
|
|
from dataclasses import dataclass, asdict
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
import os, re, subprocess, tempfile
|
|
import pandas as pd
|
|
import pdfplumber
|
|
|
|
@dataclass
|
|
class StatementMeta:
|
|
statement_id: str = ""
|
|
bank_name: str = ""
|
|
customer_name: str = ""
|
|
account_number: str = ""
|
|
customer_id: str = ""
|
|
ifsc: str = ""
|
|
period_from: str = ""
|
|
period_to: str = ""
|
|
opening_balance: Optional[float] = None
|
|
total_debit: Optional[float] = None
|
|
total_credit: Optional[float] = None
|
|
closing_balance: Optional[float] = None
|
|
source_file: str = ""
|
|
parser_name: str = ""
|
|
confidence: str = "Medium"
|
|
|
|
def to_dict(self):
|
|
return asdict(self)
|
|
|
|
STANDARD_COLUMNS = [
|
|
"statement_id", "transaction_date", "value_date", "narration", "reference_no",
|
|
"debit", "credit", "balance", "bank_name", "customer_name",
|
|
"account_number", "source_file", "source_page", "parser_name",
|
|
# Internal extraction-audit fields. These are retained for diagnostics but
|
|
# are intentionally omitted from the client-facing workbook.
|
|
"printed_debit", "printed_credit", "balance_delta", "movement_difference",
|
|
"correction_applied", "correction_reason", "extraction_confidence",
|
|
]
|
|
|
|
def amount(v):
|
|
if v is None: return None
|
|
s=str(v).strip().replace('INR','').replace('Rs.','').replace('₹','').replace(',','').replace('+','')
|
|
s=s.replace('CR','').replace('DR','').strip()
|
|
if s in ('','-'): return None
|
|
neg=s.startswith('-')
|
|
s=s.lstrip('-')
|
|
try:
|
|
x=float(s)
|
|
return -x if neg else x
|
|
except: return None
|
|
|
|
def norm(s): return re.sub(r'\s+',' ',str(s or '')).strip()
|
|
|
|
_MIN_DIGITAL_TEXT_CHARS = 120
|
|
_MIN_DIGITAL_ALPHA_CHARS = 30
|
|
_OCR_DPI = max(120, min(int(os.getenv("BANK_ANALYZER_OCR_DPI", "180")), 300))
|
|
_OCR_WORKERS = max(1, min(int(os.getenv("BANK_ANALYZER_OCR_WORKERS", "2")), 4))
|
|
|
|
|
|
def _has_meaningful_text(value: str) -> bool:
|
|
text = str(value or "").strip()
|
|
if len(text) < _MIN_DIGITAL_TEXT_CHARS:
|
|
return False
|
|
return sum(character.isalpha() for character in text) >= _MIN_DIGITAL_ALPHA_CHARS
|
|
|
|
|
|
def _run_command(arguments: list[str], timeout: int) -> subprocess.CompletedProcess[str]:
|
|
try:
|
|
return subprocess.run(
|
|
arguments,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
except FileNotFoundError as exc:
|
|
raise RuntimeError(
|
|
f"Required bank-statement extraction command is unavailable: {arguments[0]}"
|
|
) from exc
|
|
|
|
|
|
def _ocr_pdf_page(path: Path, page_number: int, work_dir: Path) -> tuple[int, str]:
|
|
prefix = work_dir / f"page_{page_number:05d}"
|
|
rendered_image = prefix.with_suffix(".png")
|
|
|
|
render = _run_command(
|
|
[
|
|
"pdftoppm",
|
|
"-f", str(page_number),
|
|
"-l", str(page_number),
|
|
"-r", str(_OCR_DPI),
|
|
"-png",
|
|
"-singlefile",
|
|
str(path),
|
|
str(prefix),
|
|
],
|
|
timeout=180,
|
|
)
|
|
if render.returncode != 0 or not rendered_image.exists():
|
|
message = render.stderr.strip() or f"Unable to render page {page_number} for OCR."
|
|
raise RuntimeError(message)
|
|
|
|
try:
|
|
environment = os.environ.copy()
|
|
environment.setdefault("OMP_THREAD_LIMIT", "1")
|
|
ocr = subprocess.run(
|
|
[
|
|
"tesseract",
|
|
str(rendered_image),
|
|
"stdout",
|
|
"-l", "eng",
|
|
"--psm", "4",
|
|
"-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"OCR failed for page {page_number}.")
|
|
return page_number, ocr.stdout or ""
|
|
finally:
|
|
rendered_image.unlink(missing_ok=True)
|
|
|
|
|
|
def _ocr_selected_pages(path: Path, page_numbers: list[int]) -> dict[int, str]:
|
|
if not page_numbers:
|
|
return {}
|
|
|
|
extracted: dict[int, str] = {}
|
|
with tempfile.TemporaryDirectory(prefix="bank_statement_ocr_") as temporary:
|
|
work_dir = Path(temporary)
|
|
with ThreadPoolExecutor(max_workers=min(_OCR_WORKERS, len(page_numbers))) as executor:
|
|
futures = {
|
|
executor.submit(_ocr_pdf_page, path, page_number, work_dir): page_number
|
|
for page_number in page_numbers
|
|
}
|
|
for future in as_completed(futures):
|
|
page_number, text = future.result()
|
|
extracted[page_number] = text
|
|
return extracted
|
|
|
|
|
|
def extract_text(path: str | Path) -> str:
|
|
"""Extract PDF text and transparently OCR image-only or mixed pages.
|
|
|
|
Existing digitally-generated statements continue to use the original fast
|
|
``pdftotext``/``pdfplumber`` path. OCR is invoked only when a page has no
|
|
meaningful embedded text, so existing parsers and workbook behaviour remain
|
|
unchanged for supported digital statements.
|
|
"""
|
|
pdf_path = Path(path)
|
|
|
|
try:
|
|
result = _run_command(
|
|
["pdftotext", "-layout", str(pdf_path), "-"],
|
|
timeout=120,
|
|
)
|
|
if result.returncode == 0 and _has_meaningful_text(result.stdout):
|
|
return result.stdout
|
|
except RuntimeError:
|
|
# pdfplumber remains the normal fallback when pdftotext is unavailable.
|
|
pass
|
|
|
|
digital_pages: list[str] = []
|
|
pages_requiring_ocr: list[int] = []
|
|
with pdfplumber.open(str(pdf_path)) as pdf:
|
|
for page_number, page in enumerate(pdf.pages, start=1):
|
|
page_text = page.extract_text(
|
|
x_tolerance=1,
|
|
y_tolerance=3,
|
|
layout=True,
|
|
) or ""
|
|
digital_pages.append(page_text)
|
|
if not _has_meaningful_text(page_text):
|
|
pages_requiring_ocr.append(page_number)
|
|
|
|
if not pages_requiring_ocr:
|
|
return "\n\f\n".join(digital_pages)
|
|
|
|
try:
|
|
ocr_pages = _ocr_selected_pages(pdf_path, pages_requiring_ocr)
|
|
except RuntimeError as exc:
|
|
if any(_has_meaningful_text(page) for page in digital_pages):
|
|
# Preserve the former mixed-PDF behaviour when OCR tooling is absent.
|
|
return "\n\f\n".join(digital_pages)
|
|
raise ValueError(
|
|
"This bank statement is image-only and OCR processing is unavailable. "
|
|
"Install poppler-utils and tesseract-ocr in the application image."
|
|
) from exc
|
|
|
|
merged_pages = [
|
|
ocr_pages.get(page_number, digital_pages[page_number - 1])
|
|
if page_number in pages_requiring_ocr
|
|
else digital_pages[page_number - 1]
|
|
for page_number in range(1, len(digital_pages) + 1)
|
|
]
|
|
return "\n\f\n".join(merged_pages)
|
|
|
|
def page_of_line(text: str, position: int) -> int:
|
|
return text[:position].count('\f')+1
|
|
|
|
|
|
|
|
def _chronological_row_indices(df: pd.DataFrame) -> list:
|
|
"""Return transaction row indices in statement chronology.
|
|
|
|
Bank exports may be oldest-first or newest-first. Preserve the bank's row
|
|
order within a date while reversing the full sequence only when the dated
|
|
rows clearly run newest-to-oldest.
|
|
"""
|
|
if df.empty:
|
|
return []
|
|
dates = pd.to_datetime(df.get("transaction_date"), errors="coerce")
|
|
valid = dates.dropna()
|
|
descending = len(valid) >= 2 and valid.iloc[0] > valid.iloc[-1]
|
|
indices = df.index.tolist()
|
|
return list(reversed(indices)) if descending else indices
|
|
|
|
|
|
def _derive_missing_statement_meta(df: pd.DataFrame, meta: StatementMeta) -> None:
|
|
"""Populate balances/totals omitted by the printed statement.
|
|
|
|
Some bank statements, including HDFC's compact monthly export, begin with
|
|
the first transaction and its post-transaction closing balance but do not
|
|
print a separate opening balance. In that case:
|
|
|
|
opening = first_balance + first_debit - first_credit
|
|
|
|
The closing balance is the last chronological running balance. Explicit
|
|
values extracted from the PDF are never overwritten.
|
|
"""
|
|
if df.empty:
|
|
return
|
|
|
|
order = _chronological_row_indices(df)
|
|
if not order:
|
|
return
|
|
|
|
if meta.opening_balance is None:
|
|
for idx in order:
|
|
balance = df.at[idx, "balance"] if "balance" in df.columns else None
|
|
if pd.isna(balance):
|
|
continue
|
|
debit = df.at[idx, "debit"] if "debit" in df.columns else None
|
|
credit = df.at[idx, "credit"] if "credit" in df.columns else None
|
|
debit_value = 0.0 if pd.isna(debit) else float(debit)
|
|
credit_value = 0.0 if pd.isna(credit) else float(credit)
|
|
if debit_value == 0.0 and credit_value == 0.0:
|
|
continue
|
|
meta.opening_balance = round(float(balance) + debit_value - credit_value, 2)
|
|
break
|
|
|
|
if meta.closing_balance is None:
|
|
for idx in reversed(order):
|
|
balance = df.at[idx, "balance"] if "balance" in df.columns else None
|
|
if pd.notna(balance):
|
|
meta.closing_balance = round(float(balance), 2)
|
|
break
|
|
|
|
|
|
def _derive_missing_statement_totals(df: pd.DataFrame, meta: StatementMeta) -> None:
|
|
"""Populate statement totals from the validated transaction population."""
|
|
if df.empty:
|
|
return
|
|
if meta.total_debit is None:
|
|
meta.total_debit = round(float(pd.to_numeric(df.get("debit"), errors="coerce").fillna(0).sum()), 2)
|
|
if meta.total_credit is None:
|
|
meta.total_credit = round(float(pd.to_numeric(df.get("credit"), errors="coerce").fillna(0).sum()), 2)
|
|
|
|
|
|
def _apply_balance_delta_validation(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
|
|
"""Validate and, when necessary, correct debit/credit using balance movement.
|
|
|
|
Printed PDF columns remain the first extraction source. The running-balance
|
|
delta is the independent accounting control. Where the printed movement and
|
|
the balance delta disagree, the delta determines the corrected side and
|
|
amount. This common routine is used by every bank parser through ``finalize``.
|
|
"""
|
|
if df.empty:
|
|
return df
|
|
|
|
tolerance = 0.01
|
|
x = df.copy()
|
|
x["printed_debit"] = pd.to_numeric(x.get("debit"), errors="coerce")
|
|
x["printed_credit"] = pd.to_numeric(x.get("credit"), errors="coerce")
|
|
x["balance_delta"] = pd.NA
|
|
x["movement_difference"] = pd.NA
|
|
x["correction_applied"] = False
|
|
x["correction_reason"] = ""
|
|
x["extraction_confidence"] = "Printed columns"
|
|
|
|
order = _chronological_row_indices(x)
|
|
|
|
previous_balance = meta.opening_balance
|
|
for idx in order:
|
|
current_balance = x.at[idx, "balance"]
|
|
if pd.isna(current_balance):
|
|
x.at[idx, "extraction_confidence"] = "Review - balance unavailable"
|
|
continue
|
|
|
|
current_balance = float(current_balance)
|
|
if previous_balance is None or pd.isna(previous_balance):
|
|
previous_balance = current_balance
|
|
x.at[idx, "extraction_confidence"] = "Printed columns - no opening delta"
|
|
continue
|
|
|
|
delta = round(current_balance - float(previous_balance), 2)
|
|
debit = float(x.at[idx, "debit"]) if pd.notna(x.at[idx, "debit"]) else 0.0
|
|
credit = float(x.at[idx, "credit"]) if pd.notna(x.at[idx, "credit"]) else 0.0
|
|
printed_movement = round(credit - debit, 2)
|
|
movement_difference = round(delta - printed_movement, 2)
|
|
x.at[idx, "balance_delta"] = delta
|
|
x.at[idx, "movement_difference"] = movement_difference
|
|
|
|
if abs(movement_difference) <= tolerance:
|
|
x.at[idx, "extraction_confidence"] = "100% - printed movement matches delta"
|
|
elif abs(delta) > tolerance:
|
|
corrected_debit = round(abs(delta), 2) if delta < 0 else 0.0
|
|
corrected_credit = round(delta, 2) if delta > 0 else 0.0
|
|
x.at[idx, "debit"] = corrected_debit
|
|
x.at[idx, "credit"] = corrected_credit
|
|
x.at[idx, "correction_applied"] = True
|
|
x.at[idx, "correction_reason"] = "Debit/credit corrected from running-balance delta"
|
|
if abs(abs(printed_movement) - abs(delta)) <= tolerance:
|
|
x.at[idx, "extraction_confidence"] = "99% - amount matched, side corrected by delta"
|
|
elif debit == 0.0 and credit == 0.0:
|
|
x.at[idx, "extraction_confidence"] = "98% - missing movement derived from delta"
|
|
else:
|
|
x.at[idx, "extraction_confidence"] = "Review - printed movement replaced by delta"
|
|
else:
|
|
x.at[idx, "extraction_confidence"] = "Review - zero balance movement"
|
|
|
|
previous_balance = current_balance
|
|
|
|
return x
|
|
|
|
|
|
def finalize(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame:
|
|
if df is None or df.empty:
|
|
return pd.DataFrame(columns=STANDARD_COLUMNS)
|
|
for c in ['debit','credit','balance']:
|
|
df[c]=pd.to_numeric(df.get(c),errors='coerce')
|
|
from .common import parse_flexible_date
|
|
for c in ['transaction_date','value_date']:
|
|
values = df.get(c)
|
|
if values is None:
|
|
df[c] = pd.NaT
|
|
else:
|
|
df[c] = values.map(parse_flexible_date)
|
|
_derive_missing_statement_meta(df, meta)
|
|
df = _apply_balance_delta_validation(df, meta)
|
|
_derive_missing_statement_totals(df, meta)
|
|
narration_values = df['narration'] if 'narration' in df.columns else pd.Series('', index=df.index)
|
|
reference_values = df['reference_no'] if 'reference_no' in df.columns else pd.Series('', index=df.index)
|
|
df['narration']=narration_values.fillna('').map(norm)
|
|
df['reference_no']=reference_values.fillna('').map(norm)
|
|
df['bank_name']=meta.bank_name
|
|
df['customer_name']=meta.customer_name
|
|
df['account_number']=meta.account_number
|
|
df['source_file']=meta.source_file
|
|
df['parser_name']=meta.parser_name
|
|
if 'source_page' not in df: df['source_page']=None
|
|
for c in STANDARD_COLUMNS:
|
|
if c not in df: df[c]=None
|
|
return df[STANDARD_COLUMNS]
|
|
|
|
class BaseParser:
|
|
bank_name='Unknown'
|
|
parser_name='BaseParser'
|
|
@classmethod
|
|
def detect(cls,text:str)->float: return 0.0
|
|
def parse(self,path:str|Path,text:str|None=None): raise NotImplementedError
|