excel_engine.py

import io
import json
import os
import re
import shutil
import subprocess
import tempfile
import time
import uuid
import zipfile
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from zoneinfo import ZoneInfo

from openpyxl import load_workbook
import policy_engine
from openpyxl.cell.cell import MergedCell


TEMPLATE_DIR = Path("templates")
CALCULATOR_TEMPLATE_DIR = TEMPLATE_DIR / "Calculator"
TEMPLATE_MANIFEST = Path(os.environ.get("TEMPLATE_MANIFEST", str(TEMPLATE_DIR / "template_manifest.json")))


def load_template_manifest() -> dict:
    try:
        data = json.loads(TEMPLATE_MANIFEST.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, dict) else {}


def template_sort_key(path: Path) -> tuple:
    numbers = tuple(int(part) for part in re.findall(r"\d+", path.stem))
    try:
        modified = path.stat().st_mtime
    except OSError:
        modified = 0
    return numbers, modified, path.name.lower()


def template_candidates(*filenames: str) -> list[Path]:
    paths = []
    for filename in filenames:
        if not filename:
            continue
        candidate = Path(filename)
        if candidate.is_absolute():
            paths.append(candidate)
        else:
            paths.extend([CALCULATOR_TEMPLATE_DIR / filename, TEMPLATE_DIR / filename])
    return paths


def latest_template(*patterns: str) -> Path | None:
    matches = []
    for base_dir in (CALCULATOR_TEMPLATE_DIR, TEMPLATE_DIR):
        if not base_dir.exists():
            continue
        for pattern in patterns:
            matches.extend(path for path in base_dir.glob(pattern) if path.is_file() and "Archive" not in path.parts)
    if not matches:
        return None
    return sorted(set(matches), key=template_sort_key)[-1]


def resolve_template(name: str, filenames: list[str], patterns: list[str] | None = None) -> Path:
    manifest = load_template_manifest()
    override = manifest.get(name)
    if isinstance(override, str) and override.strip():
        override_path = Path(override.strip())
        if not override_path.is_absolute():
            override_path = TEMPLATE_DIR / override_path
        if override_path.exists():
            return override_path

    for path in template_candidates(*filenames):
        if path.exists():
            return path

    discovered = latest_template(*(patterns or []))
    if discovered:
        return discovered

    return template_candidates(filenames[0])[0]


def resolve_latest_template(name: str, filenames: list[str], patterns: list[str]) -> Path:
    """Resolve templates that should automatically follow newly uploaded versions.

    A template_manifest.json override remains highest priority. Without an
    override, the newest matching versioned file is selected before falling back
    to explicit filenames.
    """
    manifest = load_template_manifest()
    override = manifest.get(name)
    if isinstance(override, str) and override.strip():
        override_path = Path(override.strip())
        if not override_path.is_absolute():
            override_path = TEMPLATE_DIR / override_path
        if override_path.exists():
            return override_path

    discovered = latest_template(*patterns)
    if discovered:
        return discovered

    for path in template_candidates(*filenames):
        if path.exists():
            return path

    return template_candidates(filenames[0])[0]


def template_path_from_filename(filename: str) -> Path:
    candidate = Path(str(filename or "").strip())
    if candidate.is_absolute():
        return candidate
    calculator_path = CALCULATOR_TEMPLATE_DIR / candidate
    if calculator_path.exists():
        return calculator_path
    template_path = TEMPLATE_DIR / candidate
    if template_path.exists():
        return template_path
    return calculator_path


def read_master_calculation_filenames(master_path: Path) -> dict[str, Path]:
    try:
        wb = load_workbook(master_path, data_only=True, keep_vba=True, read_only=True)
        ws = wb["Dashboard"]
    except Exception:
        return {}

    filenames = {}
    for row in range(1, min(ws.max_row, 120) + 1):
        for col in range(1, min(ws.max_column, 12) + 1):
            value = ws.cell(row=row, column=col).value
            if not isinstance(value, str) or "calculation file name" not in value.lower():
                continue
            bank_col = col
            file_col = col + 1
            for item_row in range(row + 1, min(ws.max_row, row + 40) + 1):
                bank = str(ws.cell(row=item_row, column=bank_col).value or "").strip()
                filename = str(ws.cell(row=item_row, column=file_col).value or "").strip()
                if bank and filename:
                    filenames[bank] = template_path_from_filename(filename)
            return filenames
    return filenames


def bank_template_path(bank: str, master_path: Path | None = None) -> Path:
    if master_path is not None:
        master_templates = read_master_calculation_filenames(master_path)
        aliases = {
            normalized_bank_label(name): path
            for name, path in master_templates.items()
        }
        master_template = aliases.get(normalized_bank_label(bank))
        if master_template:
            return master_template
    return BANK_CONFIGS[bank]["template"]


MASTER_TEMPLATE = resolve_latest_template(
    "Master",
    ["★EW_UMI_Calculator_v1.0.5.xlsm"],
    ["*EW_UMI_Calculator_v*.xlsm"],
)
ASB_TEMPLATE = resolve_template("ASB", ["ASB UMI Calculator v2.90.xlsx"], ["ASB UMI Calculator v*.xlsx"])

BANK_CONFIGS = {
    "ANZ": {"block": ("B", "C", "D"), "template": resolve_template("ANZ", ["ANZ_Loan_Affordability_Calculator_v11_2_Windows.xlsm"], ["ANZ*Loan*Affordability*Calculator*.xlsm"]), "result_sheet": 0, "metric": "UMI", "value_cell": "O161", "dti_cell": "F161", "status_rule": {"type": "compare_cells", "left": ("master", "Dashboard", "AY37"), "op": "<=", "right": ("bank", 0, "O176")}},
    "ASB": {"block": ("F", "G", "H"), "template": ASB_TEMPLATE, "result_sheet": "Calc", "metric": "UMI", "value_cell": "N20", "dti_cell": "N22", "status_rule": {"type": "threshold", "op": ">=", "value": 120}},
    "BNZ": {"block": ("J", "K", "L"), "template": resolve_template("BNZ", ["BNZ Affordability Calculator v12.34.xlsx", "BNZ_Affordability_Calculator_Broker_Version_12_33.xlsx"], ["BNZ*Affordability*Calculator*.xlsx"]), "result_sheet": 0, "metric": "MBS", "value_cell": "D135", "si_cell": "H132", "dti_cell": "D141", "status_rule": {"type": "threshold", "op": ">=", "value": 100}},
    "Westpac": {"block": ("N", "O", "P"), "template": resolve_template("Westpac", ["20251013 - Westpac Access Serviceability Calculator - October 2025.xlsm"], ["*Westpac*Access*Serviceability*Calculator*.xlsm"]), "result_sheet": 0, "metric": "UMI", "value_cell": "F14", "dti_cell": "D15", "status_rule": {"type": "threshold", "op": ">=", "value": 150}},
    "Kiwibank": {"block": ("R", "S", "T"), "template": resolve_template("Kiwibank", ["Kiwibank_NSR v67 - Adviser - Kainga Ora.xlsx"], ["Kiwibank*NSR*.xlsx"]), "result_sheet": 0, "metric": "NSR", "value_cell": "I136", "dti_cell": "F170", "status_rule": {"type": "cell_threshold", "cell": "F160", "op": "<=", "value": 1}},
    "BOC": {"block": ("V", "W", "X"), "template": resolve_template("BOC", ["Broker Calculator version 1.64.xlsm", "Broker Calculator version 1.62.xlsm"], ["Broker Calculator version*.xlsm"]), "result_sheet": 2, "metric": "UMI", "value_cell": "D46", "dti_cell": "D47", "status_rule": {"type": "threshold", "op": ">=", "value": 100}},
    "Avanti": {"block": ("Z", "AA", "AB"), "template": resolve_template("Avanti", ["Avanti Finance Limited - Borrowing Capacity Calculator v1.11.xlsm"], ["Avanti*Borrowing*Capacity*Calculator*.xlsm"]), "result_sheet": "Borrowing Capacity", "metric": "UMI", "value_cell": "G15", "dti_cell": None, "status_rule": {"type": "threshold", "op": ">=", "value": 0}},
    "Unity": {"block": ("AD", "AE", "AF"), "template": resolve_template("Unity", ["Unity-Loan-Serviceability-Calculator-18 June-2026.xlsx", "Unity-Loan-Serviceability-Calculator-24July-2025.xlsx"], ["Unity*Loan*Serviceability*Calculator*.xlsx"]), "result_sheet": 0, "metric": "NSR", "value_cell": "J98", "dti_cell": None, "status_cell": "J99"},
    "Co-operative": {"block": ("AH", "AI", "AJ"), "template": resolve_template("Co-operative", ["Co_operative_Bank_Servicing_Calculator_v6_9_effective_25_08_2025_2.xlsx"], ["Co*operative*Servicing*Calculator*.xlsx"]), "result_sheet": 0, "metric": "UMI", "value_cell": "E101", "dti_cell": "J101", "status_rule": {"type": "threshold", "op": ">=", "value": 50}},
    "Peppermoney": {"block": ("AL", "AM", "AN"), "template": resolve_template("Peppermoney", ["Peppermoney  BSC v1.7a.xlsm"], ["Peppermoney*BSC*.xlsm"]), "result_sheet": 0, "metric": "NSR", "value_cell": "M75", "dti_cell": None, "status_cell": "O75", "status_rule": {"type": "threshold", "op": ">=", "value": 1}},
    "SBS": {"block": ("AP", "AQ", "AR"), "template": resolve_template("SBS", ["SBS-NSR-Calculator_V16.7(Unprotect).xlsm", "SBS-NSR-Calculator_V16.7.xltm"], ["SBS*NSR*Calculator*.xlsm", "SBS*NSR*Calculator*.xltm"]), "result_sheet": "NSR Calculator", "metric": "NSR", "value_cell": "C149", "dti_cell": "I144", "status_rule": {"type": "threshold", "op": "<=", "value": 0.975}},
    "TSB": {"block": ("AT", "AU", "AV"), "template": resolve_template("TSB", ["TSB Loan Affordability Calculator v6.8 - 09 10 25.xlsx"], ["TSB*Loan*Affordability*Calculator*.xlsx"]), "result_sheet": 0, "metric": "UMI", "value_cell": "D77", "dti_cell": None, "status_rule": {"type": "threshold", "op": ">=", "value": 0}},
}

DATA_DIR = Path("/data") if Path("/data").exists() else Path("data")
CLIENTS_DIR = DATA_DIR / "clients"
JOBS_DIR = DATA_DIR / "jobs"
RETENTION_DAYS = 7
NZ_TZ = ZoneInfo("Pacific/Auckland")

ERROR_VALUES = {"#N/A", "#VALUE!", "#REF!", "#DIV/0!", "#NAME?", "#NUM!", "#NULL!"}
DEFAULT_GOOGLE_SHEET_ID = "1XfWFOiuikbyLldmR7Fyz4cJiBTWHtlRFh2hpLt_mHdo"
EXCLUDED_BANKS = {"Avanti", "TSB"}
BANK_WORKERS = 2
TEST_RATE_CELLS = {
    "ANZ": ("LAC", "Q30"),
    "ASB": ("Calc", "F22"),
    "BNZ": ("Affordability Calculator", "G23"),
    "Westpac": ("Assess Serviceability calc", "J5"),
    "Kiwibank": ("Adviser HL Worksheet", "V7"),
    "BOC": ("Data 2021", "K1"),
    "Co-operative": ("Serviceability Assessment", "C7"),
    "Peppermoney": ("Residential", "O10"),
    "SBS": ("NSR Calculator", "S9"),
    "Unity": ("Calculator", "H13"),
}
GOOGLE_HEADERS = [
    "record_id",
    "saved_at",
    "client_1",
    "client_2",
    "client_3",
    "client_label",
    "loan_amount",
    "property_value",
    "input_json",
]
GOOGLE_SETTINGS_HEADERS = ["key", "updated_at", "settings_json"]

LAST_GOOGLE_SAVE_STATUS = {
    "attempted": False,
    "saved": False,
    "message": "Google Sheet save has not run yet.",
}

