From 8fc8a90c8b35c78910a6d1ca0557146f029801cd Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Thu, 16 Jul 2026 16:40:27 +0000 Subject: [PATCH] Add WCHG Elmhurst summary ingest script (SharePoint -> S3 + epc_property) One-time ingest of Wythenshawe Community Housing Group Elmhurst *Summary* PDFs: resolve each property's SharePoint folder from the WCHG xlsx, download the summary, run the existing extractor -> mapper -> SAP10 pipeline, and (with --commit) push the PDF to S3, record an uploaded_files row, and save the EpcPropertyData to epc_property keyed on the national UPRN (property_id NULL, replace-by-(uprn,source)). Dry-run by default; also emits a SAP-accuracy CSV vs Elmhurst's reported current SAP (% within 0.5 + MAE). Adds DomnaSites.SOCIAL_HOUSING (SOCIAL_HOUSING_SHAREPOINT_ID) for the site. Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/wchg_elmhurst_ingest.py | 499 ++++++++++++++++++++++++++++++++ utils/sharepoint/domna_sites.py | 1 + 2 files changed, 500 insertions(+) create mode 100644 scripts/wchg_elmhurst_ingest.py diff --git a/scripts/wchg_elmhurst_ingest.py b/scripts/wchg_elmhurst_ingest.py new file mode 100644 index 000000000..f362d0a88 --- /dev/null +++ b/scripts/wchg_elmhurst_ingest.py @@ -0,0 +1,499 @@ +"""One-time ingest of WCHG Elmhurst *Summary* PDFs -> S3 + uploaded_files + epc_property, +with a SAP-accuracy report against Elmhurst's own reported score. + +For each row of the WCHG xlsx that has an Elmhurst "Summary" PDF in its SharePoint +folder, this: + + 1. resolves the folder from the row's `Sharepoint URL` (the `RootFolder` query + param, drive-relative), finds the summary PDF (basename contains "summary"), + 2. downloads it, runs it through the existing pipeline + `ElmhurstSiteNotesExtractor -> EpcPropertyDataMapper.from_elmhurst_site_notes + -> calculate_sap_from_inputs(cert_to_inputs(...))`, + 3. stamps the national `UPRN` from the sheet onto the EpcPropertyData (the summary + does not carry it), and compares our continuous SAP to Elmhurst's reported + `current_sap_rating`, + 4. (only with --commit) pushes the PDF to S3, records an `uploaded_files` row + (source=sharepoint, type=rd_sap_site_note), saves the EpcPropertyData to + `epc_property` keyed on UPRN (property_id NULL), replacing any prior + (uprn, source='lodged') graph, and links `epc_property.uploaded_file_id`. + +Writes an accuracy CSV and prints MAE + the share of records within 0.5 +(i.e. our rounded integer matches Elmhurst's). + +Dry run is the DEFAULT — nothing is written until you pass --commit. + +USAGE + source # SHAREPOINT_* + SOCIAL_HOUSING_SHAREPOINT_ID + PYTHONPATH=. python scripts/wchg_elmhurst_ingest.py \ + --xlsx "WCHG EPC Sharepoint URLs & UPRNs 160726.xlsx" \ + --out accuracy.csv [--limit N] [--commit] +""" + +from __future__ import annotations + +import argparse +import csv +import os +import re +import subprocess +import urllib.parse +from dataclasses import replace +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from dotenv import load_dotenv + +_REPO_ROOT = Path(__file__).resolve().parents[1] +# Load DB / bucket / AWS config before importing anything that builds an engine. +load_dotenv(_REPO_ROOT / "backend" / ".env") + +from sqlmodel import Session, col, delete, select # noqa: E402 + +from backend.app.db.connection import db_engine # noqa: E402 +from backend.documents_parser.elmhurst_extractor import ( # noqa: E402 + ElmhurstSiteNotesExtractor, +) +from datatypes.epc.domain.epc_property_data import EpcPropertyData # noqa: E402 +from datatypes.epc.domain.mapper import EpcPropertyDataMapper # noqa: E402 +from domain.sap10_calculator.calculator import calculate_sap_from_inputs # noqa: E402 +from domain.sap10_calculator.rdsap.cert_to_inputs import cert_to_inputs # noqa: E402 +from infrastructure.postgres.epc_property_table import ( # noqa: E402 + EpcBuildingPartModel, + EpcEnergyElementModel, + EpcFlatDetailsModel, + EpcFloorDimensionModel, + EpcMainHeatingDetailModel, + EpcPhotovoltaicArrayModel, + EpcPropertyEnergyPerformanceModel, + EpcPropertyModel, + EpcRenewableHeatIncentiveModel, + EpcWindowModel, +) +from infrastructure.postgres.uploaded_file_table import ( # noqa: E402 + FileSourceEnum, + FileTypeEnum, + UploadedFile, +) +from repositories.epc.epc_postgres_repository import EpcPostgresRepository # noqa: E402 +from utils.s3 import upload_file_to_s3 # noqa: E402 +from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient # noqa: E402 +from utils.sharepoint.domna_sites import DomnaSites # noqa: E402 +from utils.sharepoint.sharepoint_client import SharePointClient # noqa: E402 + +_SOURCE = "lodged" +_SHARED_DOCS_MARKER = "Shared Documents/" +# Every epc_property child table, in child-before-parent delete order (floor +# dimensions handled separately as they key on building_part, not epc_property). +_CHILD_MODELS = ( + EpcPropertyEnergyPerformanceModel, + EpcEnergyElementModel, + EpcMainHeatingDetailModel, + EpcBuildingPartModel, + EpcWindowModel, + EpcPhotovoltaicArrayModel, + EpcFlatDetailsModel, + EpcRenewableHeatIncentiveModel, +) + + +def _drive_relative_path(sharepoint_url: str) -> Optional[str]: + """Extract the drive-relative folder path from a SharePoint AllItems URL. + + The URL carries the server-relative folder in its `RootFolder` query param + (e.g. `/sites/SocialHousing/Shared Documents/.../10 Deptford Avenue M23 2XA`); + the Graph drive root is the site's "Shared Documents" library, so everything + up to and including `Shared Documents/` is stripped. + """ + query = urllib.parse.urlparse(sharepoint_url).query + params = urllib.parse.parse_qs(query) + root_values = params.get("RootFolder") + if not root_values: + return None + root = urllib.parse.unquote(root_values[0]) + idx = root.find(_SHARED_DOCS_MARKER) + return root[idx + len(_SHARED_DOCS_MARKER):] if idx >= 0 else root.lstrip("/") + + +def _find_summary_pdf( + client: SharePointClient, folder: str, depth: int = 0, maxdepth: int = 2 +) -> list[str]: + """Return drive-relative paths of every PDF whose basename contains "summary" + (case-insensitive), searching `folder` and up to `maxdepth` subfolders. + + The Elmhurst summary is named inconsistently — "Summary .pdf" or + " Summary.pdf", sometimes with stray spaces — so "summary" anywhere in + the basename is the picker, deliberately excluding " Report.pdf" (the + customer EPC report) and "Energy performance certificate (EPC) ...pdf". + """ + found: list[str] = [] + listing: dict[str, Any] = client.list_folder_contents(folder, page_size=500) + for item in listing.get("value", []): + name: str = item["name"] + if "folder" in item: + if depth < maxdepth: + found += _find_summary_pdf( + client, f"{folder}/{name}", depth + 1, maxdepth + ) + elif name.lower().endswith(".pdf") and "summary" in name.lower(): + found.append(f"{folder}/{name}") + return found + + +def _pdf_to_pages(pdf_path: Path) -> list[str]: + """Render each PDF page to the textract-style token stream the extractor + expects (copied from scripts/run_elmhurst_summary.py).""" + info = subprocess.run( + ["pdfinfo", str(pdf_path)], capture_output=True, text=True, check=True + ).stdout + match = re.search(r"Pages:\s+(\d+)", info) + if match is None: + raise RuntimeError(f"Could not parse page count from {pdf_path}") + page_count = int(match.group(1)) + pages: list[str] = [] + for i in range(1, page_count + 1): + layout = subprocess.run( + ["pdftotext", "-layout", "-f", str(i), "-l", str(i), str(pdf_path), "-"], + capture_output=True, + text=True, + check=True, + ).stdout + tokens: list[str] = [] + for line in layout.splitlines(): + if not line.strip(): + tokens.append("") + continue + tokens += [p for p in re.split(r"\s{2,}", line.strip()) if p] + pages.append("\n".join(tokens)) + return pages + + +def _delete_epc_by_uprn(session: Session, uprn: int, source: str = _SOURCE) -> int: + """Delete every epc_property graph (parent + children) for (uprn, source), + so a re-run replaces rather than duplicates. Mirrors the repository's + per-property delete, but keyed on UPRN because these rows carry no + property_id (so the repo's own delete-by-property_id never fires).""" + epc_ids = [ + i + for i in session.exec( + select(EpcPropertyModel.id) + .where(EpcPropertyModel.uprn == uprn) + .where(EpcPropertyModel.source == source) + ).all() + if i is not None + ] + if not epc_ids: + return 0 + part_ids = [ + i + for i in session.exec( + select(EpcBuildingPartModel.id).where( + col(EpcBuildingPartModel.epc_property_id).in_(epc_ids) + ) + ).all() + if i is not None + ] + if part_ids: + session.exec( # type: ignore[call-overload] + delete(EpcFloorDimensionModel).where( + col(EpcFloorDimensionModel.epc_building_part_id).in_(part_ids) + ) + ) + for child in _CHILD_MODELS: + session.exec( # type: ignore[call-overload] + delete(child).where(col(child.epc_property_id).in_(epc_ids)) + ) + session.exec( # type: ignore[call-overload] + delete(EpcPropertyModel).where(col(EpcPropertyModel.id).in_(epc_ids)) + ) + return len(epc_ids) + + +def _persist( + session: Session, + *, + epc: EpcPropertyData, + uprn: int, + pdf_path: Path, + bucket: str, +) -> tuple[int, int]: + """Push the PDF to S3, record an uploaded_files row, replace-save the EPC + graph keyed on UPRN, and link uploaded_file_id. Returns (uploaded_file_id, + epc_property_id). Caller owns the transaction.""" + key = f"documents/uprn/{uprn}/{pdf_path.name}" + upload_file_to_s3(str(pdf_path), bucket, key) + + # Replace any prior uploaded_files row for this uprn + Elmhurst-summary slot + # so re-runs stay idempotent (the S3 key is deterministic, so it overwrites). + session.exec( # type: ignore[call-overload] + delete(UploadedFile) + .where(col(UploadedFile.uprn) == uprn) + .where(col(UploadedFile.file_source) == FileSourceEnum.SHAREPOINT.value) + .where(col(UploadedFile.file_type) == FileTypeEnum.RD_SAP_SITE_NOTE.value) + ) + uploaded_file = UploadedFile( + s3_file_bucket=bucket, + s3_file_key=key, + s3_upload_timestamp=datetime.now(timezone.utc), + uprn=uprn, + file_source=FileSourceEnum.SHAREPOINT.value, + file_type=FileTypeEnum.RD_SAP_SITE_NOTE.value, + ) + session.add(uploaded_file) + session.flush() + uploaded_file_id = int(uploaded_file.id) # type: ignore[arg-type] + + _delete_epc_by_uprn(session, uprn, _SOURCE) + epc_property_id = EpcPostgresRepository(session).save( + epc, property_id=None, source=_SOURCE + ) + + model = session.get(EpcPropertyModel, epc_property_id) + if model is not None: + model.uploaded_file_id = uploaded_file_id + session.add(model) + return uploaded_file_id, epc_property_id + + +def _process_row( + client: DomnaSharepointClient, + raw_client: SharePointClient, + *, + uprn: int, + folder: str, + tmp_dir: Path, +) -> tuple[EpcPropertyData, int, Path]: + """Download -> extract -> map -> stamp UPRN. Returns + (epc, elmhurst_current_sap, local_pdf_path). Raises FileNotFoundError when + there is no summary PDF / the download fails; any other exception is an + extractor / mapper / calc failure.""" + summaries = _find_summary_pdf(raw_client, folder) + if not summaries: + raise FileNotFoundError("no summary PDF in folder") + if len(summaries) > 1: + print(f" ! {len(summaries)} summary PDFs, using first: {summaries}") + sp_path = summaries[0] + + local = tmp_dir / f"{uprn}_{Path(sp_path).name}" + if not client.download_file(sp_path, str(local)): + raise FileNotFoundError(f"download failed: {sp_path}") + + site_notes = ElmhurstSiteNotesExtractor(_pdf_to_pages(local)).extract() + epc = EpcPropertyDataMapper.from_elmhurst_site_notes(site_notes) + epc = replace(epc, uprn=uprn) # summary carries no UPRN; use the sheet's + return epc, site_notes.current_sap_rating, local + + +def _read_rows(xlsx_path: Path) -> list[dict[str, Any]]: + import openpyxl + + workbook = openpyxl.load_workbook(xlsx_path, read_only=True) + sheet = workbook.worksheets[0] + rows = list(sheet.iter_rows(values_only=True)) + header = [str(h) for h in rows[0]] + + def idx(name: str) -> int: + return header.index(name) + + i_rec, i_addr, i_url, i_uprn = ( + idx("Record ID"), + idx("Clean Address"), + idx("Sharepoint URL"), + idx("UPRN"), + ) + out: list[dict[str, Any]] = [] + for r in rows[1:]: + out.append( + { + "record_id": r[i_rec], + "address": r[i_addr], + "url": r[i_url], + "uprn": r[i_uprn], + } + ) + return out + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--xlsx", required=True, type=Path) + parser.add_argument("--out", required=True, type=Path, help="accuracy CSV path") + parser.add_argument("--limit", type=int, default=None, help="first N rows only") + parser.add_argument( + "--commit", + action="store_true", + help="write to S3 + DB (default: dry run, report only)", + ) + args = parser.parse_args() + + bucket = os.environ["ENERGY_ASSESSMENTS_BUCKET"] + site = DomnaSites.SOCIAL_HOUSING + assert site.value, "SOCIAL_HOUSING_SHAREPOINT_ID env var is not set" + + client = DomnaSharepointClient(sharepoint_location=site) + raw_client = SharePointClient( + tenant_id=os.environ["SHAREPOINT_TENANT_ID"], + client_id=os.environ["SHAREPOINT_CLIENT_ID"], + client_secret=os.environ["SHAREPOINT_CLIENT_SECRET"], + site_id=str(site.value), + ) + + rows = _read_rows(args.xlsx) + if args.limit is not None: + rows = rows[: args.limit] + + tmp_dir = Path("/tmp/wchg_elmhurst") + tmp_dir.mkdir(parents=True, exist_ok=True) + + engine = db_engine + host = str(engine.url.host) + redacted_host = host[:6] + "…" if len(host) > 6 else host + db_note = "" + try: + with Session(engine) as preflight: + existing = preflight.exec( + select(EpcPropertyModel.id).where( + col(EpcPropertyModel.source) == _SOURCE + ) + ).all() + db_note = f"{len(existing)} existing lodged epc_property rows" + except Exception as e: # dry run doesn't need the DB — just note it + if args.commit: + raise + db_note = f"UNREACHABLE ({type(e).__name__}) — fine for dry run" + print("=" * 68) + print(f" MODE : {'COMMIT (writes to prod)' if args.commit else 'DRY RUN (no writes)'}") + print(f" DB host : {redacted_host} db={engine.url.database}") + print(f" DB preflight : {db_note}") + print(f" S3 bucket : {bucket}") + print(f" rows to process: {len(rows)}") + print("=" * 68) + + results: list[dict[str, Any]] = [] + deltas: list[float] = [] + n_ok = n_within = 0 + + for n, row in enumerate(rows, 1): + rec = str(row["record_id"]) + address = str(row["address"]) + uprn_raw = row["uprn"] + url = row["url"] + base = {"record_id": rec, "address": address, "uprn": uprn_raw} + + if not url: + results.append({**base, "status": "NO_URL"}) + print(f"[{n}/{len(rows)}] NO_URL {address}") + continue + try: + uprn = int(str(uprn_raw).strip()) + except (TypeError, ValueError): + results.append({**base, "status": "NO_UPRN", "error": repr(uprn_raw)}) + print(f"[{n}/{len(rows)}] NO_UPRN {address} — {uprn_raw!r}") + continue + folder = _drive_relative_path(str(url)) + if folder is None: + results.append({**base, "status": "BAD_URL"}) + print(f"[{n}/{len(rows)}] BAD_URL {address}") + continue + + try: + epc, elmhurst_sap, pdf_path = _process_row( + client, + raw_client, + uprn=uprn, + folder=folder, + tmp_dir=tmp_dir, + ) + result = calculate_sap_from_inputs(cert_to_inputs(epc)) + computed = result.sap_score_continuous + delta = computed - elmhurst_sap + within = abs(delta) < 0.5 + deltas.append(abs(delta)) + n_ok += 1 + n_within += int(within) + + epc_property_id: Optional[int] = None + if args.commit: + with Session(engine) as session: + try: + _, epc_property_id = _persist( + session, + epc=epc, + uprn=uprn, + pdf_path=pdf_path, + bucket=bucket, + ) + session.commit() + except Exception: + session.rollback() + raise + + results.append( + { + **base, + "status": "OK", + "elmhurst_current_sap": elmhurst_sap, + "computed_continuous": round(computed, 4), + "computed_int": result.sap_score, + "delta": round(delta, 4), + "within_0_5": within, + "tfa_m2": epc.total_floor_area_m2, + "dwelling_type": epc.dwelling_type, + "epc_property_id": epc_property_id, + } + ) + flag = "✓" if within else "✗" + print( + f"[{n}/{len(rows)}] OK {flag} elm={elmhurst_sap:>3} " + f"ours={computed:6.2f} Δ={delta:+.2f} {address}" + ) + except FileNotFoundError as e: + results.append({**base, "status": "NO_SUMMARY", "error": str(e)}) + print(f"[{n}/{len(rows)}] NO_SUMMARY {address} — {e}") + except Exception as e: # extractor / mapper / calc / DB failure + results.append( + {**base, "status": "EXTRACT_ERROR", "error": f"{type(e).__name__}: {e}"} + ) + print( + f"[{n}/{len(rows)}] EXTRACT_ERROR {address} — {type(e).__name__}: {e}" + ) + + _write_report(args.out, results) + + mae = sum(deltas) / len(deltas) if deltas else 0.0 + within_pct = 100.0 * n_within / n_ok if n_ok else 0.0 + print("=" * 68) + print(f" processed OK : {n_ok}") + print(f" within 0.5 (int match): {n_within}/{n_ok} ({within_pct:.1f}%)") + print(f" MAE : {mae:.4f}") + print(f" report written : {args.out}") + if not args.commit: + print(" (dry run — pass --commit to write to S3 + DB)") + print("=" * 68) + + +def _write_report(path: Path, results: list[dict[str, Any]]) -> None: + fields = [ + "record_id", + "address", + "uprn", + "status", + "elmhurst_current_sap", + "computed_continuous", + "computed_int", + "delta", + "within_0_5", + "tfa_m2", + "dwelling_type", + "epc_property_id", + "error", + ] + with open(path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore") + writer.writeheader() + for r in results: + writer.writerow(r) + + +if __name__ == "__main__": + main() diff --git a/utils/sharepoint/domna_sites.py b/utils/sharepoint/domna_sites.py index ce579af63..32decaf67 100644 --- a/utils/sharepoint/domna_sites.py +++ b/utils/sharepoint/domna_sites.py @@ -9,4 +9,5 @@ class DomnaSites(Enum): OSMOSIS_ACD = os.getenv("OSMOSIS_ACD_SHAREPOINT_ID") PRIVATE_PAY = os.getenv("PRIVATE_PAY_SHAREPOINT_ID") SOCIAL_HOUSING_WAVE_3 = os.getenv("SOCIAL_HOUSING_WAVE_3_SHAREPOINT_ID") + SOCIAL_HOUSING = os.getenv("SOCIAL_HOUSING_SHAREPOINT_ID") ECO = os.getenv("ECO_SHAREPOINT_ID")