From 8abff7a0a1946d0f83498ad544444aa7c9522da5 Mon Sep 17 00:00:00 2001 From: Tudor Date: Wed, 15 Jul 2026 23:21:40 +0100 Subject: [PATCH] feat: ingest independent schools in Ofsted tap and dbt staging --- pipeline/meltano.yml | 3 + .../tap-uk-ofsted/tap_uk_ofsted/tap.py | 113 ++++++++++++++---- .../models/staging/stg_ofsted_inspections.sql | 8 +- 3 files changed, 99 insertions(+), 25 deletions(-) diff --git a/pipeline/meltano.yml b/pipeline/meltano.yml index cfcda23..856ac86 100644 --- a/pipeline/meltano.yml +++ b/pipeline/meltano.yml @@ -49,6 +49,9 @@ plugins: - name: mi_url kind: string description: Ofsted Management Information download URL + - name: independent_mi_url + kind: string + description: Ofsted Independent Schools Management Information download URL - name: tap-uk-fbit namespace: uk_fbit 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 eaacbd7..d58d7b5 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 @@ -2,6 +2,7 @@ from __future__ import annotations +from datetime import datetime import io import re @@ -14,20 +15,28 @@ GOV_UK_PAGE = ( "monthly-management-information-ofsteds-school-inspections-outcomes" ) +INDEPENDENT_GOV_UK_PAGE = ( + "https://www.gov.uk/government/statistical-data-sets/" + "non-association-independent-schools-inspections-and-outcomes-management-information" +) + # Column name → internal field, in priority order (first match wins). # Handles both current and older file formats. COLUMN_PRIORITY = { "urn": ["URN", "Urn", "urn"], "inspection_date": [ "Inspection start date of latest OEIF graded inspection", + "Inspection start date of latest OEIF standard inspection", "Inspection start date", "Inspection date", ], "inspection_type": [ "Inspection type of latest OEIF graded inspection", + "Inspection type of latest OEIF standard inspection", "Inspection type", ], "event_type_grouping": [ + "Event type grouping of latest OEIF standard inspection", "Event type grouping", "Inspection type grouping", ], @@ -52,10 +61,12 @@ COLUMN_PRIORITY = { "Effectiveness of leadership and management", ], "early_years_provision": [ + "Latest OEIF early years provision (where applicable)", "Latest OEIF early years provision", "Early years provision (where applicable)", ], "sixth_form_provision": [ + "Latest OEIF sixth form provision (where applicable)", "Latest OEIF sixth form provision", "Sixth form provision (where applicable)", ], @@ -68,12 +79,7 @@ COLUMN_PRIORITY = { "ungraded_inspection_date": [ "Date of latest ungraded inspection", ], - # Report Card fields (post-Nov 2025 framework). Confirmed verbatim MI - # headers per diagnose_compare_gaps.py's Task 1(c) findings. No MI column - # currently exists for early-years or sixth-form report-card grades, so - # those two fields are deliberately omitted here (see schema below) -- - # they stay absent from every record, same as the existing `report_url` - # pattern for fields with no COLUMN_PRIORITY entry. + # Report Card fields (post-Nov 2025 framework). "rc_safeguarding_met": ["Safeguarding standards"], "rc_inclusion": ["Inclusion"], "rc_curriculum_teaching": ["Curriculum and teaching"], @@ -81,6 +87,13 @@ COLUMN_PRIORITY = { "rc_attendance_behaviour": ["Attendance and behaviour"], "rc_personal_development": ["Personal development and wellbeing"], "rc_leadership_governance": ["Leadership and governance"], + "rc_early_years": ["Early years (where applicable)"], + "rc_sixth_form": ["Post-16 provision (where applicable)"], + "report_url": [ + "Web Link (opens in new window)", + "Web link to Ofsted provider page", + "Web link", + ], } @@ -103,6 +116,51 @@ def discover_csv_url() -> str | None: return matches[0] if matches else None +def discover_independent_csv_url() -> str | None: + """Scrape GOV.UK page to find the latest independent schools MI CSV download link.""" + resp = requests.get(INDEPENDENT_GOV_UK_PAGE, timeout=30) + resp.raise_for_status() + # Look for CSV attachment links + csv_links = re.findall( + r'href="(https://assets\.publishing\.service\.gov\.uk/[^"]+\.csv)"', + resp.text, + ) + if not csv_links: + # Fall back to ODS + csv_links = re.findall( + r'href="(https://assets\.publishing\.service\.gov\.uk/[^"]+\.ods)"', + resp.text, + ) + + months = { + 'january': 1, 'february': 2, 'march': 3, 'april': 4, 'may': 5, 'june': 6, + 'july': 7, 'august': 8, 'september': 9, 'october': 10, 'november': 11, 'december': 12 + } + + parsed_links = [] + for link in csv_links: + normalized_link = link.lower().replace('-', '_') + if 'most_recent' not in normalized_link: + continue + + match = re.search(r'as_at_(\d{1,2})_([a-z]+)_(\d{4})', normalized_link) + if match: + day, month_str, year = match.groups() + month = months.get(month_str) + if month: + try: + dt = datetime(int(year), month, int(day)) + parsed_links.append((dt, link)) + except ValueError: + continue + + parsed_links.sort(reverse=True) + if parsed_links: + return parsed_links[0][1] + + return csv_links[0] if csv_links else None + + class OfstedInspectionsStream(Stream): """Stream: Ofsted inspection records.""" @@ -131,8 +189,6 @@ class OfstedInspectionsStream(Stream): th.Property("rc_attendance_behaviour", th.StringType), th.Property("rc_personal_development", th.StringType), th.Property("rc_leadership_governance", th.StringType), - # No MI column exists for these yet; declared for forward - # compatibility with the mart schema, always emitted as absent/NULL. th.Property("rc_early_years", th.StringType), th.Property("rc_sixth_form", th.StringType), th.Property("report_url", th.StringType), @@ -148,15 +204,8 @@ class OfstedInspectionsStream(Stream): break return mapping - def get_records(self, context): - import pandas as pd - - url = self.config.get("mi_url") or discover_csv_url() - if not url: - self.logger.error("Could not discover Ofsted MI download URL") - return - - self.logger.info("Downloading Ofsted MI: %s", url) + def _fetch_and_parse_url(self, url: str, pd) -> list[dict]: + """Download file and parse records.""" resp = requests.get(url, timeout=120) resp.raise_for_status() @@ -172,8 +221,6 @@ class OfstedInspectionsStream(Stream): lines = text.split("\n") header_idx = 0 for i, line in enumerate(lines[:20]): - # Match lines where URN appears as a CSV field (start or after comma), - # not as a substring of words like "turn" or "return". if re.search(r'(?:^|,)\s*URN\s*(?:,|$)', line): header_idx = i break @@ -190,8 +237,14 @@ class OfstedInspectionsStream(Stream): for _, row in df.iterrows(): record = {} + for key in self.schema["properties"].keys(): + record[key] = None + for field, col in col_map.items(): - record[field] = row.get(col, None) + val = row.get(col, None) + if val == 'NULL': + val = None + record[field] = val # Cast URN try: @@ -201,6 +254,25 @@ class OfstedInspectionsStream(Stream): yield record + def get_records(self, context): + import pandas as pd + + # 1. State-funded schools + state_url = self.config.get("mi_url") or discover_csv_url() + if state_url: + self.logger.info("Downloading Ofsted state-funded MI: %s", state_url) + yield from self._fetch_and_parse_url(state_url, pd) + else: + self.logger.error("Could not discover Ofsted state-funded MI download URL") + + # 2. Independent schools + ind_url = self.config.get("independent_mi_url") or discover_independent_csv_url() + if ind_url: + self.logger.info("Downloading Ofsted independent MI: %s", ind_url) + yield from self._fetch_and_parse_url(ind_url, pd) + else: + self.logger.error("Could not discover Ofsted independent MI download URL") + class TapUKOfsted(Tap): """Singer tap for UK Ofsted Management Information.""" @@ -209,6 +281,7 @@ class TapUKOfsted(Tap): config_jsonschema = th.PropertiesList( th.Property("mi_url", th.StringType, description="Direct URL to Ofsted MI file"), + th.Property("independent_mi_url", th.StringType, description="Direct URL to Ofsted Independent Schools MI file"), ).to_dict() def discover_streams(self): diff --git a/pipeline/transform/models/staging/stg_ofsted_inspections.sql b/pipeline/transform/models/staging/stg_ofsted_inspections.sql index b4e06de..518b24a 100644 --- a/pipeline/transform/models/staging/stg_ofsted_inspections.sql +++ b/pipeline/transform/models/staging/stg_ofsted_inspections.sql @@ -46,12 +46,10 @@ renamed as ( {{ parse_report_card_grade('rc_attendance_behaviour') }}::integer as rc_attendance_behaviour, {{ parse_report_card_grade('rc_personal_development') }}::integer as rc_personal_development, {{ parse_report_card_grade('rc_leadership_governance') }}::integer as rc_leadership_governance, - -- No MI column exists for these yet (see tap.py); the tap never - -- emits rc_early_years/rc_sixth_form, so these stay NULL. - null::integer as rc_early_years, - null::integer as rc_sixth_form, + {{ parse_report_card_grade('rc_early_years') }}::integer as rc_early_years, + {{ parse_report_card_grade('rc_sixth_form') }}::integer as rc_sixth_form, - report_url + nullif(trim(report_url), 'NULL') as report_url from source where urn is not null and (