feat(pipeline): carry the report-card inspection's own date; pick newest MI file in discovery

The MI file's report-card grade columns belong to the latest FULL
inspection (col 'Inspection start date'), but inspection_date maps to the
legacy OEIF graded/ungraded dates — so report cards were being dated with
pre-Nov-2025 inspections. Also discover_csv_url() returned matches[0],
the oldest (2017) link on the GOV.UK page.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0146VHeLAWjDVE2B5uU67jCB

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

@@ -89,6 +89,10 @@ COLUMN_PRIORITY = {
     "rc_leadership_governance": ["Leadership and governance"],
     "rc_early_years": ["Early years (where applicable)"],
     "rc_sixth_form": ["Post-16 provision (where applicable)"],
+    # Date of the latest FULL inspection — in the renewed framework this is
+    # the report-card inspection's own start date (col "Inspection start
+    # date"), distinct from the legacy OEIF graded/ungraded dates above.
+    "rc_inspection_date": ["Inspection start date"],
     "report_url": [
         "Web Link (opens in new window)",
         "Web link to Ofsted provider page",
@@ -98,16 +102,44 @@ COLUMN_PRIORITY = {
 
 
 def discover_csv_url() -> str | None:
-    """Scrape GOV.UK page to find the latest MI CSV download link."""
+    """Scrape GOV.UK page to find the latest MI CSV download link.
+
+    The page lists a decade of monthly files, oldest first — take the
+    newest 'latest inspections as at <date>' link by parsing its date,
+    never matches[0] (that is a 2017 file).
+    """
     resp = requests.get(GOV_UK_PAGE, timeout=30)
     resp.raise_for_status()
-    # Look for CSV attachment links
-    matches = re.findall(
+    csv_links = re.findall(
         r'href="(https://assets\.publishing\.service\.gov\.uk/[^"]+\.csv)"',
         resp.text,
     )
-    if matches:
-        return matches[0]
+
+    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,
+        'jan': 1, 'feb': 2, 'mar': 3, 'apr': 4, 'jun': 6,
+        'jul': 7, 'aug': 8, 'sep': 9, 'oct': 10, 'nov': 11, 'dec': 12,
+    }
+    parsed_links = []
+    for link in csv_links:
+        normalized = link.lower().replace('-', '_')
+        if 'latest_inspections_as_at' not in normalized:
+            continue
+        match = re.search(r'as_at_(\d{1,2})_([a-z]+)_(\d{4})', normalized)
+        if match:
+            day, month_str, year = match.groups()
+            month = months.get(month_str)
+            if month:
+                try:
+                    parsed_links.append((datetime(int(year), month, int(day)), link))
+                except ValueError:
+                    continue
+    parsed_links.sort(reverse=True)
+    if parsed_links:
+        return parsed_links[0][1]
+    if csv_links:
+        return csv_links[-1]
     # Fall back to ODS
     matches = re.findall(
         r'href="(https://assets\.publishing\.service\.gov\.uk/[^"]+\.ods)"',
@@ -191,6 +223,7 @@ class OfstedInspectionsStream(Stream):
         th.Property("rc_leadership_governance", th.StringType),
         th.Property("rc_early_years", th.StringType),
         th.Property("rc_sixth_form", th.StringType),
+        th.Property("rc_inspection_date", th.StringType),
         th.Property("report_url", th.StringType),
     ).to_dict()
 

pipeline/transform/models/intermediate/int_ofsted_latest.sql

@@ -34,6 +34,7 @@ select
     rc_leadership_governance,
     rc_early_years,
     rc_sixth_form,
+    rc_inspection_date,
     report_url
 from ranked
 where rn = 1

pipeline/transform/models/marts/fact_ofsted_inspection.sql

@@ -23,5 +23,6 @@ select
     rc_leadership_governance,
     rc_early_years,
     rc_sixth_form,
+    rc_inspection_date,
     report_url
 from {{ ref('stg_ofsted_inspections') }}

pipeline/transform/models/staging/stg_ofsted_inspections.sql

@@ -49,6 +49,12 @@ renamed as (
         {{ parse_report_card_grade('rc_early_years') }}::integer          as rc_early_years,
         {{ parse_report_card_grade('rc_sixth_form') }}::integer           as rc_sixth_form,
 
+        -- Start date of the latest FULL inspection (the report-card
+        -- inspection in the renewed framework). Guarded in the final select:
+        -- only kept when the row actually carries report-card grades, because
+        -- in legacy-format files this column is the legacy inspection date.
+        to_date(nullif(trim(rc_inspection_date), 'NULL'), 'DD/MM/YYYY')   as rc_inspection_date_raw,
+
         nullif(trim(report_url), 'NULL')                                  as report_url
     from source
     where urn is not null
@@ -58,5 +64,17 @@ renamed as (
       )
 )
 
-select * from renamed
+select
+    *,
+    case
+        when rc_safeguarding_met is not null
+          or rc_inclusion is not null
+          or rc_curriculum_teaching is not null
+          or rc_achievement is not null
+          or rc_attendance_behaviour is not null
+          or rc_personal_development is not null
+          or rc_leadership_governance is not null
+        then rc_inspection_date_raw
+    end as rc_inspection_date
+from renamed
 where inspection_date is not null