FIELD_GROUPS = [
    {"amount": "sal_1", "dependents": ["f1", "v1"]},
    {"amount": "ovt_1", "dependents": ["f2", "v2"]},
    {"amount": "biz_1", "dependents": ["f3", "v3"]},
    {"amount": "gov_1", "dependents": ["f4"]},
    {"amount": "brd_1", "dependents": ["f5", "v5"]},
    {"amount": "rnt_1", "dependents": ["f6"]},
    {"amount": "sal_a2", "dependents": ["f1_a2", "kiwi_a2"]},
    {"amount": "ovt_a2", "dependents": ["f2_a2", "v2_a2"]},
    {"amount": "biz_a2", "dependents": ["f3_a2", "v3_a2"]},
    {"amount": "gov_a2", "dependents": ["f4_a2"]},
    {"amount": "brd_a2", "dependents": ["f5_a2", "v5_a2"]},
    {"amount": "rnt_a2", "dependents": ["f6_a2"]},
    {"amount": "sal_a3", "dependents": ["f1_a3", "kiwi_a3"]},
    {"amount": "ovt_a3", "dependents": ["f2_a3", "v2_a3"]},
    {"amount": "biz_a3", "dependents": ["f3_a3", "v3_a3"]},
    {"amount": "gov_a3", "dependents": ["f4_a3"]},
    {"amount": "rnt_a3", "dependents": ["f6_a3"]},
]

LIABILITY_FIELD_GROUPS = [
    {"amounts": ["m_b", "m_r_val"], "dependents": ["m_bk", "m_f", "m_rt", "m_tr"]},
    {"amounts": ["p_b", "p_r"], "dependents": ["p_ln", "p_f", "p_t"]},
    {"amounts": ["cc_l", "cc_b", "cc_rp"], "dependents": ["cc_bk"]},
    {"amounts": ["sl_b1", "sl_r1", "sl_b2", "sl_r2", "sl_b3", "sl_r3_amt"], "dependents": ["sl_r3"]},
    {"amounts": ["bn_l", "bn_b", "bn_r"], "dependents": ["bn_t_type", "bn_category", "bn_f"]},
]

REQUIRED_DASHBOARD_FORMULAS = {
    "AY209": "=SUBTOTAL(9,AY164:AY208)",
}


def setup_storage() -> None:
    CLIENTS_DIR.mkdir(parents=True, exist_ok=True)
    JOBS_DIR.mkdir(parents=True, exist_ok=True)


def cleanup_old_jobs() -> None:
    setup_storage()
    cutoff = datetime.now() - timedelta(days=RETENTION_DAYS)
    for path in JOBS_DIR.iterdir():
        if not path.is_dir():
            continue
        try:
            if datetime.fromtimestamp(path.stat().st_mtime) < cutoff:
                shutil.rmtree(path, ignore_errors=True)
        except OSError:
            continue


def parse_number(value):
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return value
    text = str(value).replace(",", "").replace("$", "").strip()
    if text == "":
        return None
    try:
        number = float(text)
        return int(number) if number.is_integer() else number
    except ValueError:
        return value


def parse_percent(value):
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return value
    text = str(value).replace("%", "").strip()
    if text == "":
        return None
    try:
        return float(text)
    except ValueError:
        return value


def is_blank_or_zero(value) -> bool:
    parsed = parse_number(value)
    return parsed is None or parsed == 0


def preprocess_inputs(inputs: dict) -> dict:
    cleaned = dict(inputs)
    percent_keys = {"v1", "kiwi_a2", "kiwi_a3"}

    for key, value in list(cleaned.items()):
        if key in percent_keys:
            cleaned[key] = parse_percent(value)
        elif isinstance(value, str) and any(ch.isdigit() for ch in value):
            cleaned[key] = parse_number(value)

    for group in FIELD_GROUPS:
        if is_blank_or_zero(cleaned.get(group["amount"])):
            cleaned.pop(group["amount"], None)
            for dependent in group["dependents"]:
                cleaned.pop(dependent, None)

    for group in LIABILITY_FIELD_GROUPS:
        if all(is_blank_or_zero(cleaned.get(amount)) for amount in group["amounts"]):
            for amount in group["amounts"]:
                cleaned.pop(amount, None)
            for dependent in group["dependents"]:
                cleaned.pop(dependent, None)

    return cleaned


def client_key(name: str) -> str:
    safe = "".join(ch for ch in name.strip() if ch.isalnum() or ch in (" ", "-", "_")).strip()
    return safe or "Client"


def nz_now() -> datetime:
    return datetime.now(NZ_TZ)


def parse_saved_at(value: str) -> datetime | None:
    text = str(value or "").strip()
    if not text:
        return None
    try:
        if text.endswith("Z"):
            text = text[:-1] + "+00:00"
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=NZ_TZ)
    return parsed.astimezone(NZ_TZ)


def saved_at_sort_key(value: str) -> str:
    parsed = parse_saved_at(value)
    return parsed.isoformat() if parsed else str(value or "")


def format_saved_at(value: str) -> str:
    parsed = parse_saved_at(value)
    if not parsed:
        return str(value or "")
    return parsed.strftime("%Y-%m-%d %H:%M NZT")


def google_sheet_id() -> str:
    return os.environ.get("GOOGLE_SHEET_ID", DEFAULT_GOOGLE_SHEET_ID).strip()


def google_script_url() -> str:
    return os.environ.get("GOOGLE_SCRIPT_URL", "").strip()


def google_script_token() -> str:
    return os.environ.get("GOOGLE_SCRIPT_TOKEN", "").strip()


def google_script_enabled() -> bool:
    return bool(google_script_url() and google_script_token())


def google_service_account_info() -> dict | None:
    raw = os.environ.get("GOOGLE_SERVICE_ACCOUNT_JSON", "").strip()
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return None


def google_sheet_enabled() -> bool:
    return google_script_enabled() or bool(google_sheet_id() and google_service_account_info())


def google_sheet_config_status() -> dict:
    info = google_service_account_info()
    return {
        "sheet_id": google_sheet_id(),
        "script_url": google_script_url(),
        "has_script_token": bool(google_script_token()),
        "script_enabled": google_script_enabled(),
        "has_service_account_json": bool(info),
        "client_email": str((info or {}).get("client_email") or ""),
        "enabled": google_sheet_enabled(),
    }


def last_google_save_status() -> dict:
    return dict(LAST_GOOGLE_SAVE_STATUS)


def google_script_request(action: str, payload: dict | None = None) -> dict:
    if not google_script_enabled():
        raise RuntimeError("GOOGLE_SCRIPT_URL or GOOGLE_SCRIPT_TOKEN is missing.")
    body = json.dumps(
        {
            "token": google_script_token(),
            "action": action,
            "payload": payload or {},
        },
        ensure_ascii=False,
        default=str,
    ).encode("utf-8")
    request = Request(
        google_script_url(),
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=20) as response:
            raw = response.read().decode("utf-8")
    except HTTPError as exc:
        raw = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"Apps Script HTTP {exc.code}: {raw}") from exc
    except URLError as exc:
        raise RuntimeError(f"Apps Script connection failed: {exc.reason}") from exc

    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"Apps Script returned non-JSON response: {raw[:300]}") from exc
    if not data.get("ok"):
        raise RuntimeError(str(data.get("error") or "Apps Script request failed."))
    return data


def google_records_worksheet():
    info = google_service_account_info()
    if not info:
        return None
    try:
        import gspread
        from google.oauth2.service_account import Credentials
    except ImportError:
        return None

    scopes = ["https://www.googleapis.com/auth/spreadsheets"]
    credentials = Credentials.from_service_account_info(info, scopes=scopes)
    client = gspread.authorize(credentials)
    spreadsheet = client.open_by_key(google_sheet_id())
    try:
        worksheet = spreadsheet.worksheet("records")
    except gspread.WorksheetNotFound:
        worksheet = spreadsheet.add_worksheet(title="records", rows=1000, cols=len(GOOGLE_HEADERS))

    existing_headers = worksheet.row_values(1)
    if existing_headers[: len(GOOGLE_HEADERS)] != GOOGLE_HEADERS:
        worksheet.update("A1:I1", [GOOGLE_HEADERS])
    return worksheet


def google_settings_worksheet():
    info = google_service_account_info()
    if not info:
        return None
    try:
        import gspread
        from google.oauth2.service_account import Credentials
    except ImportError:
        return None

    scopes = ["https://www.googleapis.com/auth/spreadsheets"]
    credentials = Credentials.from_service_account_info(info, scopes=scopes)
    client = gspread.authorize(credentials)
    spreadsheet = client.open_by_key(google_sheet_id())
    try:
        worksheet = spreadsheet.worksheet("rate")
    except gspread.WorksheetNotFound:
        worksheet = spreadsheet.add_worksheet(title="rate", rows=100, cols=len(GOOGLE_SETTINGS_HEADERS))

    existing_headers = worksheet.row_values(1)
    if existing_headers[: len(GOOGLE_SETTINGS_HEADERS)] != GOOGLE_SETTINGS_HEADERS:
        worksheet.update("A1:C1", [GOOGLE_SETTINGS_HEADERS])
    return worksheet


def google_named_worksheet(title: str, rows: int, cols: int, headers: list[str]):
    info = google_service_account_info()
    if not info:
        return None
    try:
        import gspread
        from google.oauth2.service_account import Credentials
    except ImportError:
        return None

    scopes = ["https://www.googleapis.com/auth/spreadsheets"]
    credentials = Credentials.from_service_account_info(info, scopes=scopes)
    client = gspread.authorize(credentials)
    spreadsheet = client.open_by_key(google_sheet_id())
    try:
        worksheet = spreadsheet.worksheet(title)
    except gspread.WorksheetNotFound:
        worksheet = spreadsheet.add_worksheet(title=title, rows=rows, cols=cols)
    existing_headers = worksheet.row_values(1)
    if existing_headers[: len(headers)] != headers:
        worksheet.update(f"A1:{chr(64 + len(headers))}1", [headers])
    return worksheet


def replace_google_sheet_rows(title: str, headers: list[str], rows: list[list]) -> None:
    worksheet = google_named_worksheet(title, max(len(rows) + 10, 100), len(headers), headers)
    if worksheet is None:
        return
    worksheet.clear()
    worksheet.update(f"A1:{chr(64 + len(headers))}1", [headers])
    if rows:
        worksheet.update(f"A2:{chr(64 + len(headers))}{len(rows) + 1}", rows, value_input_option="USER_ENTERED")


def save_readable_settings_to_google(settings: dict) -> None:
    updated_at = nz_now().strftime("%Y-%m-%d %H:%M NZT")
    replace_google_sheet_rows(
        "rate_matrix",
        ["Lender", "Source", "Product", "KO", "Offset", "Floating Rate", "Fixed Loan 1yr", "Fixed Loan 18 months", "Fixed Loan 2yr", "Fixed Loan 3yr", "updated_at"],
        [
            [
                row.get("Lender", ""),
                row.get("Source", ""),
                row.get("Product", ""),
                row.get("KO", ""),
                row.get("Offset", 0) or 0,
                row.get("Floating Rate", 0) or 0,
                row.get("Fixed Loan 1yr", 0) or 0,
                row.get("Fixed Loan 18 months", 0) or 0,
                row.get("Fixed Loan 2yr", 0) or 0,
                row.get("Fixed Loan 3yr", 0) or 0,
                updated_at,
            ]
            for row in settings.get("rate_matrix", [])
        ],
    )
    replace_google_sheet_rows(
        "pepper_rates",
        ["Lender", "Product Class", "Meaning", "Floating Rate", "Fixed Loan 1yr", "Fixed Loan 18 months", "Fixed Loan 2yr", "Fixed Loan 3yr", "updated_at"],
        [
            [
                row.get("Lender", ""),
                row.get("Product Class", ""),
                row.get("Meaning", ""),
                row.get("Floating Rate", 0) or 0,
                row.get("Fixed Loan 1yr", 0) or 0,
                row.get("Fixed Loan 18 months", 0) or 0,
                row.get("Fixed Loan 2yr", 0) or 0,
                row.get("Fixed Loan 3yr", 0) or 0,
                updated_at,
            ]
            for row in settings.get("specialist_rates", [])
        ],
    )
    replace_google_sheet_rows(
        "cashback",
        ["Lender", "Standard_Pct", "NewBuild_Pct", "Max_Limit", "FHL_Fixed", "FHL_Pct", "updated_at"],
        [
            [bank, row.get("Standard_Pct", 0) or 0, row.get("NewBuild_Pct", 0) or 0, row.get("Max_Limit", 0) or 0, row.get("FHL_Fixed", 0) or 0, row.get("FHL_Pct", 0) or 0, updated_at]
            for bank, row in (settings.get("cashback") or {}).items()
        ],
    )
    replace_google_sheet_rows(
        "lem",
        ["Lender", "80.01-85.00%", "85.01-90.00%", "90.01-95.00%", "updated_at"],
        [
            [bank, row.get("80.01-85.00%", 0) or 0, row.get("85.01-90.00%", 0) or 0, row.get("90.01-95.00%", 0) or 0, updated_at]
            for bank, row in (settings.get("lem") or {}).items()
        ],
    )
    replace_google_sheet_rows(
        "test_rate_cells",
        ["Lender", "Sheet", "Cell", "updated_at"],
        [[bank, value[0] if value else "", value[1] if len(value) > 1 else "", updated_at] for bank, value in (settings.get("test_rate_cells") or {}).items()],
    )
    replace_google_sheet_rows(
        "lvr_policy",
        ["Lender", "Existing", "Investment", "New Build", "Apt", "Work Visa", "FHL", "updated_at"],
        [
            [bank, row.get("Existing", ""), row.get("Investment", ""), row.get("New Build", ""), row.get("Apt", ""), row.get("Work Visa", ""), row.get("FHL", ""), updated_at]
            for bank, row in (settings.get("lvr_policy") or {}).items()
        ],
    )


