A R R R Associates
97 lines
3.1 KiB
Python
97 lines
3.1 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"
|
|
]
|
|
|
|
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')
|
|
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['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
|