from __future__ import annotations

from pathlib import Path
from difflib import SequenceMatcher
import re

import pandas as pd

from .parsers import parse_pdf
from .parsers.common import infer_mode

HIGH_VALUE_THRESHOLD = 50000.0
LOW_BALANCE_THRESHOLD = 1000.0

ANALYSIS_COLUMNS = [
    "statement_id", "transaction_date", "value_date", "narration", "debit", "credit", "balance",
    "reference_no", "bank_name", "customer_name", "account_number", "source_file",
    "source_page", "parser_name", "mode", "amount", "direction", "narration_key",
    "exact_key", "exact_duplicate", "possible_key", "possible_duplicate",
    "duplicate_group_id", "duplicate_reason", "duplicate_confidence",
    "transfer_bank_code", "transfer_reference", "transfer_comment",
    "auto_party", "party_match_method", "party_match_confidence",
    "auto_category", "auto_nature", "auto_group", "matched_rule_id",
    "matched_keyword", "suggested_ledger", "rule_confidence",
    "review_required", "review_note",
    "category", "counterparty",
]


def _ensure_analysis_columns(df: pd.DataFrame | None) -> pd.DataFrame:
    """Return a DataFrame with every analyzer-owned column present.

    Bank parsers may legitimately return an empty frame (for example when a PDF
    layout is detected but no transaction rows can be extracted). Downstream
    dashboard, duplicate and workbook code must still see a stable schema.
    """
    x = df.copy() if isinstance(df, pd.DataFrame) else pd.DataFrame()
    bool_columns = {"exact_duplicate", "possible_duplicate"}
    numeric_columns = {"debit", "credit", "balance", "amount"}
    for column in ANALYSIS_COLUMNS:
        if column in x.columns:
            continue
        if column in bool_columns:
            x[column] = pd.Series(False, index=x.index, dtype="bool")
        elif column in numeric_columns:
            x[column] = pd.Series(index=x.index, dtype="float64")
        else:
            x[column] = pd.Series(index=x.index, dtype="object")
    if "exact_duplicate" in x.columns:
        x["exact_duplicate"] = x["exact_duplicate"].fillna(False).astype(bool)
    if "possible_duplicate" in x.columns:
        x["possible_duplicate"] = x["possible_duplicate"].fillna(False).astype(bool)
    return x


def _duplicate_count(df: pd.DataFrame, column: str) -> int:
    if df is None or column not in df.columns or df.empty:
        return 0
    return int(df[column].fillna(False).astype(bool).sum())


def clean_key(value):
    value = re.sub(r"\s+", " ", str(value or "").upper()).strip()
    return re.sub(r"\b\d{8,}\b", "<REF>", value)


def _text(value) -> str:
    return re.sub(r"\s+", " ", str(value or "")).strip()


def _amount(row) -> float:
    return float((row.get("debit") or 0) + (row.get("credit") or 0))


def _direction(row) -> str:
    return "Debit" if pd.notna(row.get("debit")) and float(row.get("debit") or 0) > 0 else "Credit"


TRANSFER_PATTERNS = {
    "NEFT": re.compile(r"(?i)\bNEFT(?:/|\s+)([^/\s]+)?(?:/|\s+)([^/\s]+)?(?:/|\s+)([^\n]+)"),
    "RTGS": re.compile(r"(?i)\bRTGS(?:/|\s+)([^/\s]+)?(?:/|\s+)([^/\s]+)?(?:/|\s+)([^\n]+)"),
    "IMPS": re.compile(r"(?i)\bIMPS(?:/(?:P2A|P2P))?(?:/|\s+)([^/\s]+)?(?:/|\s+)([^/\s]+)?(?:/|\s+)([^\n]+)"),
}


def _rx(pattern: str) -> re.Pattern:
    return re.compile(pattern, re.IGNORECASE)