def save_policy_registry_to_google(registry: dict | None) -> None:
    """Publish the normalized policy registry to editable row-based tabs."""
    data = policy_engine.normalize_registry(registry)
    mappings = [
        ("Policy Packages", policy_engine.PACKAGE_COLUMNS, data["packages"]),
        ("Policy Rules", policy_engine.RULE_COLUMNS, data["rules"]),
        ("Policy Conditions", policy_engine.CONDITION_COLUMNS, data["conditions"]),
        ("Policy Results", policy_engine.RESULT_COLUMNS, data["results"]),
        ("Required Evidence", policy_engine.EVIDENCE_COLUMNS, data["required_evidence"]),
        ("Workflow Escalation", policy_engine.WORKFLOW_COLUMNS, data["workflows"]),
        ("Policy Sources", policy_engine.SOURCE_COLUMNS, data["sources"]),
        ("Pending Changes", policy_engine.PENDING_COLUMNS, data["pending_changes"]),
    ]
    for title, headers, records in mappings:
        replace_google_sheet_rows(title, headers, [[row.get(column, "") for column in headers] for row in records])


def load_policy_registry_from_google() -> dict | None:
    """Load row-based policy tabs when service-account access is available."""
    mappings = [
        ("packages", "Policy Packages", policy_engine.PACKAGE_COLUMNS),
        ("rules", "Policy Rules", policy_engine.RULE_COLUMNS),
        ("conditions", "Policy Conditions", policy_engine.CONDITION_COLUMNS),
        ("results", "Policy Results", policy_engine.RESULT_COLUMNS),
        ("required_evidence", "Required Evidence", policy_engine.EVIDENCE_COLUMNS),
        ("workflows", "Workflow Escalation", policy_engine.WORKFLOW_COLUMNS),
        ("sources", "Policy Sources", policy_engine.SOURCE_COLUMNS),
        ("pending_changes", "Pending Changes", policy_engine.PENDING_COLUMNS),
    ]
    registry = {}
    found = False
    for key, title, headers in mappings:
        worksheet = google_named_worksheet(title, 100, len(headers), headers)
        if worksheet is None:
            return None
        rows = worksheet.get_all_records()
        registry[key] = [{column: row.get(column, "") for column in headers} for row in rows]
        found = found or bool(rows)
    return policy_engine.normalize_registry(registry) if found else None


def apply_readable_settings_from_google(settings: dict) -> dict:
    def records(title: str, headers: list[str]) -> list[dict]:
        worksheet = google_named_worksheet(title, 100, len(headers), headers)
        if worksheet is None:
            return []
        return worksheet.get_all_records()

    rate_rows = records("rate_matrix", ["Lender", "Source", "Product", "KO", "Offset", "Floating Rate", "Fixed Loan 1yr", "Fixed Loan 18 months", "Fixed Loan 2yr", "Fixed Loan 3yr", "updated_at"])
    if rate_rows:
        settings["rate_matrix"] = [
            {key: row.get(key, "") for key in ["Lender", "Source", "Product", "KO", "Offset", "Floating Rate", "Fixed Loan 1yr", "Fixed Loan 18 months", "Fixed Loan 2yr", "Fixed Loan 3yr"]}
            for row in rate_rows
            if str(row.get("Lender") or "").strip() and str(row.get("Product") or "").strip()
        ]
    pepper_rows = records("pepper_rates", ["Lender", "Product Class", "Meaning", "Floating Rate", "Fixed Loan 1yr", "Fixed Loan 18 months", "Fixed Loan 2yr", "Fixed Loan 3yr", "updated_at"])
    if pepper_rows:
        settings["specialist_rates"] = [
            {key: row.get(key, "") for key in ["Lender", "Product Class", "Meaning", "Floating Rate", "Fixed Loan 1yr", "Fixed Loan 18 months", "Fixed Loan 2yr", "Fixed Loan 3yr"]}
            for row in pepper_rows
            if str(row.get("Lender") or "").strip() and str(row.get("Product Class") or "").strip()
        ]
    cashback_rows = records("cashback", ["Lender", "Standard_Pct", "NewBuild_Pct", "Max_Limit", "FHL_Fixed", "FHL_Pct", "updated_at"])
    if cashback_rows:
        settings["cashback"] = {row["Lender"]: {key: row.get(key, 0) for key in ["Standard_Pct", "NewBuild_Pct", "Max_Limit", "FHL_Fixed", "FHL_Pct"]} for row in cashback_rows if str(row.get("Lender") or "").strip()}
    lem_rows = records("lem", ["Lender", "80.01-85.00%", "85.01-90.00%", "90.01-95.00%", "updated_at"])
    if lem_rows:
        settings["lem"] = {row["Lender"]: {key: row.get(key, 0) for key in ["80.01-85.00%", "85.01-90.00%", "90.01-95.00%"]} for row in lem_rows if str(row.get("Lender") or "").strip()}
    test_rows = records("test_rate_cells", ["Lender", "Sheet", "Cell", "updated_at"])
    if test_rows:
        settings["test_rate_cells"] = {row["Lender"]: [row.get("Sheet", ""), row.get("Cell", "")] for row in test_rows if str(row.get("Lender") or "").strip()}
    lvr_rows = records("lvr_policy", ["Lender", "Existing", "Investment", "New Build", "Apt", "Work Visa", "FHL", "updated_at"])
    if lvr_rows:
        settings["lvr_policy"] = {row["Lender"]: {key: row.get(key, "") for key in ["Existing", "Investment", "New Build", "Apt", "Work Visa", "FHL"]} for row in lvr_rows if str(row.get("Lender") or "").strip()}
    return settings


def save_settings_to_google_sheet(settings: dict, key: str = "policy_settings") -> tuple[bool, str]:
    if google_script_enabled():
        try:
            google_script_request("save_settings", {"key": key, "settings": settings})
            try:
                if google_service_account_info():
                    save_policy_registry_to_google(settings.get("policy_registry"))
            except Exception:
                pass
            return True, "Saved to Google Sheet via Apps Script."
        except Exception as exc:
            script_error = str(exc)
        try:
            worksheet = google_settings_worksheet()
            if worksheet is None:
                return False, f"Apps Script failed ({script_error}); service account worksheet was not available."
        except Exception as exc:
            return False, f"Apps Script failed ({script_error}); service account failed ({exc})."
    else:
        worksheet = google_settings_worksheet()
        if worksheet is None:
            return False, "Google Apps Script is not configured and service account worksheet was not available."

    settings_json = json.dumps(settings, ensure_ascii=False, default=str)
    updated_at = nz_now().strftime("%Y-%m-%d %H:%M NZT")
    rows = worksheet.get_all_values()
    target_row = None
    for row_number, row in enumerate(rows[1:], start=2):
        if row and str(row[0]).strip() == key:
            target_row = row_number
            break
    values = [[key, updated_at, settings_json]]
    if target_row:
        worksheet.update(f"A{target_row}:C{target_row}", values)
    else:
        worksheet.append_row(values[0], value_input_option="USER_ENTERED")
    save_readable_settings_to_google(settings)
    save_policy_registry_to_google(settings.get("policy_registry"))
    return True, "Saved to Google Sheet via service account."


def load_settings_from_google_sheet(key: str = "policy_settings") -> tuple[dict | None, str]:
    if google_script_enabled():
        try:
            response = google_script_request("load_settings", {"key": key})
            if response.get("found"):
                loaded = json.loads(response.get("settings_json") or "{}")
                try:
                    row_registry = load_policy_registry_from_google() if google_service_account_info() else None
                    if row_registry:
                        loaded["policy_registry"] = row_registry
                except Exception:
                    pass
                return loaded, "Loaded from Google Sheet rate tab via Apps Script."
            return None, "No settings row found via Apps Script."
        except Exception as exc:
            script_error = str(exc)
        try:
            worksheet = google_settings_worksheet()
            if worksheet is None:
                return None, f"Apps Script failed ({script_error}); service account worksheet was not available."
        except Exception as exc:
            return None, f"Apps Script failed ({script_error}); service account failed ({exc})."
    else:
        worksheet = google_settings_worksheet()
        if worksheet is None:
            return None, "Google Apps Script is not configured and service account worksheet was not available."

    for row in reversed(worksheet.get_all_records()):
        if str(row.get("key") or "").strip() != key:
            continue
        raw = str(row.get("settings_json") or "").strip()
        if not raw:
            continue
        loaded = apply_readable_settings_from_google(json.loads(raw))
        row_registry = load_policy_registry_from_google()
        if row_registry:
            loaded["policy_registry"] = row_registry
        return loaded, "Loaded from Google Sheet via service account."
    return None, "No settings row found in Google Sheet rate tab."


def google_row_to_summary(row: dict) -> dict | None:
    record_id = str(row.get("record_id") or "").strip()
    input_json = str(row.get("input_json") or "").strip()
    if not record_id or not input_json:
        return None
    loan_amount = parse_number(row.get("loan_amount"))
    property_value = parse_number(row.get("property_value"))
    return {
        "id": record_id,
        "label": str(row.get("client_label") or row.get("client_1") or "Client"),
        "saved_at": str(row.get("saved_at") or ""),
        "loan_amount": loan_amount if isinstance(loan_amount, (int, float)) else 0,
        "property_value": property_value if isinstance(property_value, (int, float)) else 0,
        "path": f"google:{record_id}",
    }


def save_client_inputs_to_google(inputs: dict, saved_at: str) -> bool:
    if google_script_enabled():
        summary = client_summary(inputs, saved_at)
        response = google_script_request(
            "save",
            {
                "record": {
                    "record_id": f"{nz_now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}",
                    "saved_at": saved_at,
                    "client_1": str(inputs.get("c_name1") or ""),
                    "client_2": str(inputs.get("c_name2") or ""),
                    "client_3": str(inputs.get("c_name3") or ""),
                    "client_label": summary["label"],
                    "loan_amount": summary["loan_amount"],
                    "property_value": summary["property_value"],
                    "input_json": json.dumps(inputs, ensure_ascii=False, default=str),
                }
            },
        )
        return bool(response.get("saved", True))

    worksheet = google_records_worksheet()
    if worksheet is None:
        return False

    summary = client_summary(inputs, saved_at)
    record_id = f"{nz_now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}"
    row = [
        record_id,
        saved_at,
        str(inputs.get("c_name1") or ""),
        str(inputs.get("c_name2") or ""),
        str(inputs.get("c_name3") or ""),
        summary["label"],
        summary["loan_amount"],
        summary["property_value"],
        json.dumps(inputs, ensure_ascii=False, default=str),
    ]
    worksheet.append_row(row, value_input_option="USER_ENTERED")
    return True


