school_compare/backend/places.py

"""The place registry: what places the site publishes, and what is in each.

One module owns this question. The pages, the sitemap and the internal-link
modules all read from here, so the threshold and the collision rules exist in
exactly one place and are testable without a browser or a database.

Two namespaces, never one. 67 viable town names collide with a local
authority name, and the authority is the larger set in only 43 of them —
postal towns cross authority boundaries, so neither can absorb the other.
Keys are "<kind>:<slug>" so the collision cannot reappear in the dict.
"""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)

# Five schools with publishable data. Below this a place has nothing to say
# that a list of schools does not, and publishing it is index bloat.
MIN_SCHOOLS = 5


@dataclass(frozen=True)
class Place:
    kind: str                       # "town" | "locality" | "authority" | "outcode"
    slug: str
    name: str
    urns: tuple[int, ...]
    parent_authority: str | None    # authority NAME, for the 301 target
    # Every authority the place meaningfully sits in, largest first. A quarter
    # of outcodes and a third of towns straddle a boundary — SW19 is mostly
    # Merton but partly Wandsworth — so naming only one asserts something
    # false. parent_authority stays single because a redirect needs one
    # target; this is what the page shows.
    authorities: tuple[tuple[str, int], ...] = ()
    # URNs per phase, so the per-phase threshold can be applied without
    # re-querying. A place with 30 primaries and 2 secondaries publishes a
    # primary variant and no secondary one.
    phase_urns: dict[str, tuple[int, ...]] = field(default_factory=dict)

    def publishes_phase(self, phase: str) -> bool:
        return len(self.phase_urns.get(phase, ())) >= MIN_SCHOOLS

    @property
    def key(self) -> str:
        return f"{self.kind}:{self.slug}"


def _publishable_urns(df) -> set[int]:
    """URNs with something a page could state, deduplicated across years."""
    from backend.app import _PUBLISHABLE_FIELDS

    cols = [c for c in _PUBLISHABLE_FIELDS if c in df.columns]
    if not cols:
        return set()
    return set(df.loc[df[cols].notna().any(axis=1), "urn"].astype(int))


# The measure a phase page is built around. A page with no results in this
# column has nothing a list of school names does not already give.
_PHASE_METRIC = {
    "primary": "rwm_expected_pct",
    "secondary": "attainment_8_score",
}


def _phase_urns(group, publishable: set[int]) -> dict[str, tuple[int, ...]]:
    """URNs per phase, counting only schools with a result for that phase.

    Not merely "publishable". A school with an Ofsted grade and no results is
    worth a page of its own and belongs in the place list, but it cannot
    populate a phase page's results column — and the threshold is there to ask
    whether that column will have anything in it.

    Counting publishable schools instead let /schools/kent/primary publish
    with none of its five rows carrying a result, and left 44 phase pages
    majority-blank. It is the same rule as "no page without a local average",
    which was never extended per phase.

    All-through schools count toward both phases, matching the PHASE_GROUPS
    mapping the search filters already use.
    """
    from backend.app import PHASE_GROUPS

    if "phase" not in group.columns:
        return {}
    lowered = group["phase"].fillna("").str.lower()

    out: dict[str, tuple[int, ...]] = {}
    for phase in ("primary", "secondary"):
        wanted = PHASE_GROUPS.get(phase, set())
        subset = group[lowered.isin(wanted)]

        # The page lists every school of the phase; the threshold counts only
        # those carrying a result, so a mostly-empty table never publishes.
        metric = _PHASE_METRIC[phase]
        with_result = (
            {int(u) for u in subset.loc[subset[metric].notna(), "urn"]}
            if metric in subset.columns else set()
        )
        if len(with_result & publishable) < MIN_SCHOOLS:
            continue

        urns = tuple(sorted({int(u) for u in subset["urn"]} & publishable))
        if urns:
            out[phase] = urns
    return out


