Aislo/D03_PlanAnalysis/D03_PlanAnalysis_ForestRoad.py

"""D 노선분석 지도 올린 임도망도 겹(PLAN 584-3 · 595-4 · 요구 443 · 463).

- 출처 = 사용자가 받은 산림청 임도망도 SHP(FGIS 무료신청 · 산림빅데이터거래소 · 시군구 단위 · 선)
- 올림 = zip 하나 또는 shp · shx · dbf(· prj · cpg) 묶음 · 합계 `MAX_BYTES` · 읽어 보고 자르는 범위
  안 선이 있어야 받음(아니면 옛 것 그대로 · 까닭 = `ValueError` + 기준점 ~ 가장 가까운 임도 거리)
- 자르는 범위 = 기준점 ± `RADIUS_M` ∪ 받은 지형(`clip_box` · 지형 받기 전에도 기준점만으로)
- 원본 = 프로젝트 `D03_PlanAnalysis/임도망도.zip`(낱 파일은 zip 으로 묶음) +
  `임도망도.json`(파일 이름 · 올린 때)
- 좌표계 = prj · 없으면 EPSG:5179 · dbf 글자 = cpg · 없으면 cp949 · 자르는 범위로 잘라 가닥 · 길이 ·
  잘린 상자(`bounds` 경위도) · 기준점 거리 → 파생 저장본(열쇠 = 원본 지문 · 범위 · 기준점)

계산 자리 = ③ 서버 단독.
"""

from __future__ import annotations

import io
import json
import math
import os
import tempfile
import zipfile
from pathlib import Path
from typing import Any

from common_util.common_util_derived_store import (
    derived_key,
    file_stamp,
    read_derived,
    write_derived,
)
from common_util.common_util_json import atomic_write_json
from common_util.common_util_time import utc_now_text

Box = tuple[float, float, float, float]
XY = tuple[float, float]

FOLDER = "D03_PlanAnalysis"
SOURCE_ZIP = "임도망도.zip"
SOURCE_META = "임도망도.json"
DEFAULT_EPSG = 5179
DBF_ENCODING = "cp949"
MAX_BYTES = 100 * 1024 * 1024
SHP_PARTS = (".shp", ".shx", ".dbf", ".prj", ".cpg")
ROAD_KIND = "plan_forest_road"
ROAD_VERSION = 2  # 잘린 상자 · 기준점 거리
RADIUS_M = 5000.0  # 기준점 둘레(요구서_463 3장 18)


def clip_box(origin: XY, have: Box | None) -> Box:
    """자르는 범위 = 기준점 ± `RADIUS_M` ∪ 받은 지형."""
    x, y = origin
    box = (x - RADIUS_M, y - RADIUS_M, x + RADIUS_M, y + RADIUS_M)
    if have is None:
        return box
    return min(box[0], have[0]), min(box[1], have[1]), max(box[2], have[2]), max(box[3], have[3])


def source_dir(project_root: Path) -> Path:
    return Path(project_root) / FOLDER


def pack(files: list[tuple[str, bytes]]) -> bytes:
    """올린 파일 → zip 바이트 — zip 하나는 그대로 · 낱 파일은 shp · shx · dbf 가 다 있어야 묶음."""
    if len(files) == 1 and files[0][0].lower().endswith(".zip"):
        return files[0][1]
    parts = [(name, data) for name, data in files if Path(name).suffix.lower() in SHP_PARTS]
    have = {Path(name).suffix.lower() for name, _ in parts}
    lack = [ext for ext in (".shp", ".shx", ".dbf") if ext not in have]
    if lack:
        raise ValueError(
            f"SHP 묶음에 {' · '.join(lack)} 이(가) 없습니다 — "
            "zip 하나 또는 shp · shx · dbf · prj 를 함께 고르세요."
        )
    buffer = io.BytesIO()
    with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zf:
        for name, data in parts:
            zf.writestr(Path(name).name, data)
    return buffer.getvalue()


def read_shapes(zip_path: Path) -> list[Any]:
    """zip 안 SHP 마다 GeoDataFrame(좌표계 = prj · 없으면 5179) — 못 읽으면 `ValueError`."""
    import geopandas as gpd

    try:
        zf = zipfile.ZipFile(zip_path)
    except zipfile.BadZipFile as error:
        raise ValueError("zip 파일을 열지 못했습니다.") from error
    frames = []
    with zf, tempfile.TemporaryDirectory() as folder:
        members = [n for n in zf.namelist() if Path(n).suffix.lower() in SHP_PARTS]
        if not any(n.lower().endswith(".shp") for n in members):
            raise ValueError("zip 안에 .shp 파일이 없습니다.")
        for n in members:  # 폴더 없이 이름만 = 바깥으로 안 풂
            (Path(folder) / Path(n).name).write_bytes(zf.read(n))
        for shp in sorted(Path(folder).glob("*.shp")):
            options = {} if shp.with_suffix(".cpg").is_file() else {"encoding": DBF_ENCODING}
            try:
                frame = gpd.read_file(shp, **options)
            except Exception as error:  # noqa: BLE001 — 읽기 엔진 오류 종류가 많음
                raise ValueError(f"{shp.name} 을(를) 읽지 못했습니다 — {error}") from error
            if frame.crs is None:
                frame = frame.set_crs(DEFAULT_EPSG)
            frames.append(frame)
    return frames


