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) <noreply@anthropic.com>

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 <secrets.env>                     # 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 <addr>.pdf" or
+    "<addr> Summary.pdf", sometimes with stray spaces — so "summary" anywhere in
+    the basename is the picker, deliberately excluding "<addr> 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()

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")