A R R R Associates
108 lines
6.4 KiB
Python
108 lines
6.4 KiB
Python
from __future__ import annotations
|
|
from pathlib import Path
|
|
import re
|
|
import pandas as pd
|
|
from .parsers import parse_pdf
|
|
from parsers.common import infer_mode
|
|
|
|
|
|
def clean_key(s):
|
|
s=re.sub(r'\s+',' ',str(s or '').upper()).strip()
|
|
return re.sub(r'\b\d{8,}\b','<REF>',s)
|
|
|
|
def enrich(df):
|
|
if df.empty:return df
|
|
x=df.copy()
|
|
x['mode']=x['narration'].map(infer_mode)
|
|
x['amount']=x['debit'].fillna(0)+x['credit'].fillna(0)
|
|
x['direction']=x['debit'].notna().map({True:'Debit',False:'Credit'})
|
|
x['narration_key']=x['narration'].map(clean_key)
|
|
x['exact_key']=x.apply(lambda r:f"{r.transaction_date}|{r.value_date}|{r.debit}|{r.credit}|{r.balance}|{clean_key(r.narration)}",axis=1)
|
|
x['exact_duplicate']=x.duplicated('exact_key',keep=False)
|
|
# possible duplicate: same date, direction, amount and normalized narration across different source files
|
|
x['possible_key']=x.apply(lambda r:f"{r.transaction_date}|{r.direction}|{r.amount:.2f}|{r.narration_key}",axis=1)
|
|
x['possible_duplicate']=x.duplicated('possible_key',keep=False) & ~x['exact_duplicate']
|
|
return x
|
|
|
|
def analyze_files(paths, customer_override='', account_override=''):
|
|
metas=[]; dfs=[]
|
|
for p in paths:
|
|
meta,df=parse_pdf(p)
|
|
if customer_override: meta.customer_name=customer_override; df['customer_name']=customer_override
|
|
if account_override: meta.account_number=account_override; df['account_number']=account_override
|
|
metas.append(meta); dfs.append(df)
|
|
all_df=enrich(pd.concat(dfs,ignore_index=True) if dfs else pd.DataFrame())
|
|
# remove exact overlap duplicates, retaining first source occurrence
|
|
unique_df=all_df.drop_duplicates('exact_key',keep='first').copy() if not all_df.empty else all_df.copy()
|
|
return metas,all_df,unique_df
|
|
|
|
def reconcile(metas,all_df):
|
|
rows=[]
|
|
for m in metas:
|
|
d=all_df[all_df.source_file.eq(m.source_file)] if not all_df.empty else pd.DataFrame()
|
|
ed=float(d.debit.sum()) if not d.empty else 0
|
|
ec=float(d.credit.sum()) if not d.empty else 0
|
|
last=float(d.balance.dropna().iloc[-1]) if not d.empty and d.balance.notna().any() else None
|
|
rows.append({**m.to_dict(),'extracted_transactions':len(d),'extracted_debit':ed,'extracted_credit':ec,'extracted_closing_balance':last,
|
|
'debit_difference':None if m.total_debit is None else round(ed-m.total_debit,2),
|
|
'credit_difference':None if m.total_credit is None else round(ec-m.total_credit,2),
|
|
'closing_difference':None if m.closing_balance is None or last is None else round(last-m.closing_balance,2)})
|
|
return pd.DataFrame(rows)
|
|
|
|
def monthly_summary(df):
|
|
if df.empty:return pd.DataFrame()
|
|
x=df.copy(); x['month']=x.transaction_date.dt.to_period('M').astype(str)
|
|
return x.groupby('month',dropna=False).agg(transaction_count=('amount','size'),total_debit=('debit','sum'),total_credit=('credit','sum'),net_movement=('credit','sum')).reset_index().assign(net_movement=lambda z:z.total_credit-z.total_debit)
|
|
|
|
def mode_summary(df):
|
|
if df.empty:return pd.DataFrame()
|
|
return df.groupby(['mode','direction']).agg(transaction_count=('amount','size'),amount=('amount','sum')).reset_index()
|
|
|
|
def duplicate_summary(all_df):
|
|
return pd.DataFrame([
|
|
{'check':'All extracted rows','count':len(all_df)},
|
|
{'check':'Exact duplicate rows','count':int(all_df.exact_duplicate.sum()) if not all_df.empty else 0},
|
|
{'check':'Possible duplicate rows','count':int(all_df.possible_duplicate.sum()) if not all_df.empty else 0},
|
|
{'check':'Unique rows after exact deduplication','count':int(all_df.exact_key.nunique()) if not all_df.empty else 0},
|
|
])
|
|
|
|
def export_excel(output,metas,all_df,unique_df):
|
|
recon=reconcile(metas,all_df)
|
|
cust=next((m.customer_name for m in metas if m.customer_name),'')
|
|
acct=next((m.account_number for m in metas if m.account_number),'')
|
|
banks=', '.join(sorted({m.bank_name for m in metas}))
|
|
dashboard=pd.DataFrame([
|
|
['Customer / Account Holder',cust],['Account Number',acct],['Bank(s)',banks],
|
|
['Statements Uploaded',len(metas)],['Rows Extracted',len(all_df)],['Unique Transactions',len(unique_df)],
|
|
['Exact Duplicate Rows',int(all_df.exact_duplicate.sum()) if not all_df.empty else 0],
|
|
['Possible Duplicate Rows',int(all_df.possible_duplicate.sum()) if not all_df.empty else 0],
|
|
['Total Debit (Unique)',float(unique_df.debit.sum()) if not unique_df.empty else 0],
|
|
['Total Credit (Unique)',float(unique_df.credit.sum()) if not unique_df.empty else 0],
|
|
],columns=['Metric','Value'])
|
|
with pd.ExcelWriter(output,engine='xlsxwriter',datetime_format='dd-mmm-yyyy') as w:
|
|
dashboard.to_excel(w, sheet_name='Dashboard', index=False)
|
|
recon.to_excel(w, sheet_name='Statement Reconciliation', index=False)
|
|
all_df.to_excel(w, sheet_name='All Extracted Rows', index=False)
|
|
unique_df.to_excel(w, sheet_name='Unique Transactions', index=False)
|
|
all_df[all_df.exact_duplicate].to_excel(w, sheet_name='Exact Duplicates', index=False)
|
|
all_df[all_df.possible_duplicate].to_excel(w, sheet_name='Possible Duplicates', index=False)
|
|
monthly_summary(unique_df).to_excel(w, sheet_name='Monthly Summary', index=False)
|
|
mode_summary(unique_df).to_excel(w, sheet_name='Mode Summary', index=False)
|
|
duplicate_summary(all_df).to_excel(w, sheet_name='Duplicate Summary', index=False)
|
|
notes=pd.DataFrame({'Notes':[
|
|
'Exact duplicates use transaction date, value date, debit, credit, balance and normalized narration.',
|
|
'Possible duplicates use same date, direction, amount and normalized narration; review before deletion.',
|
|
'Bank-specific parsers are selected automatically. Customer name/account number can be manually overridden in the app.',
|
|
'The workbook is a bank-statement analysis aid, not a substitute for ledger, GST, inventory, receivable/payable and cash-book records.'
|
|
]})
|
|
notes.to_excel(w, sheet_name='Notes', index=False)
|
|
wb=w.book
|
|
head=wb.add_format({'bold':True,'bg_color':'#1F4E78','font_color':'white','border':1})
|
|
money=wb.add_format({'num_format':'#,##0.00'})
|
|
for name,ws in w.sheets.items():
|
|
ws.freeze_panes(1,0); ws.autofilter(0,0,0,max(0,ws.dim_colmax))
|
|
ws.set_row(0,22,head)
|
|
ws.set_column(0,max(0,ws.dim_colmax),18)
|
|
w.sheets['Dashboard'].set_column('A:A',32); w.sheets['Dashboard'].set_column('B:B',28)
|
|
return output
|