A R R R Associates
174 lines
6.9 KiB
Python
174 lines
6.9 KiB
Python
from __future__ import annotations
|
|
from dataclasses import dataclass, asdict
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import re, subprocess, tempfile
|
|
import pandas as pd
|
|
import pdfplumber
|
|
|
|
@dataclass
|
|
class StatementMeta:
|
|
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 = [
|
|
"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()
|
|
|
|
def extract_text(path: str|Path) -> str:
|
|
"""Prefer pdftotext layout output; fall back to pdfplumber."""
|
|
path=str(path)
|
|
try:
|
|
p=subprocess.run(['pdftotext','-layout',path,'-'], capture_output=True, text=True, timeout=120)
|
|
if p.returncode==0 and len(p.stdout.strip())>50:
|
|
return p.stdout
|
|
except Exception:
|
|
pass
|
|
parts=[]
|
|
with pdfplumber.open(path) as pdf:
|
|
for page in pdf.pages:
|
|
parts.append(page.extract_text(x_tolerance=1,y_tolerance=3,layout=True) or '')
|
|
return '\n\f\n'.join(parts)
|
|
|
|
def page_of_line(text: str, position: int) -> int:
|
|
return text[:position].count('\f')+1
|
|
|
|
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"
|
|
|
|
dated = pd.to_datetime(x.get("transaction_date"), errors="coerce")
|
|
valid_dates = dated.dropna()
|
|
descending = len(valid_dates) >= 2 and valid_dates.iloc[0] > valid_dates.iloc[-1]
|
|
order = list(reversed(x.index.tolist())) if descending else x.index.tolist()
|
|
|
|
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)
|
|
df = _apply_balance_delta_validation(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
|