Audit generator populates XLSX, uploads to S3, and records UploadedFile row 🟩

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

orchestration/audit_generator_orchestrator.py

@@ -1,13 +1,86 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from typing import TYPE_CHECKING
+from datetime import datetime, timezone
+from io import BytesIO
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Optional
 
+import openpyxl
+
+from domain.magicplan.models import Door, Plan, Room, Window
+from infrastructure.postgres.uploaded_file_table import (
+    FileSourceEnum,
+    FileTypeEnum,
+    UploadedFile,
+)
 from infrastructure.s3.s3_client import S3Client
 
 if TYPE_CHECKING:
     from orchestration.audit_generator_unit_of_work import AuditGeneratorUnitOfWork
 
+_TEMPLATE_PATH = Path(__file__).parent.parent / "applications" / "audit-generator" / "d1_ventilation_template.xlsx"
+_SHEET_NAME = "D1 Ventilation"
+_DATA_START_ROW = 6
+_MAX_ROWS = 50
+
+
+def _write_cell(sheet: Any, row: int, col: str, value: Any) -> None:
+    sheet[f"{col}{row}"] = value
+
+
+def _populate_sheet(sheet: Any, plan: Plan) -> None:
+    rooms: list[Room] = [room for floor in plan.floors for room in floor.rooms]
+    windows: list[tuple[str, Window]] = [
+        (room.name, w) for room in rooms for w in room.windows
+    ]
+    doors: list[tuple[str, Door]] = [
+        (room.name, d) for room in rooms for d in room.doors
+    ]
+
+    if len(rooms) > _MAX_ROWS:
+        raise ValueError(f"Room series exceeds {_MAX_ROWS} rows ({len(rooms)} rooms)")
+    if len(windows) > _MAX_ROWS:
+        raise ValueError(f"Window series exceeds {_MAX_ROWS} rows ({len(windows)} windows)")
+    if len(doors) > _MAX_ROWS:
+        raise ValueError(f"Door series exceeds {_MAX_ROWS} rows ({len(doors)} doors)")
+
+    for i, room in enumerate(rooms):
+        row = _DATA_START_ROW + i
+        _write_cell(sheet, row, "B", room.name)
+        _write_cell(sheet, row, "D", room.area_m2)
+
+    for i, (room_name, window) in enumerate(windows):
+        row = _DATA_START_ROW + i
+        vent = window.ventilation
+        _write_cell(sheet, row, "G", room_name)
+        _write_cell(sheet, row, "H", window.width_m)
+        _write_cell(sheet, row, "I", window.height_m)
+        # J = formula =H*I — do not write
+        _write_cell(sheet, row, "K", vent.opening_type if vent else 0)
+        _write_cell(sheet, row, "L", vent.num_openings if vent else 0)
+        pct = vent.pct_openable if vent else None
+        _write_cell(sheet, row, "M", (pct / 100) if pct is not None else 0)
+        # N = formula =J*M — do not write
+        # O, P = blank (visual check by auditor)
+        _write_cell(sheet, row, "Q", vent.trickle_vent_area_mm2 if vent else 0)
+        _write_cell(sheet, row, "R", vent.num_trickle_vents if vent else 0)
+        # S = formula =Q*R — do not write
+
+    for i, (room_name, door) in enumerate(doors):
+        row = _DATA_START_ROW + i
+        vent = door.ventilation
+        _write_cell(sheet, row, "V", room_name)
+        _write_cell(sheet, row, "W", door.width_mm)
+        _write_cell(sheet, row, "X", vent.undercut_mm if vent else 0)
+        # Y = formula =W*X — do not write
+
+
+def _serialise_workbook(wb: Any) -> bytes:
+    buf = BytesIO()
+    wb.save(buf)
+    return buf.getvalue()
+
 
 class AuditGeneratorOrchestrator:
     def __init__(
@@ -21,4 +94,39 @@ class AuditGeneratorOrchestrator:
         self._uow_factory = uow_factory
 
     def run(self) -> None:
-        raise NotImplementedError
+        with self._uow_factory() as uow:
+            uploaded_file = uow.uploaded_file.get_latest_by_hubspot_deal_id(
+                self._hubspot_deal_id, FileTypeEnum.MAGIC_PLAN_JSON
+            )
+            if uploaded_file is None:
+                raise ValueError(
+                    f"No MagicPlan JSON has been uploaded for deal {self._hubspot_deal_id!r}"
+                )
+
+            plan = uow.magic_plan.get_plan_by_uploaded_file_id(uploaded_file.id)
+            if plan is None:
+                raise ValueError(
+                    f"MagicPlan JSON exists for deal {self._hubspot_deal_id!r} "
+                    "but the plan is not yet parsed into the database"
+                )
+
+            wb = openpyxl.load_workbook(_TEMPLATE_PATH)
+            sheet = wb[_SHEET_NAME]
+            _populate_sheet(sheet, plan)
+            xlsx_bytes = _serialise_workbook(wb)
+
+            s3_key = (
+                f"documents/hubspot_deal_id/{self._hubspot_deal_id}/ventilation_audit.xlsx"
+            )
+            self._s3_client.put_object(s3_key, xlsx_bytes)
+
+            new_row = UploadedFile(
+                s3_file_bucket=self._s3_client.bucket,
+                s3_file_key=s3_key,
+                s3_upload_timestamp=datetime.now(timezone.utc),
+                hubspot_deal_id=self._hubspot_deal_id,
+                file_type=FileTypeEnum.VENTILATION_AUDIT.value,
+                file_source=FileSourceEnum.AUDIT_GENERATOR.value,
+            )
+            uow.uploaded_file.insert(new_row)
+            uow.commit()