# Phase 1 global narration rules. Specific rules must appear before generic rules.
# Every rule remains reviewable in Excel through the manual override columns.
GLOBAL_CLASSIFICATION_RULES = [
    # Google sub-rules must precede generic Google/software rules.
    {"id": "GLB-GOOGLE-TOLL", "pattern": _rx(r"\b(?:GPAY[- ]?TOLL|FASTAG|TOLL(?:\s+PLAZA)?|PARKING)\b"), "debit": "Toll & Parking Expenses", "credit": "Toll / Parking Refund", "ledger": "Toll & Parking Expenses", "confidence": 98.0},
    {"id": "GLB-GOOGLE-INTERNET", "pattern": _rx(r"\b(?:GOOGLEBBPSINTERNET|GOOGLE.*INTERNET BILL)\b"), "debit": "Telephone & Internet Expenses", "credit": "Telephone / Internet Refund", "ledger": "Telephone & Internet Expenses", "confidence": 97.0},
    {"id": "GLB-GOOGLE-REFUND", "pattern": _rx(r"\b(?:GPAYONLINEREFUNDS|GOOGLE.*REFUND|REFUND.*GOOGLE)\b"), "debit": "Refund Paid / Review", "credit": "Refund / Reversal Receipt", "ledger": "Refunds & Reversals", "confidence": 98.0, "review": True},

    # Bank charges and bank interest.
    {"id": "GLB-BANK-CHARGES", "pattern": _rx(r"\b(?:ATM\s*(?:WDL|ENQ)?\s*CHARGES?|ATM\s*AMC|FOLIO\s*CHARGES?|IMPS\s*COMMISSION\s*CHARGES?|NEFT\s*CHARGES?|RTGS\s*CHARGES?|SMS\s*CHARGES?|BANK\s*CHARGES?|SERVICE\s*CHARGES?|CHEQUE\s*RETURN\s*CHARGES?|BOUNCE\s*CHARGES?|RETURN\s*CHARGES?)\b"), "debit": "Bank Charges", "credit": "Bank Charges Reversal", "ledger": "Bank Charges", "confidence": 99.0},
    {"id": "GLB-BANK-INTEREST-DEBIT", "pattern": _rx(r"\b(?:DEBIT\s*INTEREST|INTEREST\s*DEBITED|OD\s*INTEREST|CC\s*INTEREST)\b"), "debit": "Interest on Bank / OD / CC", "credit": "Interest Reversal", "ledger": "Interest on Bank Borrowings", "confidence": 99.0},
    {"id": "GLB-BANK-INTEREST-CREDIT", "pattern": _rx(r"\b(?:CREDIT\s*INTEREST|INTEREST\s*CREDITED|SB\s*INTEREST)\b"), "debit": "Interest Reversal", "credit": "Interest Income", "ledger": "Interest Income", "confidence": 99.0},

    # Telecom, utilities and vehicle operating costs.
    {"id": "GLB-TELECOM", "pattern": _rx(r"\b(?:AIRTEL|VODAFONE|VI\s*(?:PREPAID|POSTPAID)?|RELIANCE\s*JIO|JIOFIBER|BSNL|HATHWAY|AIRTEL\s*XSTREAM|ACT\s*FIBERNET|POSTPAID|BROADBAND)\b"), "debit": "Telephone & Internet Expenses", "credit": "Telephone / Internet Refund", "ledger": "Telephone & Internet Expenses", "confidence": 96.0},
    {"id": "GLB-ELECTRICITY", "pattern": _rx(r"\b(?:TNEB|TANGEDCO|BESCOM|KSEB|MSEB|APSPDCL|ELECTRICITY|POWER\s*BILL|EB\s*BILL)\b"), "debit": "Electricity Charges", "credit": "Electricity Deposit / Refund", "ledger": "Electricity Charges", "confidence": 98.0},
    {"id": "GLB-FUEL", "pattern": _rx(r"\b(?:BPCL|HPCL|INDIAN\s*OIL|IOC|SHELL|NAYARA|PETROL|DIESEL|FUEL|CNG)\b"), "debit": "Fuel & Vehicle Running Expenses", "credit": "Fuel Refund / Reimbursement", "ledger": "Fuel & Vehicle Running Expenses", "confidence": 96.0},
    {"id": "GLB-VEHICLE-REPAIR", "pattern": _rx(r"\b(?:TYRE|TIRE|BATTERY|SPARES?|GARAGE|WORKSHOP|MECHANIC|VEHICLE\s*REPAIR|MOTOR\s*SERVICE|AUTO\s*SERVICE)\b"), "debit": "Repairs & Maintenance - Vehicle", "credit": "Vehicle Repair Refund", "ledger": "Repairs & Maintenance - Vehicle", "confidence": 94.0},

    # Travel, accommodation, food and personal-review merchants.
    {"id": "GLB-TRAVEL", "pattern": _rx(r"\b(?:IRCTC|INDIGO|AIR\s*INDIA|AKASA|SPICEJET|VISTARA|UBER|OLA|RAPIDO|REDBUS)\b"), "debit": "Travelling Expenses", "credit": "Travel Cancellation Refund", "ledger": "Travelling Expenses", "confidence": 96.0},
    {"id": "GLB-ACCOMMODATION-BRAND", "pattern": _rx(r"\b(?:OYO|FABHOTELS?|MARRIOTT|TAJ\s+HOTELS?|LEMON\s+TREE|HOLIDAY\s+INN|ACCOMMODATION|LODGING)\b"), "debit": "Travelling & Accommodation", "credit": "Accommodation Refund", "ledger": "Travelling & Accommodation", "confidence": 97.0},
    {"id": "GLB-ACCOMMODATION-GENERIC", "pattern": _rx(r"\bHOTEL\b"), "debit": "Travelling & Accommodation", "credit": "Accommodation / Restaurant Refund", "ledger": "Travelling & Accommodation", "confidence": 70.0, "review": True},
    {"id": "GLB-FOOD", "pattern": _rx(r"\b(?:SWIGGY|ZOMATO|EATCLUB|RESTAURANT|CAFE|FOOD\s*COURT)\b"), "debit": "Staff Welfare / Possible Personal Expense", "credit": "Food Order Refund", "ledger": "Staff Welfare / Personal Review", "confidence": 70.0, "review": True},
    {"id": "GLB-MEDICAL", "pattern": _rx(r"\b(?:PHARMACY|MEDPLUS|APOLLO\s*PHARMACY|HOSPITAL|CLINIC|MEDICAL)\b"), "debit": "Medical / Possible Personal Expense", "credit": "Medical Refund / Reimbursement", "ledger": "Medical / Personal Review", "confidence": 75.0, "review": True},
    {"id": "GLB-ENTERTAINMENT", "pattern": _rx(r"\b(?:NETFLIX|JIOHOTSTAR|HOTSTAR|PRIME\s*VIDEO|SONYLIV|ZEE5|PVR|INOX)\b"), "debit": "Subscription / Possible Personal Expense", "credit": "Subscription Refund", "ledger": "Subscriptions / Personal Review", "confidence": 75.0, "review": True},
    {"id": "GLB-RETAIL-REVIEW", "pattern": _rx(r"\b(?:DMART|D[- ]?MART|RELIANCE\s*SMART|SUPERMARKET|SHOPPING\s*MALL|LIFESTYLE)\b"), "debit": "General Purchase / Possible Personal Expense", "credit": "Retail Refund", "ledger": "General Purchases / Personal Review", "confidence": 68.0, "review": True},

    # Insurance and subscriptions.
    {"id": "GLB-INSURANCE", "pattern": _rx(r"\b(?:INSURANCE|PREMIUM|LIC|HDFC\s*ERGO|ICICI\s*LOMBARD|STAR\s*HEALTH|NIVA\s*BUPA|UNIVERSAL\s*SOMPO)\b"), "debit": "Insurance Expenses", "credit": "Insurance Claim / Refund", "ledger": "Insurance Expenses", "confidence": 96.0},
    {"id": "GLB-HOSTING", "pattern": _rx(r"\b(?:HOSTINGER|GODADDY|NAMECHEAP|CLOUDFLARE|BIGROCK|RESELLERCLUB)\b"), "debit": "Website & Hosting Expenses", "credit": "Hosting Refund", "ledger": "Website & Hosting Expenses", "confidence": 98.0},
    {"id": "GLB-SOFTWARE", "pattern": _rx(r"\b(?:AMAZON\s*WEB\s*SERVICES|AWS|AZURE|GOOGLE\s*CLOUD|DIGITALOCEAN|LINODE|VULTR|HETZNER|OPENAI|CHATGPT|ANTHROPIC|CLAUDE|GITHUB|ATLASSIAN|SLACK|ZOOM|CANVA|ADOBE|FIGMA|GOOGLE\s*PLAY)\b"), "debit": "Software Subscription", "credit": "Software / Subscription Refund", "ledger": "Software Subscription", "confidence": 95.0},
    {"id": "GLB-COURIER", "pattern": _rx(r"\b(?:DTDC|BLUE\s*DART|DELHIVERY|PROFESSIONAL\s*COURIER|INDIA\s*POST|COURIER)\b"), "debit": "Courier & Postage", "credit": "Courier Refund", "ledger": "Courier & Postage", "confidence": 97.0},

    # Employment, occupancy and financing.
    {"id": "GLB-SALARY", "pattern": _rx(r"\b(?:SALARY|WAGES|PAYROLL)\b"), "debit": "Salary & Wages", "credit": "Salary Reversal / Employee Recovery", "ledger": "Salary & Wages", "confidence": 95.0},
    {"id": "GLB-RENT", "pattern": _rx(r"\b(?:RENT|LEASE|RENTAL)\b"), "debit": "Rent Expense", "credit": "Rent Receipt", "ledger": "Rent Expense / Rent Income", "confidence": 92.0},
    {"id": "GLB-LOAN-EMI", "pattern": _rx(r"\b(?:LOAN\s*EMI|BAJAJ\s*EMI|EMI|NACH|ECS|AUTO\s*DEBIT)\b"), "debit": "Loan Repayment", "credit": "Loan Receipt / Funding", "ledger": "Loan Account", "confidence": 82.0, "review": True, "group": "Contra / Balance Sheet"},
    {"id": "GLB-CREDIT-CARD", "pattern": _rx(r"\b(?:CREDIT\s*CARD\s*PAYMENT|CARD\s*PAYMENT|CC\s*PAYMENT)\b"), "debit": "Credit Card Payment / Contra", "credit": "Credit Card Refund / Reversal", "ledger": "Credit Card Account", "confidence": 90.0, "review": True, "group": "Contra / Balance Sheet"},

    # Taxes and statutory payments.
    {"id": "GLB-GST", "pattern": _rx(r"\b(?:GST\s*PMT|GSTN|GST\s*PAYMENT|CPIN)\b"), "debit": "GST Payment", "credit": "GST Refund / Receipt", "ledger": "GST Payable / Receivable", "confidence": 98.0, "group": "Contra / Balance Sheet"},
    {"id": "GLB-INCOME-TAX", "pattern": _rx(r"\b(?:INCOME\s*TAX|ITNS|OLTAS|ADVANCE\s*TAX|SELF\s*ASSESSMENT\s*TAX)\b"), "debit": "Income Tax Payment", "credit": "Income Tax Refund", "ledger": "Income Tax", "confidence": 98.0, "group": "Contra / Balance Sheet"},
    {"id": "GLB-TDS", "pattern": _rx(r"\b(?:TDS|CHALLAN\s*281|TRACES)\b"), "debit": "TDS Payment", "credit": "TDS Refund / Reversal", "ledger": "TDS Payable / Receivable", "confidence": 98.0, "group": "Contra / Balance Sheet"},
    {"id": "GLB-PF", "pattern": _rx(r"\b(?:EPFO|PROVIDENT\s*FUND|PF\s*PAYMENT)\b"), "debit": "Provident Fund Payment", "credit": "Provident Fund Refund", "ledger": "Provident Fund Payable", "confidence": 97.0, "group": "Contra / Balance Sheet"},
    {"id": "GLB-ESIC", "pattern": _rx(r"\b(?:ESIC|ESI\s*PAYMENT)\b"), "debit": "ESIC Payment", "credit": "ESIC Refund", "ledger": "ESIC Payable", "confidence": 97.0, "group": "Contra / Balance Sheet"},
    {"id": "GLB-PROF-TAX", "pattern": _rx(r"\b(?:PROFESSIONAL\s*TAX|PROF\s*TAX|PTAX)\b"), "debit": "Professional Tax Payment", "credit": "Professional Tax Refund", "ledger": "Professional Tax Payable", "confidence": 97.0, "group": "Contra / Balance Sheet"},
    {"id": "GLB-MCA-ROC", "pattern": _rx(r"\b(?:MCA|ROC\s*FEE|REGISTRAR\s*OF\s*COMPANIES)\b"), "debit": "ROC / MCA Filing Fees", "credit": "ROC / MCA Refund", "ledger": "ROC / MCA Filing Fees", "confidence": 96.0},

    # Transport, handling, advances and commission.
    {"id": "GLB-FREIGHT", "pattern": _rx(r"\b(?:FREIGHT|LR\s*PAYMENT|LORRY\s*RECEIPT|CARRIAGE|TRANSPORT\s*(?:CHARGE|PAYMENT|RECEIPT))\b"), "debit": "Freight / Carriage Expenses", "credit": "Freight / Transport Receipt", "ledger": "Freight / Carriage", "confidence": 88.0, "review": True},
    {"id": "GLB-HANDLING", "pattern": _rx(r"\b(?:LOADING|UNLOADING|HAMALI|HANDLING\s*CHARGES?)\b"), "debit": "Loading & Unloading Charges", "credit": "Loading / Handling Receipt", "ledger": "Loading & Unloading Charges", "confidence": 93.0},
    {"id": "GLB-ADVANCE", "pattern": _rx(r"\b(?:VEHICLE\s*ADVANCE|DRIVER\s*ADVANCE|TRIP\s*ADVANCE|DIESEL\s*ADVANCE|ADVANCE)\b"), "debit": "Advance Paid", "credit": "Advance Received", "ledger": "Advances", "confidence": 72.0, "review": True, "group": "Contra / Balance Sheet"},
    {"id": "GLB-COMMISSION", "pattern": _rx(r"\b(?:COMMISSION|BROKERAGE|AGENT\s*COMMISSION)\b"), "debit": "Commission & Brokerage Expense", "credit": "Commission Income", "ledger": "Commission & Brokerage", "confidence": 90.0, "review": True},

    # Refunds, reversals and failed transactions.
    {"id": "GLB-REFUND", "pattern": _rx(r"\b(?:REFUND|REVERSAL|RVSL|REVERSED|CANCELLED|CANCELLATION)\b"), "debit": "Refund Paid / Review", "credit": "Refund / Reversal Receipt", "ledger": "Refunds & Reversals", "confidence": 92.0, "review": True},
    {"id": "GLB-DISHONOUR", "pattern": _rx(r"\b(?:CHEQUE\s*RETURN|BOUNCE|DISHONOURED|FAILED\s*TRANSACTION|RETURNED)\b"), "debit": "Dishonour / Returned Transaction", "credit": "Dishonour / Returned Transaction", "ledger": "Dishonoured Transactions", "confidence": 91.0, "review": True},

    # Payment gateway and marketplace settlement rules.
    {"id": "GLB-PAYMENT-GATEWAY", "pattern": _rx(r"\b(?:RAZORPAY|CASHFREE|PAYU|CCAVENUE|PHONEPE\s*PG)\b"), "debit": "Payment Gateway Charges / Settlement Adjustment", "credit": "Payment Gateway Settlement", "ledger": "Payment Gateway Settlement", "confidence": 88.0, "review": True},
    {"id": "GLB-MARKETPLACE", "pattern": _rx(r"\b(?:AMAZON\s*SELLER|FLIPKART\s*SELLER|MEESHO|MYNTRA|AJIO)\b"), "debit": "Marketplace Charges / Review", "credit": "Sales Receipt / Marketplace Settlement", "ledger": "Marketplace Settlement", "confidence": 88.0, "review": True},
]


