Model/scripts/wchg_elmhurst_ingest.py

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