diff --git a/app/modules/bank_statement_analyzer/analyzer.py b/app/modules/bank_statement_analyzer/analyzer.py index dd2e154..9fbd2a0 100644 --- a/app/modules/bank_statement_analyzer/analyzer.py +++ b/app/modules/bank_statement_analyzer/analyzer.py @@ -619,6 +619,56 @@ def _safe_sheet_name(value: str) -> str: return re.sub(r"[][\\:*?/]", "_", value)[:31] +def _statement_rows_in_chronology(tx: pd.DataFrame) -> pd.DataFrame: + """Return transaction rows in statement chronology for reconciliation. + + The transaction-classification sheet preserves parser order. The balance + reconciliation sheet must instead follow each statement from opening to + closing, including for bank exports that list newest transactions first. + """ + if tx.empty: + return tx.copy() + + frames: list[pd.DataFrame] = [] + statement_values = tx.get("statement_id", pd.Series("STMT-001", index=tx.index)).fillna("STMT-001") + for statement_id in pd.unique(statement_values): + part = tx.loc[statement_values.eq(statement_id)].copy() + dates = pd.to_datetime(part.get("transaction_date"), errors="coerce") + valid_dates = dates.dropna() + if len(valid_dates) >= 2 and valid_dates.iloc[0] > valid_dates.iloc[-1]: + part = part.iloc[::-1].copy() + part["statement_row_no"] = range(1, len(part) + 1) + frames.append(part) + return pd.concat(frames, ignore_index=True) if frames else tx.copy() + + +def _statement_opening_balances(metas) -> dict[str, float]: + """Map statement IDs to their printed or commonly-derived opening balance.""" + result: dict[str, float] = {} + for position, meta in enumerate(metas, start=1): + statement_id = str(getattr(meta, "statement_id", "") or f"STMT-{position:03d}") + opening = getattr(meta, "opening_balance", None) + result[statement_id] = float(opening or 0.0) + return result + + +def _reorder_workbook_sheets(workbook, preferred_order: list[str]) -> None: + """Place the primary working papers first without recreating any sheet. + + XlsxWriter serialises worksheets in ``worksheets_objs`` order. Reordering + that list before workbook close preserves every worksheet object, formula, + table, hidden state and formatting while changing only the visible tab order. + """ + rank = {name: position for position, name in enumerate(preferred_order)} + original = {worksheet.get_name(): position for position, worksheet in enumerate(workbook.worksheets_objs)} + workbook.worksheets_objs.sort( + key=lambda worksheet: ( + rank.get(worksheet.get_name(), len(preferred_order)), + original.get(worksheet.get_name(), len(original)), + ) + ) + + def _write_summary_reconciliation( worksheet, start_row: int, @@ -751,6 +801,101 @@ def export_excel(output, metas, all_df, unique_df, financial_year="", selected_b # Raw and Python-analysis sheets. recon.to_excel(writer, sheet_name="Statement Reconciliation", index=False) + + # Formula-driven transaction-level reconciliation for every bank/layout. + # It retains the original transaction row number and source PDF page so + # a reviewer can return directly to the line that first breaks continuity. + balance_rows = _statement_rows_in_chronology(tx) + opening_by_statement = _statement_opening_balances(metas) + balance_columns = [ + "Statement ID", "Statement Row No.", "Transaction Row No.", + "Source File", "Page No.", "Transaction Date", "Value Date", + "Narration", "Debit", "Credit", "Statement Balance", + "Calculated Balance", "Difference", "Status", "Review Note", + ] + balance_export = pd.DataFrame(columns=balance_columns) + if not balance_rows.empty: + balance_export = pd.DataFrame({ + "Statement ID": balance_rows.get("statement_id", ""), + "Statement Row No.": balance_rows.get("statement_row_no", ""), + "Transaction Row No.": balance_rows.get("row_id", ""), + "Source File": balance_rows.get("source_file", ""), + "Page No.": balance_rows.get("source_page", ""), + "Transaction Date": balance_rows.get("transaction_date", pd.NaT), + "Value Date": balance_rows.get("value_date", pd.NaT), + "Narration": balance_rows.get("narration", ""), + "Debit": pd.to_numeric(balance_rows.get("debit"), errors="coerce"), + "Credit": pd.to_numeric(balance_rows.get("credit"), errors="coerce"), + "Statement Balance": pd.to_numeric(balance_rows.get("balance"), errors="coerce"), + "Calculated Balance": "", + "Difference": "", + "Status": "", + "Review Note": "", + }) + balance_export.to_excel(writer, sheet_name="Balance Reconciliation", index=False) + ws_balance = writer.sheets["Balance Reconciliation"] + balance_col = {name: index for index, name in enumerate(balance_columns)} + previous_statement = None + previous_calc_excel_row = None + for row_index in range(len(balance_export)): + excel_row = row_index + 2 + statement_id = str(balance_export.iloc[row_index]["Statement ID"] or "STMT-001") + debit_cell = f'{_excel_col(balance_col["Debit"])}{excel_row}' + credit_cell = f'{_excel_col(balance_col["Credit"])}{excel_row}' + statement_balance_cell = f'{_excel_col(balance_col["Statement Balance"])}{excel_row}' + calculated_col_letter = _excel_col(balance_col["Calculated Balance"]) + calculated_cell = f'{calculated_col_letter}{excel_row}' + difference_cell = f'{_excel_col(balance_col["Difference"])}{excel_row}' + + if statement_id != previous_statement: + opening_balance = float(opening_by_statement.get(statement_id, 0.0)) + calculated_formula = f'={opening_balance}+IFERROR({credit_cell},0)-IFERROR({debit_cell},0)' + else: + calculated_formula = f'={calculated_col_letter}{previous_calc_excel_row}+IFERROR({credit_cell},0)-IFERROR({debit_cell},0)' + + statement_balance = balance_export.iloc[row_index]["Statement Balance"] + debit_value = balance_export.iloc[row_index]["Debit"] + credit_value = balance_export.iloc[row_index]["Credit"] + if statement_id != previous_statement: + cached_calculated = float(opening_by_statement.get(statement_id, 0.0)) + (0.0 if pd.isna(credit_value) else float(credit_value)) - (0.0 if pd.isna(debit_value) else float(debit_value)) + else: + prior_cached = float(balance_export.iloc[row_index - 1].get("_cached_calculated", 0.0)) + cached_calculated = prior_cached + (0.0 if pd.isna(credit_value) else float(credit_value)) - (0.0 if pd.isna(debit_value) else float(debit_value)) + balance_export.loc[balance_export.index[row_index], "_cached_calculated"] = cached_calculated + cached_difference = 0.0 if pd.isna(statement_balance) else float(statement_balance) - cached_calculated + cached_status = "Reconciled" if pd.notna(statement_balance) and abs(cached_difference) <= 0.01 else "Review Required" + cached_note = "" if cached_status == "Reconciled" else ("Statement balance unavailable" if pd.isna(statement_balance) else "Running balance differs from calculated balance") + + ws_balance.write_formula(row_index + 1, balance_col["Calculated Balance"], calculated_formula, money, cached_calculated) + ws_balance.write_formula(row_index + 1, balance_col["Difference"], f'=IF({statement_balance_cell}="","",{statement_balance_cell}-{calculated_cell})', money, cached_difference if pd.notna(statement_balance) else "") + ws_balance.write_formula(row_index + 1, balance_col["Status"], f'=IF({statement_balance_cell}="","Review Required",IF(ABS({difference_cell})<=0.01,"Reconciled","Review Required"))', text_fmt, cached_status) + ws_balance.write(row_index + 1, balance_col["Review Note"], cached_note, text_fmt) + + previous_statement = statement_id + previous_calc_excel_row = excel_row + + if len(balance_export): + ws_balance.add_table(0, 0, len(balance_export), len(balance_columns) - 1, { + "name": "tblBalanceReconciliation", + "columns": [{"header": column} for column in balance_columns], + "style": "Table Style Medium 2", + }) + ws_balance.conditional_format(1, balance_col["Difference"], len(balance_export), balance_col["Difference"], { + "type": "cell", "criteria": "not between", "minimum": -0.01, "maximum": 0.01, "format": warning_fmt, + }) + ws_balance.conditional_format(1, balance_col["Status"], len(balance_export), balance_col["Status"], { + "type": "text", "criteria": "containing", "value": "Review", "format": warning_fmt, + }) + ws_balance.freeze_panes(1, 8) + ws_balance.set_column(balance_col["Statement ID"], balance_col["Statement ID"], 13) + ws_balance.set_column(balance_col["Statement Row No."], balance_col["Transaction Row No."], 14) + ws_balance.set_column(balance_col["Source File"], balance_col["Source File"], 28) + ws_balance.set_column(balance_col["Page No."], balance_col["Page No."], 10) + ws_balance.set_column(balance_col["Transaction Date"], balance_col["Value Date"], 13, date_format) + ws_balance.set_column(balance_col["Narration"], balance_col["Narration"], 60) + ws_balance.set_column(balance_col["Debit"], balance_col["Difference"], 18, money) + ws_balance.set_column(balance_col["Status"], balance_col["Review Note"], 24) + raw_export.to_excel(writer, sheet_name="All Extracted Rows", index=False) exact_export.to_excel(writer, sheet_name="Exact Duplicates", index=False) possible_export.to_excel(writer, sheet_name="Possible Duplicates", index=False) @@ -964,7 +1109,8 @@ def export_excel(output, metas, all_df, unique_df, financial_year="", selected_b worksheet.set_row(0, 22, header) max_col = max(0, worksheet.dim_colmax) if worksheet.dim_rowmax >= 0 and name not in { - "Transaction Classification", "Category Summary", "Party Summary", "Category Party Summary" + "Transaction Classification", "Balance Reconciliation", + "Category Summary", "Party Summary", "Category Party Summary" }: worksheet.autofilter(0, 0, worksheet.dim_rowmax, max_col) worksheet.set_column(0, max_col, 18) @@ -974,4 +1120,29 @@ def export_excel(output, metas, all_df, unique_df, financial_year="", selected_b writer.sheets["Masters"].set_column("A:D", 34) # Formula/dropdown support only; not exposed as a visible report sheet. writer.sheets["Masters"].hide() + + # Keep the five principal working sheets first, then retain every other + # existing report in its established relative order. + _reorder_workbook_sheets(workbook, [ + "Dashboard", + "Category Summary", + "Party Summary", + "Category Party Summary", + "Trial Balance", + "Statement Reconciliation", + "Balance Reconciliation", + "All Extracted Rows", + "Exact Duplicates", + "Possible Duplicates", + "Duplicate Summary", + "Masters", + "Transaction Classification", + "Monthly Summary", + "Transfer Comments", + "Review Items", + "Mode Summary", + "Draft Financials", + ]) + writer.sheets["Dashboard"].activate() + writer.sheets["Dashboard"].set_first_sheet() return output