def _default_group(category: str, direction: str) -> str:
    if any(token in category for token in ("Contra", "Loan", "Advance", "GST", "Income Tax", "TDS", "Provident Fund", "ESIC", "Professional Tax")):
        return "Contra / Balance Sheet"
    if direction == "Credit" or any(token in category for token in ("Receipt", "Income", "Refund", "Deposit", "Settlement")):
        return "Income / Receipt"
    if category.endswith("/ Review") or "Possible Personal" in category:
        return "Unclassified"
    return "Expense / Payment"


def _apply_global_rule(narration: str, direction: str) -> dict | None:
    searchable = _text(narration).upper()
    for rule in GLOBAL_CLASSIFICATION_RULES:
        match = rule["pattern"].search(searchable)
        if not match:
            continue
        category = rule["debit"] if direction == "Debit" else rule["credit"]
        return {
            "category": category,
            "rule_id": rule["id"],
            "keyword": match.group(0),
            "ledger": rule["ledger"],
            "confidence": float(rule.get("confidence", 90.0)),
            "review": bool(rule.get("review", False)),
            "group": rule.get("group") or _default_group(category, direction),
        }
    return None


def _normalize_party(value: str) -> str:
    text = _text(value).upper()
    text = re.sub(r"\b(?:PVT\.?\s*LTD\.?|PRIVATE\s+LIMITED)\b", "PRIVATE LIMITED", text)
    text = re.sub(r"\bLTD\.?\b", "LIMITED", text)
    text = re.sub(r"\bM/S\.?\b", "", text)
    text = re.sub(r"[^A-Z0-9& ]+", " ", text)
    text = re.sub(r"\b(?:MR|MRS|MS|MISS|SHRI|SRI|SMT)\b", "", text)
    return re.sub(r"\s+", " ", text).strip()[:120]


def _extract_transfer_details(narration: str) -> tuple[str, str, str]:
    raw = _text(narration)
    upper = raw.upper()
    for mode, pattern in TRANSFER_PATTERNS.items():
        if mode not in upper:
            continue
        match = pattern.search(raw)
        if match:
            parts = [(_text(v) if v else "") for v in match.groups()]
            bank_code = parts[0][:30]
            reference = parts[1][:80]
            comment = parts[2][:240]
            comment = re.sub(r"(?i)/?BRANCH\s*:\s*.*$", "", comment).strip(" /.-")
            return bank_code, reference, comment
    return "", "", ""


def _party_group_key(value: str) -> str:
    return _normalize_party(value)


def _group_similar_parties(values: list[str]) -> dict[str, tuple[str, str, float]]:
    canonical: list[str] = []
    result: dict[str, tuple[str, str, float]] = {}
    for original in values:
        key = _party_group_key(original) or "UNIDENTIFIED"
        if key in result:
            continue
        best = None
        best_score = 0.0
        for existing in canonical:
            score = SequenceMatcher(None, key, existing).ratio()
            tokens_a = set(key.split())
            tokens_b = set(existing.split())
            containment = min(len(tokens_a & tokens_b) / max(1, len(tokens_a)), len(tokens_a & tokens_b) / max(1, len(tokens_b)))
            if key in existing or existing in key:
                score = max(score, 0.96 if min(len(key), len(existing)) >= 6 else score)
            score = max(score, containment)
            if score > best_score:
                best, best_score = existing, score
        if best and best_score >= 0.94:
            result[key] = (best.title(), "Fuzzy/partial", round(best_score * 100, 1))
        else:
            canonical.append(key)
            result[key] = (key.title(), "Normalized exact", 100.0)
    return result


