feat(ofsted): extract Ofsted's further education and skills MI (H4)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/fe_mi.py

@@ -0,0 +1,137 @@
+"""Ofsted's further education and skills management information (audit H4).
+
+Ofsted publishes FE inspection outcomes on one gov.uk page per academic year
+("Further education and skills inspections and outcomes: management
+information from November 2025 to August 2026"). Each page carries "as at"
+files (one row per provider, with its latest inspection under the previous
+framework) and cumulative files of renewed-framework inspections, i.e. report
+cards ("…_10_November_2025_to_31_January_2026.csv", "…_to_28_February_2026.csv",
+…). Monitoring-visit files are a different series.
+
+Free of the Singer SDK so CI's pytest, which installs only the backend's
+requirements, can load it.
+"""
+from __future__ import annotations
+
+import io
+import re
+from datetime import date
+from typing import Callable, Iterator
+
+import pandas as pd
+
+SEARCH_URL = "https://www.gov.uk/api/search.json"
+SEARCH_QUERY = "further education and skills inspections and outcomes management information"
+PAGE_TITLE_PREFIX = "Further education and skills inspections and outcomes: management information from "
+# Renewed-framework (report card) inspections began on 10 November 2025.
+RENEWED_FRAMEWORK_START = date(2025, 11, 1)
+
+_MONTHS = {name: number for number, name in enumerate(
+    ["january", "february", "march", "april", "may", "june", "july",
+     "august", "september", "october", "november", "december"], start=1)}
+_TITLE_START = re.compile(r"from ([A-Za-z]+) (\d{4})")
+_CSV_LINK = re.compile(r'https://assets\.publishing\.service\.gov\.uk/[^"\s<>]+\.csv')
+_AS_AT = re.compile(r"further_education_and_skills_-_+as_at_(\d{1,2})_([A-Za-z]+)_(\d{4})\.csv$")
+_PERIOD = re.compile(
+    r"further_education_and_skills_-_+(\d{1,2})_([A-Za-z]+)_(\d{4})_to_(\d{1,2})_([A-Za-z]+)_(\d{4})\.csv$")
+
+# Ofsted column → Singer field.
+PROVIDER_COLUMNS = {
+    "Provider URN": "provider_urn",
+    "Provider UKPRN": "provider_ukprn",
+    "Provider name": "provider_name",
+    "Provider type": "provider_type",
+    "Has the provider received a full inspection under the renewed EIF?": "renewed_eif_full_inspection",
+    "First day of inspection of latest inspection before renewed EIF": "graded_inspection_date",
+    "Latest overall effectiveness before renewed EIF": "overall_effectiveness",
+    "Date of latest short inspection before renewed EIF (if relevant)": "short_inspection_date",
+}
+INSPECTION_COLUMNS = {
+    "Provider URN": "provider_urn",
+    "Provider UKPRN": "provider_ukprn",
+    "Inspection number": "inspection_number",
+    "First day of inspection": "first_day_of_inspection",
+    "Date published": "date_published",
+    "Safeguarding standards": "safeguarding_standards",
+}
+
+
+def _date(day: str, month: str, year: str) -> date | None:
+    number = _MONTHS.get(month.lower())
+    return date(int(year), number, int(day)) if number else None
+
+
+def _page_start(title: str) -> date | None:
+    match = _TITLE_START.search(title)
+    number = _MONTHS.get(match.group(1).lower()) if match else None
+    return date(int(match.group(2)), number, 1) if number else None
+
+
+def _dated_pages(results: list[dict]) -> list[tuple[date, str]]:
+    pages = []
+    for result in results:
+        title = result.get("title", "")
+        start = _page_start(title) if title.startswith(PAGE_TITLE_PREFIX) else None
+        if start:
+            pages.append((start, result["link"]))
+    return sorted(pages, reverse=True)
+
+
+def fe_pages(results: list[dict]) -> list[str]:
+    """gov.uk paths of the FE MI pages, newest academic year first."""
+    return [link for _, link in _dated_pages(results)]
+
+
+def renewed_framework_pages(results: list[dict]) -> list[str]:
+    """The pages that can hold report cards: from November 2025 onwards."""
+    return [link for start, link in _dated_pages(results) if start >= RENEWED_FRAMEWORK_START]
+
+
+def _latest(html: str, pattern: re.Pattern, date_groups: slice) -> str | None:
+    best = None
+    for link in set(_CSV_LINK.findall(html)):
+        match = pattern.search(link)
+        when = _date(*match.groups()[date_groups]) if match else None
+        if when and (best is None or when > best[0]):
+            best = (when, link)
+    return best[1] if best else None
+
+
+def latest_as_at_csv(html: str) -> str | None:
+    """The newest "as at" CSV on a page."""
+    return _latest(html, _AS_AT, slice(0, 3))
+
+
+def latest_period_csv(html: str) -> str | None:
+    """The renewed-framework inspections CSV with the latest end date on a page."""
+    return _latest(html, _PERIOD, slice(3, 6))
+
+
+def providers_csv_url(pages: list[str], fetch: Callable[[str], str]) -> str | None:
+    """The newest "as at" CSV, from the newest page that has one."""
+    for page in pages:
+        url = latest_as_at_csv(fetch(page))
+        if url:
+            return url
+    return None
+
+
+def read_mi_csv(content: bytes) -> pd.DataFrame:
+    """Ofsted's MI CSVs: UTF-8 with a BOM, or Windows-1252. The header is the
+    line naming "Provider URN"."""
+    try:
+        text = content.decode("utf-8-sig")
+    except UnicodeDecodeError:
+        text = content.decode("cp1252")
+    lines = text.splitlines()
+    start = next((i for i, line in enumerate(lines[:20]) if "Provider URN" in line), 0)
+    return pd.read_csv(io.StringIO("\n".join(lines[start:])), dtype=str, keep_default_na=False)
+
+
+def records(df: pd.DataFrame, columns: dict[str, str], key: str) -> Iterator[dict]:
+    """Singer records: the mapped columns, stripped. Rows without `key` are skipped."""
+    present = {source: field for source, field in columns.items() if source in df.columns}
+    for _, row in df.iterrows():
+        record = {field: str(row[source]).strip() for source, field in present.items()}
+        if record.get(key):
+            yield record

pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/tap.py

@@ -10,6 +10,19 @@ import requests
 from singer_sdk import Stream, Tap
 from singer_sdk import typing as th
 
+from tap_uk_ofsted.fe_mi import (
+    INSPECTION_COLUMNS,
+    PROVIDER_COLUMNS,
+    SEARCH_QUERY,
+    SEARCH_URL,
+    fe_pages,
+    latest_period_csv,
+    providers_csv_url,
+    read_mi_csv,
+    records,
+    renewed_framework_pages,
+)
+
 GOV_UK_PAGE = (
     "https://www.gov.uk/government/statistical-data-sets/"
     "monthly-management-information-ofsteds-school-inspections-outcomes"
@@ -304,6 +317,71 @@ class OfstedInspectionsStream(Stream):
             self.logger.error("Could not discover Ofsted independent MI download URL")
 
 
+def _fe_search_results() -> list[dict]:
+    """gov.uk search results for Ofsted's FE MI pages (one per academic year)."""
+    resp = requests.get(SEARCH_URL, params={"q": SEARCH_QUERY, "count": 50}, timeout=60)
+    resp.raise_for_status()
+    return resp.json().get("results", [])
+
+
+def _gov_uk_page(path: str) -> str:
+    url = path if path.startswith("http") else "https://www.gov.uk" + path
+    resp = requests.get(url, timeout=60)
+    resp.raise_for_status()
+    return resp.text
+
+
+class OfstedFEProvidersStream(Stream):
+    """Ofsted's further education and skills providers: the newest "as at"
+    file, one row per provider with its latest inspection under the previous
+    framework (audit H4). Page and file selection: fe_mi.py."""
+
+    name = "ofsted_fe_providers"
+    primary_keys = ["provider_urn"]
+    replication_key = None
+
+    schema = th.PropertiesList(
+        th.Property("provider_urn", th.StringType, required=True),
+        *[th.Property(field, th.StringType) for field in PROVIDER_COLUMNS.values() if field != "provider_urn"],
+    ).to_dict()
+
+    def get_records(self, context):
+        url = providers_csv_url(fe_pages(_fe_search_results()), _gov_uk_page)
+        if not url:
+            self.logger.error("No Ofsted FE 'as at' file found")
+            return
+        self.logger.info("Downloading Ofsted FE providers MI: %s", url)
+        resp = requests.get(url, timeout=120)
+        resp.raise_for_status()
+        yield from records(read_mi_csv(resp.content), PROVIDER_COLUMNS, "provider_urn")
+
+
+class OfstedFEInspectionsStream(Stream):
+    """Ofsted's renewed-framework FE inspections (report cards), from the
+    latest cumulative file on every page since November 2025, so a report
+    card survives the next academic year's page (audit H4)."""
+
+    name = "ofsted_fe_inspections"
+    primary_keys = ["inspection_number"]
+    replication_key = None
+
+    schema = th.PropertiesList(
+        th.Property("inspection_number", th.StringType, required=True),
+        *[th.Property(field, th.StringType) for field in INSPECTION_COLUMNS.values() if field != "inspection_number"],
+    ).to_dict()
+
+    def get_records(self, context):
+        for page in renewed_framework_pages(_fe_search_results()):
+            url = latest_period_csv(_gov_uk_page(page))
+            if not url:
+                self.logger.warning("No renewed-framework FE inspections file on %s", page)
+                continue
+            self.logger.info("Downloading Ofsted FE inspections MI: %s", url)
+            resp = requests.get(url, timeout=120)
+            resp.raise_for_status()
+            yield from records(read_mi_csv(resp.content), INSPECTION_COLUMNS, "inspection_number")
+
+
 class TapUKOfsted(Tap):
     """Singer tap for UK Ofsted Management Information."""
 
@@ -315,7 +393,7 @@ class TapUKOfsted(Tap):
     ).to_dict()
 
     def discover_streams(self):
