chore: remove the code the legacy CSV importer left behind
PR Checks / Frontend Typecheck + Tests (pull_request) Successful in 1m11s
PR Checks / Backend Smoke (pull_request) Successful in 9s
PR Checks / Build Backend (no push) (pull_request) Successful in 18s
PR Checks / Build Frontend (no push) (pull_request) Successful in 1m18s
PR Checks / Build Pipeline (no push) (pull_request) Successful in 11s
PR Checks / AI Code Review (Claude) (pull_request) Successful in 1m2s

`backend/migration.py` and `scripts/migrate_csv_to_db.py` import `School`,
`SchoolResult`, `init_db` and `set_db_schema_version` — names that no longer
exist. `scripts/geocode_schools.py` imports the same removed ORM model. None of
the three can be imported against the current backend, so they were not dormant
utilities anyone could fall back on; they were files that would fail on the
first line. `backend/version.py` existed only to hand `SCHEMA_VERSION` to that
importer, and the FastAPI lifespan performs no version-triggered import.

Three symbols go with them, each confirmed to have no caller: the unvectorised
`haversine_distance`, superseded by the inline NumPy calculation in search;
`fetcher`, an SWR helper for a dependency this project does not install; and
`kmToMiles`. `calculateDistance` stays — CutoffMapPanel uses it.

Two comments pointed at `migrate_csv_to_db.py --drop` to explain why Payload
owns its own schema. The reason survives the script: blog content must stay
clear of the school marts and Airflow's metadata. Reworded rather than deleted,
so the constraint keeps its justification.

docs/LEGACY_CODE.md records what was removed and where to find it in history. It
also records what was deliberately *not* removed, which is the more useful half:
unused UI components awaiting a design decision, manual data utilities whose
operators a repository search cannot see, and fallbacks that look obsolete but
are load-bearing — `data_loader.py`'s older-mart branches, the generated GIAS
dictionary copies, and the `legacy`-named dbt models that annual DAG selectors
explicitly include. A zero-import count is evidence, not a verdict.

The scripts that fetch DfE CSVs are marked historical and kept, pending
confirmation that nobody runs them by hand.

Checked: 190 backend tests, 429 frontend tests, `tsc --noEmit` clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016y2J6bs8gbuSJbH18w7Tan
This commit is contained in:
TudorandClaude Opus 5 committed 2026-09-14 23:01:22 +01:00
1 parent eaf5e5d180
commit 1d8858fbda
11 files changed
+98 -832

No files matched your search