def _apply_party_grouping(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return df
    x = df.copy()
    mapping = _group_similar_parties([str(v or "") for v in x["auto_party"].tolist()])
    grouped, methods, scores = [], [], []
    for value in x["auto_party"].tolist():
        key = _party_group_key(str(value or "")) or "UNIDENTIFIED"
        party, method, score = mapping.get(key, (str(value or "Unidentified"), "Original", 0.0))
        grouped.append(party)
        methods.append(method)
        scores.append(score)
    x["auto_party"] = grouped
    x["party_match_method"] = methods
    x["party_match_confidence"] = scores
    return x


def _extract_counterparty(narration: str, direction: str) -> str:
    raw = _text(narration)
    upper = raw.upper()
    if not raw:
        return "Unidentified"
    if "CASH DEP" in upper or "CASH DEPOSIT" in upper:
        return "Cash Deposit"
    if "CASH WITHDRAW" in upper or "ATM WDL" in upper or "ATM/CASH" in upper and direction == "Debit":
        return "Cash Withdrawal"
    if "BANK CHARGE" in upper or "CHARGES" in upper or "COMMISSION" in upper:
        return "Bank Charges"
    if "INTEREST" in upper:
        return "Bank Interest"
    if "/UPI/" in upper or upper.startswith("UPI/"):
        parts = [p.strip() for p in re.split(r"/", raw) if p.strip()]
        ignored = {"UPI", "DR", "CR", "NA", "PAYMENT", "PHONEPE", "PAYTM"}
        for part in parts[1:]:
            cleaned = re.sub(r"^X+\d+$", "", part, flags=re.I).strip()
            if cleaned and cleaned.upper() not in ignored and not re.fullmatch(r"\d{6,}", cleaned):
                return cleaned[:120]
        return "UPI / Mobile"
    for marker in ("NEFT/", "RTGS/", "IMPS/", "TRANSFER TO", "TRANSFER FROM"):
        if marker in upper:
            parts = [p.strip() for p in re.split(r"/", raw) if p.strip()]
            for part in parts[1:]:
                if re.fullmatch(r"[A-Z0-9]{10,}", part.upper()):
                    continue
                if re.fullmatch(r"\d+(?:\.\d+)?", part):
                    continue
                if part.upper() in {"NEFT", "RTGS", "IMPS", "P2A", "P2P", "BRANCH : ATM SERVICE BRANCH"}:
                    continue
                return part[:120]
    if "CTS-CHQ" in upper or "CLEARING" in upper:
        return "Cheque Clearing"
    return raw[:120]


def _classify_row(row) -> tuple[str, str, str, str, str, str, str, float, bool]:
    narration = _text(row.get("narration"))
    upper = narration.upper()
    direction = row.get("direction") or _direction(row)
    mode = str(row.get("mode") or "Other")
    counterparty = _extract_counterparty(narration, direction)

    # Nominal credits are often bank-account validation entries rather than income.
    amount = float(row.get("amount") or 0)
    if direction == "Credit" and 0 < amount <= 10:
        category = "Bank Account Validation / Test Transaction"
        return category, counterparty, "Receipt", "Unclassified", "GLB-NOMINAL-VALIDATION", f"Amount {amount:,.2f}", "Validation / Test Transaction", 75.0, True

    # Strong global rules take priority over generic mode classifications.
    matched = _apply_global_rule(narration, direction)
    if matched:
        nature = "Payment" if direction == "Debit" else "Receipt"
        if matched["group"] == "Contra / Balance Sheet":
            nature = "Contra / Balance Sheet"
        elif "Refund" in matched["category"] or "Reversal" in matched["category"]:
            nature = "Refund / Reversal"
        return (
            matched["category"], counterparty, nature, matched["group"], matched["rule_id"],
            matched["keyword"], matched["ledger"], matched["confidence"], matched["review"],
        )

    if "CASH DEP" in upper or "CASH DEPOSIT" in upper:
        return "Cash Deposit / Cash Sales", "Cash Deposit", "Receipt", "Income / Receipt", "GLB-CASH-DEPOSIT", "CASH DEP", "Cash / Sales Receipts", 95.0, True
    if "CASH WITHDRAW" in upper or "ATM WDL" in upper or ("ATM/CASH" in upper and direction == "Debit"):
        return "Cash Withdrawal", "Cash Withdrawal", "Payment", "Contra / Balance Sheet", "GLB-CASH-WITHDRAWAL", "ATM/CASH", "Cash Account", 98.0, True
    if any(token in upper for token in ("SELF", "OWN ACCOUNT", "SELF TRANSFER")) or ("TRANSFER FROM" in upper and "MOBILE TRANSFER" in upper):
        return "Self Transfer / Contra", counterparty, "Contra", "Contra / Balance Sheet", "GLB-SELF-CONTRA", "SELF/OWN ACCOUNT", "Contra / Own Account", 80.0, True
    if mode == "UPI" or "/UPI/" in upper or upper.startswith("UPI/"):
        category = "UPI Payment / Expense" if direction == "Debit" else "UPI Customer Receipt"
        return category, counterparty, "Payment" if direction == "Debit" else "Receipt", _default_group(category, direction), "FALLBACK-UPI", "UPI", "UPI - Manual Classification", 45.0, True
    if mode == "NEFT" or "NEFT" in upper:
        category = "NEFT Payment" if direction == "Debit" else "NEFT Receipt"
        return category, counterparty, "Payment" if direction == "Debit" else "Receipt", _default_group(category, direction), "FALLBACK-NEFT", "NEFT", "NEFT - Manual Classification", 50.0, True
    if mode == "RTGS" or "RTGS" in upper:
        category = "RTGS Payment" if direction == "Debit" else "RTGS Receipt"
        return category, counterparty, "Payment" if direction == "Debit" else "Receipt", _default_group(category, direction), "FALLBACK-RTGS", "RTGS", "RTGS - Manual Classification", 50.0, True
    if mode == "IMPS" or "IMPS" in upper:
        category = "IMPS Payment" if direction == "Debit" else "IMPS Receipt"
        return category, counterparty, "Payment" if direction == "Debit" else "Receipt", _default_group(category, direction), "FALLBACK-IMPS", "IMPS", "IMPS - Manual Classification", 50.0, True
    if "CTS-CHQ" in upper or "CLEARING" in upper or mode == "Cheque":
        category = "Cheque Payment" if direction == "Debit" else "Cheque Deposit"
        return category, "Cheque Clearing", "Payment" if direction == "Debit" else "Receipt", _default_group(category, direction), "FALLBACK-CHEQUE", "CHEQUE/CLEARING", "Cheque Clearing", 55.0, True
    category = "Other Bank Payment / Review" if direction == "Debit" else "Other Bank Receipt / Review"
    return category, counterparty, "Payment" if direction == "Debit" else "Receipt", "Unclassified", "FALLBACK-UNCLASSIFIED", "", "Unclassified / Review", 0.0, True


def _review_note(row) -> str:
    notes: list[str] = []
    amount = float(row.get("amount") or 0)
    balance = row.get("balance")
    category = str(row.get("category") or "")
    counterparty = str(row.get("counterparty") or "")
    if bool(row.get("review_required")):
        notes.append("Global rule requires manual review")
    if amount >= HIGH_VALUE_THRESHOLD:
        notes.append(f"High value transaction >= {HIGH_VALUE_THRESHOLD:,.0f}")
    if category == "Cash Deposit / Cash Sales":
        notes.append("Cash deposit - verify cash book/source")
    elif category == "Cash Withdrawal":
        notes.append("Cash withdrawal - verify cash book/use")
    if category.endswith("/ Review") or counterparty == "Unidentified":
        notes.append("Narration requires manual classification")
    if amount >= 10000 and amount % 10000 == 0:
        notes.append("Round/high amount - verify nature")
    if pd.notna(balance) and float(balance) < LOW_BALANCE_THRESHOLD:
        notes.append("Low balance after transaction")
    if bool(row.get("exact_duplicate")):
        notes.append("Exact duplicate / overlapping statement row")
    elif bool(row.get("possible_duplicate")):
        notes.append("Possible duplicate - review before exclusion")
    return "; ".join(dict.fromkeys(notes))


def enrich(df, classification_enabled: bool = True):
    x = _ensure_analysis_columns(df)
    if x.empty:
        return x
    x["mode"] = x["narration"].map(infer_mode)
    x["amount"] = x["debit"].fillna(0) + x["credit"].fillna(0)
    x["direction"] = x.apply(_direction, axis=1)
    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)
    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"]
    x["duplicate_group_id"] = x.groupby("exact_key", dropna=False).ngroup().map(lambda n: f"EX-{n+1:06d}")
    x.loc[~x["exact_duplicate"], "duplicate_group_id"] = ""
    x["duplicate_reason"] = x.apply(lambda r: "Exact same date/value/narration/balance" if bool(r.get("exact_duplicate")) else ("Same date/direction/amount/normalized narration" if bool(r.get("possible_duplicate")) else ""), axis=1)
    x["duplicate_confidence"] = x.apply(lambda r: 100.0 if bool(r.get("exact_duplicate")) else (85.0 if bool(r.get("possible_duplicate")) else 0.0), axis=1)
    transfer = x["narration"].map(_extract_transfer_details)
    x["transfer_bank_code"] = transfer.map(lambda t: t[0])
    x["transfer_reference"] = transfer.map(lambda t: t[1])
    x["transfer_comment"] = transfer.map(lambda t: t[2])
    if classification_enabled:
        classified = x.apply(_classify_row, axis=1, result_type="expand")
        classified.columns = [
            "auto_category", "auto_party", "auto_nature", "auto_group",
            "matched_rule_id", "matched_keyword", "suggested_ledger",
            "rule_confidence", "review_required",
        ]
        for column in classified.columns:
            x[column] = classified[column]
        x = _apply_party_grouping(x)
        x["category"] = x["auto_category"]
        x["counterparty"] = x["auto_party"]
        x["review_note"] = x.apply(_review_note, axis=1)
    else:
        x["auto_category"] = "Classification disabled"
        x["auto_party"] = x["narration"].map(lambda value: _text(value)[:120] or "Unidentified")
        x["party_match_method"] = "Original"
        x["party_match_confidence"] = 0.0
        x["auto_nature"] = x["direction"].map(lambda d: "Payment" if d == "Debit" else "Receipt")
        x["auto_group"] = "Unclassified"
        x["matched_rule_id"] = "CLASSIFICATION-DISABLED"
        x["matched_keyword"] = ""
        x["suggested_ledger"] = ""
        x["rule_confidence"] = 0.0
        x["review_required"] = True
        x["category"] = x["auto_category"]
        x["counterparty"] = x["auto_party"]
        x["review_note"] = x.apply(_review_note, axis=1)
    return _ensure_analysis_columns(x)


