school_compare/backend/app.py

"""
SchoolCompare.co.uk API
Serves primary and secondary school performance data for comparing schools.
Uses real data from UK Government Compare School Performance downloads.
"""

import hashlib
import re
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Optional

import numpy as np
import pandas as pd
from fastapi import FastAPI, HTTPException, Query, Request, Depends, Header
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, Response
from fastapi.staticfiles import StaticFiles
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from starlette.middleware.base import BaseHTTPMiddleware

import asyncio
from .config import settings
from .data_loader import (
    clear_cache,
    compute_benchmarks,
    load_school_data,
    load_latest_school_data,
    geocode_single_postcode,
    get_supplementary_data,
    get_supplementary_data_batch,
    search_schools_typesense,
)
from .data_loader import get_data_info as get_db_info
from .schemas import METRIC_DEFINITIONS, RANKING_COLUMNS, SCHOOL_COLUMNS
from .utils import clean_for_json, convert_to_native

# Values to exclude from filter dropdowns (empty strings, non-applicable labels)
EXCLUDED_FILTER_VALUES = {"", "Not applicable", "Does not apply"}

# Maps user-facing phase filter values to the GIAS PhaseOfEducation values they include.
# All-through schools appear in both primary and secondary results.
PHASE_GROUPS: dict[str, set[str]] = {
    "primary": {"primary", "middle deemed primary", "all-through"},
    "secondary": {"secondary", "middle deemed secondary", "all-through", "16 plus"},
    "all-through": {"all-through"},
}

# Must match SITE_URL in nextjs-app/lib/site.ts. The apex 301s to www, and a
# sitemap <loc> that redirects wastes a crawl on every URL it lists.
BASE_URL = "https://www.schoolcompare.co.uk"
MAX_SLUG_LENGTH = 60

# In-memory sitemap cache: name -> XML. Populated on startup and by the admin
# regenerate endpoint after a pipeline run.
_sitemaps: dict[str, str] | None = None


def _slugify(text: str) -> str:
    text = text.lower()
    text = re.sub(r"[^\w\s-]", "", text)
    text = re.sub(r"\s+", "-", text)
    text = re.sub(r"-+", "-", text)
    return text.strip("-")


def _school_url(urn: int, school_name: str) -> str:
    slug = _slugify(school_name)
    if len(slug) > MAX_SLUG_LENGTH:
        slug = slug[:MAX_SLUG_LENGTH].rstrip("-")
    return f"/school/{urn}-{slug}"


# Routes worth submitting that are not a school page. /admissions was missing
# from the sitemap entirely despite being a static, indexable guide.
STATIC_SITEMAP_PATHS = ("/", "/rankings", "/compare", "/admissions")


def _has_publishable_data(row) -> bool:
    """True when a school page has something a search result could state.

    A school with no results in any year and no Ofsted grade renders an empty
    page. Submitting it spends crawl budget and drags the corpus-wide quality
    signal down, so it stays out of the sitemap. The page itself still resolves
    for anyone who has the URL.
    """
    for field in ("rwm_expected_pct", "attainment_8_score", "ofsted_grade"):
        value = row.get(field)
        if value is not None and not pd.isna(value):
            return True
    return False


def _url_element(loc: str, lastmod: str | None = None) -> str:
    """One <url> entry. No priority or changefreq — Google ignores both."""
    body = f"<loc>{loc}</loc>"
    if lastmod:
        body += f"<lastmod>{lastmod}</lastmod>"
    return f"  <url>{body}</url>"


def _school_sitemap_rows(df) -> list[str]:
    """A <url> element per school that has something to show.

    lastmod comes from the school's Ofsted date where there is one and is
    omitted otherwise. An always-now lastmod is a claim Google learns to
    distrust; an absent one honestly means "unknown".
    """
    if df.empty or "urn" not in df.columns or "school_name" not in df.columns:
        return []

    rows: list[str] = []
    seen: set[int] = set()

    # Latest row per URN first, so a school's most recent Ofsted date wins.
    ordered = df.sort_values("year", ascending=False) if "year" in df.columns else df

    for _, row in ordered.iterrows():
        urn = int(row["urn"])
        if urn in seen:
            continue
        seen.add(urn)
        if not _has_publishable_data(row):
            continue

        lastmod = None
        ofsted_date = row.get("ofsted_date")
        if ofsted_date is not None and not pd.isna(ofsted_date):
            lastmod = pd.Timestamp(ofsted_date).date().isoformat()

        rows.append(_url_element(
            BASE_URL + _school_url(urn, str(row["school_name"])), lastmod))

    return rows


# Sitemaps cap at 50,000 URLs per file. 10,000 keeps a child small enough to
# scan by eye in Search Console, which is the point of splitting at all:
# coverage is reported per submitted sitemap, so one file per page family is
# what makes an indexation problem attributable to a family.
SITEMAP_CHUNK_SIZE = 10_000

# Children are served under /sitemaps/ because Next.js only treats a whole
# bracketed path segment as dynamic — a route folder named "sitemap-[...parts]"
# is read as a literal static segment and never matches.
SITEMAP_CHILD_PREFIX = "/sitemaps"


def _urlset(rows: list[str]) -> str:
    return "\n".join([
        '<?xml version="1.0" encoding="UTF-8"?>',
        '<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">',
        *rows,
        "</urlset>",
    ])


