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"])