diff --git a/pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/fe_mi.py b/pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/fe_mi.py new file mode 100644 index 0000000..9ba7df2 --- /dev/null +++ b/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 diff --git a/pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/tap.py b/pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/tap.py index ff49cf2..ae91d29 100644 --- a/pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/tap.py +++ b/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__": diff --git a/pipeline/tests/test_ofsted_fe_mi.py b/pipeline/tests/test_ofsted_fe_mi.py new file mode 100644 index 0000000..7fd9310 --- /dev/null +++ b/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''' +x +x +x +x +x +x +x +x +x +''' + +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': '

No files yet

', '/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': '-', + }]