173 lines
10 KiB
Python
173 lines
10 KiB
Python
"""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()
|