+3
View File
@@ -1,5 +1,8 @@
#!/usr/bin/env python3
"""
Historical standalone CSV utility; not part of the managed Meltano/dbt pipeline.
See docs/LEGACY_CODE.md before using it for current school data.
Data Download Helper Script
This script provides instructions and utilities for downloading
+3
View File
@@ -1,5 +1,8 @@
#!/usr/bin/env python3
"""
Historical standalone CSV utility; not part of the managed Meltano/dbt pipeline.
See docs/LEGACY_CODE.md before using it for current school data.
Fetch real school performance data from UK Government sources.
This script downloads KS2 (Key Stage 2) primary school data from:
-184
View File
@@ -1,184 +0,0 @@
#!/usr/bin/env python3
"""
Geocode all school postcodes and update the database.
This script should be run as a weekly cron job to ensure all schools
have up-to-date latitude/longitude coordinates.
Usage:
python scripts/geocode_schools.py [--force]
Options:
--force Re-geocode all postcodes, even if already geocoded
Crontab example (run every Sunday at 2am):
0 2 * * 0 cd /path/to/school_compare && /path/to/venv/bin/python scripts/geocode_schools.py >> /var/log/geocode_schools.log 2>&1
"""
import argparse
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, Tuple
import requests
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.database import SessionLocal
from backend.models import School
def geocode_postcodes_bulk(postcodes: list) -> Dict[str, Tuple[float, float]]:
"""
Geocode postcodes in bulk using postcodes.io API.
Returns dict of postcode -> (latitude, longitude).
"""
results = {}
valid_postcodes = [
p.strip().upper()
for p in postcodes
if p and isinstance(p, str) and len(p.strip()) >= 5
]
valid_postcodes = list(set(valid_postcodes))
if not valid_postcodes:
return results
batch_size = 100
total_batches = (len(valid_postcodes) + batch_size - 1) // batch_size
for i, batch_start in enumerate(range(0, len(valid_postcodes), batch_size)):
batch = valid_postcodes[batch_start : batch_start + batch_size]
print(f" Geocoding batch {i + 1}/{total_batches} ({len(batch)} postcodes)...")
try:
response = requests.post(
"https://api.postcodes.io/postcodes",
json={"postcodes": batch},
timeout=30,
)
if response.status_code == 200:
data = response.json()
for item in data.get("result", []):
if item and item.get("result"):
pc = item["query"].upper()
lat = item["result"].get("latitude")
lon = item["result"].get("longitude")
if lat and lon:
results[pc] = (lat, lon)
else:
print(f" Warning: API returned status {response.status_code}")
except Exception as e:
print(f" Warning: Geocoding batch failed: {e}")
return results
def geocode_schools(force: bool = False) -> None:
"""
Geocode all schools in the database.
Args:
force: If True, re-geocode all postcodes even if already geocoded
"""
print(f"\n{'='*60}")
print(f"School Geocoding Job - {datetime.now().isoformat()}")
print(f"{'='*60}\n")
db = SessionLocal()
try:
# Get schools that need geocoding
if force:
schools = db.query(School).filter(School.postcode.isnot(None)).all()
print(f"Force mode: Processing all {len(schools)} schools with postcodes")
else:
schools = db.query(School).filter(
School.postcode.isnot(None),
(School.latitude.is_(None)) | (School.longitude.is_(None))
).all()
print(f"Found {len(schools)} schools without coordinates")
if not schools:
print("No schools to geocode. Exiting.")
return
# Extract unique postcodes
postcodes = list(set(
s.postcode.strip().upper()
for s in schools
if s.postcode
))
print(f"Unique postcodes to geocode: {len(postcodes)}")
# Geocode in bulk
print("\nGeocoding postcodes...")
geocoded = geocode_postcodes_bulk(postcodes)
print(f"Successfully geocoded: {len(geocoded)} postcodes")
# Update database
print("\nUpdating database...")
updated_count = 0
failed_count = 0
for school in schools:
if not school.postcode:
continue
pc_upper = school.postcode.strip().upper()
coords = geocoded.get(pc_upper)
if coords:
school.latitude = coords[0]
school.longitude = coords[1]
updated_count += 1
else:
failed_count += 1
db.commit()
print(f"\nResults:")
print(f" - Updated: {updated_count} schools")
print(f" - Failed (invalid/not found): {failed_count} postcodes")
# Summary stats
total_with_coords = db.query(School).filter(
School.latitude.isnot(None),
School.longitude.isnot(None)
).count()
total_schools = db.query(School).count()
print(f"\nDatabase summary:")
print(f" - Total schools: {total_schools}")
print(f" - Schools with coordinates: {total_with_coords}")
print(f" - Coverage: {100*total_with_coords/total_schools:.1f}%")
except Exception as e:
print(f"Error during geocoding: {e}")
db.rollback()
raise
finally:
db.close()
print(f"\n{'='*60}")
print(f"Geocoding job completed - {datetime.now().isoformat()}")
print(f"{'='*60}\n")
def main():
parser = argparse.ArgumentParser(
description="Geocode school postcodes and update database"
)
parser.add_argument(
"--force",
action="store_true",
help="Re-geocode all postcodes, even if already geocoded"
)
args = parser.parse_args()
geocode_schools(force=args.force)
if __name__ == "__main__":
main()
-68
View File
@@ -1,68 +0,0 @@
#!/usr/bin/env python3
"""
CLI script for manual database migration.
Usage:
python scripts/migrate_csv_to_db.py [--drop] [--geocode]
Options:
--drop Drop existing tables before migration (full reimport)
--geocode Geocode postcodes (requires network access)
"""
import sys
from pathlib import Path
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
import argparse
from backend.config import settings
from backend.database import Base, engine, init_db, set_db_schema_version
from backend.migration import load_csv_data, migrate_data, run_full_migration
from backend.version import SCHEMA_VERSION
def main():
parser = argparse.ArgumentParser(
description="Migrate CSV data to PostgreSQL database"
)
parser.add_argument(
"--drop", action="store_true", help="Drop existing tables before migration"
)
parser.add_argument("--geocode", action="store_true", help="Geocode postcodes")
args = parser.parse_args()
print("=" * 60)
print("School Data Migration: CSV -> PostgreSQL")
print("=" * 60)
print(f"\nDatabase: {settings.database_url.split('@')[-1]}")
print(f"Data directory: {settings.data_dir}")
print(f"Target schema version: {SCHEMA_VERSION}")
if args.drop:
print("\nRunning full migration (drop + reimport)...")
success = run_full_migration(geocode=args.geocode)
else:
print("\nCreating tables (preserving existing data)...")
init_db()
print("\nLoading CSV data...")
df = load_csv_data(settings.data_dir)
if df.empty:
print("No data found to migrate!")
return 1
migrate_data(df, geocode=args.geocode)
success = True
if success:
# Ensure schema_version table exists
init_db()
set_db_schema_version(SCHEMA_VERSION)
print(f"\nSchema version set to {SCHEMA_VERSION}")
return 0 if success else 1
if __name__ == "__main__":
sys.exit(main())