gendesign/data/sql/56_fetch_ekb_districts.py

"""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())