def analyze_files(paths, customer_override="", account_override="", bank_hint="auto", classification_enabled=True):
    metas = []
    frames = []
    for path in paths:
        meta, df = parse_pdf(path, bank_hint=bank_hint)
        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
        statement_id = f"STMT-{len(metas) + 1:03d}"
        df = df.copy()
        df["statement_id"] = statement_id
        meta.statement_id = statement_id
        metas.append(meta)
        frames.append(df)
    combined = pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
    all_df = _ensure_analysis_columns(enrich(combined, classification_enabled))

    # IMPORTANT ACCOUNTING CONTROL:
    # Duplicate detection is advisory only.  A bank may legitimately contain two
    # transactions with the same date, amount and narration.  Removing a flagged
    # row changes debit/credit totals and creates a false closing-balance
    # difference.  Therefore every extracted row remains in the accounting table;
    # exact/possible duplicate flags are retained for reviewer action.
    unique_df = _ensure_analysis_columns(all_df.copy())
    if metas and all_df.empty:
        raise ValueError(
            "The bank statement format was identified, but no transaction rows could be extracted. "
            "Please verify that the PDF is text-readable and that this statement layout is supported."
        )
    return metas, all_df, unique_df


def reconcile(metas, all_df):
    """Reconcile each uploaded statement independently.

    The routine never mixes rows from different source statements.  Duplicate
    flags remain review indicators and do not remove transactions from the
    accounting totals.
    """
    rows = []
    for position, meta in enumerate(metas, start=1):
        statement_id = getattr(meta, "statement_id", f"STMT-{position:03d}")
        if all_df.empty:
            data = pd.DataFrame()
        elif "statement_id" in all_df.columns and all_df["statement_id"].notna().any():
            data = all_df[all_df.statement_id.eq(statement_id)].copy()
        else:
            data = all_df[all_df.source_file.eq(meta.source_file)].copy()

        extracted_debit = round(float(pd.to_numeric(data.get("debit"), errors="coerce").fillna(0).sum()), 2) if not data.empty else 0.0
        extracted_credit = round(float(pd.to_numeric(data.get("credit"), errors="coerce").fillna(0).sum()), 2) if not data.empty else 0.0
        computed_closing = None
        if meta.opening_balance is not None:
            computed_closing = round(float(meta.opening_balance) + extracted_credit - extracted_debit, 2)

        balance_values = pd.to_numeric(data.get("balance"), errors="coerce").dropna() if not data.empty else pd.Series(dtype=float)
        printed_last_balance = float(balance_values.iloc[-1]) if not balance_values.empty else None

        debit_diff = None if meta.total_debit is None else round(extracted_debit - float(meta.total_debit), 2)
        credit_diff = None if meta.total_credit is None else round(extracted_credit - float(meta.total_credit), 2)
        closing_diff = None if meta.closing_balance is None or computed_closing is None else round(computed_closing - float(meta.closing_balance), 2)
        last_balance_diff = None if meta.closing_balance is None or printed_last_balance is None else round(printed_last_balance - float(meta.closing_balance), 2)
        discontinuities = int((pd.to_numeric(data.get("movement_difference"), errors="coerce").abs() > 0.01).sum()) if not data.empty and "movement_difference" in data.columns else 0

        checks = (debit_diff, credit_diff, closing_diff, last_balance_diff)
        status = "Reconciled" if all(value is None or abs(float(value)) <= 0.01 for value in checks) else "Review"
        rows.append({
            "statement_id": statement_id,
            **meta.to_dict(),
            "extracted_transactions": len(data),
            "extracted_debit": extracted_debit,
            "debit_difference": debit_diff,
            "extracted_credit": extracted_credit,
            "credit_difference": credit_diff,
            "computed_closing_balance": computed_closing,
            "printed_last_running_balance": printed_last_balance,
            "statement_closing_difference": closing_diff,
            "last_running_balance_difference": last_balance_diff,
            "balance_discontinuities": discontinuities,
            "status": status,
        })
    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)
    summary = x.groupby("month", dropna=False).agg(
        transaction_count=("amount", "size"),
        opening_balance=("balance", "first"),
        total_debit=("debit", "sum"),
        total_credit=("credit", "sum"),
        closing_balance=("balance", "last"),
    ).reset_index()
    summary["net_movement"] = summary.total_credit - summary.total_debit
    return summary


def mode_summary(df):
    if df.empty:
        return pd.DataFrame()
    return df.groupby(["mode", "direction"], dropna=False).agg(transaction_count=("amount", "size"), amount=("amount", "sum")).reset_index()


def category_summary(df):
    if df.empty:
        return pd.DataFrame()
    result = df.groupby("category", dropna=False).agg(
        transaction_count=("amount", "size"), debit=("debit", "sum"), credit=("credit", "sum")
    ).reset_index()
    result["net_credit_debit"] = result.credit - result.debit
    return result.sort_values(["transaction_count", "category"], ascending=[False, True])


def party_summary(df):
    if df.empty:
        return pd.DataFrame()
    result = df.groupby(["counterparty", "category"], dropna=False).agg(
        transaction_count=("amount", "size"), debit=("debit", "sum"), credit=("credit", "sum")
    ).reset_index()
    result["net_credit_debit"] = result.credit - result.debit
    return result.sort_values(["transaction_count", "counterparty"], ascending=[False, True])


def duplicate_summary(all_df):
    return pd.DataFrame(
        [
            {"check": "All extracted rows", "count": len(all_df)},
            {"check": "Exact duplicate rows", "count": _duplicate_count(all_df, "exact_duplicate")},
            {"check": "Possible duplicate rows", "count": _duplicate_count(all_df, "possible_duplicate")},
            {"check": "Unique rows after exact deduplication", "count": int(all_df["exact_key"].nunique()) if not all_df.empty and "exact_key" in all_df.columns else 0},
        ]
    )


def draft_financials(df, metas):
    if df.empty:
        return pd.DataFrame(columns=["particulars", "amount", "treatment_notes", "finalisation_requirement"])
    categories = df.groupby("category", dropna=False).agg(debit=("debit", "sum"), credit=("credit", "sum"))
    receipt_categories = ["Cash Deposit / Cash Sales", "UPI Customer Receipt", "Payment Aggregator Settlement", "NEFT Receipt", "RTGS Receipt", "IMPS Receipt", "Cheque Deposit"]
    payment_categories = ["UPI Payment / Expense", "NEFT Payment", "RTGS Payment", "IMPS Payment", "Cheque Payment"]
    business_receipts = float(sum(categories.loc[c, "credit"] for c in receipt_categories if c in categories.index))
    other_receipts = float(sum(categories.loc[c, "credit"] for c in categories.index if c not in receipt_categories and categories.loc[c, "credit"] > 0))
    classified_payments = float(sum(categories.loc[c, "debit"] for c in payment_categories if c in categories.index))
    other_payments = float(sum(categories.loc[c, "debit"] for c in categories.index if c not in payment_categories and categories.loc[c, "debit"] > 0))
    opening = next((meta.opening_balance for meta in metas if meta.opening_balance is not None), None)
    closing = next((meta.closing_balance for meta in reversed(metas) if meta.closing_balance is not None), None)
    return pd.DataFrame(
        [
            ["Potential business receipts / collections", business_receipts, "Narration-based grouping of customer, cash and settlement credits", "Verify with GST/sales register and cash book"],
            ["Other / unclassified receipts", other_receipts, "May include capital, loans, contra and other receipts", "Classify with books and supporting records"],
            ["Classified bank payments", classified_payments, "Narration-based payment classification", "Verify expense, purchase and drawings treatment"],
            ["Other / unclassified bank payments", other_payments, "Requires review", "Classify before final accounts"],
            ["Opening bank balance", opening, "Per first uploaded statement", "Verify statement continuity"],
            ["Closing bank balance", closing, "Per last uploaded statement", "Agree with bank reconciliation"],
        ],
        columns=["particulars", "amount", "treatment_notes", "finalisation_requirement"],
    )


def _workbook_columns(df):
    preferred = [
        "statement_id", "transaction_date", "value_date", "narration", "transfer_bank_code", "transfer_reference",
        "transfer_comment", "auto_party", "party_match_method", "party_match_confidence",
        "auto_category", "auto_nature", "auto_group", "matched_rule_id", "matched_keyword",
        "suggested_ledger", "rule_confidence", "review_required", "mode", "direction", "debit", "credit",
        "balance", "reference_no", "bank_name", "customer_name", "account_number", "source_file",
        "source_page", "parser_name", "exact_duplicate", "possible_duplicate", "duplicate_group_id",
        "duplicate_reason", "duplicate_confidence", "review_note",
    ]
    return [column for column in preferred if column in df.columns]


