Backfill uniform secondary-glazing overrides misfiled as Single glazing

39 Hyde-796 property_overrides rows ('100% secondary glazing (sap 9.94)')
plus the classifier-cache row; the cache phase self-gates on the
'Secondary glazing' pgEnum label landing (assessment-model#345).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

scripts/lisasrequest/reclassify_secondary_glazing.py

@@ -0,0 +1,197 @@
+"""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())