Model/scripts/lisasrequest/reclassify_dominant_single_glazing.py

"""Backfill glazing overrides where a >= 90%-single dwelling was flattened onto its
*minority* double type.

Sibling to ``scripts/reclassify_mixed_glazing.py``. That script rescued genuine
mixes (no type >= 90%) to ``MIXED``; it deliberately left near-uniform rows to the
LLM classifier. But for a dominant-*single* split ("4% Double glazing 2002 or
later, 96% Single Glazing") the LLM had latched onto the 4% minority and written a
"Double glazing…" value. ``glazing_mix_guard`` now resolves a dominant-single
split deterministically to ``SINGLE`` (single glazing is era-free), so this
backfill fixes the rows written before that guard change.

Uses the SAME guard as the live path, so the backfill and the classifier cannot
drift. SCOPED TO ONE PORTFOLIO (``--portfolio``, default 796 = Hyde) and only
touches ``property_overrides.override_value`` (TEXT — what the modelling reads);
the global ``landlord_glazing_overrides`` classifier cache is left alone.

DRY-RUN BY DEFAULT: prints what it would change and writes nothing. Pass
``--apply`` to execute inside a transaction; it also writes an audit CSV of every
row changed (property_id, uprn, old value, new value) so the change is reversible.
Idempotent — only rows whose stored value differs from the guard's target member
are touched.

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

from __future__ import annotations

import argparse
import csv
import sys
from collections import Counter
from pathlib import Path

from sqlalchemy import text

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

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

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


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

    audit: list[tuple[int, object, str, str, str]] = []  # pid, uprn, descr, old, new
    tally: Counter[str] = Counter()
    with engine.begin() as conn:
        conn.execute(text("SET statement_timeout = 120000"))
        for property_id, uprn, description, value in conn.execute(
            _SELECT, {"portfolio": args.portfolio}
        ):
            member = glazing_mix_guard(description or "")
            if member is None or value == member.value:
                continue
            tally[f"{value!r} -> {member.value!r}"] += 1
            audit.append((property_id, uprn, description, value, member.value))
            if args.apply:
                conn.execute(
                    _UPDATE,
                    {
                        "portfolio": args.portfolio,
                        "property_id": property_id,
                        "new_value": member.value,
                    },
                )
        if not args.apply:
            conn.rollback()

    verb = "re-classified" if args.apply else "would re-classify"
    print(f"portfolio {args.portfolio}: {verb} {len(audit)} glazing override row(s)")
    for change, n in tally.most_common():
        print(f"  {n:5d}  {change}")

    if args.apply and audit:
        out = _REPO_ROOT / "scripts" / "lisasrequest" / (
            f"reclassify_dominant_single_glazing_{args.portfolio}_audit.csv"
        )
        with out.open("w", newline="") as fh:
            w = csv.writer(fh)
            w.writerow(["property_id", "uprn", "description", "old_value", "new_value"])
            w.writerows(audit)
        print(f"\naudit trail: {out}")
    if not args.apply:
        print("\nDRY-RUN — nothing written. Re-run with --apply to execute.")
    return 0


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