-        return [OfstedInspectionsStream(self)]
+        return [OfstedInspectionsStream(self), OfstedFEProvidersStream(self), OfstedFEInspectionsStream(self)]
 
 
 if __name__ == "__main__":

pipeline/tests/test_ofsted_fe_mi.py

@@ -0,0 +1,138 @@
+"""Ofsted publishes further-education inspection outcomes on a gov.uk page per
+academic year, separate from the school MI the tap already reads (audit H4).
+The links below are the real ones from the 2025/26 page; the headers are
+Ofsted's own.
+"""
+import importlib.util
+from pathlib import Path
+
+import pandas as pd
+import pytest
+
+MODULE = (Path(__file__).resolve().parents[1] / 'plugins' / 'extractors' / 'tap-uk-ofsted'
+          / 'tap_uk_ofsted' / 'fe_mi.py')
+
+A = 'https://assets.publishing.service.gov.uk/media/'
+PAGE_2025_26 = f'''
+<a href="{A}6989b16dc106cc5501b6e4c3/Management_information_-_further_education_and_skills_-_as_at_31_January_2026.csv">x</a>
+<a href="{A}6989b18cd3f57710b50a9b92/Management_information_-_further_education_and_skills_-_10_November_2025_to_31_January_2026.csv">x</a>
+<a href="{A}69aec8d6d620c14fa183eeb6/Management_information_-_further_education_and_skills_-__10_November_2025_to_28_February_2026.csv">x</a>
+<a href="{A}69aec8fcc78869bf8eb8a54a/Management_information_-_further_education_and_skills_-_as_at_28_February_2026.csv">x</a>
+<a href="{A}69aef091bde9c3f213c89a1e/Management_information_-_further_education_and_skills_-_as_at_31_January_2026.ods">x</a>
+<a href="{A}6a9fec1e92e72b8ac437eef3/Management_information_-_further_education_and_skills_-_as_at_31_August_2026.csv">x</a>
+<a href="{A}6a9fed6b5f6e942efe37eed4/Management_information_-_new_provider_monitoring_inspections_-_10_November_2025_to_31_August_2026.csv">x</a>
+<a href="{A}6a9fed815f6e942efe37eed5/Management_information_-_monitoring_inspections_-_10_November_2025_to_31_August_2026.csv">x</a>
+<a href="{A}6a9feda65f6e942efe37eed6/Management_information_-_further_education_and_skills_-_10_November_2025_to_31_August_2026.csv">x</a>
+'''
+
+SEARCH_RESULTS = [
+    {'title': 'Further education and skills inspections and outcomes: management information from September 2024 to August 2025',
+     'link': '/government/statistical-data-sets/fe-2024-25'},
+    {'title': 'Further education and skills inspections and outcomes: management information from November 2025 to August 2026',
+     'link': '/government/statistical-data-sets/fe-2025-26'},
+    {'title': 'Further education and skills inspections and outcomes: management information from September 2023 to August 2024',
+     'link': '/government/statistical-data-sets/fe-2023-24'},
+    {'title': 'Monthly management information: Ofsted\'s school inspections outcomes',
+     'link': '/government/statistical-data-sets/schools'},
+]
+
+PROVIDER_HEADER = (
+    'Provider URN', 'Provider UKPRN', 'Provider name', 'Provider type', 'Provider group',
+    'Local authority', 'Region', 'Ofsted region',
+    'Has the provider received a full inspection under the renewed EIF?',
+    'Has the provider received a new provider monitoring inspection?',
+    'Date of latest short inspection before renewed EIF (if relevant)',
+    'Inspection number of latest full inspection before renewed EIF',
+    'First day of inspection of latest inspection before renewed EIF',
+    'Latest overall effectiveness before renewed EIF',
+    'Latest quality of education before renewed EIF',
+    'Latest behaviour and attitudes before renewed EIF',
+    'Latest personal development before renewed EIF',
+    'Latest leadership and management before renewed EIF',
+    'Latest contribution to meeting skills needs before renewed EIF',
+    'Latest safeguarding before renewed EIF', '',
+)
+
+INSPECTION_HEADER = (
+    'Provider URN', 'Provider UKPRN', 'Provider name', 'Provider type', 'Provider group',
+    'Local authority', 'Region', 'Ofsted region', 'Inspection number', 'First day of inspection',
+    'Date published', 'Safeguarding standards', 'Inclusion', 'Leadership and governance',
+    'Contribution to meeting skills needs',
+    'Education programmes for young people - curriculum, teaching and training',
+    'Education programmes for young people - achievement',
+    'Education programmes for young people - participation and development',
+    'Adult learning programmes - curriculum, teaching and training',
+    'Adult learning programmes - achievement',
+    'Adult learning programmes - participation and development',
+    'Apprenticeships - curriculum, teaching and training', 'Apprenticeships - achievement',
+    'Apprenticeships - participation and development',
+    'Provision for learners and apprentices with high needs - curriculum, teaching and training',
+    'Provision for learners and apprentices with high needs - achievement',
+    'Provision for learners and apprentices with high needs - participation and development', '',
+)
+
+
+@pytest.fixture
+def fe():
+    spec = importlib.util.spec_from_file_location('fe_mi', MODULE)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_the_newest_as_at_file_is_chosen(fe):
+    assert fe.latest_as_at_csv(PAGE_2025_26).endswith('_as_at_31_August_2026.csv')
+
+
+def test_the_latest_cumulative_inspections_file_is_chosen_not_a_monitoring_one(fe):
+    assert fe.latest_period_csv(PAGE_2025_26) == (
+        A + '6a9feda65f6e942efe37eed6/Management_information_-_further_education_and_skills_'
+            '-_10_November_2025_to_31_August_2026.csv')
+
+
+def test_a_doubled_underscore_file_name_still_parses(fe):
+    page = '\n'.join(line for line in PAGE_2025_26.splitlines() if 'August' not in line)
+    assert fe.latest_period_csv(page).endswith('__10_November_2025_to_28_February_2026.csv')
+
+
+def test_monitoring_files_alone_give_nothing(fe):
+    page = '\n'.join(line for line in PAGE_2025_26.splitlines() if 'monitoring' in line)
+    assert fe.latest_period_csv(page) is None
+    assert fe.latest_as_at_csv(page) is None
+
+
+def test_pages_are_the_fe_series_newest_first(fe):
+    assert fe.fe_pages(SEARCH_RESULTS) == [
+        '/government/statistical-data-sets/fe-2025-26',
+        '/government/statistical-data-sets/fe-2024-25',
+        '/government/statistical-data-sets/fe-2023-24',
+    ]
+
+
+def test_report_cards_are_read_only_from_renewed_framework_pages(fe):
+    # Earlier pages hold previous-framework inspection lists, not report cards.
+    assert fe.renewed_framework_pages(SEARCH_RESULTS) == ['/government/statistical-data-sets/fe-2025-26']
+
+
+def test_providers_fall_back_to_the_previous_page(fe):
+    pages = {'/new': '<p>No files yet</p>', '/old': PAGE_2025_26}
+    assert fe.providers_csv_url(['/new', '/old'], pages.__getitem__).endswith('_as_at_31_August_2026.csv')
+
+
+@pytest.mark.parametrize('columns, header', [('PROVIDER_COLUMNS', PROVIDER_HEADER),
+                                             ('INSPECTION_COLUMNS', INSPECTION_HEADER)])
+def test_every_mapped_column_is_in_ofsteds_file(fe, columns, header):
+    assert set(getattr(fe, columns)) - set(header) == set()
+
+
+def test_csv_reading_and_records(fe):
+    content = ('' + ','.join(f'"{c}"' for c in PROVIDER_HEADER[:14]) + '\n'
+               + '145175,10005429,Queen Mary\'s College,Sixth form college,Sixth form colleges,Hampshire,'
+                 'South East,South East,No,No,-,10227183,06/12/2022,2\n'
+               + ',,Blank provider,,,,,,,,,,,\n').encode('utf-8')
+    rows = list(fe.records(fe.read_mi_csv(content), fe.PROVIDER_COLUMNS, 'provider_urn'))
+    assert rows == [{
+        'provider_urn': '145175', 'provider_ukprn': '10005429', 'provider_name': "Queen Mary's College",
+        'provider_type': 'Sixth form college', 'renewed_eif_full_inspection': 'No',
+        'graded_inspection_date': '06/12/2022', 'overall_effectiveness': '2', 'short_inspection_date': '-',
+    }]