| """Build the Leanstral-8B project workbook from the JSON sheets in ./data. |
| |
| The xlsx is a BUILD ARTEFACT: edit data/*.json (or use the `journal` / `status` |
| subcommands below), never the workbook, or the next build will overwrite your edit. |
| |
| Style follows the Duplex training workbook: one header row (bold white Calibri on |
| #305496, centred, wrapped), body cells top-aligned and wrapped, frozen header row, |
| autofilter over the used range, and Source / Confidence / Status columns. |
| |
| python build_workbook.py # rebuild Leanstral_8B_Workbook.xlsx |
| python build_workbook.py journal "event" "result" "evidence" "next action" |
| python build_workbook.py status WBS 4.1 "Done" # set the Status cell of one row by Id |
| |
| Docs tooling, not align runtime: it is outside the dependency firewall's scan and |
| may import openpyxl. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import contextlib |
| import datetime as dt |
| import fcntl |
| import json |
| import os |
| import sys |
| from pathlib import Path |
|
|
| HERE = Path(__file__).resolve().parent |
| DATA = HERE / "data" |
| OUT = HERE / "Leanstral_8B_Workbook.xlsx" |
| LOCK = DATA / ".workbook.lock" |
|
|
|
|
| @contextlib.contextmanager |
| def locked(): |
| """One writer at a time: journal/status are read-modify-write on data/*.json, and two concurrent |
| CLI calls would otherwise read the same snapshot and drop each other's change.""" |
| LOCK.parent.mkdir(parents=True, exist_ok=True) |
| with open(LOCK, "w") as fh: |
| fcntl.flock(fh, fcntl.LOCK_EX) |
| try: |
| yield |
| finally: |
| fcntl.flock(fh, fcntl.LOCK_UN) |
|
|
| |
| SHEETS = ["README", "WBS", "Recipe_Stack", "Datasets", "Data_Gaps", "Evals_Gates", |
| "Experiment_Plan", "Journal", "Sources"] |
|
|
|
|
| def load(name: str) -> dict: |
| d = json.loads((DATA / f"{name}.json").read_text(encoding="utf-8")) |
| cols, widths = d["columns"], d["widths"] |
| if len(cols) != len(widths): |
| raise SystemExit(f"{name}: {len(cols)} columns but {len(widths)} widths") |
| for i, r in enumerate(d["rows"]): |
| if len(r) != len(cols): |
| raise SystemExit(f"{name} row {i + 2}: {len(r)} cells, expected {len(cols)}: {r[:2]}") |
| return d |
|
|
|
|
| def save(name: str, d: dict) -> None: |
| p = DATA / f"{name}.json" |
| tmp = p.with_name(f"{p.name}.tmp{os.getpid()}") |
| tmp.write_text(json.dumps(d, indent=1, ensure_ascii=False) + "\n", encoding="utf-8") |
| os.replace(tmp, p) |
|
|
|
|
| def build() -> Path: |
| from openpyxl import Workbook |
| from openpyxl.styles import Alignment, Font, PatternFill |
| from openpyxl.utils import get_column_letter |
|
|
| head_font = Font(name="Calibri", bold=True, color="FFFFFFFF") |
| head_fill = PatternFill("solid", fgColor="FF305496") |
| head_align = Alignment(horizontal="center", vertical="center", wrap_text=True) |
| body_font = Font(name="Calibri", size=11) |
| body_align = Alignment(vertical="top", wrap_text=True) |
|
|
| wb = Workbook() |
| wb.remove(wb.worksheets[0]) |
| for name in SHEETS: |
| d = load(name) |
| ws = wb.create_sheet(name) |
| ws.append(d["columns"]) |
| for c in ws[1]: |
| c.font, c.fill, c.alignment = head_font, head_fill, head_align |
| for r in d["rows"]: |
| ws.append(r) |
| for row in ws.iter_rows(min_row=2): |
| for c in row: |
| c.font, c.alignment = body_font, body_align |
| for i, w in enumerate(d["widths"], start=1): |
| ws.column_dimensions[get_column_letter(i)].width = w |
| ws.freeze_panes = "A2" |
| ws.auto_filter.ref = f"A1:{get_column_letter(len(d['columns']))}{ws.max_row}" |
| tmp = OUT.with_name(f"{OUT.stem}.tmp{os.getpid()}.xlsx") |
| wb.save(tmp) |
| os.replace(tmp, OUT) |
| return OUT |
|
|
|
|
| def journal(event: str, result: str, evidence: str, nxt: str) -> None: |
| d = load("Journal") |
| d["rows"].append([dt.date.today().isoformat(), event, result, evidence, nxt]) |
| save("Journal", d) |
|
|
|
|
| def set_status(sheet: str, row_id: str, status: str) -> None: |
| d = load(sheet) |
| col = d["columns"].index("Status") |
| hits = [r for r in d["rows"] if r[0] == row_id] |
| if len(hits) != 1: |
| raise SystemExit(f"{sheet}: {len(hits)} rows with Id {row_id!r}") |
| hits[0][col] = status |
| save(sheet, d) |
|
|
|
|
| def main(argv: list[str]) -> int: |
| with locked(): |
| return _main(argv) |
|
|
|
|
| def _main(argv: list[str]) -> int: |
| if argv[:1] == ["journal"]: |
| if len(argv) != 5: |
| raise SystemExit("usage: journal EVENT RESULT EVIDENCE NEXT") |
| journal(*argv[1:]) |
| elif argv[:1] == ["status"]: |
| if len(argv) != 4: |
| raise SystemExit("usage: status SHEET ID STATUS") |
| set_status(*argv[1:]) |
| elif argv: |
| raise SystemExit(__doc__) |
| print(build()) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main(sys.argv[1:])) |
|
|