# A place is described by an authority when it holds at least a tenth of the
# schools, and at least two. GIAS carries occasional postcode errors — EN6
# lists two Shropshire schools among fourteen in Hertfordshire — and a bare
# "any authority present" rule would print those as though they were real.
# There is deliberately no cap on how many are named. An earlier cut stopped
# at three, which silently dropped the fourth in exactly the case where the
# information matters most — a genuinely fragmented place. The share rule is
# the only limit, and it already bounds the list at ten.
_AUTHORITY_MIN_SHARE = 0.10
_AUTHORITY_MIN_SCHOOLS = 2


def _authorities(group) -> tuple[tuple[str, int], ...]:
    """Authorities this place meaningfully sits in, largest first."""
    from backend.app import EXCLUDED_FILTER_VALUES

    if "local_authority" not in group.columns:
        return ()
    counts = group["local_authority"].dropna().value_counts()
    total = int(counts.sum())
    if not total:
        return ()

    kept = [
        (str(name), int(n)) for name, n in counts.items()
        if str(name) not in EXCLUDED_FILTER_VALUES
        and n >= _AUTHORITY_MIN_SCHOOLS
        and n / total >= _AUTHORITY_MIN_SHARE
    ]
    # A place too small or too fragmented for the share rule still names its
    # largest authority, or the page would say nothing about where it is.
    if not kept:
        for name, n in counts.items():
            if str(name) not in EXCLUDED_FILTER_VALUES:
                return ((str(name), int(n)),)
        return ()
    return tuple(kept)


def _parent_authority(authorities: tuple[tuple[str, int], ...]) -> str | None:
    """The 301 target: the largest authority a place sits in.

    Derived from `authorities` rather than computed separately. The first cut
    used `mode()` here while `authorities` used `value_counts()`, and on an
    exact tie pandas does not guarantee the two pick the same name — so the
    redirect could have pointed somewhere other than the authority the page
    named first. One computation, one answer.

    Deriving it also inherits the sentinel filter, so a place can no longer
    redirect to /schools/authority/does-not-apply.
    """
    return authorities[0][0] if authorities else None


def _group(df, column: str, kind: str, publishable: set[int]) -> dict[str, Place]:
    """One Place per distinct SLUG in `column` that clears the threshold.

    Grouped by slug, not by raw value, because GIAS spells the same place
    several ways and they all resolve to one URL. Five town slugs come from
    more than one spelling: "London" (1,819 schools) and "LONDON" (12) both
    slugify to `london`; Weston-super-Mare is split 14/19 across two
    spellings; Newcastle-under-Lyme across three.

    Grouping by raw value meant the later group simply overwrote the earlier
    one in this dict — so /schools/london could have shown twelve schools
    instead of 1,819, silently and depending on row order.

    The display name is the most common spelling, which is the one a reader
    expects to see.
    """
    from backend.app import _slugify

    if column not in df.columns:
        return {}

    working = df.assign(_slug=df[column].map(
        lambda v: _slugify(str(v).strip()) if isinstance(v, str) and v.strip() else None))
    working = working[working["_slug"].notna() & (working["_slug"] != "")]

    out: dict[str, Place] = {}
    for slug, group in working.groupby("_slug"):
        slug = str(slug)
        urns = tuple(sorted({int(u) for u in group["urn"]} & publishable))
        if len(urns) < MIN_SCHOOLS:
            continue
        spellings = group[column].dropna().value_counts()
        if spellings.empty:
            continue
        name = str(spellings.index[0]).strip()
        authorities = () if kind == "authority" else _authorities(group)
        place = Place(
            kind=kind, slug=slug, name=name, urns=urns,
            parent_authority=_parent_authority(authorities),
            authorities=authorities,
            phase_urns=_phase_urns(group, publishable),
        )
        out[place.key] = place
    return out


