feat: GIAS code->name dictionaries generated from live bulk CSV

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

pipeline/scripts/generate_gias_codes.py

@@ -0,0 +1,134 @@
+"""Generate GIAS code->name dictionaries from the live bulk CSV.
+
+Writes:
+  - backend/gias_codes.py                        (canonical Python module)
+  - pipeline/scripts/gias_codes.py               (byte-identical copy)
+  - pipeline/transform/seeds/gias_code_names.csv (dbt seed for drift test)
+
+Run from the repo root whenever the dbt drift test warns that DfE
+added/renamed a value:  python pipeline/scripts/generate_gias_codes.py
+"""
+
+from __future__ import annotations
+
+import io
+import sys
+from datetime import date, timedelta
+from pathlib import Path
+
+import pandas as pd
+import requests
+
+GIAS_URL = (
+    "https://ea-edubase-api-prod.azurewebsites.net"
+    "/edubase/downloads/public/edubasealldata{date}.csv"
+)
+
+# (CSV code column, CSV name column, python dict name, seed field key)
+FIELDS = [
+    ("TypeOfEstablishment (code)", "TypeOfEstablishment (name)", "SCHOOL_TYPE", "school_type"),
+    ("EstablishmentStatus (code)", "EstablishmentStatus (name)", "ESTABLISHMENT_STATUS", "establishment_status"),
+    ("PhaseOfEducation (code)", "PhaseOfEducation (name)", "PHASE_OF_EDUCATION", "phase_of_education"),
+    ("OfficialSixthForm (code)", "OfficialSixthForm (name)", "OFFICIAL_SIXTH_FORM", "official_sixth_form"),
+    ("ReligiousCharacter (code)", "ReligiousCharacter (name)", "RELIGIOUS_CHARACTER", "religious_character"),
+    ("AdmissionsPolicy (code)", "AdmissionsPolicy (name)", "ADMISSIONS_POLICY", "admissions_policy"),
+]
+
+MODULE_HEADER = '''"""GIAS code -> name dictionaries.
+
+GENERATED by pipeline/scripts/generate_gias_codes.py from the GIAS bulk CSV
+— do not edit by hand; rerun the script when the dbt drift test warns.
+The canonical file is backend/gias_codes.py; pipeline/scripts/gias_codes.py
+must be byte-identical (enforced by backend/tests/test_gias_codes.py).
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+
+logger = logging.getLogger(__name__)
+
+'''
+
+MODULE_FOOTER = '''
+
+def translate(code, mapping: dict[int, str]) -> str | None:
+    """Translate a GIAS code to its display name.
+
+    None/NaN -> None (column absent or suppressed). Unknown codes degrade to
+    "Unknown (<code>)" with a warning so a new DfE value never blanks the UI.
+    """
+    if code is None or (isinstance(code, float) and math.isnan(code)):
+        return None
+    code = int(code)
+    if code not in mapping:
+        logger.warning("Unknown GIAS code %s (not in dictionary)", code)
+        return f"Unknown ({code})"
+    return mapping[code]
+'''
+
+
+def download_csv() -> pd.DataFrame:
+    for day in (date.today(), date.today() - timedelta(days=1)):
+        url = GIAS_URL.format(date=day.strftime("%Y%m%d"))
+        print(f"Downloading {url}")
+        resp = requests.get(url, timeout=300)
+        if resp.status_code == 404:
+            continue
+        resp.raise_for_status()
+        return pd.read_csv(
+            io.StringIO(resp.content.decode("latin-1")),
+            dtype=str, keep_default_na=False,
+        )
+    sys.exit("GIAS CSV not available for today or yesterday")
+
+
+def main() -> None:
+    repo = Path(__file__).resolve().parents[2]
+    df = download_csv()
+
+    module_parts = [MODULE_HEADER]
+    seed_rows: list[tuple[str, int, str]] = []
+
+    for code_col, name_col, dict_name, field_key in FIELDS:
+        pairs = (
+            df[[code_col, name_col]]
+            .loc[lambda d: (d[code_col] != "") & (d[name_col] != "")]
+            .drop_duplicates()
+        )
+        mapping = sorted((int(c), n) for c, n in pairs.itertuples(index=False))
+        dupes = len(mapping) - len({c for c, _ in mapping})
+        if dupes:
+            sys.exit(f"{code_col}: {dupes} codes map to multiple names — investigate before generating")
+        lines = [f"{dict_name}: dict[int, str] = {{"]
+        for code, name in mapping:
+            escaped = name.replace('"', '\\"')
+            lines.append(f'    {code}: "{escaped}",')
+        lines.append("}\n")
+        module_parts.append("\n".join(lines))
+        seed_rows += [(field_key, code, name) for code, name in mapping]
+
+    module = "\n".join(module_parts) + MODULE_FOOTER
+
+    (repo / "backend" / "gias_codes.py").write_text(module)
+    (repo / "pipeline" / "scripts" / "gias_codes.py").write_text(module)
+
+    seed_path = repo / "pipeline" / "transform" / "seeds" / "gias_code_names.csv"
+    with open(seed_path, "w", newline="") as fh:
+        import csv
+        w = csv.writer(fh)
+        w.writerow(["field", "code", "name"])
+        w.writerows(seed_rows)
+
+    print(f"Wrote backend/gias_codes.py, pipeline/scripts/gias_codes.py, {seed_path.name}")
+    print("\nKey codes for the dbt work (Task 3):")
+    for field in ("establishment_status", "phase_of_education", "official_sixth_form"):
+        print(f"  {field}:")
+        for f, code, name in seed_rows:
+            if f == field:
+                print(f"    {code} = {name}")
+
+
+if __name__ == "__main__":
+    main()

