"""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 datetime import datetime 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 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": datetime.now().isoformat(timespec="seconds"), }, ) return road_layer(project_root, box, epsg)