SPipe/backend/gisio.py

"""Vector GIS input via GDAL (pyogrio): any supported format -> GeoJSON in WGS84.

Shapefiles must arrive zipped (they are several files); everything else is a
single file. Files are read straight from a temp copy (zip through GDAL's
/vsizip/, never extracted), so archive paths cannot escape the temp dir.
"""
from __future__ import annotations

import os
import tempfile

import numpy as np
import pyogrio
import shapely
from pyogrio import raw as ogr_raw
from pyproj import CRS, Transformer
from shapely.geometry import mapping

from .batch import BatchError

VECTOR_EXTENSIONS = {
    ".zip": "ESRI Shapefile (zip)",
    ".geojson": "GeoJSON",
    ".json": "GeoJSON",
    ".gpkg": "GeoPackage",
    ".dxf": "DXF",
    ".dgn": "MicroStation DGN",
    ".mif": "MapInfo MIF",
    ".tab": "MapInfo TAB",
    ".gml": "GML",
    ".kml": "KML",
    ".e00": "ArcInfo E00",
    ".csv": "CSV (WKT/X,Y)",
}
MAX_FEATURES = 200_000


def _gdal_path(path: str, ext: str) -> str:
    if ext == ".zip":
        return "/vsizip/" + path.replace("\\", "/")
    return path


def _clean(v):
    if isinstance(v, (np.floating, float)):
        f = float(v)
        return None if f != f or f in (float("inf"), float("-inf")) else f
    if isinstance(v, np.integer):
        return int(v)
    if isinstance(v, bytes):
        try:
            return v.decode("utf-8")
        except UnicodeDecodeError:
            return v.decode("cp949", errors="replace")
    if hasattr(v, "isoformat"):
        return v.isoformat()
    return v


def read_vector(contents: bytes, filename: str, layer: str | None = None, src_epsg: int | None = None, default_crs: str | None = None) -> dict:
    """`default_crs` applies only when the file itself carries no CRS (DXF, bare CSV ...)."""
    ext = os.path.splitext(filename.lower())[1]
    if ext not in VECTOR_EXTENSIONS:
        raise BatchError(f"지원하지 않는 형식입니다: {ext or filename}. 지원: {', '.join(sorted(VECTOR_EXTENSIONS))}")
    with tempfile.TemporaryDirectory() as tmp:
        path = os.path.join(tmp, "input" + ext)
        with open(path, "wb") as f:
            f.write(contents)
        src = _gdal_path(path, ext)
        try:
            layers = [str(row[0]) for row in pyogrio.list_layers(src)]
        except Exception as e:  # GDAL raises its own error types
            raise BatchError(f"파일을 읽을 수 없습니다: {e}") from e
        if not layers:
            raise BatchError("파일에 레이어가 없습니다.")
        name = layer if layer in layers else layers[0]
        try:
            info = pyogrio.read_info(src, layer=name, encoding=None)
            meta, _fids, geometry, field_data = ogr_raw.read(src, layer=name, max_features=MAX_FEATURES + 1)
        except Exception as e:
            raise BatchError(f"레이어 {name}을(를) 읽을 수 없습니다: {e}") from e

    count = len(geometry)
    if count > MAX_FEATURES:
        raise BatchError(f"도형이 {MAX_FEATURES:,}개를 넘습니다.")

    crs_text = meta.get("crs") or info.get("crs")
    assumed = False
    src_crs = CRS.from_epsg(src_epsg) if src_epsg else (CRS.from_user_input(crs_text) if crs_text else None)
    if src_crs is None and default_crs:
        try:
            src_crs = CRS.from_user_input(default_crs)
            assumed = True
        except Exception as e:
            raise BatchError(f"기본 좌표계를 읽을 수 없습니다: {default_crs[:60]}") from e
    src_code = src_crs.to_epsg() if src_crs else None

    geoms = shapely.from_wkb(np.array(geometry, dtype=object))
    if src_crs is not None and not src_crs.equals(CRS.from_epsg(4326)):
        tr = Transformer.from_crs(src_crs, CRS.from_epsg(4326), always_xy=True)
        geoms = shapely.transform(geoms, lambda xy: np.column_stack(tr.transform(xy[:, 0], xy[:, 1])))
    fields = list(meta.get("fields", []))
    features = []
    for i in range(count):
        g = geoms[i]
        props = {fields[k]: _clean(field_data[k][i]) for k in range(len(fields))}
        features.append({"type": "Feature", "geometry": mapping(g) if g is not None else None, "properties": props})

    return {
        "layers": layers,
        "layer": name,
        "crs": f"EPSG:{src_code}" if src_code else (src_crs.to_string() if src_crs else None),
        "reprojected": src_crs is not None,
        "crsAssumed": assumed,
        "geometryType": meta.get("geometry_type"),
        "count": count,
        "fields": fields,
        "featureCollection": {"type": "FeatureCollection", "features": features},
    }