def _excel_col(index: int) -> str:
    result = ""
    n = index + 1
    while n:
        n, remainder = divmod(n - 1, 26)
        result = chr(65 + remainder) + result
    return result


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,
    opening_balance: float,
    statement_closing: float,
    money_format,
    text_format,
    warning_format,
):
    """Add a formula-driven opening-to-closing reconciliation block."""
    title_row = start_row
    opening_row = start_row + 1
    credit_row = start_row + 2
    debit_row = start_row + 3
    computed_row = start_row + 4
    statement_row = start_row + 5
    difference_row = start_row + 6
    status_row = start_row + 7

    worksheet.write(title_row, 0, "Bank Balance Reconciliation", text_format)
    worksheet.write(opening_row, 0, "Opening Balance", text_format)
    worksheet.write_number(opening_row, 1, float(opening_balance or 0.0), money_format)
    worksheet.write(credit_row, 0, "Add: Total Credits", text_format)
    worksheet.write_formula(credit_row, 1, "=SUM(tblTransactions[credit])", money_format, 0)
    worksheet.write(debit_row, 0, "Less: Total Debits", text_format)
    worksheet.write_formula(debit_row, 1, "=SUM(tblTransactions[debit])", money_format, 0)
    worksheet.write(computed_row, 0, "Computed Closing Balance", text_format)
    worksheet.write_formula(computed_row, 1, f"=B{opening_row + 1}+B{credit_row + 1}-B{debit_row + 1}", money_format, float(opening_balance or 0.0))
    worksheet.write(statement_row, 0, "Statement Closing Balance", text_format)
    worksheet.write_number(statement_row, 1, float(statement_closing or 0.0), money_format)
    worksheet.write(difference_row, 0, "Difference", text_format)
    worksheet.write_formula(difference_row, 1, f"=B{computed_row + 1}-B{statement_row + 1}", money_format, 0)
    worksheet.write(status_row, 0, "Reconciliation Status", text_format)
    worksheet.write_formula(status_row, 1, f'=IF(ABS(B{difference_row + 1})<=0.01,"Reconciled","RECONCILIATION ERROR")', text_format, "Reconciled")
    worksheet.conditional_format(difference_row, 1, difference_row, 1, {"type": "cell", "criteria": "not between", "minimum": -0.01, "maximum": 0.01, "format": warning_format})
    worksheet.conditional_format(status_row, 1, status_row, 1, {"type": "text", "criteria": "containing", "value": "ERROR", "format": warning_format})