def find_client_matches_in_google(name: str, limit: int = 20) -> list[dict]:
    query = str(name or "").strip().casefold()
    if not query:
        return []
    if google_script_enabled():
        response = google_script_request("search", {"query": name, "limit": limit})
        matches = []
        for row in response.get("records", []):
            summary = google_row_to_summary(row)
            if summary:
                matches.append(summary)
        matches.sort(key=lambda item: saved_at_sort_key(item.get("saved_at", "")), reverse=True)
        return matches[:limit]

    worksheet = google_records_worksheet()
    if worksheet is None:
        return []

    matches = []
    for row in worksheet.get_all_records():
        names = [
            str(row.get("client_1") or ""),
            str(row.get("client_2") or ""),
            str(row.get("client_3") or ""),
            str(row.get("client_label") or ""),
        ]
        if any(query in item.strip().casefold() for item in names):
            summary = google_row_to_summary(row)
            if summary:
                matches.append(summary)
    matches.sort(key=lambda item: saved_at_sort_key(item.get("saved_at", "")), reverse=True)
    return matches[:limit]


def load_client_from_google_record(record_id: str) -> dict | None:
    if google_script_enabled():
        response = google_script_request("load", {"record_id": record_id})
        record = response.get("record") or {}
        input_json = str(record.get("input_json") or "").strip()
        if input_json:
            try:
                return json.loads(input_json)
            except json.JSONDecodeError:
                return None
        inputs = record.get("inputs")
        return inputs if isinstance(inputs, dict) else None

    worksheet = google_records_worksheet()
    if worksheet is None:
        return None
    for row in worksheet.get_all_records():
        if str(row.get("record_id") or "").strip() == record_id:
            try:
                return json.loads(str(row.get("input_json") or "{}"))
            except json.JSONDecodeError:
                return None
    return None


def save_client_inputs(inputs: dict) -> None:
    setup_storage()
    name = client_key(str(inputs.get("c_name1") or inputs.get("client_name") or "Client"))
    folder = CLIENTS_DIR / name
    folder.mkdir(parents=True, exist_ok=True)
    now = nz_now()
    payload = {"saved_at": now.isoformat(timespec="seconds"), "inputs": inputs}
    payload["summary"] = client_summary(inputs, payload["saved_at"])
    timestamp = now.strftime("%Y%m%d_%H%M%S")
    (folder / "latest.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")
    (folder / f"{timestamp}.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")
    LAST_GOOGLE_SAVE_STATUS.update({
        "attempted": True,
        "saved": False,
        "message": "",
    })
    if not google_sheet_enabled():
        config = google_sheet_config_status()
        if not config["script_url"]:
            message = "GOOGLE_SCRIPT_URL secret is missing."
        elif not config["has_script_token"]:
            message = "GOOGLE_SCRIPT_TOKEN secret is missing."
        elif not config["has_service_account_json"]:
            message = "Google Apps Script is not configured, and GOOGLE_SERVICE_ACCOUNT_JSON is missing or invalid."
        elif not config["sheet_id"]:
            message = "GOOGLE_SHEET_ID is missing."
        else:
            message = "Google Sheet is not configured."
        LAST_GOOGLE_SAVE_STATUS.update({"message": message})
        return
    try:
        saved = save_client_inputs_to_google(inputs, payload["saved_at"])
        LAST_GOOGLE_SAVE_STATUS.update({
            "saved": bool(saved),
            "message": "Saved to Google Sheet." if saved else "Google Sheet worksheet was not available.",
        })
    except Exception as exc:
        LAST_GOOGLE_SAVE_STATUS.update({
            "saved": False,
            "message": f"{type(exc).__name__}: {exc}",
        })


def client_summary(inputs: dict, saved_at: str, path: Path | None = None) -> dict:
    names = [str(inputs.get(key) or "").strip() for key in ("c_name1", "c_name2", "c_name3")]
    names = [name for name in names if name]
    loan_amount = parse_number(inputs.get("loan_amt"))
    property_value = parse_number(inputs.get("prop_val"))
    return {
        "id": path.stem if path else "",
        "label": " / ".join(names) or str(inputs.get("client_name") or "Client"),
        "saved_at": saved_at,
        "loan_amount": loan_amount if isinstance(loan_amount, (int, float)) else 0,
        "property_value": property_value if isinstance(property_value, (int, float)) else 0,
        "path": str(path) if path else "",
    }