pipeline/scripts/gias_codes.py

@@ -0,0 +1,152 @@
+"""GIAS code -> name dictionaries.
+
+GENERATED by pipeline/scripts/generate_gias_codes.py from the GIAS bulk CSV
+— do not edit by hand; rerun the script when the dbt drift test warns.
+The canonical file is backend/gias_codes.py; pipeline/scripts/gias_codes.py
+must be byte-identical (enforced by backend/tests/test_gias_codes.py).
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+
+logger = logging.getLogger(__name__)
+
+
+SCHOOL_TYPE: dict[int, str] = {
+    1: "Community school",
+    2: "Voluntary aided school",
+    3: "Voluntary controlled school",
+    5: "Foundation school",
+    6: "City technology college",
+    7: "Community special school",
+    8: "Non-maintained special school",
+    10: "Other independent special school",
+    11: "Other independent school",
+    12: "Foundation special school",
+    14: "Pupil referral unit",
+    15: "Local authority nursery school",
+    18: "Further education",
+    24: "Secure units",
+    25: "Offshore schools",
+    26: "Service children's education",
+    27: "Miscellaneous",
+    28: "Academy sponsor led",
+    29: "Higher education institutions",
+    30: "Welsh establishment",
+    31: "Sixth form centres",
+    32: "Special post 16 institution",
+    33: "Academy special sponsor led",
+    34: "Academy converter",
+    35: "Free schools",
+    36: "Free schools special",
+    37: "British schools overseas",
+    38: "Free schools alternative provision",
+    39: "Free schools 16 to 19",
+    40: "University technical college",
+    41: "Studio schools",
+    42: "Academy alternative provision converter",
+    43: "Academy alternative provision sponsor led",
+    44: "Academy special converter",
+    45: "Academy 16-19 converter",
+    46: "Academy 16 to 19 sponsor led",
+    49: "Online provider",
+    56: "Institution funded by other government department",
+    57: "Academy secure 16 to 19",
+}
+
+ESTABLISHMENT_STATUS: dict[int, str] = {
+    1: "Open",
+    2: "Closed",
+    3: "Open, but proposed to close",
+    4: "Proposed to open",
+}
+
+PHASE_OF_EDUCATION: dict[int, str] = {
+    0: "Not applicable",
+    1: "Nursery",
+    2: "Primary",
+    3: "Middle deemed primary",
+    4: "Secondary",
+    5: "Middle deemed secondary",
+    6: "16 plus",
+    7: "All-through",
+}
+
+OFFICIAL_SIXTH_FORM: dict[int, str] = {
+    0: "Not applicable",
+    1: "Has a sixth form",
+    2: "Does not have a sixth form",
+}
+
+RELIGIOUS_CHARACTER: dict[int, str] = {
+    0: "Does not apply",
+    2: "Church of England",
+    3: "Roman Catholic",
+    4: "Methodist",
+    5: "Jewish",
+    6: "None",
+    7: "Muslim",
+    8: "Seventh Day Adventist",
+    9: "Church of England/Methodist",
+    10: "Methodist/Church of England",
+    11: "Church of England/Roman Catholic",
+    12: "Church of England/United Reformed Church",
+    13: "Roman Catholic/Church of England",
+    14: "Quaker",
+    15: "Christian",
+    16: "United Reformed Church",
+    17: "Congregational Church",
+    18: "Free Church",
+    19: "Church of England/Free Church",
+    20: "Church of England/Christian",
+    21: "Sikh",
+    22: "Greek Orthodox",
+    24: "Buddhist",
+    25: "Hindu",
+    26: "Moravian",
+    28: "Inter- / non- denominational",
+    29: "Multi-faith",
+    30: "Church of England/Methodist/United Reform Church/Baptist",
+    31: "Anglican",
+    32: "Anglican/Christian",
+    33: "Anglican/Evangelical",
+    34: "Anglican/Church of England",
+    35: "Catholic",
+    36: "Charadi Jewish",
+    37: "Christian/Evangelical",
+    38: "Christian Science",
+    39: "Christian/Methodist",
+    40: "Christian/non-denominational",
+    41: "Church of England/Evangelical",
+    42: "Islam",
+    43: "Orthodox Jewish",
+    44: "Plymouth Brethren Christian Church",
+    45: "Protestant",
+    46: "Protestant/Evangelical",
+    47: "Reformed Baptist",
+    48: "Roman Catholic/Anglican",
+    49: "Sunni Deobandi",
+}
+
+ADMISSIONS_POLICY: dict[int, str] = {
+    0: "Not applicable",
+    2: "Selective",
+    4: "Non-selective",
+}
+
+
+def translate(code, mapping: dict[int, str]) -> str | None:
+    """Translate a GIAS code to its display name.
+
+    None/NaN -> None (column absent or suppressed). Unknown codes degrade to
+    "Unknown (<code>)" with a warning so a new DfE value never blanks the UI.
+    """
+    if code is None or (isinstance(code, float) and math.isnan(code)):
+        return None
+    code = int(code)
+    if code not in mapping:
+        logger.warning("Unknown GIAS code %s (not in dictionary)", code)
+        return f"Unknown ({code})"
+    return mapping[code]