def export_excel(output, metas, all_df, unique_df, financial_year="", selected_bank="auto", classification_enabled=True):
    all_df = _ensure_analysis_columns(all_df)
    unique_df = _ensure_analysis_columns(unique_df)
    recon = reconcile(metas, all_df)
    customer = next((meta.customer_name for meta in metas if meta.customer_name), "")
    account = next((meta.account_number for meta in metas if meta.account_number), "")
    banks = ", ".join(sorted({meta.bank_name for meta in metas}))
    # Sum each statement's opening and closing independently.  This is an
    # accounting aggregate, not a first-statement/last-statement shortcut.
    opening = round(sum(float(meta.opening_balance) for meta in metas if meta.opening_balance is not None), 2)
    statement_closing = round(sum(float(meta.closing_balance) for meta in metas if meta.closing_balance is not None), 2)

    tx = unique_df[_workbook_columns(unique_df)].copy() if not unique_df.empty else pd.DataFrame(columns=_workbook_columns(unique_df))
    tx.insert(0, "row_id", range(1, len(tx) + 1))
    tx["manual_party"] = ""
    tx["final_party"] = tx.get("auto_party", "")
    tx["manual_category"] = ""
    tx["final_category"] = tx.get("auto_category", "")
    tx["manual_nature"] = ""
    tx["final_nature"] = tx.get("auto_nature", "")
    tx["manual_group"] = ""
    tx["final_group"] = tx.get("auto_group", "")
    tx["manual_review_note"] = ""

    raw_export = all_df[_workbook_columns(all_df)].copy() if not all_df.empty else all_df.copy()
    exact_export = raw_export[all_df["exact_duplicate"].fillna(False).astype(bool).to_numpy()].copy() if not all_df.empty else raw_export.copy()
    possible_export = raw_export[all_df["possible_duplicate"].fillna(False).astype(bool).to_numpy()].copy() if not all_df.empty else raw_export.copy()

    categories = sorted({str(v) for v in tx.get("auto_category", pd.Series(dtype=str)).dropna() if str(v).strip()} | {
        "Advance Paid", "Advance Received", "Bank Charges", "Capital Introduction", "Cash Deposit / Cash Sales",
        "Cash Withdrawal", "Customer / Business Receipt", "Drawings", "Electricity Charges",
        "Fuel & Vehicle Running Expenses", "Freight / Carriage Expenses", "Insurance Expenses",
        "Interest Income", "Interest on Bank / OD / CC", "Loading & Unloading Charges",
        "Loan Receipt / Funding", "Loan Repayment", "Medical / Possible Personal Expense",
        "Purchase / Supplier Payment", "Refund / Reversal Receipt", "Rent Expense",
        "Repairs & Maintenance - Vehicle", "Salary & Wages", "Self Transfer / Contra",
        "Software Subscription", "Staff Welfare / Possible Personal Expense",
        "Telephone & Internet Expenses", "Toll & Parking Expenses", "Travelling & Accommodation",
        "Travelling Expenses", "Website & Hosting Expenses", "Unclassified / Review",
    })
    parties = sorted({str(v) for v in tx.get("auto_party", pd.Series(dtype=str)).dropna() if str(v).strip()})
    natures = ["Receipt", "Payment", "Contra", "Loan", "Capital", "Refund", "Unclassified"]
    groups = ["Income / Receipt", "Expense / Payment", "Contra / Balance Sheet", "Unclassified"]

    notes = pd.DataFrame({
        "Assumption / Method": [
            "Python analysis engine", "Excel reporting engine", "Date parsing", "Duplicate detection",
            "Party grouping", "Transfer comments", "Global rules", "Manual overrides", "Trial balance limitation", "Files analysed",
        ],
        "Details": [
            "PDF extraction, transaction reconstruction, duplicate detection, references, comments and automatic classification are performed in Python.",
            "Dashboard, category, party, category-party, monthly and trial-balance outputs are formula-driven and recalculate after manual overrides.",
            "Bank parsers accept multiple supported date layouts and validate transaction dates against statement periods where available.",
            "Exact and possible duplicate flags are Python-generated review indicators. No transaction is automatically removed from accounting totals or closing-balance reconciliation.",
            "Party names are normalized using regex cleanup and conservative fuzzy/partial matching. Review suggested groupings before finalisation.",
            "NEFT, RTGS and IMPS bank codes, references and narration comments are preserved in dedicated columns.",
            "Phase 1 classification uses deterministic global regex rules. Matched Rule ID, keyword, suggested ledger, confidence and review requirement are preserved for audit traceability.",
            "Enter corrections only in Manual Party, Manual Category, Manual Nature, Manual Group and Manual Review Note. Final columns use Excel formulas.",
            "The Trial Balance is a bank-movement working paper, not a final accounting trial balance. Verify with ledgers, invoices, GST, loans, capital and supporting records.",
            "; ".join(meta.source_file for meta in metas),
        ],
    })

    rules_export = pd.DataFrame([
        {
            "Rule ID": rule["id"],
            "Priority": position + 1,
            "Regex Pattern": rule["pattern"].pattern,
            "Debit Category": rule["debit"],
            "Credit Category": rule["credit"],
            "Suggested Ledger": rule["ledger"],
            "Confidence": float(rule.get("confidence", 90.0)),
            "Review Required": bool(rule.get("review", False)),
            "Group Override": rule.get("group", ""),
        }
        for position, rule in enumerate(GLOBAL_CLASSIFICATION_RULES)
    ])

    with pd.ExcelWriter(output, engine="xlsxwriter", datetime_format="dd-mmm-yyyy", engine_kwargs={"options": {"strings_to_formulas": True}}) as writer:
        workbook = writer.book
        workbook.set_calc_mode("auto")
        header = workbook.add_format({"bold": True, "bg_color": "#1F4E78", "font_color": "white", "border": 1})
        subheader = workbook.add_format({"bold": True, "bg_color": "#D9EAF7", "border": 1})
        money = workbook.add_format({"num_format": "#,##0.00;[Red](#,##0.00)", "border": 1})
        integer = workbook.add_format({"num_format": "0", "border": 1})
        percent = workbook.add_format({"num_format": "0.0%", "border": 1})
        date_format = workbook.add_format({"num_format": "dd-mmm-yyyy", "border": 1})
        input_fmt = workbook.add_format({"bg_color": "#FFF2CC", "border": 1})
        formula_fmt = workbook.add_format({"bg_color": "#E2F0D9", "border": 1})
        text_fmt = workbook.add_format({"border": 1})
        warning_fmt = workbook.add_format({"bg_color": "#FCE4D6", "font_color": "#9C0006", "border": 1})

        # 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)
        duplicate_summary(all_df).to_excel(writer, sheet_name="Duplicate Summary", index=False)

        # Masters first so validation ranges exist.
        max_master = max(len(categories), len(parties), len(natures), len(groups), 1)
        masters = pd.DataFrame({
            "Categories": categories + [""] * (max_master - len(categories)),
            "Parties": parties + [""] * (max_master - len(parties)),
            "Natures": natures + [""] * (max_master - len(natures)),
            "Groups": groups + [""] * (max_master - len(groups)),
        })
        masters.to_excel(writer, sheet_name="Masters", index=False)
        workbook.define_name("CategoryList", f"=Masters!$A$2:$A${len(categories)+1}")
        workbook.define_name("PartyList", f"=Masters!$B$2:$B${max(2, len(parties)+1)}")
        workbook.define_name("NatureList", f"=Masters!$C$2:$C${len(natures)+1}")
        workbook.define_name("GroupList", f"=Masters!$D$2:$D${len(groups)+1}")

        # Transaction classification table with manual inputs and formula-driven final fields.
        tx.to_excel(writer, sheet_name="Transaction Classification", index=False)
        ws_tx = writer.sheets["Transaction Classification"]
        columns = list(tx.columns)
        col_index = {name: i for i, name in enumerate(columns)}
        for row in range(1, len(tx) + 1):
            excel_row = row + 1
            for manual in ("manual_party", "manual_category", "manual_nature", "manual_group", "manual_review_note"):
                ws_tx.write_blank(row, col_index[manual], None, input_fmt)
            formulas = {
                "final_party": f'=IF({ _excel_col(col_index["manual_party"]) }{excel_row}<>"",{ _excel_col(col_index["manual_party"]) }{excel_row},{ _excel_col(col_index["auto_party"]) }{excel_row})',
                "final_category": f'=IF({ _excel_col(col_index["manual_category"]) }{excel_row}<>"",{ _excel_col(col_index["manual_category"]) }{excel_row},{ _excel_col(col_index["auto_category"]) }{excel_row})',
                "final_nature": f'=IF({ _excel_col(col_index["manual_nature"]) }{excel_row}<>"",{ _excel_col(col_index["manual_nature"]) }{excel_row},{ _excel_col(col_index["auto_nature"]) }{excel_row})',
                "final_group": f'=IF({ _excel_col(col_index["manual_group"]) }{excel_row}<>"",{ _excel_col(col_index["manual_group"]) }{excel_row},{ _excel_col(col_index["auto_group"]) }{excel_row})',
            }
            cached = {
                "final_party": str(tx.iloc[row-1].get("auto_party", "")),
                "final_category": str(tx.iloc[row-1].get("auto_category", "")),
                "final_nature": str(tx.iloc[row-1].get("auto_nature", "")),
                "final_group": str(tx.iloc[row-1].get("auto_group", "")),
            }
            for name, formula in formulas.items():
                ws_tx.write_formula(row, col_index[name], formula, formula_fmt, cached[name])
        last_row = max(2, len(tx) + 1)
        ws_tx.data_validation(1, col_index["manual_category"], last_row-1, col_index["manual_category"], {"validate": "list", "source": "=CategoryList"})
        ws_tx.data_validation(1, col_index["manual_party"], last_row-1, col_index["manual_party"], {"validate": "list", "source": "=PartyList"})
        ws_tx.data_validation(1, col_index["manual_nature"], last_row-1, col_index["manual_nature"], {"validate": "list", "source": "=NatureList"})
        ws_tx.data_validation(1, col_index["manual_group"], last_row-1, col_index["manual_group"], {"validate": "list", "source": "=GroupList"})
        if len(tx):
            ws_tx.add_table(0, 0, len(tx), len(columns)-1, {"name": "tblTransactions", "columns": [{"header": c} for c in columns], "style": "Table Style Medium 2"})
        ws_tx.freeze_panes(1, 4)
        ws_tx.set_column(col_index["narration"], col_index["narration"], 60)
        for c in ("transfer_comment", "auto_party", "manual_party", "final_party", "auto_category", "manual_category", "final_category", "matched_rule_id", "matched_keyword", "suggested_ledger", "review_note", "manual_review_note"):
            ws_tx.set_column(col_index[c], col_index[c], 28)
        for c in ("debit", "credit", "balance"):
            ws_tx.set_column(col_index[c], col_index[c], 15, money)
        for c in ("transaction_date", "value_date"):
            ws_tx.set_column(col_index[c], col_index[c], 13, date_format)

        # Formula-driven Dashboard.
        exact_excel_col = _excel_col(list(raw_export.columns).index("exact_duplicate")) if "exact_duplicate" in raw_export.columns else "A"
        possible_excel_col = _excel_col(list(raw_export.columns).index("possible_duplicate")) if "possible_duplicate" in raw_export.columns else "A"
        dashboard_rows = [
            ("Customer / Account Holder", customer, "Extracted or override"),
            ("Account Number", account, "Extracted or override"),
            ("Bank(s)", banks, "Detected/selected parser"),
            ("Financial Year", financial_year or "Not specified", "User input"),
            ("Statements Uploaded", len(metas), "PDF files processed"),
            ("Transactions Analysed", "=ROWS(tblTransactions[row_id])", "All extracted rows retained; duplicate flags are review-only"),
            ("Exact Duplicate Rows", f'=COUNTIF(\'All Extracted Rows\'!${exact_excel_col}:${exact_excel_col},TRUE)', "Python detected"),
            ("Possible Duplicate Rows", f'=COUNTIF(\'All Extracted Rows\'!${possible_excel_col}:${possible_excel_col},TRUE)', "Python detected"),
            ("Combined Opening Balance", opening, "Sum of each statement opening; statement-wise details in reconciliation sheet"),
            ("Total Debit", "=SUM(tblTransactions[debit])", "Formula-driven"),
            ("Total Credit", "=SUM(tblTransactions[credit])", "Formula-driven"),
            ("Computed Closing Balance", "=B10+B12-B11", "Opening + Credits - Debits"),
            ("Combined Statement Closing", statement_closing, "Sum of each statement closing; statement-wise details in reconciliation sheet"),
            ("Closing Difference", "=B13-B14", "Should be zero"),
            ("Review Items", '=COUNTIF(tblTransactions[review_note],"?*")+COUNTIF(tblTransactions[manual_review_note],"?*")', "Formula-driven"),
        ]
        pd.DataFrame(dashboard_rows, columns=["Metric", "Value", "Remarks"]).to_excel(writer, sheet_name="Dashboard", index=False)
        ws_dash = writer.sheets["Dashboard"]
        ws_dash.set_column("A:A", 32)
        ws_dash.set_column("B:B", 24)
        ws_dash.set_column("C:C", 45)
        ws_dash.conditional_format("B15", {"type": "cell", "criteria": "!=", "value": 0, "format": warning_fmt})

        # Category Summary formulas.
        cat_df = pd.DataFrame({"Category": categories})
        cat_df["Debit"] = ""
        cat_df["Credit"] = ""
        cat_df["Net Credit/(Debit)"] = ""
        cat_df["Transaction Count"] = ""
        cat_df["View Transactions"] = ""
        cat_df.to_excel(writer, sheet_name="Category Summary", index=False)
        ws = writer.sheets["Category Summary"]
        for r in range(1, len(cat_df)+1):
            er = r+1
            ws.write_formula(r, 1, f'=SUMIFS(tblTransactions[debit],tblTransactions[final_category],A{er})', money, 0)
            ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[credit],tblTransactions[final_category],A{er})', money, 0)
            ws.write_formula(r, 3, f'=C{er}-B{er}', money, 0)
            ws.write_formula(r, 4, f'=COUNTIF(tblTransactions[final_category],A{er})', integer, 0)
            ws.write_formula(r, 5, '=HYPERLINK("#\'Transaction Classification\'!A1","View / Filter")', text_fmt, "View / Filter")

        if len(cat_df):
            ws.autofilter(0, 0, len(cat_df), len(cat_df.columns) - 1)
        _write_summary_reconciliation(ws, len(cat_df) + 3, opening, statement_closing, money, text_fmt, warning_fmt)

        # Party Summary formulas.
        party_df = pd.DataFrame({"Party / Counterparty": parties})
        for c in ("Debit", "Credit", "Net Credit/(Debit)", "Transaction Count", "First Date", "Last Date", "View Transactions"):
            party_df[c] = ""
        party_df.to_excel(writer, sheet_name="Party Summary", index=False)
        ws = writer.sheets["Party Summary"]
        for r in range(1, len(party_df)+1):
            er = r+1
            ws.write_formula(r, 1, f'=SUMIFS(tblTransactions[debit],tblTransactions[final_party],A{er})', money, 0)
            ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[credit],tblTransactions[final_party],A{er})', money, 0)
            ws.write_formula(r, 3, f'=C{er}-B{er}', money, 0)
            ws.write_formula(r, 4, f'=COUNTIF(tblTransactions[final_party],A{er})', integer, 0)
            ws.write_formula(r, 5, f'=IFERROR(MINIFS(tblTransactions[transaction_date],tblTransactions[final_party],A{er}),"")', date_format, "")
            ws.write_formula(r, 6, f'=IFERROR(MAXIFS(tblTransactions[transaction_date],tblTransactions[final_party],A{er}),"")', date_format, "")
            ws.write_formula(r, 7, '=HYPERLINK("#\'Transaction Classification\'!A1","View / Filter")', text_fmt, "View / Filter")

        if len(party_df):
            ws.autofilter(0, 0, len(party_df), len(party_df.columns) - 1)
        _write_summary_reconciliation(ws, len(party_df) + 3, opening, statement_closing, money, text_fmt, warning_fmt)

        # Category + Party formula-driven summary and trial balance detail.
        pairs = tx[["auto_category", "auto_party"]].drop_duplicates().rename(columns={"auto_category": "Main Category", "auto_party": "Party / Counterparty"}) if len(tx) else pd.DataFrame(columns=["Main Category", "Party / Counterparty"])
        for c in ("Debit", "Credit", "Net Credit/(Debit)", "Transaction Count", "Source Modes", "Review"):
            pairs[c] = ""
        pairs.to_excel(writer, sheet_name="Category Party Summary", index=False)
        ws = writer.sheets["Category Party Summary"]
        for r in range(1, len(pairs)+1):
            er = r+1
            ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[debit],tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', money, 0)
            ws.write_formula(r, 3, f'=SUMIFS(tblTransactions[credit],tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', money, 0)
            ws.write_formula(r, 4, f'=D{er}-C{er}', money, 0)
            ws.write_formula(r, 5, f'=COUNTIFS(tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', integer, 0)
            ws.write_formula(r, 6, f'=IF(E{er}>0,"See Transaction Classification","")', text_fmt, "See Transaction Classification")
            ws.write_formula(r, 7, f'=IF(OR(A{er}="Unclassified / Review",A{er}="Other Bank Receipt / Review",A{er}="Other Bank Payment / Review"),"Review","")', text_fmt, "")
        if len(pairs):
            ws.autofilter(0, 0, len(pairs), len(pairs.columns) - 1)
        _write_summary_reconciliation(ws, len(pairs) + 3, opening, statement_closing, money, text_fmt, warning_fmt)

        # Formula-driven bank movement Trial Balance.
        trial = pd.DataFrame({
            "Main Category": ["Opening Balance"] + list(pairs["Main Category"]) + ["Total Period Movement", "Computed Closing Balance", "Statement Closing Balance", "Difference"],
            "Party / Counterparty": ["Bank Account"] + list(pairs["Party / Counterparty"]) + ["", "Bank Account", "Bank Account", "Reconciliation"],
            "Debit": [""] * (len(pairs)+5),
            "Credit": [""] * (len(pairs)+5),
            "Net Credit/(Debit)": [""] * (len(pairs)+5),
            "Reference": [""] * (len(pairs)+5),
        })
        trial.to_excel(writer, sheet_name="Trial Balance", index=False)
        ws = writer.sheets["Trial Balance"]
        ws.write_number(1, 4, opening, money)
        ws.write(1, 5, "Extracted statement opening signed balance", text_fmt)
        for i in range(len(pairs)):
            r = i + 2
            er = r + 1
            ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[debit],tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', money, 0)
            ws.write_formula(r, 3, f'=SUMIFS(tblTransactions[credit],tblTransactions[final_category],A{er},tblTransactions[final_party],B{er})', money, 0)
            ws.write_formula(r, 4, f'=D{er}-C{er}', money, 0)
            ws.write_formula(r, 5, '=HYPERLINK("#\'Transaction Classification\'!A1","View / Filter")', text_fmt, "View / Filter")
        total_row = len(pairs) + 2
        computed_row = total_row + 1
        statement_row = total_row + 2
        diff_row = total_row + 3
        ws.write_formula(total_row, 2, f'=SUM(C3:C{total_row})', money, 0)
        ws.write_formula(total_row, 3, f'=SUM(D3:D{total_row})', money, 0)
        ws.write_formula(total_row, 4, f'=D{total_row+1}-C{total_row+1}', money, 0)
        ws.write_formula(computed_row, 4, f'=E2+E{total_row+1}', money, opening)
        ws.write_number(statement_row, 4, statement_closing, money)
        ws.write_formula(diff_row, 4, f'=E{computed_row+1}-E{statement_row+1}', money, 0)
        ws.conditional_format(diff_row, 4, diff_row, 4, {"type": "cell", "criteria": "!=", "value": 0, "format": warning_fmt})

        # Formula-driven monthly summary.
        dates = pd.to_datetime(tx.get("transaction_date", pd.Series(dtype="datetime64[ns]")), errors="coerce").dropna()
        months = pd.period_range(dates.min().to_period("M"), dates.max().to_period("M"), freq="M") if not dates.empty else []
        monthly = pd.DataFrame({"Month": [p.to_timestamp() for p in months]})
        for c in ("Opening Balance", "Debit", "Credit", "Net Movement", "Computed Closing", "Transaction Count"):
            monthly[c] = ""
        monthly.to_excel(writer, sheet_name="Monthly Summary", index=False)
        ws = writer.sheets["Monthly Summary"]
        for r in range(1, len(monthly)+1):
            er = r+1
            if r == 1:
                ws.write_number(r, 1, opening, money)
            else:
                ws.write_formula(r, 1, f'=E{er-1}', money, 0)
            ws.write_formula(r, 2, f'=SUMIFS(tblTransactions[debit],tblTransactions[transaction_date],">="&A{er},tblTransactions[transaction_date],"<"&EDATE(A{er},1))', money, 0)
            ws.write_formula(r, 3, f'=SUMIFS(tblTransactions[credit],tblTransactions[transaction_date],">="&A{er},tblTransactions[transaction_date],"<"&EDATE(A{er},1))', money, 0)
            ws.write_formula(r, 4, f'=D{er}-C{er}', money, 0)
            ws.write_formula(r, 5, f'=B{er}+E{er}', money, 0)
            ws.write_formula(r, 6, f'=COUNTIFS(tblTransactions[transaction_date],">="&A{er},tblTransactions[transaction_date],"<"&EDATE(A{er},1))', integer, 0)
            ws.write_datetime(r, 0, pd.Timestamp(monthly.iloc[r-1]["Month"]).to_pydatetime(), workbook.add_format({"num_format": "mmm-yyyy", "border": 1}))

        # Transfer Comments and Review Items derived from Python analysis.
        transfer_cols = [c for c in ["transaction_date", "mode", "transfer_bank_code", "transfer_reference", "transfer_comment", "auto_party", "auto_category", "debit", "credit", "review_note"] if c in tx.columns]
        tx.loc[tx.get("transfer_comment", pd.Series(index=tx.index, dtype=str)).fillna("").ne(""), transfer_cols].to_excel(writer, sheet_name="Transfer Comments", index=False)
        review_mask = tx.get("review_note", pd.Series(index=tx.index, dtype=str)).fillna("").ne("")
        tx.loc[review_mask].to_excel(writer, sheet_name="Review Items", index=False)

        # Retain existing mode and draft sheets for backward compatibility.
        mode_summary(unique_df).to_excel(writer, sheet_name="Mode Summary", index=False)
        draft_financials(unique_df, metas).to_excel(writer, sheet_name="Draft Financials", index=False)

        # Shared formatting.
        for name, worksheet in writer.sheets.items():
            worksheet.freeze_panes(1, 0)
            worksheet.set_row(0, 22, header)
            max_col = max(0, worksheet.dim_colmax)
            if worksheet.dim_rowmax >= 0 and name not in {
                "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)
        for name in ("Category Summary", "Party Summary", "Category Party Summary", "Trial Balance"):
            writer.sheets[name].set_column("A:B", 30)
        writer.sheets["Trial Balance"].set_column("C:E", 18, money)
        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
