perfect-postcode/analysis/og_preflight.py

#!/usr/bin/env python3
"""OG / share-card preflight for Perfect Postcode growth findings.

Before publishing growth findings we must confirm that each finding's Open
Graph share card actually renders, and that the filter name baked into its
deep-link map query still matches a live feature. The OG render has no guard
today, so a wrong/renamed filter silently ships a card whose map contradicts
the headline.

For every analysis/out/findings/*.json this script:
  1. Fetches the OG render at {BASE}/api/screenshot?og=1&{map_query} and checks
     it is a non-trivial image (HTTP 200, image/*, > MIN_IMAGE_BYTES, and
     ~1200x630 when Pillow is installed).
  2. Cross-checks every `filter=<NAME>:...` in the map query against the live
     feature list from {BASE}/api/features, failing on any drifted name.

Exits non-zero if any finding fails, so it can gate a publish step.

Usage:
    source .venv/bin/activate && python analysis/og_preflight.py --base https://perfect-postcode.co.uk

Needs the server (local dev or prod) reachable at --base / $OG_BASE
(default http://localhost:8001); it makes live HTTP requests.
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.parse
import urllib.request
from io import BytesIO
from pathlib import Path

# Pillow is optional: when present we additionally decode the image and verify
# its dimensions; otherwise we fall back to status/content-type/size checks.
try:
    from PIL import Image  # type: ignore

    _HAVE_PIL = True
except Exception:
    _HAVE_PIL = False

FINDINGS_DIR = Path(__file__).resolve().parent / "out" / "findings"
DEFAULT_BASE = os.environ.get("OG_BASE", "http://localhost:8001")
MIN_IMAGE_BYTES = 5 * 1024  # a blank/error render is far smaller than this
OG_SIZE = (1200, 630)  # screenshot service VIEWPORT, device scale factor 1
SIZE_TOLERANCE = 4  # px slack when checking decoded dimensions


def http_get(url: str, timeout: float):
    """GET a URL, returning (status, content_type, body_bytes). Raises on error."""
    req = urllib.request.Request(url, headers={"User-Agent": "og-preflight"})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return resp.status, resp.headers.get("Content-Type", ""), resp.read()


def filter_names(map_query: str) -> list[str]:
    """Extract the feature name from each `filter=<name>:...` query param.

    Mirrors the frontend parser (url-state.ts): the name is everything before
    the first colon of a URL-decoded filter value, e.g. "Est. price per sqm".
    Non-filter params (school=, crime=, ...) carry no feature name and are
    ignored here.
    """
    names = []
    for key, value in urllib.parse.parse_qsl(map_query, keep_blank_values=True):
        if key == "filter" and ":" in value:
            names.append(value.split(":", 1)[0])
    return names


def live_feature_names(base: str, timeout: float) -> set[str]:
    """Collect every feature `name` from the grouped {base}/api/features response."""
    status, _ctype, body = http_get(f"{base}/api/features", timeout)
    if status != 200:
        raise RuntimeError(f"/api/features returned HTTP {status}")
    payload = json.loads(body)
    names: set[str] = set()
    for group in payload.get("groups", []):
        for feature in group.get("features", []):
            name = feature.get("name")
            if name:
                names.add(name)
    return names


def check_og(map_query: str, base: str, timeout: float) -> tuple[bool, str]:
    """Render the OG card and validate it looks like a real image."""
    url = f"{base}/api/screenshot?og=1&{map_query}"
    try:
        status, ctype, body = http_get(url, timeout)
    except Exception as exc:  # network / HTTP error
        return False, f"request failed: {exc}"
    if status != 200:
        return False, f"HTTP {status}"
    if not ctype.lower().startswith("image/"):
        return False, f"content-type {ctype!r} is not an image"
    if len(body) < MIN_IMAGE_BYTES:
        return False, f"image too small ({len(body)} B) - likely blank/error render"
    if _HAVE_PIL:
        try:
            img = Image.open(BytesIO(body))
            img.load()  # force a full decode; raises on a corrupt image
            w, h = img.size
        except Exception as exc:
            return False, f"undecodable image: {exc}"
        if abs(w - OG_SIZE[0]) > SIZE_TOLERANCE or abs(h - OG_SIZE[1]) > SIZE_TOLERANCE:
            return False, f"unexpected size {w}x{h} (want ~{OG_SIZE[0]}x{OG_SIZE[1]})"
        return True, f"{len(body)} B, {w}x{h}"
    return True, f"{len(body)} B"


def main() -> int:
    parser = argparse.ArgumentParser(description="Preflight OG cards for growth findings.")
    parser.add_argument(
        "--base", default=DEFAULT_BASE, help=f"server base URL (default {DEFAULT_BASE}; or $OG_BASE)"
    )
    parser.add_argument(
        "--findings-dir", default=str(FINDINGS_DIR), help="directory of finding *.json files"
    )
    parser.add_argument(
        "--timeout", type=float, default=30.0, help="per-request timeout in seconds (default 30)"
    )
    args = parser.parse_args()
    base = args.base.rstrip("/")

    finding_paths = sorted(Path(args.findings_dir).glob("*.json"))
    if not finding_paths:
        print(f"No findings found in {args.findings_dir}", file=sys.stderr)
        return 2

    # The drift guard needs the live feature list; if we cannot fetch it the
    # whole gate is inconclusive, so bail out non-zero before checking cards.
    try:
        live_names = live_feature_names(base, args.timeout)
    except Exception as exc:
        print(f"FATAL: could not load {base}/api/features: {exc}", file=sys.stderr)
        return 2
    print(f"Loaded {len(live_names)} live feature names from {base}/api/features")
    if not _HAVE_PIL:
        print("(Pillow not installed - skipping image dimension check)")

    passed = failed = 0
    for path in finding_paths:
        finding = json.loads(path.read_text())
        slug = finding.get("slug", path.stem)
        map_query = finding.get("map_query")
        if not map_query:
            print(f"FAIL  {slug}: no map_query field")
            failed += 1
            continue

        # Drift guard first: a renamed filter is the silent-ship bug we most
        # want to catch, and it makes the rendered card meaningless anyway.
        drifted = [n for n in filter_names(map_query) if n not in live_names]
        og_ok, og_reason = check_og(map_query, base, args.timeout)

        if drifted:
            print(f"FAIL  {slug}: unknown filter name(s): {', '.join(drifted)}")
            failed += 1
        elif not og_ok:
            print(f"FAIL  {slug}: OG render - {og_reason}")
            failed += 1
        else:
            print(f"PASS  {slug}: OG {og_reason}")
            passed += 1

    print(f"\n{passed} passed / {failed} failed (of {passed + failed})")
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())