def _plain(value: Any) -> Any:
    if isinstance(value, float) and not math.isfinite(value):
        return None
    return value if isinstance(value, (str, int, float, bool)) or value is None else str(value)


def clip_roads(zip_path: Path, box: Box, epsg: int, origin: XY | None = None) -> dict[str, Any]:
    """SHP 선 → 범위로 잘라 경위도 선(원 속성 + `length_m`) · 가닥 · 길이 · 원본 선 수 · 좌표계 ·
    잘린 상자(`bounds` [서, 남, 동, 북] · 없으면 None) · 기준점 ~ 가장 가까운 임도(`distance_m` ·
    자르기 전 모든 선)."""
    from pyproj import Transformer
    from shapely.geometry import Point, mapping
    from shapely.geometry import box as rect
    from shapely.ops import transform

    frame = rect(*box)
    here = Point(origin) if origin else None
    to_ll = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True).transform
    features: list[dict[str, Any]] = []
    total, length, crs = 0, 0.0, []
    near: float | None = None
    edges: list[tuple[float, float, float, float]] = []
    for shapes in read_shapes(zip_path):
        crs.append(shapes.crs.to_string())
        shapes = shapes[shapes.geometry.notna()]
        shapes = shapes[shapes.geometry.geom_type.isin(("LineString", "MultiLineString"))]
        total += len(shapes)
        metric = shapes.to_crs(epsg)
        for (_, row), geometry in zip(
            shapes.drop(columns=shapes.geometry.name).iterrows(), metric.geometry
        ):
            if here is not None:
                gap = geometry.distance(here)
                near = gap if near is None else min(near, gap)
            part = geometry.intersection(frame)
            if part.is_empty or part.length <= 0:
                continue
            length += part.length
            props = {str(k): _plain(v) for k, v in row.items()}
            props["length_m"] = round(part.length, 1)
            lonlat = transform(to_ll, part)
            edges.append(lonlat.bounds)
            features.append({"type": "Feature", "geometry": mapping(lonlat), "properties": props})
    if not total:
        raise ValueError("SHP 에 선 지물이 없습니다 — 임도망도(선) 파일인지 보세요.")
    return {
        "count": len(features),
        "length_m": round(length, 1),
        "total": total,
        "crs": sorted(set(crs)),
        "bounds": [
            min(e[0] for e in edges),
            min(e[1] for e in edges),
            max(e[2] for e in edges),
            max(e[3] for e in edges),
        ]
        if edges
        else None,
        "distance_m": None if near is None else round(near, 1),
        "collection": {"type": "FeatureCollection", "features": features},
    }


def road_layer(project_root: Path, box: Box, epsg: int, origin: XY | None = None) -> dict[str, Any]:
    """올린 임도망도 — 원본 지문 · 범위 · 기준점 같으면 저장본 · 올린 적 없으면
    `FileNotFoundError`."""
    path = source_dir(project_root) / SOURCE_ZIP
    stamp = file_stamp(path)
    if stamp is None:
        raise FileNotFoundError(
            "올린 임도망도가 없습니다 — [임도망 올리기]로 시군구 SHP 를 올리세요."
        )
    at = [round(v, 1) for v in origin] if origin else None
    key = derived_key(ROAD_KIND, ROAD_VERSION, stamp, int(epsg), [round(v, 1) for v in box], at)
    saved = read_derived(project_root, ROAD_KIND, key)
    if saved is not None:
        return saved["value"]
    value = clip_roads(path, box, epsg, origin)
    try:
        value["source"] = json.loads((source_dir(project_root) / SOURCE_META).read_text("utf-8"))
    except (OSError, ValueError):
        value["source"] = None
    write_derived(project_root, ROAD_KIND, key, value)
    return value


def save_roads(
    project_root: Path,
    files: list[tuple[str, bytes]],
    box: Box,
    epsg: int,
    origin: XY | None = None,
) -> dict[str, Any]:
    """올림 — 묶어 읽어 보고 자르는 범위 안 선이 있으면 원본을 바꿈 · 아니면 `ValueError`
    (옛 것 그대로 · 기준점 ~ 가장 가까운 임도 거리)."""
    data = pack(files)
    folder = source_dir(project_root)
    folder.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(dir=folder, suffix=".zip.new", delete=False) as temporary:
        temporary.write(data)
    trial = Path(temporary.name)
    try:
        got = clip_roads(trial, box, epsg, origin)
        if not got["count"]:
            gap = got["distance_m"]
            where = f" — 가장 가까운 임도 {gap / 1000:.1f} km" if gap is not None else ""
            raise ValueError(
                f"임도가 기준점 ±{RADIUS_M / 1000:.0f} km · 받은 지형 밖뿐입니다{where}"
                f"(파일 선 {got['total']}가닥) — 사업지 시군구 파일인지 보세요."
            )
        os.replace(trial, folder / SOURCE_ZIP)
    finally:
        trial.unlink(missing_ok=True)
    atomic_write_json(
        folder / SOURCE_META,
        {
            "files": [Path(name).name for name, _ in files],
            "bytes": len(data),
            "uploaded_at": utc_now_text(),
        },
    )
    return road_layer(project_root, box, epsg, origin)