feat: ingest independent schools in Ofsted tap and dbt staging

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

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
+    # Report Card fields (post-Nov 2025 framework).
-    # 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.
     "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):
+    def _fetch_and_parse_url(self, url: str, pd) -> list[dict]:
-        import pandas as pd
+        """Download file and parse records."""
-
-        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)
         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):

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
+        {{ parse_report_card_grade('rc_early_years') }}::integer          as rc_early_years,
-        -- emits rc_early_years/rc_sixth_form, so these stay NULL.
+        {{ parse_report_card_grade('rc_sixth_form') }}::integer           as rc_sixth_form,
-        null::integer as rc_early_years,
-        null::integer as rc_sixth_form,
 
-        report_url
+        nullif(trim(report_url), 'NULL')                                  as report_url
     from source
     where urn is not null
       and (