diff --git a/app/modules/bank_statement_analyzer/parsers/base.py b/app/modules/bank_statement_analyzer/parsers/base.py index 870ce60..69ad3e6 100644 --- a/app/modules/bank_statement_analyzer/parsers/base.py +++ b/app/modules/bank_statement_analyzer/parsers/base.py @@ -203,6 +203,75 @@ def extract_text(path: str | Path) -> str: def page_of_line(text: str, position: int) -> int: return text[:position].count('\f')+1 + + +def _chronological_row_indices(df: pd.DataFrame) -> list: + """Return transaction row indices in statement chronology. + + Bank exports may be oldest-first or newest-first. Preserve the bank's row + order within a date while reversing the full sequence only when the dated + rows clearly run newest-to-oldest. + """ + if df.empty: + return [] + dates = pd.to_datetime(df.get("transaction_date"), errors="coerce") + valid = dates.dropna() + descending = len(valid) >= 2 and valid.iloc[0] > valid.iloc[-1] + indices = df.index.tolist() + return list(reversed(indices)) if descending else indices + + +def _derive_missing_statement_meta(df: pd.DataFrame, meta: StatementMeta) -> None: + """Populate balances/totals omitted by the printed statement. + + Some bank statements, including HDFC's compact monthly export, begin with + the first transaction and its post-transaction closing balance but do not + print a separate opening balance. In that case: + + opening = first_balance + first_debit - first_credit + + The closing balance is the last chronological running balance. Explicit + values extracted from the PDF are never overwritten. + """ + if df.empty: + return + + order = _chronological_row_indices(df) + if not order: + return + + if meta.opening_balance is None: + for idx in order: + balance = df.at[idx, "balance"] if "balance" in df.columns else None + if pd.isna(balance): + continue + debit = df.at[idx, "debit"] if "debit" in df.columns else None + credit = df.at[idx, "credit"] if "credit" in df.columns else None + debit_value = 0.0 if pd.isna(debit) else float(debit) + credit_value = 0.0 if pd.isna(credit) else float(credit) + if debit_value == 0.0 and credit_value == 0.0: + continue + meta.opening_balance = round(float(balance) + debit_value - credit_value, 2) + break + + if meta.closing_balance is None: + for idx in reversed(order): + balance = df.at[idx, "balance"] if "balance" in df.columns else None + if pd.notna(balance): + meta.closing_balance = round(float(balance), 2) + break + + +def _derive_missing_statement_totals(df: pd.DataFrame, meta: StatementMeta) -> None: + """Populate statement totals from the validated transaction population.""" + if df.empty: + return + if meta.total_debit is None: + meta.total_debit = round(float(pd.to_numeric(df.get("debit"), errors="coerce").fillna(0).sum()), 2) + if meta.total_credit is None: + meta.total_credit = round(float(pd.to_numeric(df.get("credit"), errors="coerce").fillna(0).sum()), 2) + + def _apply_balance_delta_validation(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame: """Validate and, when necessary, correct debit/credit using balance movement. @@ -224,10 +293,7 @@ def _apply_balance_delta_validation(df: pd.DataFrame, meta: StatementMeta) -> pd 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() + order = _chronological_row_indices(x) previous_balance = meta.opening_balance for idx in order: @@ -285,7 +351,9 @@ def finalize(df: pd.DataFrame, meta: StatementMeta) -> pd.DataFrame: df[c] = pd.NaT else: df[c] = values.map(parse_flexible_date) + _derive_missing_statement_meta(df, meta) df = _apply_balance_delta_validation(df, meta) + _derive_missing_statement_totals(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)