MEGAINFRA-WEBAPP/scripts/import-management-workbook.py

174 lines
10 KiB
Python
Raw Normal View History

"""Read saved Excel values and colours into the management snapshots.
Run with the bundled Python runtime (openpyxl required). Never edits the workbook.
Usage: python import-management-workbook.py "/path/to/Turnover sheet.xlsx"
"""
import hashlib
import json
import re
import sys
from datetime import datetime
from pathlib import Path
import openpyxl
def number(value):
if value is None:
return None
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"Expected a number, got {value!r}")
return value
def colour(cell):
value = cell.font.color
if value is None or (value.type == "theme" and value.theme == 1):
return "black"
if value.type == "rgb":
rgb = value.rgb[-6:].upper()
return {"000000": "black", "7030A0": "purple", "FF0000": "red"}.get(rgb, rgb)
raise ValueError(f"Unsupported font colour at {cell.coordinate}: {value}")
def schedule(sheet):
months = {}
date = None
for row in range(1, sheet.max_row + 1):
heading = sheet.cell(row, 2).value
if isinstance(heading, datetime):
date = heading.strftime("%Y-%m-01")
months.setdefault(date, [])
name, amount = sheet.cell(row, 3).value, sheet.cell(row, 5).value
if date and isinstance(name, str) and isinstance(amount, (int, float)):
months[date].append({"id": f"retention-{row}", "project": name.strip(),
"debtor": (sheet.cell(row, 4).value or "").strip() or None,
"amount": amount})
# Preserve empty months, including gaps in the source's date headings.
first, last = min(months), max(months)
year, month = map(int, first[:7].split("-"))
while f"{year:04d}-{month:02d}-01" <= last:
months.setdefault(f"{year:04d}-{month:02d}-01", [])
year, month = (year + 1, 1) if month == 12 else (year, month + 1)
return [{"date": date, "entries": entries} for date, entries in sorted(months.items())]
def main():
source = Path(sys.argv[1])
workbook = openpyxl.load_workbook(source, data_only=True)
formulas = openpyxl.load_workbook(source, data_only=False)
root = Path(__file__).resolve().parents[1] / "src/lib/management"
historical = workbook[" 2024 - 2025"]
comparisons = []
for year, rows, result_row in [(2024, [23, 25, 28, 30], 32), (2025, [60, 62, 65, 67], 69)]:
comparisons.append({"year": year, **{key: [number(historical.cell(row, col).value) for col in range(3, 15)]
for key, row in zip(["income", "costs", "retentionAdded", "retentionPayouts"], rows)},
"reportedCashResult": number(historical.cell(result_row, 15).value),
"reportedResultWithRetention": number(historical.cell(result_row + 1, 15).value),
"retentionClosing": number(historical.cell(rows[2] + 1, 15).value)})
# December is the annual payout subtotal, not another monthly payout.
comparisons[-1]["retentionPayouts"][-1] = None
balances = []
for date_row, first_row, retention_row, wip_row, net_row in [(72, 74, 83, 84, 86), (88, 89, 98, 99, 101)]:
balances.append({"label": historical.cell(date_row, 5).value.strftime("%B %Y"),
"liabilities": [{"name": historical.cell(r, 5).value, "amount": number(historical.cell(r, 6).value)} for r in range(first_row, first_row + 7)],
"retention": number(historical.cell(retention_row, 6).value), "wip": number(historical.cell(wip_row, 6).value),
"reportedNet": number(historical.cell(net_row, 6).value)})
comparison_sheet = workbook["Income 2026"]
income_comparisons = []
for year, date_row, income_row in [(2025, 114, 116), (2026, 119, 121)]:
for month, col in enumerate(range(5, 17), 1):
date = comparison_sheet.cell(date_row, col).value
if not isinstance(date, datetime) or (date.year, date.month) != (year, month):
raise ValueError(f"Invalid comparison period at {comparison_sheet.cell(date_row, col).coordinate}")
income_comparisons.append({"year": year,
"income": [number(comparison_sheet.cell(income_row, col).value) for col in range(5, 17)]})
for year in [2025, 2026, 2027]:
name = "Income per month 2025" if year == 2025 else f"Income {year}"
sheet, formula = workbook[name], formulas[name]
start = 8 if year == 2025 else 7
def labelled_row(label):
matches = [r for r in range(1, 50) if str(sheet.cell(r, 3).value or "").strip() == label]
if not matches:
raise ValueError(f"Missing {label} in {sheet.title}")
return matches[0]
retention_row = labelled_row("RETENTION")
payout_row = labelled_row("RETENTION PAYOUTS")
income_row = labelled_row("TOTAL INCOME INCLUDING VAT")
cost_row = labelled_row("TOTAL COST INCLUDING VAT")
columns = range(start, start + 12)
cash_factors, forecast_cost_rates = [], []
for col in columns:
income_expression = str(formula.cell(income_row, col).value)
factor = re.fullmatch(r"=SUM\([^)]*\)\*(0\.\d+)", income_expression)
cash_factors.append(float(factor[1]) if factor else 1)
cost_expression = str(formula.cell(cost_row, col).value)
rate = re.fullmatch(rf"=[A-Z]+{income_row}\*(0\.\d+)", cost_expression)
forecast_cost_rates.append(float(rate[1]) if rate else None)
rates = []
for col in columns:
expression = formula.cell(retention_row, col).value
match = re.fullmatch(r"=SUM\([A-Z]+(\d+):[A-Z]+(\d+)\)\*(0\.\d+)", expression)
if not match:
raise ValueError(f"Unrecognised retention formula: {name}!{formula.cell(retention_row, col).coordinate} {expression}")
rates.append((int(match[1]), int(match[2]), float(match[3])))
projects = []
for row in range(6, retention_row - 1):
name_value = sheet.cell(row, 3).value
if not isinstance(name_value, str) or not isinstance(sheet.cell(row, 2).value, (int, float)):
continue
future = colour(sheet.cell(row, 3)) == "red"
income = [number(sheet.cell(row, col).value) for col in columns]
actual, forecast, future_values = [], [], []
for col, value in zip(columns, income):
kind = colour(sheet.cell(row, col)) if value is not None else None
if kind not in (None, "black", "purple", "red"):
raise ValueError(f"Unknown income colour: {name}!{sheet.cell(row, col).coordinate} {kind}")
future_values.append(value if kind == "red" or future else None)
actual.append(value if kind == "black" and not future else None)
forecast.append(value if kind == "purple" and not future else None)
project_rates = [rate if lo <= row <= hi and not future else 0 for lo, hi, rate in rates]
costs = None
if year == 2025:
cost_sheet = workbook["Cost per month"]
matches = [r for r in range(7, 22) if cost_sheet.cell(r, 3).value == name_value]
if len(matches) > 1:
raise ValueError(f"Duplicate cost row for {name_value}")
if matches:
costs = [number(cost_sheet.cell(matches[0], c).value) for c in range(8, 20)]
projects.append({"id": re.sub(r"[^a-z0-9]+", "-", name_value.lower()).strip("-"), "name": name_value,
"manager": sheet.cell(row, 4).value if year == 2025 else None,
"client": sheet.cell(row, 5 if year == 2025 else 4).value,
"engineer": sheet.cell(row, 6 if year == 2025 else 5).value,
"statedValue": number(sheet.cell(row, start - 1).value),
"remainingValue": number(sheet.cell(row, 19).value) if year == 2026 else None, "income": income,
"forecastIncome": forecast, "actualIncome": actual, "futureIncome": future_values, "excludedFromCashflow": future,
"costs": costs, "retentionEligible": any(project_rates), "retentionRates": project_rates})
snapshot = {"sourceFile": source.name, "sourceSheet": sheet.title, "planYear": year,
"sourceSha256": hashlib.sha256(source.read_bytes()).hexdigest(),
"periodLabel": f"March {year}–February {year + 1}", "months": [sheet.cell(3, c).value.strip() for c in columns],
"projects": projects,
"reportedIncome": [number(sheet.cell(income_row, c).value) for c in columns],
"cashflowIncomeFactors": cash_factors,
"forecastCostRates": forecast_cost_rates,
"reportedCosts": [number(sheet.cell(cost_row, c).value) for c in columns],
"reportedCostAnnual": number(workbook["Cost per month"]["G27"].value) if year == 2025 else number(sheet.cell(cost_row, start - 1).value),
"reportedRetention": [number(sheet.cell(retention_row, c).value) for c in columns],
"reportedRetentionClosing": [number(sheet.cell(retention_row + 1, c).value) for c in columns],
"reportedCumulativeBalance": [number(sheet.cell(37, c).value) for c in columns] if year == 2025 else [None] * 12,
"openingRetention": number(sheet.cell(retention_row + 1, start - 1).value),
"retentionPayouts": [number(sheet.cell(payout_row, c).value) for c in columns],
"headOfficeCosts": [number(workbook["Cost per month"].cell(22, c).value) for c in range(8, 20)] if year == 2025 else [0] * 12,
"costDetailAvailable": year == 2025,
"retentionSchedule": schedule(workbook[f"Retention {year}"]) if year != 2025 else [],
"comparisons": comparisons, "incomeComparisons": income_comparisons, "balanceSnapshots": balances}
target = root / ("workbook-snapshot.json" if year == 2026 else f"workbook-snapshot-{year}.json")
target.write_text(json.dumps(snapshot, indent=2) + "\n")
included = [p for p in projects if not p["excludedFromCashflow"]]
print(year, len(projects), "jobs", "included income", sum(sum(v or 0 for v in p["income"]) for p in included),
"excluded", [p["name"] for p in projects if p["excludedFromCashflow"]])
if __name__ == "__main__":
main()