Model/scripts/lisasrequest/reclassify_secondary_glazing.py

"""Backfill uniform secondary-glazing overrides misfiled as ``Single glazing``.

Hyde-796 landlords assert ``100% secondary glazing (sap 9.94)``; before
``GlazingType.SECONDARY`` existed the classifier's least-bad option was
``Single glazing``, which models the windows at U = 4.8 instead of secondary's
2.9 (SAP10 glazing code 5) — a material distortion (Model#1416).

Fixes BOTH stores in one pass:

* ``property_overrides.override_value`` (TEXT — what the modelling reads), so
  re-modelling picks the correct code; and
* the ``landlord_glazing_overrides`` classifier cache, so the stale value can't
  be re-served to new properties. The cache column is the FE-owned ``glazing``
  pgEnum — this phase requires the ``Secondary glazing`` label
  (assessment-model#345 + ``drizzle-kit migrate``) and is skipped with a warning
  until the label exists.

A *uniform* secondary assertion has exactly one glazing type at 100%, so it is
fully determined and safe to fix by description match (the live
``glazing_mix_guard`` deliberately leaves uniform assertions to the LLM, which
can now target SECONDARY). Dominant secondary *splits* are the guard's job and
are already covered by ``reclassify_dominant_glazing``.

DRY-RUN BY DEFAULT: prints what it would change and writes nothing. Pass
``--apply`` to execute inside a transaction; writes audit CSVs of every row
changed so the change is reversible. Idempotent.

    python -m scripts.lisasrequest.reclassify_secondary_glazing            # dry run
    python -m scripts.lisasrequest.reclassify_secondary_glazing --apply    # write
"""

from __future__ import annotations

import argparse
import csv
import re
import sys
from pathlib import Path
from typing import Any, Optional

from sqlalchemy import text
from sqlalchemy.engine import Connection

_REPO_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(_REPO_ROOT))

from domain.epc.property_overrides.glazing_type import GlazingType  # noqa: E402
from scripts.e2e_common import ENV_PATH, build_engine, load_env  # noqa: E402

# "100% secondary glazing", optionally trailed by "(sap 9.94)" or similar noise.
# One type, 100% — a fully-determined uniform assertion.
_UNIFORM_SECONDARY = re.compile(r"^\s*100%\s*secondary\s*glazing\b", re.IGNORECASE)

_SELECT_OVERRIDES = text(
    """
    SELECT po.property_id, pr.uprn,
           po.original_spreadsheet_description AS description,
           po.override_value AS value
    FROM property_overrides po
    JOIN property pr ON pr.id = po.property_id
    WHERE po.portfolio_id = :portfolio
      AND po.override_component = 'glazing'
    """
)
_UPDATE_OVERRIDE = text(
    """
    UPDATE property_overrides
       SET override_value = :new_value
     WHERE portfolio_id = :portfolio
       AND override_component = 'glazing'
       AND property_id = :property_id
       AND override_value <> :new_value
    """
)
_SELECT_CACHE = text(
    """
    SELECT id, description, value::text AS value
    FROM landlord_glazing_overrides
    WHERE portfolio_id = :portfolio
    """
)
_UPDATE_CACHE = text(
    """
    UPDATE landlord_glazing_overrides
       SET value = CAST(:new_value AS glazing),
           updated_at = now()
     WHERE id = :id
       AND value::text <> :new_value
    """
)
_ENUM_HAS_LABEL = text(
    """
    SELECT EXISTS (
        SELECT 1 FROM pg_enum e
        JOIN pg_type t ON t.oid = e.enumtypid
        WHERE t.typname = 'glazing' AND e.enumlabel = :label
    )
    """
)


def _target(description: Optional[str]) -> Optional[str]:
    if description and _UNIFORM_SECONDARY.match(description):
        return GlazingType.SECONDARY.value
    return None


def _write_audit(name: str, portfolio: int, header: list[str], rows: list[Any]) -> None:
    out = _REPO_ROOT / "scripts" / "lisasrequest" / f"{name}_{portfolio}_audit.csv"
    with out.open("w", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(header)
        w.writerows(rows)
    print(f"audit trail: {out}")


def _fix_property_overrides(conn: Connection, portfolio: int, apply: bool) -> int:
    audit: list[tuple[int, object, str, str, str]] = []
    for property_id, uprn, description, value in conn.execute(
        _SELECT_OVERRIDES, {"portfolio": portfolio}
    ):
        new_value = _target(description)
        if new_value is None or value == new_value:
            continue
        audit.append((property_id, uprn, description, value, new_value))
        if apply:
            conn.execute(
                _UPDATE_OVERRIDE,
                {"portfolio": portfolio, "property_id": property_id, "new_value": new_value},
            )
    verb = "re-classified" if apply else "would re-classify"
    print(f"property_overrides: {verb} {len(audit)} row(s)")
    if apply and audit:
        _write_audit(
            "reclassify_secondary_glazing",
            portfolio,
            ["property_id", "uprn", "description", "old_value", "new_value"],
            audit,
        )
    return len(audit)


def _fix_cache(conn: Connection, portfolio: int, apply: bool) -> int:
    label = GlazingType.SECONDARY.value
    if not conn.execute(_ENUM_HAS_LABEL, {"label": label}).scalar():
        print(
            f"cache: SKIPPED — the glazing pgEnum has no {label!r} label yet; "
            "merge assessment-model#345 and run drizzle-kit migrate first, then re-run"
        )
        return 0
    audit: list[tuple[str, str, str, str]] = []
    for row_id, description, value in conn.execute(_SELECT_CACHE, {"portfolio": portfolio}):
        new_value = _target(description)
        if new_value is None or value == new_value:
            continue
        audit.append((str(row_id), description, value, new_value))
        if apply:
            conn.execute(_UPDATE_CACHE, {"id": row_id, "new_value": new_value})
    verb = "re-classified" if apply else "would re-classify"
    print(f"cache: {verb} {len(audit)} row(s)")
    if apply and audit:
        _write_audit(
            "reclassify_secondary_glazing_cache",
            portfolio,
            ["id", "description", "old_value", "new_value"],
            audit,
        )
    return len(audit)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--portfolio", type=int, default=796)
    parser.add_argument(
        "--apply",
        action="store_true",
        help="execute the updates (default: dry-run, writes nothing)",
    )
    args = parser.parse_args()

    load_env(ENV_PATH)
    engine = build_engine()

    with engine.begin() as conn:
        conn.execute(text("SET statement_timeout = 120000"))
        changed = _fix_property_overrides(conn, args.portfolio, args.apply)
        changed += _fix_cache(conn, args.portfolio, args.apply)
        if not args.apply:
            conn.rollback()

    if not args.apply:
        print("\nDRY-RUN — nothing written. Re-run with --apply to execute.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())