Aislo/D03_PlanAnalysis/D03_PlanAnalysis_ForestRoad.py

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

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

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

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]

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 = 1


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) -> dict[str, Any]:
    """SHP 선 → 받은 범위로 잘라 경위도 선(원 속성 + `length_m`) · 가닥 · 길이 · 원본 선 수 ·
    좌표계."""
    from pyproj import Transformer
    from shapely.geometry import box as rect
    from shapely.geometry import mapping
    from shapely.ops import transform

    frame = rect(*box)
    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, []
    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
        ):
            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)
            features.append(
                {
                    "type": "Feature",
                    "geometry": mapping(transform(to_ll, part)),
                    "properties": props,
                }
            )
    if not total:
        raise ValueError("SHP 에 선 지물이 없습니다 — 임도망도(선) 파일인지 보세요.")
    return {
        "count": len(features),
        "length_m": round(length, 1),
        "total": total,
        "crs": sorted(set(crs)),
        "collection": {"type": "FeatureCollection", "features": features},
    }


def road_layer(project_root: Path, box: Box, epsg: int) -> dict[str, Any]:
    """올린 임도망도 — 원본 지문 · 범위 같으면 저장본 · 올린 적 없으면 `FileNotFoundError`."""
    path = source_dir(project_root) / SOURCE_ZIP
    stamp = file_stamp(path)
    if stamp is None:
        raise FileNotFoundError(
            "올린 임도망도가 없습니다 — 옆 올리기 단추로 시군구 SHP 를 올리세요."
        )
    key = derived_key(ROAD_KIND, ROAD_VERSION, stamp, int(epsg), [round(v, 1) for v in box])
    saved = read_derived(project_root, ROAD_KIND, key)
    if saved is not None:
        return saved["value"]
    value = clip_roads(path, box, epsg)
    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
) -> 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)
        if not got["count"]:
            raise ValueError(
                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)