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','',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,'Dashboard',index=False) recon.to_excel(w,'Statement Reconciliation',index=False) all_df.to_excel(w,'All Extracted Rows',index=False) unique_df.to_excel(w,'Unique Transactions',index=False) all_df[all_df.exact_duplicate].to_excel(w,'Exact Duplicates',index=False) all_df[all_df.possible_duplicate].to_excel(w,'Possible Duplicates',index=False) monthly_summary(unique_df).to_excel(w,'Monthly Summary',index=False) mode_summary(unique_df).to_excel(w,'Mode Summary',index=False) duplicate_summary(all_df).to_excel(w,'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,'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