feat(places): London localities and postcode districts
The GIAS town field puts 1,819 London schools under the single value 'London', so it cannot answer 'schools in Battersea' — a query that appears in the baseline. No single field can: parliamentary constituency gives Battersea but not Canary Wharf, admin_ward gives Canary Wharf but not Battersea, and neither gives Clapham or Shoreditch. So a locality is curated, defined by the postcode districts it covers, which needs no new ingestion. A locality may not shadow a published town: the registry raises rather than silently costing a page that carries real demand. One below the threshold is logged rather than raising, because a locality can legitimately be too small. The pipeline seed mirrors the module, with a test guarding the drift — the same arrangement gias_codes has, and for the same reason: the backend image does not contain pipeline/. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015mWQnpye9F299NVRCCSRvj
This commit is contained in:
1 parent
759d9f5cea
commit
de853b90b3
4 files changed
+234
-1
No files matched your search
+76
-1
@@ -12,8 +12,12 @@ Keys are "<kind>:<slug>" so the collision cannot reappear in the dict.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
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
|
||||
@@ -80,6 +84,71 @@ def _group(df, column: str, kind: str, publishable: set[int]) -> dict[str, 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".
|
||||
"""
|
||||
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
|
||||
place = Place(kind="outcode", slug=str(oc).lower(), name=str(oc),
|
||||
urns=urns, parent_authority=_parent_authority(group))
|
||||
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:
|
||||
raise ValueError(
|
||||
f"locality {slug!r} collides with a published town of the same "
|
||||
"slug; publishing both would shadow the town silently"
|
||||
)
|
||||
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
|
||||
place = Place(kind="locality", slug=slug, name=name, urns=urns,
|
||||
parent_authority=_parent_authority(group))
|
||||
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:
|
||||
@@ -88,5 +157,11 @@ def build_place_registry(df) -> dict[str, Place]:
|
||||
publishable = _publishable_urns(df)
|
||||
registry: dict[str, Place] = {}
|
||||
registry.update(_group(df, "local_authority", "authority", publishable))
|
||||
registry.update(_group(df, "town", "town", 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
|
||||
Reference in new issue
Block a user