"""Fetch ЕКБ district polygons from OSM Overpass and load into PostGIS. Usage: cd backend && uv run python ../data/sql/56_fetch_ekb_districts.py [--force-refetch] Without --force-refetch использует кэш `data/raw/ekb_districts.geojson` если он существует. Иначе дёргает Overpass API один раз и сохраняет. После Overpass fetch: 1. Переводит OSM relation members в GeoJSON MultiPolygon (через shapely). 2. UPSERT в `ekb_districts_geom (district_name, geom, osm_id, fetched_at)`. OSM-to-our naming: "Верх-Исетский район" → "Верх-Исетский" "Академический район" → "Академический" ... """ from __future__ import annotations import argparse import json import logging import sys import urllib.parse import urllib.request from pathlib import Path # Add backend to path so we can import app.* in standalone CLI mode. ROOT = Path(__file__).resolve().parent.parent.parent sys.path.insert(0, str(ROOT / "backend")) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger(__name__) # Cache lives at repo root, committed to git. GEOJSON_CACHE = ROOT / "data" / "raw" / "ekb_districts.geojson" # OSM admin_level=9 in RU = city districts. ЕКБ has 8 of them. # Bbox covers all of Ekaterinburg city + adjacent. OVERPASS_QUERY = """ [out:json][timeout:90]; ( relation['boundary'='administrative']['admin_level'='9']['name'~'район'] (56.71,60.42,57.00,60.95); ); out geom; """ OVERPASS_URL = "https://overpass-api.de/api/interpreter" # OSM display name → ekb_districts.district_name (exact match required). DISTRICT_NAME_MAP = { "Академический район": "Академический", "Верх-Исетский район": "Верх-Исетский", "Железнодорожный район": "Железнодорожный", "Кировский район": "Кировский", "Ленинский район": "Ленинский", "Октябрьский район": "Октябрьский", "Орджоникидзевский район": "Орджоникидзевский", "Чкаловский район": "Чкаловский", } def fetch_overpass() -> dict: """Hit Overpass API; return raw JSON response.""" log.info("Querying Overpass...") req = urllib.request.Request( OVERPASS_URL, data=urllib.parse.urlencode({"data": OVERPASS_QUERY}).encode(), headers={"User-Agent": "gendesign-recommend/1.0 (district mapping)"}, ) with urllib.request.urlopen(req, timeout=120) as resp: return json.loads(resp.read().decode()) def osm_relation_to_geojson(relation: dict) -> dict | None: """Convert one Overpass relation (way-members + tags) to a GeoJSON MultiPolygon Feature. Outer + inner rings are assembled into polygons. OSM relations of type=multipolygon have 'members' which are 'way'-objects each carrying their own 'geometry' (list of {lat, lon}). We build outer rings, inner rings (holes), and assemble Polygon parts → MultiPolygon. """ from shapely.geometry import MultiPolygon, Polygon, mapping from shapely.ops import polygonize, unary_union members = relation.get("members") or [] outer_lines: list[list[tuple[float, float]]] = [] inner_lines: list[list[tuple[float, float]]] = [] for m in members: if m.get("type") != "way": continue coords = [(pt["lon"], pt["lat"]) for pt in (m.get("geometry") or [])] if len(coords) < 2: continue role = m.get("role") or "outer" (inner_lines if role == "inner" else outer_lines).append(coords) if not outer_lines: log.warning("Relation %s has no outer members", relation.get("id")) return None # Use shapely.polygonize to stitch open way-segments into closed rings. # This handles real-world OSM cases where a single ring is split across # multiple way-members at admin-boundary intersections. from shapely.geometry import LineString outer_polys = list(polygonize([LineString(seg) for seg in outer_lines])) inner_polys = list(polygonize([LineString(seg) for seg in inner_lines])) if not outer_polys: log.warning( "Relation %s outer members did not form closed rings", relation.get("id") ) return None # Subtract holes from outers (best-effort). if inner_polys: holes_union = unary_union(inner_polys) outer_polys = [p.difference(holes_union) for p in outer_polys] multipoly = MultiPolygon( [p for poly in outer_polys for p in (poly.geoms if poly.geom_type == "MultiPolygon" else [poly])] if any(p.geom_type == "MultiPolygon" for p in outer_polys) else outer_polys ) # Ensure all geometry is valid; buffer(0) is the standard fix for # self-intersections that OSM data sometimes carries. if not multipoly.is_valid: multipoly = multipoly.buffer(0) if multipoly.geom_type == "Polygon": multipoly = MultiPolygon([multipoly]) return { "type": "Feature", "properties": { "name": relation.get("tags", {}).get("name"), "osm_id": relation.get("id"), "admin_level": relation.get("tags", {}).get("admin_level"), }, "geometry": mapping(multipoly), } def build_geojson(overpass_data: dict) -> dict: """Build a FeatureCollection from raw Overpass response.""" features: list[dict] = [] for el in overpass_data.get("elements", []): if el.get("type") != "relation": continue feature = osm_relation_to_geojson(el) if feature: features.append(feature) return {"type": "FeatureCollection", "features": features} def load_or_fetch(force: bool = False) -> dict: if GEOJSON_CACHE.exists() and not force: log.info("Using cached GeoJSON: %s", GEOJSON_CACHE) return json.loads(GEOJSON_CACHE.read_text(encoding="utf-8")) raw = fetch_overpass() geojson = build_geojson(raw) GEOJSON_CACHE.parent.mkdir(parents=True, exist_ok=True) GEOJSON_CACHE.write_text( json.dumps(geojson, ensure_ascii=False, indent=2), encoding="utf-8" ) log.info("Cached %d features → %s", len(geojson["features"]), GEOJSON_CACHE) return geojson def upsert_into_db(geojson: dict) -> int: from sqlalchemy import text from app.core.db import SessionLocal db = SessionLocal() try: loaded = 0 for feat in geojson.get("features", []): osm_name = feat["properties"].get("name") our_name = DISTRICT_NAME_MAP.get(osm_name) if not our_name: log.warning("Skipping unmapped OSM district: %s", osm_name) continue geom_text = json.dumps(feat["geometry"]) db.execute( text( """ INSERT INTO ekb_districts_geom (district_name, geom, osm_id, fetched_at) VALUES ( :dn, ST_Multi(ST_GeomFromGeoJSON(:gj))::geometry(MultiPolygon, 4326), :osm, NOW() ) ON CONFLICT (district_name) DO UPDATE SET geom = EXCLUDED.geom, osm_id = EXCLUDED.osm_id, fetched_at = NOW() """ ), { "dn": our_name, "gj": geom_text, "osm": feat["properties"].get("osm_id"), }, ) loaded += 1 log.info(" ✓ %s (osm_id=%s)", our_name, feat["properties"].get("osm_id")) db.commit() return loaded finally: db.close() def main() -> int: p = argparse.ArgumentParser(description="Fetch+load EKB district polygons") p.add_argument( "--force-refetch", action="store_true", help="Re-hit Overpass even if cache exists", ) p.add_argument( "--dry-run", action="store_true", help="Fetch+parse only, skip DB upsert", ) args = p.parse_args() geojson = load_or_fetch(force=args.force_refetch) log.info("Have %d features in GeoJSON", len(geojson["features"])) if args.dry_run: return 0 n = upsert_into_db(geojson) log.info("Loaded %d districts into ekb_districts_geom", n) return 0 if __name__ == "__main__": sys.exit(main())