def build_sitemaps() -> dict[str, str]:
    """Build the sitemap index and every child, keyed by name."""
    df = load_school_data()

    children: dict[str, str] = {
        "static.xml": _urlset(
            [_url_element(BASE_URL + path) for path in STATIC_SITEMAP_PATHS]),
    }

    school_rows = _school_sitemap_rows(df)
    # Always emit at least one school child, so the index shape is stable even
    # on an empty database.
    chunks = [school_rows[i:i + SITEMAP_CHUNK_SIZE]
              for i in range(0, len(school_rows), SITEMAP_CHUNK_SIZE)] or [[]]
    for n, chunk in enumerate(chunks, start=1):
        children[f"schools-{n}.xml"] = _urlset(chunk)

    # On a sitemap index, lastmod means "when this sitemap file last changed",
    # so generation time is the correct value here — unlike on a <url>, where
    # it would be a claim about content we cannot support.
    generated = datetime.now(timezone.utc).date().isoformat()
    index_rows = [
        f"  <sitemap><loc>{BASE_URL}{SITEMAP_CHILD_PREFIX}/{name}</loc>"
        f"<lastmod>{generated}</lastmod></sitemap>"
        for name in children
    ]
    index = "\n".join([
        '<?xml version="1.0" encoding="UTF-8"?>',
        '<sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">',
        *index_rows,
        "</sitemapindex>",
    ])
    return {**children, "sitemap.xml": index}


def build_sitemap() -> str:
    """The sitemap index. Kept for `lifespan` and the admin endpoint."""
    return build_sitemaps()["sitemap.xml"]


def clean_filter_values(series: pd.Series) -> list[str]:
    """Return sorted unique values from a Series, excluding NaN and junk labels."""
    return sorted(
        v for v in series.dropna().unique().tolist()
        if v not in EXCLUDED_FILTER_VALUES
    )


# =============================================================================
# SECURITY MIDDLEWARE & HELPERS
# =============================================================================

# Rate limiter
limiter = Limiter(key_func=get_remote_address)


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """Add security headers to all responses."""

    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)

        # Prevent clickjacking
        response.headers["X-Frame-Options"] = "DENY"

        # Prevent MIME type sniffing
        response.headers["X-Content-Type-Options"] = "nosniff"

        # XSS Protection (legacy browsers)
        response.headers["X-XSS-Protection"] = "1; mode=block"

        # Referrer policy
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"

        # Permissions policy (restrict browser features)
        response.headers["Permissions-Policy"] = (
            "geolocation=(), microphone=(), camera=(), payment=()"
        )

        # Content Security Policy
        response.headers["Content-Security-Policy"] = (
            "default-src 'self'; "
            "script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net https://unpkg.com https://analytics.schoolcompare.co.uk; "
            "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com https://cdn.jsdelivr.net https://unpkg.com; "
            "font-src 'self' https://fonts.gstatic.com; "
            "img-src 'self' data: https://*.tile.openstreetmap.org https://unpkg.com; "
            "connect-src 'self' https://cdn.jsdelivr.net https://*.tile.openstreetmap.org https://unpkg.com https://analytics.schoolcompare.co.uk; "
            "frame-ancestors 'none'; "
            "base-uri 'self'; "
            "form-action 'self' https://formsubmit.co;"
        )

        # HSTS (only enable if using HTTPS in production)
        response.headers["Strict-Transport-Security"] = (
            "max-age=31536000; includeSubDomains"
        )

        return response


# Per-path Cache-Control rules. Keys are matched as path prefixes (longest wins).
# Values: (max_age, s_maxage, stale_while_revalidate)
CACHE_RULES: list[tuple[str, tuple[int, int, int]]] = [
    ("/api/filters",            (300,  86400, 604800)),
    ("/api/metrics",            (300,  86400, 604800)),
    ("/api/national-averages",  (300,  86400, 604800)),
    ("/api/la-averages",        (300,  86400, 604800)),
    ("/api/data-info",          (300,  86400, 604800)),
    ("/api/schools/",           (300,  3600,  86400)),  # /api/schools/{urn}
    ("/api/rankings",           (60,   600,   3600)),
    ("/api/compare",            (60,   600,   3600)),
    ("/api/schools",            (30,   300,   1800)),   # search list
]


def _cache_control_for_path(path: str) -> Optional[str]:
    # Longest-prefix match
    best: Optional[tuple[int, tuple[int, int, int]]] = None
    for prefix, vals in CACHE_RULES:
        if path.startswith(prefix) and (best is None or len(prefix) > best[0]):
            best = (len(prefix), vals)
    if best is None:
        return None
    max_age, s_maxage, swr = best[1]
    return f"public, max-age={max_age}, s-maxage={s_maxage}, stale-while-revalidate={swr}"


class CacheAndETagMiddleware(BaseHTTPMiddleware):
    """Set Cache-Control on cacheable API responses and serve 304s via ETag."""

    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)

        # Only cache GETs that succeeded.
        if request.method != "GET" or response.status_code != 200:
            return response

        cache_header = _cache_control_for_path(request.url.path)
        if cache_header is None:
            return response

        # Drain body so we can hash it for ETag.
        body_chunks = []
        async for chunk in response.body_iterator:
            body_chunks.append(chunk)
        body = b"".join(body_chunks)

        etag = '"' + hashlib.md5(body).hexdigest() + '"'
        headers = dict(response.headers)
        headers["Cache-Control"] = cache_header
        headers["ETag"] = etag
        headers["Vary"] = ", ".join(filter(None, [headers.get("Vary"), "Accept-Encoding"]))

        inm = request.headers.get("if-none-match")
        if inm and inm == etag:
            # Strip content headers on 304.
            for h in ("Content-Length", "content-length", "Content-Type", "content-type"):
                headers.pop(h, None)
            return Response(status_code=304, headers=headers)

        return Response(content=body, status_code=200, headers=headers, media_type=response.media_type)