# "SW11 2AA" -> "SW11". Two letters max, one or two digits, optional letter.
_OUTCODE_RE = re.compile(r"^([A-Z]{1,2}\d{1,2}[A-Z]?)\s")


def _outcode(postcode) -> str | None:
    if not isinstance(postcode, str):
        return None
    m = _OUTCODE_RE.match(postcode.upper().strip())
    return m.group(1) if m else None


def _outcode_places(df, publishable: set[int]) -> dict[str, Place]:
    """One Place per postcode district clearing the threshold.

    These carry no phase variants: nobody searches "primary schools in SW11",
    so the spec gives them no /primary or /secondary route. `phase_urns` is
    left empty rather than computed and then filtered downstream — the
    registry is the one place that decides which phases a place publishes,
    and the page links whatever it reports.

    Computing them here put a link to a route that does not exist on every one
    of the 1,720 outcode pages.
    """
    if "postcode" not in df.columns:
        return {}
    working = df.assign(_oc=df["postcode"].map(_outcode))
    working = working[working["_oc"].notna()]

    out: dict[str, Place] = {}
    for oc, group in working.groupby("_oc"):
        urns = tuple(sorted({int(u) for u in group["urn"]} & publishable))
        if len(urns) < MIN_SCHOOLS:
            continue
        authorities = _authorities(group)
        place = Place(kind="outcode", slug=str(oc).lower(), name=str(oc),
                      urns=urns, parent_authority=_parent_authority(authorities),
                      authorities=authorities)
        out[place.key] = place
    return out


def _locality_places(df, publishable: set[int],
                     town_slugs: set[str]) -> dict[str, Place]:
    """One Place per curated locality clearing the threshold."""
    from backend.localities import LOCALITY_OUTCODES

    if "postcode" not in df.columns:
        return {}
    working = df.assign(_oc=df["postcode"].map(_outcode))

    out: dict[str, Place] = {}
    for slug, (name, outcodes) in LOCALITY_OUTCODES.items():
        if slug in town_slugs:
            # Skip, do not raise. The guard exists so a locality never
            # silently shadows a town — skipping achieves that, and the error
            # log makes it loud.
            #
            # Raising here took down sitemap generation for all 25,000 school
            # pages when "richmond" met the GIAS town Richmond in North
            # Yorkshire. Worse, GIAS town names change without any code change,
            # so a raise means curated data can break the site spontaneously.
            # A curation mistake must cost one page, not the sitemap.
            logger.error(
                "locality %r collides with the published town of the same "
                "slug and has been skipped; rename it or remove it", slug)
            continue
        group = working[working["_oc"].isin(outcodes)]
        urns = tuple(sorted({int(u) for u in group["urn"]} & publishable))
        if len(urns) < MIN_SCHOOLS:
            # Not an error — a locality can legitimately be too small. Logged
            # because one you meant to publish quietly vanishing is the
            # failure worth hearing about.
            logger.warning(
                "locality %s (%s) has %d publishable schools, below the "
                "threshold of %d - not published",
                slug, ", ".join(outcodes), len(urns), MIN_SCHOOLS)
            continue
        authorities = _authorities(group)
        place = Place(kind="locality", slug=slug, name=name, urns=urns,
                      parent_authority=_parent_authority(authorities),
                      authorities=authorities,
                      phase_urns=_phase_urns(group, publishable))
        out[place.key] = place
    return out


def build_place_registry(df) -> dict[str, Place]:
    """Every place the site publishes, keyed by "<kind>:<slug>"."""
    if df.empty or "urn" not in df.columns:
        return {}

    publishable = _publishable_urns(df)
    registry: dict[str, Place] = {}
    registry.update(_group(df, "local_authority", "authority", publishable))

    towns = _group(df, "town", "town", publishable)
    registry.update(towns)

    town_slugs = {p.slug for p in towns.values()}
    registry.update(_locality_places(df, publishable, town_slugs))
    registry.update(_outcode_places(df, publishable))
    return registry