Add bank statement analyzer with automatic work storage

2026-07-13 10:12:19 +05:30
parent d391e9b443
commit bbc5afe1c0
23 changed files with 1047 additions and 0 deletions
@@ -0,0 +1,91 @@
+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"
+]
+
+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 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')
+    for c in ['transaction_date','value_date']:
+        df[c]=pd.to_datetime(df.get(c),errors='coerce',dayfirst=True)
+    df['narration']=df.get('narration','').fillna('').map(norm)
+    df['reference_no']=df.get('reference_no','').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