Aislo/B04_PreProcess/B04_PreProcess_Engine_SheetParser.py

# B04_PreProcess_Engine_SheetParser.py
# 1:5,000 수치지형도 도엽 zip 파서 + 9매 병합 → 레이어별 GeoJSON 산출.
#
# 레이어 코드 (보유 도엽 2매 실측 확정, 2026-07-26):
#   N3L_E0020000 하천중심선(구분: 세류/소하천/지방하천), N3P_F0020000 표고점(수치),
#   N3L_F0030000 성/절토 비탈면, N3L_F0040000 옹벽·석축, N3P_E0042326 유수방향,
#   N3P_H0020000 기준점(삼각점/통합기준점, 표고), N3P_H0040000 지명(계곡·부락 등)
# 벼랑바위·너덜바위·산산맥은 표본 도엽에 미출현 — 코드 확정 시 SHEET_LAYERS에 추가.
#
# 좌표계: zip 내부 PRJ는 오라벨 사례가 있어 map_sheets_index.json의 교정 crs를 사용한다.

from __future__ import annotations

import logging
import tempfile
import zipfile
from pathlib import Path

import geopandas as gpd
import pandas as pd

from .B04_PreProcess_Engine_SheetStore import _load_index, get_sheet_path

logger = logging.getLogger(__name__)

# 레이어 키 → (지오메트리 접두사, 지형지물 코드)
SHEET_LAYERS: dict[str, tuple[str, str]] = {
    "등고선": ("N3L", "F0010000"),
    "하천중심선": ("N3L", "E0020000"),
    "표고점": ("N3P", "F0020000"),
    "성절토": ("N3L", "F0030000"),
    "옹벽석축": ("N3L", "F0040000"),
    "유수방향": ("N3P", "E0042326"),
    "기준점": ("N3P", "H0020000"),
    "지명": ("N3P", "H0040000"),
    "도로중심선": ("N3L", "A0020000"),
}

# 산출 파일 접두사 (기존 processed/의 국가 GIS geojson과 구분)
_OUTPUT_PREFIX = "도엽"
# 파서 판 번호 — 병합 캐시 열쇠에 넣음 · 읽는 법이 바뀌면 올려 옛 병합본을 다시 지음
PARSER_VERSION = 2
# .cpg 없는 도엽 dbf 의 글자 인코딩(국토지리정보원 배포본)
DBF_FALLBACK_ENCODING = "cp949"


def _read_layer_from_zip(
    zip_path: Path, prefix: str, code: str, crs: str
) -> gpd.GeoDataFrame | None:
    """도엽 zip에서 단일 레이어 SHP를 읽어 WGS84 GeoDataFrame으로 반환. 없으면 None."""
    member_base = f"{prefix}_{code}"
    with zipfile.ZipFile(zip_path) as zf:
        names = zf.namelist()
        wanted = {
            n
            for n in names
            if Path(n).stem.upper() == member_base
            and Path(n).suffix.lower() in (".shp", ".shx", ".dbf", ".prj", ".cpg")
        }
        if not any(n.lower().endswith(".shp") for n in wanted):
            return None
        with tempfile.TemporaryDirectory() as td:
            for n in wanted:
                (Path(td) / Path(n).name).write_bytes(zf.read(n))
            has_cpg = any(Path(n).suffix.lower() == ".cpg" for n in wanted)
            options = {} if has_cpg else {"encoding": DBF_FALLBACK_ENCODING}
            gdf = gpd.read_file(Path(td) / f"{member_base}.shp", **options)

    if gdf.empty:
        return None
    # 내부 PRJ 오라벨 대비: 인덱스의 교정 crs로 강제 지정
    gdf = gdf.set_crs(crs, allow_override=True)
    return gdf.to_crs("EPSG:4326")


def parse_and_merge_sheets(
    store_dir: str | Path,
    sheet_nos: list[str],
    output_dir: str | Path,
    layers: list[str] | None = None,
) -> dict:
    """도엽 여러 매를 레이어별로 병합해 GeoJSON으로 저장한다.

    - 도곽 경계 중복 지물은 UFID 기준 dedup (UFID 없으면 지오메트리 WKB 기준).
    - 산출: {output_dir}/도엽_{레이어}.geojson (EPSG:4326)
    - 반환: {레이어: {"count": n, "file": 경로}} (지물 0건 레이어는 제외)
    """
    store = Path(store_dir)
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    index = _load_index(store)["sheets"]
    target_layers = layers or list(SHEET_LAYERS)

    result: dict[str, dict] = {}
    for layer_key in target_layers:
        if layer_key not in SHEET_LAYERS:
            logger.warning("도엽 파서: 미정의 레이어 요청 무시: %s", layer_key)
            continue
        prefix, code = SHEET_LAYERS[layer_key]

        parts: list[gpd.GeoDataFrame] = []
        for sheet_no in sheet_nos:
            sheet_no = str(sheet_no)
            zip_path = get_sheet_path(store, sheet_no)
            entry = index.get(sheet_no)
            if zip_path is None or entry is None:
                logger.warning("도엽 파서: 영구저장소에 없는 도엽 스킵: %s", sheet_no)
                continue
            try:
                gdf = _read_layer_from_zip(zip_path, prefix, code, entry["crs"])
            except Exception as exc:
                logger.warning("도엽 파서: %s %s 읽기 실패: %s", sheet_no, layer_key, exc)
                continue
            if gdf is None:
                continue
            gdf["도엽번호"] = sheet_no
            parts.append(gdf)

        if not parts:
            continue

        merged = gpd.GeoDataFrame(pd.concat(parts, ignore_index=True), crs="EPSG:4326")
        # 도곽 경계에 걸친 지물은 인접 도엽에 중복 수록됨 → UFID 우선 dedup
        if "UFID" in merged.columns and merged["UFID"].notna().any():
            merged = merged.drop_duplicates(subset="UFID", keep="first")
        else:
            merged = merged.loc[~merged.geometry.to_wkb().duplicated(keep="first")]

        out_path = out / f"{_OUTPUT_PREFIX}_{layer_key}.geojson"
        merged.to_file(out_path, driver="GeoJSON")
        result[layer_key] = {"count": int(len(merged)), "file": str(out_path)}
        logger.info("도엽 병합 완료: %s %d건 → %s", layer_key, len(merged), out_path.name)

    return result