class RequestSizeLimitMiddleware(BaseHTTPMiddleware):
    """Limit request body size to prevent DoS attacks."""

    async def dispatch(self, request: Request, call_next):
        content_length = request.headers.get("content-length")
        if content_length:
            if int(content_length) > settings.max_request_size:
                return Response(
                    content="Request too large",
                    status_code=413,
                )
        return await call_next(request)


def verify_admin_api_key(x_api_key: str = Header(None)) -> bool:
    """Verify admin API key for protected endpoints."""
    if not x_api_key or x_api_key != settings.admin_api_key:
        raise HTTPException(
            status_code=401,
            detail="Invalid or missing API key",
            headers={"WWW-Authenticate": "ApiKey"},
        )
    return True


# Input validation helpers
def sanitize_search_input(value: Optional[str], max_length: int = 100) -> Optional[str]:
    """Sanitize search input to prevent injection attacks."""
    if value is None:
        return None
    # Strip whitespace and limit length
    value = value.strip()[:max_length]
    # Remove potentially dangerous characters (allow alphanumeric, spaces, common punctuation)
    value = re.sub(r"[^\w\s\-\',\.]", "", value)
    return value if value else None


def validate_postcode(postcode: Optional[str]) -> Optional[str]:
    """Validate and normalize UK postcode format."""
    if not postcode:
        return None
    postcode = postcode.strip().upper()
    # UK postcode pattern
    pattern = r"^[A-Z]{1,2}[0-9][A-Z0-9]?\s*[0-9][A-Z]{2}$"
    if not re.match(pattern, postcode):
        return None
    return postcode


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan - startup and shutdown events."""
    global _sitemaps
    print("Loading school data from marts...")
    df = load_school_data()
    if df.empty:
        print("Warning: No data in marts. Run the annual EES pipeline to populate KS2 data.")
    else:
        print(f"Data loaded successfully: {len(df)} records.")
        # Pre-compute the latest-year snapshot so the first search request is fast
        await asyncio.to_thread(load_latest_school_data)
    try:
        _sitemaps = build_sitemaps()
        n = sum(x.count("<url>") for x in _sitemaps.values())
        print(f"Sitemaps built: {len(_sitemaps)} files, {n} URLs.")
    except Exception as e:
        print(f"Warning: sitemap build failed on startup: {e}")

    yield

    print("Shutting down...")


app = FastAPI(
    title="SchoolCompare API",
    description="API for comparing primary and secondary school performance data - schoolcompare.co.uk",
    version="2.0.0",
    lifespan=lifespan,
    # Disable docs in production for security
    docs_url="/docs" if settings.debug else None,
    redoc_url="/redoc" if settings.debug else None,
    openapi_url="/openapi.json" if settings.debug else None,
)

# Add rate limiter
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# Middleware (Starlette runs the last-added middleware first on the way out,
# so list outermost-last: GZip wraps everything and compresses the final body).
app.add_middleware(CacheAndETagMiddleware)
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(RequestSizeLimitMiddleware)
app.add_middleware(GZipMiddleware, minimum_size=512)

# CORS middleware - restricted for production
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=False,  # Don't allow credentials unless needed
    allow_methods=["GET", "POST"],  # Only allow needed methods
    allow_headers=["Content-Type", "X-API-Key"],  # Only allow needed headers
)


@app.get("/")
async def root():
    """Serve the frontend."""
    return FileResponse(settings.frontend_dir / "index.html")


@app.get("/compare")
async def serve_compare():
    """Serve the frontend for /compare route (SPA routing)."""
    return FileResponse(settings.frontend_dir / "index.html")


@app.get("/rankings")
async def serve_rankings():
    """Serve the frontend for /rankings route (SPA routing)."""
    return FileResponse(settings.frontend_dir / "index.html")


@app.get("/api/config")
async def get_config():
    """Return public configuration for the frontend."""
    return {
        "ga_measurement_id": settings.ga_measurement_id
    }


@app.get("/api/schools")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_schools(
    request: Request,
    search: Optional[str] = Query(None, description="Search by school name", max_length=100),
    local_authority: Optional[str] = Query(
        None, description="Filter by local authority", max_length=100
    ),
    school_type: Optional[str] = Query(None, description="Filter by school type", max_length=100),
    phase: Optional[str] = Query(None, description="Filter by phase: primary, secondary, all-through", max_length=50),
    postcode: Optional[str] = Query(None, description="Search near postcode", max_length=10),
    radius: float = Query(5.0, ge=0.1, le=5, description="Search radius in miles"),
    page: int = Query(1, ge=1, le=1000, description="Page number"),
    page_size: int = Query(25, ge=1, le=500, description="Results per page"),
    gender: Optional[str] = Query(None, description="Filter by gender (Mixed/Boys/Girls)", max_length=50),
    admissions_policy: Optional[str] = Query(None, description="Filter by admissions policy", max_length=100),
    has_sixth_form: Optional[str] = Query(None, description="Filter by sixth form presence: yes/no", max_length=3),
):
    """
    Get list of schools with pagination.

    Returns paginated results with total count for efficient loading.
    Supports location-based search using postcode and phase filtering.
    """
    # Sanitize inputs
    search = sanitize_search_input(search)
    local_authority = sanitize_search_input(local_authority)
    school_type = sanitize_search_input(school_type)
    phase = sanitize_search_input(phase)
    postcode = validate_postcode(postcode)

    # Load the pre-computed latest-year snapshot (cached after first request / startup).
    # This avoids rebuilding the expensive groupby + prev-year merge on every search.
    df_latest = load_latest_school_data()

    if df_latest.empty:
        return {"schools": [], "total": 0, "page": page, "page_size": 0}

    # Use configured default if not specified
    if page_size is None:
        page_size = settings.default_page_size

    # Phase filter — uses PHASE_GROUPS so all-through/middle schools appear
    # in the correct phase(s) rather than being invisible to both filters.
    if phase:
        phase_lower = phase.lower().replace("_", "-")
        allowed = PHASE_GROUPS.get(phase_lower)
        if allowed:
            df_latest = df_latest[df_latest["phase"].str.lower().isin(allowed)]

    # Secondary-specific filters (after phase filter)
    if gender:
        df_latest = df_latest[df_latest["gender"].str.lower() == gender.lower()]
    if admissions_policy:
        df_latest = df_latest[df_latest["admissions_policy"].str.lower() == admissions_policy.lower()]
    # GIAS OfficialSixthForm flag (dim_school.has_sixth_form). NULL (flag not
    # yet populated by the pipeline) is treated as "no sixth form".
    if has_sixth_form in ("yes", "no"):
        if "has_sixth_form" in df_latest.columns:
            flag = df_latest["has_sixth_form"].eq(True)
        else:  # Defensive fallback only — data_loader now always synthesizes
            # has_sixth_form as NULL when the DB predates the pipeline re-run,
            # so this branch shouldn't normally trigger. Falls back to age
            # range if the column is somehow absent anyway.
            flag = df_latest["age_range"].str.contains("18", na=False)
        df_latest = df_latest[flag if has_sixth_form == "yes" else ~flag]

    # Include key result metrics for display on cards
    location_cols = ["latitude", "longitude"]
    result_cols = [
        "phase",
        "year",
        "rwm_expected_pct",
        "rwm_high_pct",
        "prev_rwm_expected_pct",
        "prev_attainment_8_score",
        "reading_expected_pct",
        "writing_expected_pct",
        "maths_expected_pct",
        "total_pupils",
        "attainment_8_score",
        "english_maths_standard_pass_pct",
    ]
    available_cols = [
        c
        for c in SCHOOL_COLUMNS + location_cols + result_cols
        if c in df_latest.columns
    ]
    # fact_performance guarantees one row per (urn, year); df_latest has one row per urn.
    schools_df = df_latest[available_cols]

    # Location-based search (uses pre-geocoded data from database)
    search_coords = None
    if postcode:
        # Offload the synchronous HTTP call to a thread so the event loop stays free
        coords = await asyncio.to_thread(geocode_single_postcode, postcode)
        if coords:
            search_coords = coords
            schools_df = schools_df.copy()

            # Filter by distance using pre-geocoded lat/long from database
            # Use vectorized haversine calculation for better performance
            lat1, lon1 = search_coords

            # Handle potential duplicate columns by taking first occurrence
            lat_col = schools_df.loc[:, "latitude"]
            lon_col = schools_df.loc[:, "longitude"]
            if isinstance(lat_col, pd.DataFrame):
                lat_col = lat_col.iloc[:, 0]
            if isinstance(lon_col, pd.DataFrame):
                lon_col = lon_col.iloc[:, 0]

            lat2 = lat_col.values
            lon2 = lon_col.values

            # Vectorized haversine formula
            R = 3959  # Earth's radius in miles
            lat1_rad = np.radians(lat1)
            lat2_rad = np.radians(lat2)
            dlat = np.radians(lat2 - lat1)
            dlon = np.radians(lon2 - lon1)

            a = np.sin(dlat / 2) ** 2 + np.cos(lat1_rad) * np.cos(lat2_rad) * np.sin(dlon / 2) ** 2
            c = 2 * np.arctan2(np.sqrt(a), np.sqrt(1 - a))
            distances = R * c

            # Handle missing coordinates
            has_coords = ~(pd.isna(lat_col) | pd.isna(lon_col))
            distances = np.where(has_coords.values, distances, float("inf"))
            schools_df["distance"] = distances
            schools_df = schools_df[schools_df["distance"] <= radius]
            schools_df = schools_df.sort_values("distance")

    # Apply filters
    if search:
        ts_urns = search_schools_typesense(search)
        if ts_urns:
            urn_order = {urn: i for i, urn in enumerate(ts_urns)}
            schools_df = schools_df[schools_df["urn"].isin(set(ts_urns))].copy()
            schools_df["_ts_rank"] = schools_df["urn"].map(urn_order)
            schools_df = schools_df.sort_values("_ts_rank").drop(columns=["_ts_rank"])
        else:
            # Fallback: Typesense unavailable, use substring match
            search_lower = search.lower()
            mask = schools_df["school_name"].str.lower().str.contains(search_lower, na=False)
            if "address" in schools_df.columns:
                mask = mask | schools_df["address"].str.lower().str.contains(search_lower, na=False)
            schools_df = schools_df[mask]

    if local_authority:
        schools_df = schools_df[
            schools_df["local_authority"].str.lower() == local_authority.lower()
        ]

    if school_type:
        schools_df = schools_df[
            schools_df["school_type"].str.lower() == school_type.lower()
        ]

    # Compute result-scoped filter values (before pagination).
    # Gender and admissions are secondary-only filters — scope them to schools
    # with KS4 data so they don't appear for purely primary result sets.
    _sec_mask = schools_df["attainment_8_score"].notna() if "attainment_8_score" in schools_df.columns else pd.Series(False, index=schools_df.index)
    result_filters = {
        "local_authorities": clean_filter_values(schools_df["local_authority"]) if "local_authority" in schools_df.columns else [],
        "school_types": clean_filter_values(schools_df["school_type"]) if "school_type" in schools_df.columns else [],
        "phases": clean_filter_values(schools_df["phase"]) if "phase" in schools_df.columns else [],
        "genders": clean_filter_values(schools_df.loc[_sec_mask, "gender"]) if "gender" in schools_df.columns and _sec_mask.any() else [],
        "admissions_policies": clean_filter_values(schools_df.loc[_sec_mask, "admissions_policy"]) if "admissions_policy" in schools_df.columns and _sec_mask.any() else [],
    }

    # Pagination
    total = len(schools_df)
    start_idx = (page - 1) * page_size
    end_idx = start_idx + page_size
    schools_df = schools_df.iloc[start_idx:end_idx]

    return {
        "schools": clean_for_json(schools_df),
        "total": total,
        "page": page,
        "page_size": page_size,
        "total_pages": (total + page_size - 1) // page_size if page_size > 0 else 0,
        "result_filters": result_filters,
        "location_info": {
            "postcode": postcode,
            "radius": radius * 1.60934,  # Convert miles to km for frontend display
            "coordinates": [search_coords[0], search_coords[1]]
        }
        if search_coords
        else None,
    }


@app.get("/api/schools/{urn}")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_school_details(request: Request, urn: int):
    """Get detailed performance data for a specific school across all years."""
    # Validate URN range (UK school URNs are 6 digits)
    if not (100000 <= urn <= 999999):
        raise HTTPException(status_code=400, detail="Invalid URN format")

    df = load_school_data()

    if df.empty:
        raise HTTPException(status_code=404, detail="No data available")

    school_data = df[df["urn"] == urn]

    if school_data.empty:
        raise HTTPException(status_code=404, detail="School not found")

    # Sort by year
    school_data = school_data.sort_values("year")

    # Get latest info for the school
    latest = school_data.iloc[-1]

    # Fetch supplementary data (Ofsted, admissions, etc.)
    from .database import SessionLocal
    supplementary = {}
    try:
        db = SessionLocal()
        supplementary = get_supplementary_data(db, urn)
        db.close()
    except Exception:
        pass

    # Schools with no performance rows (post-16 institutions, PRUs, new
    # schools) carry NaN in every LEFT-JOINed numeric column; NaN reaching
    # JSONResponse raises ValueError, so school_info needs the same
    # conversion yearly_data gets from clean_for_json.
    school_info = {
        k: convert_to_native(v)
        for k, v in {
            "urn": urn,
            "school_name": latest.get("school_name", ""),
            "local_authority": latest.get("local_authority", ""),
            "school_type": latest.get("school_type", ""),
            "address": latest.get("address", ""),
            "religious_denomination": latest.get("religious_denomination", ""),
            "age_range": latest.get("age_range", ""),
            "has_sixth_form": latest.get("has_sixth_form"),
            "nursery_provision": latest.get("nursery_provision"),
            "status": latest.get("status"),
            "latitude": latest.get("latitude"),
            "longitude": latest.get("longitude"),
            "phase": latest.get("phase"),
            # GIAS fields
            "website": latest.get("website"),
            "telephone": latest.get("telephone"),
            "headteacher_name": latest.get("headteacher_name"),
            "capacity": latest.get("capacity"),
            "total_pupils": latest.get("gias_total_pupils"),
            "trust_name": latest.get("trust_name"),
            "gender": latest.get("gender"),
            "county": latest.get("county"),
            "parliamentary_constituency": latest.get("parliamentary_constituency"),
        }.items()
    }

    return {
        "school_info": school_info,
        "yearly_data": clean_for_json(school_data),
        # Supplementary data (null if not yet populated by Kestra)
        "ofsted": supplementary.get("ofsted"),
        "census": supplementary.get("census"),
        "admissions": supplementary.get("admissions"),
        "admissions_history": supplementary.get("admissions_history") or [],
        "admission_distance": supplementary.get("admission_distance"),
        "sen_detail": supplementary.get("sen_detail"),
        "phonics": supplementary.get("phonics"),
        "deprivation": supplementary.get("deprivation"),
        "finance": supplementary.get("finance"),
    }


@app.get("/api/compare")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def compare_schools(
    request: Request,
    urns: str = Query(..., description="Comma-separated URNs", max_length=100)
):
    """Compare multiple schools side by side."""
    df = load_school_data()

    if df.empty:
        raise HTTPException(status_code=404, detail="No data available")

    try:
        urn_list = [int(u.strip()) for u in urns.split(",")]
        # Limit number of schools to compare
        if len(urn_list) > 10:
            raise HTTPException(status_code=400, detail="Maximum 10 schools can be compared")
        # Validate URN format
        for urn in urn_list:
            if not (100000 <= urn <= 999999):
                raise HTTPException(status_code=400, detail="Invalid URN format")
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid URN format")

    comparison_data = df[df["urn"].isin(urn_list)]

    if comparison_data.empty:
        raise HTTPException(status_code=404, detail="No schools found")

    # One session for all schools' supplementary blocks; failures degrade
    # to empty blocks rather than failing a working comparison (mirrors
    # the detail endpoint's defensive pattern).
    from . import database

    _EMPTY_SUPPLEMENTARY = {
        "ofsted": None,
        "census": None,
        "admissions": None,
        "admissions_history": [],
        "deprivation": None,
    }
    supplementary_by_urn: dict = {}
    census_benchmarks = None
    db = None
    try:
        db = database.SessionLocal()
        # One query per table for all schools, not ~5 queries per school.
        batch = get_supplementary_data_batch(db, urn_list)
        for urn in urn_list:
            supp = batch.get(urn, {})
            supplementary_by_urn[urn] = {
                key: supp.get(key, default)
                for key, default in _EMPTY_SUPPLEMENTARY.items()
            }
        # Import-time census context benchmarks (fact_census_benchmarks);
        # absent mart → None, and compute_benchmarks leaves those fields null.
        try:
            from .models import CensusBenchmark

            rows = db.query(CensusBenchmark).all()
            by_phase = {
                r.phase: {
                    "year": r.year,
                    "fsm_pct": r.fsm_pct,
                    "eal_pct": r.eal_pct,
                    "median_pupils": r.median_pupils,
                }
                for r in rows
                if getattr(r, "phase", None) in ("primary", "secondary")
            }
            if by_phase:
                census_benchmarks = by_phase
        except Exception:
            # Missing mart (or a stubbed session in tests) must never break
            # the compare payload — and not every session has rollback().
            try:
                db.rollback()
            except Exception:
                pass
    except Exception:
        supplementary_by_urn = {}
    finally:
        if db is not None:
            db.close()

    result = {}
    for urn in urn_list:
        school_data = comparison_data[comparison_data["urn"] == urn].sort_values("year")
        if not school_data.empty:
            latest = school_data.iloc[-1]
            result[str(urn)] = {
                "school_info": {
                    "urn": urn,
                    "school_name": latest.get("school_name", ""),
                    "local_authority": latest.get("local_authority", ""),
                    "school_type": latest.get("school_type", ""),
                    "address": latest.get("address", ""),
                    "phase": latest.get("phase", ""),
                    "attainment_8_score": float(latest["attainment_8_score"]) if pd.notna(latest.get("attainment_8_score")) else None,
                    "rwm_expected_pct": float(latest["rwm_expected_pct"]) if pd.notna(latest.get("rwm_expected_pct")) else None,
                    # GIAS facts the compare "Who goes there" section needs
                    # (same fields the detail endpoint exposes)
                    "religious_denomination": convert_to_native(latest.get("religious_denomination")),
                    "age_range": convert_to_native(latest.get("age_range")),
                    "gender": convert_to_native(latest.get("gender")),
                    # Needed by the admissions "What this means" copy: selective
                    # schools get entrance-test framing, never the distance template.
                    "admissions_policy": convert_to_native(latest.get("admissions_policy")),
                    "has_sixth_form": convert_to_native(latest.get("has_sixth_form")),
                    "capacity": convert_to_native(latest.get("capacity")),
                    "gias_total_pupils": convert_to_native(latest.get("gias_total_pupils")),
                    "trust_name": convert_to_native(latest.get("trust_name")),
                },
                "yearly_data": clean_for_json(school_data),
                **supplementary_by_urn.get(urn, dict(_EMPTY_SUPPLEMENTARY)),
            }

    return {
        "comparison": result,
        # Official DfE anchors + computed state-school benchmarks so the
        # compare UI can label provenance correctly (spec §8.6).
        "national_averages": _national_averages_payload(df),
        "benchmarks": compute_benchmarks(df, census_benchmarks=census_benchmarks),
    }


@app.get("/api/filters")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_filter_options(request: Request):
    """Get available filter options (local authorities, school types, years)."""
    df = load_school_data()

    if df.empty:
        return {
            "local_authorities": [],
            "school_types": [],
            "years": [],
        }

    # Phases: return values from data, ordered sensibly
    phases = clean_filter_values(df["phase"]) if "phase" in df.columns else []

    secondary_df = df[df["attainment_8_score"].notna()] if "attainment_8_score" in df.columns else df.iloc[0:0]
    genders = clean_filter_values(secondary_df["gender"]) if "gender" in secondary_df.columns else []
    admissions_policies = clean_filter_values(secondary_df["admissions_policy"]) if "admissions_policy" in secondary_df.columns else []

    return {
        "local_authorities": clean_filter_values(df["local_authority"]) if "local_authority" in df.columns else [],
        "school_types": clean_filter_values(df["school_type"]) if "school_type" in df.columns else [],
        "years": sorted(df["year"].dropna().unique().tolist()),
        "phases": phases,
        "genders": genders,
        "admissions_policies": admissions_policies,
    }


@app.get("/api/la-averages")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_la_averages(request: Request):
    """Get per-LA average Attainment 8 score for secondary schools in the latest year."""
    df = load_school_data()
    if df.empty:
        return {"year": 0, "secondary": {"attainment_8_by_la": {}}}
    latest_year = int(df["year"].max())
    sec_df = df[(df["year"] == latest_year) & df["attainment_8_score"].notna()]
    la_avg = sec_df.groupby("local_authority")["attainment_8_score"].mean().round(1).to_dict()
    return {"year": latest_year, "secondary": {"attainment_8_by_la": la_avg}}


_KS2_NATIONAL_METRICS = [
    "rwm_expected_pct", "rwm_high_pct",
    "reading_expected_pct", "writing_expected_pct", "maths_expected_pct",
    # Per-subject higher-standard nationals: reading/maths reach the "higher
    # standard" in the tests; writing is teacher-assessed at "greater depth"
    # (writing_gd_pct). Needed so each SATs bar compares to its own benchmark.
    "reading_high_pct", "writing_gd_pct", "maths_high_pct",
    "gps_expected_pct", "gps_high_pct", "science_expected_pct",
    "reading_avg_score", "maths_avg_score", "gps_avg_score",
    "reading_progress", "writing_progress", "maths_progress",
    "overall_absence_pct", "persistent_absence_pct",
    "disadvantaged_gap", "disadvantaged_pct", "sen_support_pct", "eal_pct",
]
_KS4_NATIONAL_METRICS = [
    "attainment_8_score", "progress_8_score",
    "english_maths_standard_pass_pct", "english_maths_strong_pass_pct",
    "ebacc_entry_pct", "ebacc_standard_pass_pct", "ebacc_strong_pass_pct",
    "ebacc_avg_score", "gcse_grade_91_pct",
]


def _national_averages_payload(df: pd.DataFrame) -> dict:
    """National-averages payload shared by /api/national-averages and
    /api/compare.

    Both series are persisted marts computed at import time: official DfE
    KS2 figures (fact_ks2_national_averages) and official DfE KS4 figures
    (fact_ks4_national_averages) — the API never aggregates the performance
    dataframe per request. If the KS4 mart hasn't been built yet, the
    secondary series is empty — never a computed stand-in, because the UI
    labels these figures as official DfE data.
    """
    if df.empty:
        return {"primary": {}, "secondary": {}}

    latest_year = int(df["year"].max())

    from . import database
    from .models import Ks2NationalAverage, Ks4NationalAverage

    def _row_metrics(row, metric_list):
        out = {}
        for col in metric_list:
            val = getattr(row, col, None)
            if val is not None:
                out[col] = val
        return out

    ks2_rows: list = []
    ks4_rows: list = []
    db = None
    try:
        db = database.SessionLocal()
        try:
            ks2_rows = db.query(Ks2NationalAverage).order_by(Ks2NationalAverage.year).all()
        except Exception:
            db.rollback()
        try:
            ks4_rows = db.query(Ks4NationalAverage).order_by(Ks4NationalAverage.year).all()
        except Exception:
            db.rollback()
    except Exception:
        pass
    finally:
        if db is not None:
            db.close()

    primary_by_year = {r.year: _row_metrics(r, _KS2_NATIONAL_METRICS) for r in ks2_rows}
    secondary_by_year = {r.year: _row_metrics(r, _KS4_NATIONAL_METRICS) for r in ks4_rows}

    all_years = sorted(set(primary_by_year) | set(secondary_by_year))
    by_year = [
        {
            "year": yr,
            "primary": primary_by_year.get(yr, {}),
            "secondary": secondary_by_year.get(yr, {}),
        }
        for yr in all_years
    ]

    latest_primary = next((e["primary"] for e in reversed(by_year) if e["primary"]), {})
    latest_secondary = next((e["secondary"] for e in reversed(by_year) if e["secondary"]), {})

    return {
        "year": latest_year,
        "primary": latest_primary,
        "secondary": latest_secondary,
        "by_year": by_year,
    }


@app.get("/api/national-averages")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_national_averages(request: Request):
    """
    National averages: official DfE KS2 figures per year plus computed
    KS4 averages, derived from the loaded DataFrame and the
    fact_ks2_national_averages mart.
    """
    return _national_averages_payload(load_school_data())


@app.get("/api/metrics")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_available_metrics(request: Request):
    """
    Get list of available performance metrics for schools.

    This is the single source of truth for metric definitions.
    Frontend should consume this to avoid duplication.
    """
    df = load_school_data()

    available = []
    for key, info in METRIC_DEFINITIONS.items():
        if df.empty or key in df.columns:
            available.append({"key": key, **info})

    return {"metrics": available}


@app.get("/api/rankings")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_rankings(
    request: Request,
    metric: str = Query("rwm_expected_pct", description="Metric to rank by", max_length=50),
    year: Optional[int] = Query(
        None,
        description="Academic year code, e.g. 201819 (defaults to most recent)",
        ge=2000,
        le=210100,
    ),
    limit: int = Query(20, ge=1, le=100, description="Number of schools to return"),
    local_authority: Optional[str] = Query(
        None, description="Filter by local authority", max_length=100
    ),
    phase: Optional[str] = Query(
        None, description="Filter by phase: primary or secondary", max_length=20
    ),
):
    """Get school rankings by a specific metric."""
    # Sanitize local authority input
    local_authority = sanitize_search_input(local_authority)

    # Validate metric name (only allow alphanumeric and underscore)
    if not re.match(r"^[a-z0-9_]+$", metric):
        raise HTTPException(status_code=400, detail="Invalid metric name")

    df = load_school_data()

    if df.empty:
        return {"metric": metric, "year": None, "rankings": [], "total": 0}

    if metric not in df.columns:
        raise HTTPException(status_code=400, detail=f"Metric '{metric}' not available")

    # Filter by year
    if year:
        df = df[df["year"] == year]
    else:
        # Use most recent year
        max_year = df["year"].max()
        df = df[df["year"] == max_year]

    # Filter by local authority if specified
    if local_authority:
        df = df[df["local_authority"].str.lower() == local_authority.lower()]

    # Filter by phase
    if phase == "primary" and "rwm_expected_pct" in df.columns:
        df = df[df["rwm_expected_pct"].notna()]
    elif phase == "secondary" and "attainment_8_score" in df.columns:
        df = df[df["attainment_8_score"].notna()]

    # Sort and rank (exclude rows with no data for this metric)
    df = df.dropna(subset=[metric])
    total = len(df)

    # For progress scores, higher is better. For percentages, higher is also better.
    df = df.sort_values(metric, ascending=False).head(limit)

    # Return only relevant fields for rankings
    available_cols = [c for c in RANKING_COLUMNS if c in df.columns]
    df = df[available_cols].copy()

    # Surface the requested metric under a stable `value` key so the
    # frontend doesn't need to know each metric's column name. The raw
    # metric column is also kept in the row for callers that want it.
    df["value"] = df[metric]

    return {
        "metric": metric,
        "year": int(df["year"].iloc[0]) if not df.empty else None,
        "rankings": clean_for_json(df),
        "total": total,
    }


@app.get("/api/data-info")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_data_info(request: Request):
    """Get information about loaded data."""
    # Get info directly from database
    db_info = get_db_info()

    if db_info["total_schools"] == 0:
        return {
            "status": "no_data",
            "message": "No data in marts. Run the annual EES pipeline to load KS2 data.",
            "data_source": "PostgreSQL",
        }

    # Also get DataFrame-based stats for backwards compatibility
    df = load_school_data()

    if df.empty:
        return {
            "status": "no_data",
            "message": "No data available",
            "data_source": "PostgreSQL",
        }

    years = [int(y) for y in sorted(df["year"].dropna().unique())]
    schools_per_year = {
        str(int(k)): int(v)
        for k, v in df.dropna(subset=["year"]).groupby("year")["urn"].nunique().to_dict().items()
    }
    la_counts = {
        str(k): int(v)
        for k, v in df["local_authority"].value_counts().to_dict().items()
    }

    return {
        "status": "loaded",
        "data_source": "PostgreSQL",
        "total_records": int(len(df)),
        "unique_schools": int(df["urn"].nunique()),
        "years_available": years,
        "schools_per_year": schools_per_year,
        "local_authorities": la_counts,
    }


@app.post("/api/admin/reload")
@limiter.limit("5/minute")
async def reload_data(
    request: Request,
    _: bool = Depends(verify_admin_api_key)
):
    """
    Admin endpoint to force data reload (useful after data updates).
    Requires X-API-Key header with valid admin API key.
    """
    clear_cache()
    await asyncio.to_thread(load_school_data)
    await asyncio.to_thread(load_latest_school_data)
    return {"status": "reloaded"}




# =============================================================================
# SEO FILES
# =============================================================================


@app.get("/favicon.svg")
async def favicon():
    """Serve favicon."""
    return FileResponse(settings.frontend_dir / "favicon.svg", media_type="image/svg+xml")


@app.get("/robots.txt")
async def robots_txt():
    """Serve robots.txt for search engine crawlers."""
    return FileResponse(settings.frontend_dir / "robots.txt", media_type="text/plain")


def _serve_sitemap(name: str) -> Response:
    global _sitemaps
    if _sitemaps is None:
        try:
            _sitemaps = build_sitemaps()
        except Exception as e:
            raise HTTPException(status_code=503, detail=f"Sitemap unavailable: {e}")
    if name not in _sitemaps:
        raise HTTPException(status_code=404, detail="No such sitemap")
    return Response(content=_sitemaps[name], media_type="application/xml")


@app.get("/sitemap.xml")
async def sitemap_xml():
    """Serve the sitemap index."""
    return _serve_sitemap("sitemap.xml")


@app.get("/sitemaps/{name}")
async def sitemap_child(name: str):
    """Serve a child sitemap (static.xml, or schools-N.xml)."""
    return _serve_sitemap(name)


@app.post("/api/admin/regenerate-sitemap")
@limiter.limit("10/minute")
async def regenerate_sitemap(
    request: Request,
    _: bool = Depends(verify_admin_api_key),
):
    """Rebuild and cache the sitemap from current school data. Called by Airflow after data updates."""
    global _sitemaps
    _sitemaps = build_sitemaps()
    n = sum(x.count("<url>") for x in _sitemaps.values())
    return {"status": "ok", "urls": n, "sitemaps": len(_sitemaps)}


# Mount static files directly (must be after all routes to avoid catching API calls)
if settings.frontend_dir.exists():
    app.mount("/static", StaticFiles(directory=settings.frontend_dir), name="static")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host=settings.host, port=settings.port)