def find_client_matches(name: str, limit: int = 20) -> list[dict]:
    setup_storage()
    query = str(name or "").strip().casefold()
    if not query:
        return []
    try:
        google_matches = find_client_matches_in_google(name, limit)
        if google_matches:
            return google_matches
    except Exception:
        pass

    matches = []
    for json_path in CLIENTS_DIR.glob("*/*.json"):
        if json_path.name == "latest.json":
            continue
        try:
            payload = json.loads(json_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        inputs = payload.get("inputs", {})
        names = [
            str(inputs.get("c_name1") or ""),
            str(inputs.get("c_name2") or ""),
            str(inputs.get("c_name3") or ""),
            str(inputs.get("client_name") or ""),
        ]
        if any(query in item.strip().casefold() for item in names):
            summary = payload.get("summary") or client_summary(inputs, payload.get("saved_at", ""), json_path)
            summary["path"] = str(json_path)
            summary["id"] = json_path.stem
            matches.append(summary)
    matches.sort(key=lambda item: saved_at_sort_key(item.get("saved_at", "")), reverse=True)
    return matches[:limit]


def load_client_by_path(path: str) -> dict | None:
    if str(path).startswith("google:"):
        try:
            return load_client_from_google_record(str(path).split(":", 1)[1])
        except Exception:
            return None

    json_path = Path(path)
    if not json_path.exists() or not json_path.is_file():
        return None
    try:
        payload = json.loads(json_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    return payload.get("inputs", {})


def saved_client_record_count() -> int:
    setup_storage()
    return sum(1 for path in CLIENTS_DIR.glob("*/*.json") if path.name != "latest.json")


def storage_status() -> dict:
    setup_storage()
    google_config = google_sheet_config_status()
    return {
        "path": str(DATA_DIR),
        "persistent": DATA_DIR == Path("/data"),
        "record_count": saved_client_record_count(),
        "google_sheet_enabled": google_config["enabled"],
        "google_sheet_id": google_config["sheet_id"],
        "google_script_url": google_config["script_url"],
        "google_script_enabled": google_config["script_enabled"],
        "google_has_script_token": google_config["has_script_token"],
        "google_client_email": google_config["client_email"],
        "google_has_service_account_json": google_config["has_service_account_json"],
        "last_google_save": last_google_save_status(),
    }


def create_client_backup_zip() -> bytes:
    setup_storage()
    buffer = io.BytesIO()
    with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        for json_path in CLIENTS_DIR.glob("*/*.json"):
            if json_path.is_file():
                zf.write(json_path, arcname=str(json_path.relative_to(DATA_DIR)))
    return buffer.getvalue()


def restore_client_backup_zip(data: bytes) -> int:
    setup_storage()
    restored = 0
    with zipfile.ZipFile(io.BytesIO(data), "r") as zf:
        for info in zf.infolist():
            path = Path(info.filename)
            if info.is_dir() or path.is_absolute() or ".." in path.parts:
                continue
            parts = path.parts
            if len(parts) >= 3 and parts[0] == "clients":
                relative = Path(*parts[1:])
            else:
                relative = path
            if relative.suffix.lower() != ".json" or len(relative.parts) < 2:
                continue
            target = CLIENTS_DIR / relative
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(zf.read(info))
            restored += 1
    return restored


def load_latest_client(name: str) -> dict | None:
    setup_storage()
    query = str(name or "").strip().casefold()
    if not query:
        return None

    try:
        google_matches = find_client_matches_in_google(name, 1)
        if google_matches:
            return load_client_from_google_record(google_matches[0]["id"])
    except Exception:
        pass

    path = CLIENTS_DIR / client_key(name) / "latest.json"
    if path.exists():
        payload = json.loads(path.read_text(encoding="utf-8"))
        return payload.get("inputs", {})

    newest_match = None
    newest_mtime = 0.0
    for latest_path in CLIENTS_DIR.glob("*/latest.json"):
        try:
            payload = json.loads(latest_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        inputs = payload.get("inputs", {})
        names = [
            str(inputs.get("c_name1") or ""),
            str(inputs.get("c_name2") or ""),
            str(inputs.get("c_name3") or ""),
            str(inputs.get("client_name") or ""),
        ]
        if any(item.strip().casefold() == query for item in names):
            mtime = latest_path.stat().st_mtime
            if mtime >= newest_mtime:
                newest_match = inputs
                newest_mtime = mtime

    return newest_match


def find_soffice() -> str:
    for name in ("soffice", "libreoffice"):
        path = shutil.which(name)
        if path:
            return path
    raise RuntimeError("LibreOffice/soffice was not found.")


def recalc_with_libreoffice(source_file: Path, output_dir: Path, output_format: str = "xlsx") -> Path:
    soffice = find_soffice()
    output_dir.mkdir(parents=True, exist_ok=True)
    completed = subprocess.run(
        [
            soffice,
            "--headless",
            "--convert-to",
            output_format,
            "--outdir",
            str(output_dir),
            str(source_file),
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        timeout=180,
    )
    if completed.returncode != 0:
        raise RuntimeError(completed.stderr or completed.stdout)
    output_file = output_dir / source_file.with_suffix(f".{output_format}").name
    if not output_file.exists():
        raise RuntimeError(f"LibreOffice did not create output file. {completed.stdout or completed.stderr}")
    return output_file


def recalc_many_with_libreoffice(source_files: list[Path], output_dir: Path, output_format: str = "xlsx") -> dict[Path, Path]:
    if not source_files:
        return {}
    soffice = find_soffice()
    output_dir.mkdir(parents=True, exist_ok=True)
    completed = subprocess.run(
        [
            soffice,
            "--headless",
            "--convert-to",
            output_format,
            "--outdir",
            str(output_dir),
            *[str(path) for path in source_files],
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        timeout=max(180, 60 * len(source_files)),
    )
    if completed.returncode != 0:
        raise RuntimeError(completed.stderr or completed.stdout)

    outputs = {}
    for source_file in source_files:
        output_file = output_dir / source_file.with_suffix(f".{output_format}").name
        if output_file.exists():
            outputs[source_file] = output_file
    missing = [path.name for path in source_files if path not in outputs]
    if missing:
        raise RuntimeError(f"LibreOffice did not create output files for: {', '.join(missing)}")
    return outputs


def variable_to_dashboard_cell(master_path: Path) -> dict:
    wb = load_workbook(master_path, data_only=True, keep_vba=True)
    ws = wb["Mapping"]
    mapping = {}
    for row in range(2, ws.max_row + 1):
        key = ws[f"E{row}"].value
        cell = ws[f"F{row}"].value
        if key and cell:
            mapping[str(key)] = str(cell)
    return mapping


MASTER_TEST_RATE_SCAN_COLUMNS = ("BN", "BO", "BP", "BQ")


def normalized_bank_label(value) -> str:
    text = str(value or "").strip().lower()
    text = re.sub(r"[^a-z0-9]+", "", text)
    aliases = {
        "coop": "cooperative",
        "cooperative": "cooperative",
        "thecooperativebank": "cooperative",
        "pepper": "peppermoney",
        "peppermoney": "peppermoney",
        "bankofchina": "boc",
    }
    return aliases.get(text, text)


def write_dashboard_test_rates(ws, test_rates: dict | None) -> int:
    if not isinstance(test_rates, dict) or not test_rates:
        return 0

    rates_by_bank = {}
    for bank, rate in test_rates.items():
        parsed_rate = parse_number(rate)
        if parsed_rate is None:
            continue
        rates_by_bank[normalized_bank_label(bank)] = parsed_rate

    if not rates_by_bank:
        return 0

    written = 0
    for row in range(1, ws.max_row + 1):
        for column in MASTER_TEST_RATE_SCAN_COLUMNS:
            cell = ws[f"{column}{row}"]
            bank_key = normalized_bank_label(cell.value)
            if bank_key not in rates_by_bank:
                continue
            target = ws.cell(row=row, column=cell.column + 1)
            target.value = rates_by_bank[bank_key]
            written += 1
    return written


def write_dashboard_inputs(master_path: Path, inputs: dict) -> None:
    wb = load_workbook(master_path, keep_vba=True)
    ws = wb["Dashboard"]
    mapping = variable_to_dashboard_cell(master_path)

    for key, value in inputs.items():
        cell = mapping.get(key)
        if not cell or value is None or value == "":
            continue
        ws[cell] = value

    student_loan_repayments = [
        ("sl_r1", "sl_f1"),
        ("sl_r2", "sl_f2"),
        ("sl_r3_amt", "sl_f3"),
    ]
    for amount_key, frequency_key in student_loan_repayments:
        cell = mapping.get(amount_key)
        if cell and inputs.get(amount_key) not in (None, ""):
            ws[cell] = monthlyize_engine(inputs.get(amount_key), inputs.get(frequency_key) or "Monthly")
    frequency_cell = mapping.get("sl_r3")
    if frequency_cell:
        ws[frequency_cell] = "Monthly"

    deposit_value = parse_number(inputs.get("dep_val"))
    if deposit_value is None or deposit_value == 0:
        property_value = parse_number(inputs.get("prop_val")) or 0
        loan_amount = parse_number(inputs.get("loan_amt")) or 0
        deposit_value = max(property_value - loan_amount, 0) if property_value and loan_amount else 0
    ws["AY36"] = deposit_value

    for cell, formula in REQUIRED_DASHBOARD_FORMULAS.items():
        ws[cell] = formula

    write_dashboard_test_rates(ws, inputs.get("_test_rates"))

    force_workbook_recalculation(wb)
    wb.save(master_path)


def build_recalculated_master(job_dir: Path, inputs: dict, timestamp: str) -> tuple[Path, Path]:
    master_label = safe_filename_part(f"{MASTER_TEMPLATE.stem}_{client_names_label(inputs)}_{timestamp}")
    working_master = job_dir / f"{master_label}.xlsm"
    shutil.copy2(MASTER_TEMPLATE, working_master)
    write_dashboard_inputs(working_master, inputs)
    recalculated_master = recalc_with_libreoffice(working_master, job_dir / "master_recalculated", "xlsx")
    return working_master, recalculated_master


def should_write_mapping_value(value) -> bool:
    if value is None:
        return False
    if isinstance(value, str) and value.strip() == "":
        return False
    return True


def force_workbook_recalculation(workbook) -> None:
    try:
        workbook.calculation.fullCalcOnLoad = True
        workbook.calculation.forceFullCalc = True
        workbook.calculation.calcMode = "auto"
    except AttributeError:
        pass


ASB_OPTIONAL_ZERO_CELLS = {
    # Optional income fields. ASB formulas often test for blank cells; writing 0
    # makes lookup formulas run with blank type/frequency cells and can produce #N/A.
    "C36", "C39", "C42", "C45", "C50", "C53",
    "F32", "F36", "F39", "F42", "F45", "F50", "F53",
    "I32", "I36", "I39", "I45", "I50", "I53",
    # Optional liability/expense fields where the bank template treats blanks as absent.
    "F78", "F82", "F92", "F93", "F106", "F108", "F110", "F112", "F114",
    "F118", "F120", "F122", "F124", "F130", "F132", "F134", "F136",
}


ASB_OPTIONAL_ZERO_SHEET_CELLS = {
    ("LoanDetails", "B10"), ("LoanDetails", "C10"), ("LoanDetails", "D10"), ("LoanDetails", "E10"),
    ("LoanDetails", "F10"), ("LoanDetails", "H10"), ("LoanDetails", "I10"),
    ("LoanDetails", "B11"), ("LoanDetails", "C11"), ("LoanDetails", "D11"), ("LoanDetails", "E11"),
    ("LoanDetails", "F11"), ("LoanDetails", "H11"), ("LoanDetails", "I11"),
    ("LoanDetails", "B12"), ("LoanDetails", "C12"), ("LoanDetails", "D12"), ("LoanDetails", "E12"),
    ("LoanDetails", "F12"), ("LoanDetails", "G12"),
    ("Buy Now Pay Later", "B15"), ("Buy Now Pay Later", "C15"), ("Buy Now Pay Later", "D15"),
}


def should_write_bank_mapping_value(bank: str, sheet_name: str, target_cell: str, value) -> bool:
    if not should_write_mapping_value(value):
        return False
    if bank == "ASB" and parse_number(value) == 0:
        cell = str(target_cell).upper()
        sheet = str(sheet_name)
        if sheet == "Calc" and cell in ASB_OPTIONAL_ZERO_CELLS:
            return False
        if (sheet, cell) in ASB_OPTIONAL_ZERO_SHEET_CELLS:
            return False
    if bank == "Westpac" and str(sheet_name) == "Assess Serviceability calc" and str(target_cell).upper() == "I7":
        if parse_number(value) == 0:
            return False
    return True


def patch_boc_formula_guards(workbook) -> None:
    guarded_pmt = '=IF(OR($E$5<=0,$E$9<=0,$B$20<=0),0,-PMT($B$20/100/12,$E$9*$E$7,$E$5,,0))'
    for sheet_name in [f"Loan {idx:02d}" for idx in range(1, 9)]:
        if sheet_name in workbook.sheetnames:
            workbook[sheet_name]["D20"] = guarded_pmt


def clear_cells_if_amount_blank(sheet, amount_cell: str, cells: list[str]) -> None:
    if is_blank_or_zero(sheet[amount_cell].value):
        for cell in cells:
            if not isinstance(sheet[cell], MergedCell):
                sheet[cell] = None


def clear_cells(sheet, cells: list[str]) -> None:
    for cell in cells:
        if not isinstance(sheet[cell], MergedCell):
            sheet[cell] = None


def monthlyize_engine(amount, frequency: str) -> float:
    number = parse_number(amount) or 0
    freq = str(frequency or "").strip().lower()
    if freq.startswith("week"):
        return number * 52 / 12
    if freq.startswith("fort"):
        return number * 26 / 12
    if freq.startswith("annual") or freq.startswith("year"):
        return number / 12
    return number


def asb_frequency_value(amount, frequency: str) -> tuple[float, str]:
    freq = str(frequency or "").strip()
    if freq in {"Weekly", "Fortnightly", "Monthly", "Quarterly"}:
        return parse_number(amount) or 0, freq
    return monthlyize_engine(amount, freq), "Monthly"


def lender_cells(lender: str) -> tuple[str, str]:
    lender_name = str(lender or "").strip()
    if lender_name == "ASB":
        return "Yes", ""
    return "No", lender_name


def has_item_value(item: dict, amount_keys: tuple[str, ...]) -> bool:
    return bool(str(item.get("lender") or item.get("issuer") or item.get("facility") or "").strip()) or any(
        parse_number(item.get(key)) not in (None, 0) for key in amount_keys
    )


def patch_asb_itemized_liabilities(workbook, inputs: dict) -> None:
    if "LoanDetails" in workbook.sheetnames:
        ws = workbook["LoanDetails"]
        for row in range(10, 51):
            for col in "BCDEFGHIJK":
                if not isinstance(ws[f"{col}{row}"], MergedCell):
                    ws[f"{col}{row}"] = None

        loan_rows = []
        for item in inputs.get("_mortgage_items") or []:
            if not isinstance(item, dict) or not has_item_value(item, ("balance", "repayment")):
                continue
            repayment, frequency = asb_frequency_value(item.get("repayment"), item.get("frequency"))
            asb_facility, other_lender = lender_cells(item.get("lender"))
            loan_rows.append({
                "loan_type": "Home Loan",
                "revolving": "No",
                "asb_facility": asb_facility,
                "other_lender": other_lender,
                "balance": parse_number(item.get("balance")) or 0,
                "limit": None,
                "repayment": repayment,
                "frequency": frequency,
                "term": parse_number(item.get("term")) or None,
            })
        for item in inputs.get("_personal_items") or []:
            if not isinstance(item, dict) or not has_item_value(item, ("balance", "repayment")):
                continue
            repayment, frequency = asb_frequency_value(item.get("repayment"), item.get("frequency"))
            asb_facility, other_lender = lender_cells(item.get("lender"))
            loan_rows.append({
                "loan_type": "Personal Loan",
                "revolving": "No",
                "asb_facility": asb_facility,
                "other_lender": other_lender,
                "balance": parse_number(item.get("balance")) or 0,
                "limit": None,
                "repayment": repayment,
                "frequency": frequency,
                "term": parse_number(item.get("term")) or None,
            })
        for item in inputs.get("_credit_items") or []:
            if not isinstance(item, dict) or not has_item_value(item, ("balance", "limit")):
                continue
            liability_type = str(item.get("type") or "Credit Card")
            asb_facility, other_lender = lender_cells(item.get("issuer"))
            if liability_type == "Overdraft / Revolving":
                loan_type = "Home Loan"
                revolving = "Yes"
            else:
                loan_type = "Credit Card"
                revolving = None
            loan_rows.append({
                "loan_type": loan_type,
                "revolving": revolving,
                "asb_facility": asb_facility,
                "other_lender": other_lender,
                "balance": parse_number(item.get("balance")) or 0,
                "limit": parse_number(item.get("limit")) or 0,
                "repayment": None,
                "frequency": None,
                "term": None,
            })

        for offset, item in enumerate(loan_rows[:41]):
            row = 10 + offset
            ws[f"B{row}"] = item["loan_type"]
            if item.get("revolving") is not None:
                ws[f"C{row}"] = item["revolving"]
            ws[f"D{row}"] = item["asb_facility"]
            ws[f"E{row}"] = item["other_lender"]
            if item.get("balance") not in (None, 0):
                ws[f"F{row}"] = item["balance"]
            if item.get("limit") not in (None, 0):
                ws[f"G{row}"] = item["limit"]
            if item.get("repayment") not in (None, 0):
                ws[f"H{row}"] = item["repayment"]
            if item.get("frequency"):
                ws[f"I{row}"] = item["frequency"]
            if item.get("term") not in (None, 0):
                ws[f"J{row}"] = item["term"]

    if "Buy Now Pay Later" in workbook.sheetnames:
        ws = workbook["Buy Now Pay Later"]
        for row in range(15, 56):
            for col in "BCDE":
                if not isinstance(ws[f"{col}{row}"], MergedCell):
                    ws[f"{col}{row}"] = None
        for offset, item in enumerate((inputs.get("_bnpl_items") or [])[:41]):
            if not isinstance(item, dict) or not has_item_value(item, ("limit", "balance", "repayment")):
                continue
            row = 15 + offset
            ws[f"B{row}"] = item.get("facility") or None
            limit = parse_number(item.get("limit")) or 0
            if limit:
                ws[f"C{row}"] = limit
            ws[f"D{row}"] = item.get("category") or "Household"


def patch_westpac_blank_optional_rows(workbook) -> None:
    sheet_name = "Assess Serviceability calc"
    if sheet_name not in workbook.sheetnames:
        return
    ws = workbook[sheet_name]
    if is_blank_or_zero(ws["I7"].value):
        clear_cells(ws, ["I7"])
    for row in list(range(3, 20)) + list(range(21, 24)) + list(range(25, 43)):
        clear_cells_if_amount_blank(ws, f"Q{row}", [f"Q{row}", f"R{row}", f"S{row}"])

    if "Workings" not in workbook.sheetnames:
        return
    workings = workbook["Workings"]
    for row in list(range(166, 174)) + list(range(176, 184)):
        workings[f"H{row}"] = f'=IF(OR(E{row}="",E{row}=0,F{row}=""),0,E{row}*VLOOKUP(F{row},$B$21:$D$27,2,FALSE))'
        workings[f"I{row}"] = f'=IF(E{row}=0,0,IF($G{row}="Gross",$H{row},IFERROR($H{row}/VLOOKUP(H$174,$B$293:$G$24294,6),0)))'
        workings[f"J{row}"] = f'=IF(E{row}=0,0,IF($G{row}="Net",$H{row},IFERROR($H{row}*VLOOKUP(H$174,$F$293:$G$24294,2),0)))'
        workings[f"K{row}"] = f'=IF(OR(G{row}=0,G{row}=""),0,IFERROR(J{row}*VLOOKUP($B{row},$B$66:$C$76,2,FALSE)/12,0))'
    for row in range(186, 189):
        workings[f"H{row}"] = f'=IF(OR(E{row}="",E{row}=0,F{row}=""),0,E{row}*VLOOKUP(F{row},$B$21:$D$27,2,FALSE))'
        workings[f"I{row}"] = f'=IF(E{row}=0,0,H{row}/12)'
        workings[f"J{row}"] = f'=IF(E{row}=0,0,IFERROR(H{row}*VLOOKUP(B{row},B$66:C$75,2,FALSE),0))'
        workings[f"K{row}"] = f'=IF(E{row}=0,0,J{row}/12)'
    for row in list(range(191, 198)) + list(range(200, 206)) + list(range(208, 211)):
        workings[f"H{row}"] = f'=IF(OR(F{row}="",F{row}=0,G{row}=""),0,F{row}*VLOOKUP(G{row},$B$21:$D$27,2,FALSE)/12)'


def patch_asb_blank_optional_rows(workbook) -> None:
    if "Calc" in workbook.sheetnames:
        ws = workbook["Calc"]
        for col in ("C", "F", "I", "L"):
            for start_row in (34, 37, 40, 43):
                clear_cells_if_amount_blank(
                    ws,
                    f"{col}{start_row + 2}",
                    [f"{col}{start_row}", f"{col}{start_row + 1}", f"{col}{start_row + 2}"],
                )

    if "LoanDetails" in workbook.sheetnames:
        ws = workbook["LoanDetails"]
        for row in (10, 11, 12):
            numeric_cells = [f"F{row}", f"G{row}", f"H{row}"]
            if all(is_blank_or_zero(ws[cell].value) for cell in numeric_cells):
                for col in "BCDEFGHIJKL":
                    ws[f"{col}{row}"] = None

    if "Buy Now Pay Later" in workbook.sheetnames:
        ws = workbook["Buy Now Pay Later"]
        if is_blank_or_zero(ws["C15"].value):
            for cell in ("B15", "C15", "D15"):
                ws[cell] = None


def patch_bnz_blank_optional_rows(workbook) -> None:
    sheet_name = "Affordability Calculator"
    if sheet_name not in workbook.sheetnames:
        return
    ws = workbook[sheet_name]

    for row in (32, 34, 40, 44, 46, 53, 59, 60):
        clear_cells_if_amount_blank(ws, f"G{row}", [f"B{row}", f"E{row}", f"G{row}", f"H{row}"])

    for row in (80, 90):
        balance_cell = "B80" if row == 80 else "C90"
        if is_blank_or_zero(ws[balance_cell].value):
            for col in "BCDEFGIJK":
                cell = f"{col}{row}"
                if not isinstance(ws[cell], MergedCell):
                    ws[cell] = None

    for row in range(99, 109):
        clear_cells_if_amount_blank(ws, f"F{row}", [f"B{row}", f"F{row}", f"G{row}"])

    for row in range(109, 115):
        if is_blank_or_zero(ws[f"E{row}"].value):
            for col in "EFG":
                cell = f"{col}{row}"
                if not isinstance(ws[cell], MergedCell):
                    ws[cell] = None


def patch_sbs_blank_optional_rows(workbook) -> None:
    sheet_name = "NSR Calculator"
    if sheet_name not in workbook.sheetnames:
        return
    ws = workbook[sheet_name]

    for cell in ("M23", "R23"):
        if is_blank_or_zero(ws[cell].value):
            clear_cells(ws, [cell])

    income_groups = [
        ("D", "F"),
        ("I", "K"),
        ("N", "P"),
        ("S", None),
    ]
    for amount_col, ks_col in income_groups:
        for row in (27, 30, 33, 36, 39, 42, 45, 48, 50, 52):
            cells = [f"{amount_col}{row}"]
            if ks_col:
                cells.append(f"{ks_col}{row}")
            clear_cells_if_amount_blank(ws, f"{amount_col}{row}", cells)

    for row in (72, 77, 82, 88, 98, 107, 116, 125, 127):
        for col in ("C", "D", "I", "M", "N"):
            cell = f"{col}{row}"
            if cell in ws and is_blank_or_zero(ws[cell].value):
                clear_cells(ws, [cell])


def patch_unity_blank_optional_rows(workbook) -> None:
    sheet_name = "Calculator"
    if sheet_name not in workbook.sheetnames:
        return
    ws = workbook[sheet_name]

    if is_new_unity_layout(workbook):
        for row in (35, 36, 37, 38, 39, 40, 41):
            for amount_col, freq_col in (("G", "I"), ("H", "I"), ("J", "L"), ("K", "L")):
                clear_cells_if_amount_blank(ws, f"{amount_col}{row}", [f"{amount_col}{row}"])
            for freq_col in ("I", "L"):
                if is_blank_or_zero(ws[f"{freq_col}{row}"].value):
                    clear_cells(ws, [f"{freq_col}{row}"])

        for row in range(58, 73):
            clear_cells_if_amount_blank(ws, f"H{row}", [f"H{row}", f"I{row}"])

        for row in (85, 86, 87, 88):
            clear_cells_if_amount_blank(ws, f"H{row}", [f"H{row}"])
        for row in (89, 90, 91, 92):
            clear_cells_if_amount_blank(ws, f"H{row}", [f"H{row}", f"I{row}", f"J{row}"])
        return

    for income_col, helper_cols in (("H", ["H"]), ("I", ["I"])):
        clear_cells_if_amount_blank(ws, f"{income_col}34", [f"{income_col}34", f"{income_col}35", f"{income_col}36"])
    for row in range(45, 50):
        clear_cells_if_amount_blank(ws, f"H{row}", [f"H{row}"])

    for row in range(59, 74):
        clear_cells_if_amount_blank(ws, f"H{row}", [f"H{row}", f"I{row}"])
        ws[f"J{row}"] = f'=IF(OR(H{row}="",H{row}=0,I{row}=""),0,H{row}*VLOOKUP(I{row},Lookups!$A$43:$B$47,2,0)/12)'

    for row in range(87, 95):
        clear_cells_if_amount_blank(ws, f"H{row}", [f"H{row}", f"I{row}"])


def is_new_unity_layout(workbook) -> bool:
    if "Calculator" not in workbook.sheetnames:
        return False
    ws = workbook["Calculator"]
    marker = str(ws["I101"].value or ws["J101"].value or "")
    status_formula = str(ws["J99"].value or "")
    return "2026" in marker or "J98" in status_formula or str(ws["C57"].value or "").strip().upper() == "LIVING EXPENSES"


def apply_bank_mapping(recalculated_master: Path, bank: str, template_path: Path, output_path: Path, inputs: dict | None = None) -> int:
    sheet_col, cell_col, value_col = BANK_CONFIGS[bank]["block"]
    master = load_workbook(recalculated_master, data_only=True)
    dashboard = master["Dashboard"]
    target = load_workbook(template_path, keep_vba=template_path.suffix.lower() == ".xlsm")
    applied = 0

    for row in range(26, dashboard.max_row + 1):
        sheet_name = dashboard[f"{sheet_col}{row}"].value
        target_cell = dashboard[f"{cell_col}{row}"].value
        value = dashboard[f"{value_col}{row}"].value
        if not sheet_name or not target_cell or str(target_cell).startswith("="):
            continue
        if not should_write_bank_mapping_value(bank, str(sheet_name), str(target_cell), value):
            continue
        if sheet_name not in target.sheetnames:
            continue
        try:
            target[str(sheet_name)][str(target_cell)] = value
            applied += 1
        except Exception:
            continue

    if bank == "BOC":
        patch_boc_formula_guards(target)
    if bank == "Westpac":
        patch_westpac_blank_optional_rows(target)
    if bank == "ASB":
        patch_asb_itemized_liabilities(target, inputs or {})
        patch_asb_blank_optional_rows(target)
    if bank == "BNZ":
        patch_bnz_blank_optional_rows(target)
    if bank == "SBS":
        patch_sbs_blank_optional_rows(target)
    if bank == "Unity":
        patch_unity_blank_optional_rows(target)

    force_workbook_recalculation(target)
    target.save(output_path)
    return applied


def normalize_excel_value(value):
    if isinstance(value, str) and value.strip().upper() in ERROR_VALUES:
        return value.strip().upper()
    return value


def get_sheet(workbook, sheet_ref):
    if isinstance(sheet_ref, int):
        return workbook.worksheets[sheet_ref]
    return workbook[str(sheet_ref)]


def compare_values(left, op: str, right) -> bool:
    left_num = parse_number(left)
    right_num = parse_number(right)
    if not isinstance(left_num, (int, float)) or not isinstance(right_num, (int, float)):
        return False
    if op == ">=":
        return left_num >= right_num
    if op == "<=":
        return left_num <= right_num
    if op == ">":
        return left_num > right_num
    if op == "<":
        return left_num < right_num
    if op == "=":
        return left_num == right_num
    return False


def resolve_rule_value(rule_ref, bank_wb, master_wb):
    source, sheet_ref, cell = rule_ref
    wb = master_wb if source == "master" else bank_wb
    return get_sheet(wb, sheet_ref)[cell].value


def normalize_rate_percent(value):
    normalized = normalize_excel_value(value)
    number = parse_number(normalized)
    if isinstance(number, (int, float)):
        return number * 100 if 0 < number < 1 else number
    return normalized


def read_test_rate(bank: str, bank_wb):
    sheet_name, cell = TEST_RATE_CELLS.get(bank, (None, None))
    if not sheet_name or not cell:
        return None
    try:
        sheet = get_sheet(bank_wb, sheet_name)
    except Exception:
        return None
    return normalize_rate_percent(sheet[cell].value)


def status_for_bank(bank: str, value, bank_wb, master_wb) -> str:
    config = BANK_CONFIGS[bank]
    status_cell = config.get("status_cell")
    if status_cell:
        status_value = get_sheet(bank_wb, config["result_sheet"])[status_cell].value
        text = str(status_value or "").strip().upper()
        if "PASS" in text:
            return "PASS"
        if "FAIL" in text:
            return "FAIL"

    rule = config.get("status_rule", {})
    if rule.get("type") == "threshold":
        return "PASS" if compare_values(value, rule["op"], rule["value"]) else "FAIL"
    if rule.get("type") == "cell_threshold":
        cell_value = get_sheet(bank_wb, config["result_sheet"])[rule["cell"]].value
        return "PASS" if compare_values(cell_value, rule["op"], rule["value"]) else "FAIL"
    if rule.get("type") == "compare_cells":
        left = resolve_rule_value(rule["left"], bank_wb, master_wb)
        right = resolve_rule_value(rule["right"], bank_wb, master_wb)
        return "PASS" if compare_values(left, rule["op"], right) else "FAIL"
    return "FAIL"


def policy_number(value) -> float | None:
    parsed = parse_number(value)
    return float(parsed) if isinstance(parsed, (int, float)) else None


FREQ_TO_ANNUAL = {"Weekly": 52, "Fortnightly": 26, "Monthly": 12, "Annually": 1}


def annualized_policy_amount(inputs: dict, amount_key: str, freq_key: str | None = None, *, default_annual: bool = False) -> float:
    amount = policy_number(inputs.get(amount_key)) or 0.0
    if default_annual:
        return amount
    frequency = str(inputs.get(freq_key or "") or "Annually")
    return amount * FREQ_TO_ANNUAL.get(frequency, 1)


def income_mix(inputs: dict) -> dict:
    applicant_fields = [
        ("_1", "f1", "f2", "f4", "f5", "f6"),
        ("_a2", "f1_a2", "f2_a2", "f4_a2", "f5_a2", "f6_a2"),
        ("_a3", "f1_a3", "f2_a3", "f4_a3", "f5_a3", "f6_a3"),
    ]
    total = 0.0
    government = 0.0

    for suffix, salary_freq, overtime_freq, government_freq, boarder_freq, rental_freq in applicant_fields:
        total += annualized_policy_amount(inputs, f"sal{suffix}", salary_freq)
        total += annualized_policy_amount(inputs, f"ovt{suffix}", overtime_freq)
        total += annualized_policy_amount(inputs, f"biz{suffix}", default_annual=True)
        gov = annualized_policy_amount(inputs, f"gov{suffix}", government_freq)
        government += gov
        total += gov
        total += annualized_policy_amount(inputs, f"brd{suffix}", boarder_freq)
        total += annualized_policy_amount(inputs, f"rnt{suffix}", rental_freq)

    return {
        "total_annual_income": total,
        "government_annual_income": government,
        "government_is_majority": government > 0 and total > 0 and government / total >= 0.5,
    }


def normalize_deal_type(value: str, legacy_property_type: str = "Existing") -> str:
    text = str(value or "").strip()
    if not text:
        return {
            "New Build": "Construction",
            "Apt": "Apartment",
        }.get(legacy_property_type, "Existing property")
    return {
        "Not selected yet - pre-approval": "Pre-approval - property not selected",
        "Purchase - Existing home": "Existing property",
        "Purchase - New build / turnkey": "Off-plan turnkey",
        "Completed new build - waiting for CCC": "Completed turnkey",
        "Completed new build - CCC issued": "Completed turnkey",
        "Completed new build - waiting CCC/title": "Completed turnkey",
        "Completed new build - CCC/title issued": "Completed turnkey",
        "Construction loan": "Construction",
        "Refinance - no extra borrowing": "Dollar-for-dollar refinance",
        "Refinance - extra borrowing or debt consolidation": "Refinance with cash-out / debt consolidation",
        "Labour-only construction": "Construction",
    }.get(text, text)


def policy_context(inputs: dict) -> dict:
    loan_amount = policy_number(inputs.get("loan_amt") or inputs.get("loan_amount")) or 0.0
    property_value = policy_number(inputs.get("prop_val") or inputs.get("property_value")) or 0.0
    lvr = policy_number(inputs.get("lvr"))
    if lvr is None:
        lvr = loan_amount / property_value * 100 if property_value else 0.0

    legacy_property_type = str(inputs.get("prop_type") or inputs.get("property_type") or "Existing")
    purpose = str(inputs.get("loan_purpose") or "").strip()
    if not purpose:
        purpose = "Investor" if legacy_property_type == "Investment" else "Owner Occupied"

    deal_type = normalize_deal_type(str(inputs.get("deal_type") or "").strip(), legacy_property_type)

    relationship = str(inputs.get("existing_bank_relationship") or "None").strip()
    existing_banks = set()
    if relationship in {"ASB", "BNZ"}:
        existing_banks.add(relationship)
    elif relationship == "Both":
        existing_banks.update({"ASB", "BNZ"})

    mix = income_mix(inputs)
    return {
        "lvr": lvr,
        "purpose": purpose,
        "deal_type": deal_type,
        "existing_banks": existing_banks,
        "pre_approval": bool(inputs.get("pre_approval")),
        "kainga_ora": str(inputs.get("fhl_YN") or inputs.get("fhl") or "No").strip().lower() in {"yes", "y", "true", "1"},
        **mix,
    }


def set_policy_result(
    result: dict,
    *,
    status: str,
    metric: str,
    value,
    threshold: str,
    rule: str,
    note: str = "",
    detail: str = "",
) -> dict:
    result.update({
        "status": status,
        "metric": metric,
        "value": value,
        "criteria_threshold": threshold,
        "criteria_rule": rule,
        "criteria_note": note,
        "criteria_detail": detail,
    })
    return result


def policy_value_label(metric: str, value) -> str:
    if value is None:
        return "not available"
    if metric in {"UMI", "MBS", "SI"}:
        return f"${value:,.0f}" if metric in {"UMI", "MBS"} else f"{value:,.0f} SI"
    return f"{value:.3f}"


def threshold_result_detail(metric: str, value, op: str, threshold: float, *, label: str | None = None) -> str:
    name = label or metric
    if value is None:
        return f"{name} result was not available, so this servicing check could not be verified."

    passed = compare_values(value, op, threshold)
    threshold_label = policy_value_label(metric, threshold)
    value_label = policy_value_label(metric, value)
    is_money = metric in {"UMI", "MBS"}

    if op == ">=":
        if passed:
            buffer = value - threshold
            extra = f" with ${buffer:,.0f} buffer" if is_money and buffer > 0 else ""
            return f"Current {name} is {value_label}, meeting the required minimum {threshold_label}{extra}."
        shortfall = threshold - value
        gap = f"${shortfall:,.0f}" if is_money else f"{shortfall:.3f}"
        return f"Current {name} is {value_label}, which is {gap} below the required minimum {threshold_label}."

    if op == "<=":
        if passed:
            return f"Current {name} is {value_label}, within the allowed maximum {threshold_label}."
        excess = value - threshold
        gap = f"${excess:,.0f}" if is_money else f"{excess:.3f}"
        return f"Current {name} is {value_label}, which is {gap} above the allowed maximum {threshold_label}."

    if op == "<":
        if passed:
            return f"Current {name} is {value_label}, meeting the requirement to be below {threshold_label}."
        excess = value - threshold
        gap = f"${excess:,.0f}" if is_money else f"{excess:.3f}"
        return f"Current {name} is {value_label}, which is {gap} above the required level below {threshold_label}."

    if op == ">":
        if passed:
            return f"Current {name} is {value_label}, meeting the requirement to be above {threshold_label}."
        shortfall = threshold - value
        gap = f"${shortfall:,.0f}" if is_money else f"{shortfall:.3f}"
        return f"Current {name} is {value_label}, which is {gap} below the required level above {threshold_label}."

    if passed:
        return f"Current {name} is {value_label}, meeting the required value {threshold_label}."
    return f"Current {name} is {value_label}, but the required value is {threshold_label}."


def minimum_result_detail(metric: str, value, minimum: float, *, label: str | None = None) -> str:
    return threshold_result_detail(metric, value, ">=", minimum, label=label)


def generic_criteria_detail(bank: str, result: dict) -> str:
    config = BANK_CONFIGS.get(bank, {})
    status = str(result.get("status") or "FAIL").upper()
    metric = result.get("metric", config.get("metric", "Result"))
    value = policy_number(result.get("value"))
    rule = config.get("status_rule", {})

    if config.get("status_cell"):
        return (
            f"The {bank} calculator returned {status}. "
            "This lender uses its own calculator status, so check the generated workbook if you need the detailed internal breakdown."
        )

    if rule.get("type") == "threshold":
        return threshold_result_detail(metric, value, rule.get("op", ">="), float(rule.get("value", 0)))

    if rule.get("type") == "cell_threshold":
        check_value = policy_number(result.get("status_check_value"))
        check_metric = "calculator ratio"
        return threshold_result_detail(check_metric, check_value, rule.get("op", "<="), float(rule.get("value", 0)))

    if rule.get("type") == "compare_cells":
        left = policy_number(result.get("status_check_left"))
        right = policy_number(result.get("status_check_right"))
        if left is None or right is None:
            return f"The {bank} calculator could not provide enough values to explain the servicing result."
        if compare_values(left, rule.get("op", "<="), right):
            buffer = right - left
            return f"The {bank} calculator shows available servicing of ${right:,.0f} against required servicing of ${left:,.0f}, leaving ${buffer:,.0f} buffer."
        shortfall = left - right
        return f"The {bank} calculator shows required servicing of ${left:,.0f} but available servicing of ${right:,.0f}, leaving a ${shortfall:,.0f} shortfall."

    if status == "PASS":
        return f"The {bank} calculator result meets the servicing requirement."
    return f"The {bank} calculator result does not meet the servicing requirement."


def evaluate_asb_criteria(result: dict, context: dict) -> dict:
    umi = policy_number(result.get("value"))
    dti = policy_number(result.get("dti"))
    lvr = context["lvr"]
    investor = context["purpose"] == "Investor"
    existing = "ASB" in context["existing_banks"]
    pre_approval = context["pre_approval"]
    kainga_ora = context["kainga_ora"]

    high_lvr = lvr > 80
    dti_limit = 7 if investor else 6
    high_dti = dti is not None and dti > dti_limit

    if kainga_ora and high_lvr:
        threshold = 750 if dti is not None and dti > 7 else 200
        rule = f"ASB RBNZ-exempt / Kainga Ora LVR >80% and DTI {'>7' if dti is not None and dti > 7 else '<=7'}"
        return set_policy_result(
            result,
            status="PASS" if umi is not None and umi >= threshold else "FAIL",
            metric="UMI",
            value=umi,
            threshold=f">= ${threshold:,.0f}",
            rule=rule,
            note="ASB Home Happenings Update dated 13 July 2026: RBNZ high-LVR/DTI exemptions, including Kainga Ora First Home Loans, are unchanged.",
            detail=minimum_result_detail("UMI", umi, threshold),
        )

    if investor:
        if lvr > 70:
            return set_policy_result(
                result,
                status="REVIEW",
                metric="UMI",
                value=umi,
                threshold="ASB Traffic Light confirmation required",
                rule="ASB Investor LVR >70%",
                note="The supplied text does not include the Investor high-LVR traffic-light table. Cashback eligibility is not a lending approval criterion.",
                detail=f"Investor LVR is {lvr:.2f}%. ASB investor lending above 70% LVR needs traffic-light availability confirmation before the result can be treated as a clear pass.",
            )
        threshold = 120 if high_dti else 0
        rule = f"ASB Investor {'High DTI > 7' if high_dti else 'Standard'}"
    else:
        threshold = 300 if high_lvr else (120 if high_dti else 0)
        if high_lvr and high_dti:
            rule = "ASB Owner Occupied High LVR & High DTI"
        elif high_lvr:
            rule = "ASB Owner Occupied High LVR > 80%"
        elif high_dti:
            rule = "ASB Owner Occupied High DTI > 6"
        else:
            rule = "ASB Owner Occupied Standard"

        if pre_approval and high_lvr and not existing:
            return set_policy_result(
                result,
                status="INELIGIBLE",
                metric="UMI",
                value=umi,
                threshold=f">= ${threshold:,.0f}",
                rule=rule,
                note="ASB Home Happenings Update dated 13 July 2026 moved >80% LVR pre-approvals for new owner-occupied customers to RED.",
                detail="This application is not eligible because it is a new-to-ASB owner-occupied high-LVR pre-approval. RBNZ-exempt / Kainga Ora scenarios are assessed under the separate exemption rule.",
            )

    status = "PASS" if umi is not None and umi >= threshold else "FAIL"
    if threshold <= 0:
        detail = "No extra ASB UMI buffer applies to this standard scenario. The calculator result is treated as acceptable for this policy check."
    else:
        detail = minimum_result_detail("UMI", umi, threshold)
    return set_policy_result(
        result,
        status=status,
        metric="UMI",
        value=umi,
        threshold=f">= ${threshold:,.0f}",
        rule=rule,
        detail=detail,
    )


def evaluate_bnz_criteria(result: dict, context: dict) -> dict:
    mbs = policy_number(result.get("value"))
    si = policy_number(result.get("si"))
    dti = policy_number(result.get("dti"))
    lvr = context["lvr"]
    investor = context["purpose"] == "Investor"
    existing = "BNZ" in context["existing_banks"]
    pre_approval = context["pre_approval"]
    deal_type = context["deal_type"]

    metric = "MBS"
    value = mbs
    minimum = 100.0
    dti_max = None
    rule = "BNZ Standard"
    note = ""
    max_lvr = 70 if investor else 80
    existing_only = False
    live_only_for_all = False
    live_only_if_new = False

    if deal_type in {"Construction", "Labour-only construction"}:
        metric, value, minimum = "SI", si, 110.0
        max_lvr = 80 if deal_type == "Labour-only construction" else 90
        rule = f"BNZ {deal_type}"
        existing_only = True
        live_only_for_all = True
    elif deal_type == "Off-plan turnkey":
        max_lvr = 90 if investor else 95
        rule = "BNZ Off-plan turnkey"
        live_only_for_all = True
    elif deal_type == "Completed turnkey":
        max_lvr = 90 if investor else 95
        rule = "BNZ Completed turnkey"
        live_only_for_all = True
        note = "The completed property must receive CCC and settle within 90 days."
    elif deal_type == "Dollar-for-dollar refinance":
        max_lvr = 90 if investor else 95
        rule = "BNZ RBNZ-exempt dollar-for-dollar refinance"
        live_only_for_all = True
    elif investor and lvr > 70:
        max_lvr = 90
        minimum = 500.0
        dti_max = 7.0
        rule = "BNZ Investor high-LVR funding deal"
        existing_only = True
        live_only_for_all = True
        note = "RBNZ funding deal; existing BNZ live application only."
    elif not investor and lvr > 80:
        max_lvr = 95
        rule = "BNZ Owner Occupied high-LVR funding deal"
        live_only_for_all = True
        note = "RBNZ funding deal; live application only."
    else:
        rule = f"BNZ {'Investor' if investor else 'Owner Occupied'} standard"
        live_only_if_new = True

    threshold = f">= {minimum:.0f} {'SI' if metric == 'SI' else 'MBS'}"
    if dti_max is not None:
        threshold += f" and DTI < {dti_max:g}"

    if lvr > max_lvr:
        return set_policy_result(
            result,
            status="INELIGIBLE",
            metric=metric,
            value=value,
            threshold=threshold,
            rule=rule,
            note=f"Maximum LVR for this scenario is {max_lvr:.0f}%.",
            detail=f"Current LVR is {lvr:.2f}%, which is above BNZ's allowed maximum of {max_lvr:.0f}% for this scenario.",
        )
    if existing_only and not existing:
        return set_policy_result(
            result,
            status="INELIGIBLE",
            metric=metric,
            value=value,
            threshold=threshold,
            rule=rule,
            note="This scenario is not available to new-to-BNZ customers.",
            detail="This BNZ scenario requires an existing BNZ customer. Select BNZ or Both under Existing Bank Relationship only if the client meets BNZ's existing-customer definition.",
        )
    if pre_approval and (live_only_for_all or (live_only_if_new and not existing)):
        return set_policy_result(
            result,
            status="INELIGIBLE",
            metric=metric,
            value=value,
            threshold=threshold,
            rule=rule,
            note="This scenario is available for live deals only.",
            detail="Pre-approval was selected, but BNZ only allows this scenario for a live deal.",
        )

    numeric_pass = value is not None and value >= minimum
    detail_parts = [minimum_result_detail(metric, value, minimum)]
    if dti_max is not None:
        if dti is None:
            detail_parts.append(f"DTI result was not available, so the DTI < {dti_max:g} requirement could not be verified.")
        elif dti < dti_max:
            detail_parts.append(f"Current DTI is {dti:.2f}, meeting the DTI < {dti_max:g} requirement.")
        else:
            detail_parts.append(f"Current DTI is {dti:.2f}, which is above the required DTI < {dti_max:g}.")
        numeric_pass = numeric_pass and dti is not None and dti < dti_max
    status = "PASS" if numeric_pass else "FAIL"
    return set_policy_result(
        result,
        status=status,
        metric=metric,
        value=value,
        threshold=threshold,
        rule=rule,
        note=note,
        detail=" ".join(detail_parts),
    )


def evaluate_pepper_criteria(result: dict, context: dict) -> dict:
    nsr = policy_number(result.get("value"))
    government_majority = bool(context.get("government_is_majority"))
    government_income = context.get("government_annual_income") or 0.0
    total_income = context.get("total_annual_income") or 0.0
    threshold = 1.25 if government_majority else 1.00
    rule = "Peppermoney NSR - Government allowance majority income" if government_majority else "Peppermoney NSR - All applications"
    threshold_text = f">= {threshold:.2f}"
    status = "PASS" if nsr is not None and nsr >= threshold else "FAIL"

    detail = threshold_result_detail("NSR", nsr, ">=", threshold)
    if government_majority:
        detail += (
            f" Government benefit income appears to be the majority income "
            f"(${government_income:,.0f} of ${total_income:,.0f} annual gross income), so Pepper's higher NSR 1.25 rule applies."
        )
    elif government_income > 0 and total_income > 0:
        detail += (
            f" Government benefit income was included (${government_income:,.0f} of ${total_income:,.0f} annual gross income), "
            "but it does not appear to be the majority income, so the standard NSR 1.00 rule applies."
        )
    else:
        detail += " No government-benefit majority income was identified, so the standard NSR 1.00 rule applies."

    return set_policy_result(
        result,
        status=status,
        metric="NSR",
        value=nsr,
        threshold=threshold_text,
        rule=rule,
        note="Interest-rate buffer eligibility is a separate manual check because it depends on transaction type, cash-out, Equifax score and RHI/default history.",
        detail=detail,
    )


def apply_servicing_criteria(bank: str, result: dict, inputs: dict) -> dict:
    context = policy_context(inputs)
    if bank == "ASB":
        return evaluate_asb_criteria(result, context)
    if bank == "BNZ":
        return evaluate_bnz_criteria(result, context)
    if bank == "Peppermoney":
        return evaluate_pepper_criteria(result, context)

    result.setdefault("criteria_threshold", "Calculator result")
    result.setdefault("criteria_rule", f"{bank} calculator")
    result.setdefault("criteria_note", "")
    result.setdefault("criteria_detail", generic_criteria_detail(bank, result))
    return result


def read_bank_results(bank: str, workbook_path: Path, recalculated_master: Path, policy_inputs: dict | None = None) -> dict:
    wb = load_workbook(workbook_path, data_only=True)
    master_wb = load_workbook(recalculated_master, data_only=True)
    config = BANK_CONFIGS[bank]
    sheet = get_sheet(wb, config["result_sheet"])
    value = normalize_excel_value(sheet[config["value_cell"]].value)
    dti = None
    if config.get("dti_cell"):
        dti = normalize_excel_value(sheet[config["dti_cell"]].value)
    si = None
    if config.get("si_cell"):
        si = normalize_excel_value(sheet[config["si_cell"]].value)
    result = {
        "bank": bank,
        "metric": config["metric"],
        "value": value,
        "dti": dti,
        "si": si,
        "test_rate": read_test_rate(bank, wb),
        "status": status_for_bank(bank, value, wb, master_wb),
        "result_cells": {"value": f"{config['result_sheet']}!{config['value_cell']}", "dti": config.get("dti_cell"), "si": config.get("si_cell")},
    }
    rule = config.get("status_rule", {})
    if rule.get("type") == "cell_threshold":
        result["status_check_value"] = normalize_excel_value(sheet[rule["cell"]].value)
    elif rule.get("type") == "compare_cells":
        result["status_check_left"] = normalize_excel_value(resolve_rule_value(rule["left"], wb, master_wb))
        result["status_check_right"] = normalize_excel_value(resolve_rule_value(rule["right"], wb, master_wb))
    return apply_servicing_criteria(bank, result, policy_inputs or {})


def create_zip(files: list[Path]) -> bytes:
    buffer = io.BytesIO()
    with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        for file_path in files:
            if file_path.exists():
                zf.write(file_path, arcname=file_path.name)
    return buffer.getvalue()


def safe_filename_part(value: str) -> str:
    text = str(value or "").strip()
    for char in '<>:"/\\|?*':
        text = text.replace(char, " ")
    return " ".join(text.split()) or "Client"


def client_names_label(inputs: dict) -> str:
    names = [
        str(inputs.get("c_name1") or "").strip(),
        str(inputs.get("c_name2") or "").strip(),
        str(inputs.get("c_name3") or "").strip(),
    ]
    names = [name for name in names if name]
    return safe_filename_part(" & ".join(names) if names else str(inputs.get("client_name") or "Client"))


def bank_file_label(bank: str, inputs: dict, timestamp: str) -> str:
    return f"{safe_filename_part(bank)}_UMI Calculator_{client_names_label(inputs)}_{timestamp}"


def package_file_label(inputs: dict, timestamp: str) -> str:
    loan_amount = int(parse_number(inputs.get("loan_amt")) or 0)
    return f"{client_names_label(inputs)}_Loan_{loan_amount:,.0f}_{timestamp}"


def populate_bank_file(recalculated_master: Path, bank: str, inputs: dict, run_timestamp: str, generated_dir: Path) -> dict:
    config = BANK_CONFIGS[bank]
    source_template = bank_template_path(bank, recalculated_master)
    bank_label = bank_file_label(bank, inputs, run_timestamp)
    populated = generated_dir / f"{bank_label}_input{source_template.suffix}"
    shutil.copy2(source_template, populated)
    applied = apply_bank_mapping(recalculated_master, bank, populated, populated, inputs)
    return {"bank": bank, "source": populated, "applied": applied}


def run_all_banks_engine(
    raw_inputs: dict,
    target_banks: list[str] | None = None,
    build_package: bool = True,
    save_inputs_record: bool = True,
) -> dict:
    total_started = time.perf_counter()
    timings = {}
    cleanup_old_jobs()
    setup_storage()

    inputs = preprocess_inputs(raw_inputs)
    if save_inputs_record:
        save_client_inputs(inputs)

    run_timestamp = nz_now().strftime("%Y%m%d_%H%M%S")
    job_id = f"{run_timestamp}_{uuid.uuid4().hex[:8]}"
    job_dir = JOBS_DIR / job_id
    generated_dir = job_dir / "generated"
    generated_dir.mkdir(parents=True, exist_ok=True)

    started = time.perf_counter()
    working_master, recalculated_master = build_recalculated_master(job_dir, inputs, run_timestamp)
    timings["Master recalc"] = time.perf_counter() - started

    client = client_key(str(inputs.get("c_name1") or "Client"))
    package_label = package_file_label(inputs, run_timestamp)
    target_banks = target_banks or list(BANK_CONFIGS)
    target_banks = [bank for bank in target_banks if bank not in EXCLUDED_BANKS]
    generated_files = [working_master] if build_package else []
    results = []
    pending = []

    banks_to_prepare = []
    for bank in target_banks:
        config = BANK_CONFIGS.get(bank)
        source_template = bank_template_path(bank, recalculated_master) if config else None
        if not config or not source_template or not source_template.exists():
            missing_name = source_template.name if source_template else bank
            results.append({"bank": bank, "metric": "Result", "value": f"Template missing: {missing_name}", "dti": None, "status": "FAIL"})
            continue
        banks_to_prepare.append(bank)

    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=BANK_WORKERS) as executor:
        future_to_bank = {
            executor.submit(populate_bank_file, recalculated_master, bank, inputs, run_timestamp, generated_dir): bank
            for bank in banks_to_prepare
        }
        for future in as_completed(future_to_bank):
            bank = future_to_bank[future]
            try:
                pending.append(future.result())
            except Exception as exc:
                results.append({"bank": bank, "metric": "Result", "value": f"Prepare failed: {exc}", "dti": None, "status": "FAIL"})
    timings["Bank file mapping"] = time.perf_counter() - started

    pending.sort(key=lambda item: target_banks.index(item["bank"]) if item["bank"] in target_banks else 999)

    started = time.perf_counter()
    try:
        converted = recalc_many_with_libreoffice([item["source"] for item in pending], generated_dir / "recalculated", "xlsx")
    except Exception as exc:
        converted = {}
        batch_error = exc
    else:
        batch_error = None
    timings["Bank LibreOffice recalc"] = time.perf_counter() - started

    started = time.perf_counter()
    for item in pending:
        bank = item["bank"]
        source_file = item["source"]
        try:
            if source_file in converted:
                recalculated = converted[source_file]
            else:
                if batch_error is not None:
                    raise RuntimeError(batch_error)
                raise RuntimeError("Batch recalculation output missing")
            if build_package:
                final_file = generated_dir / f"{bank_file_label(bank, inputs, run_timestamp)}{recalculated.suffix}"
                shutil.copy2(recalculated, final_file)
                result = read_bank_results(bank, final_file, recalculated_master, inputs)
                if source_file.suffix.lower() == ".xlsm":
                    download_file = generated_dir / f"{bank_file_label(bank, inputs, run_timestamp)}{source_file.suffix}"
                    shutil.copy2(source_file, download_file)
                    generated_files.append(download_file)
                else:
                    generated_files.append(final_file)
            else:
                result = read_bank_results(bank, recalculated, recalculated_master, inputs)
        except Exception as exc:
            if build_package:
                generated_files.append(source_file)
            result = {"bank": bank, "metric": "Result", "value": f"Read failed: {exc}", "dti": None, "status": "FAIL"}
        result["mapping_rows_applied"] = item["applied"]
        results.append(result)
    timings["Result read"] = time.perf_counter() - started

    started = time.perf_counter()
    package_bytes = b""
    package_path = None
    if build_package:
        package_bytes = create_zip(generated_files)
        package_path = job_dir / f"{package_label}.zip"
        package_path.write_bytes(package_bytes)
    timings["Package build"] = time.perf_counter() - started

    (job_dir / "input.json").write_text(json.dumps(inputs, indent=2, default=str), encoding="utf-8")
    (job_dir / "results.json").write_text(json.dumps(results, indent=2, default=str), encoding="utf-8")
    timings["Total"] = time.perf_counter() - total_started

    return {
        "job_id": job_id,
        "inputs": inputs,
        "results": results,
        "timings": timings,
        "package_bytes": package_bytes,
        "package_name": package_path.name if package_path else "",
        "package_path": str(package_path) if package_path else "",
    }


def run_asb_engine(raw_inputs: dict) -> dict:
    return run_all_banks_engine(raw_inputs, ["ASB"])