Aislo/D03_PlanAnalysis/D03_PlanAnalysis_SheetTiles.py

"""도엽 조각 — 수치지형도 선 겹을 도엽(1:5,000) 한 장씩 공용 창고에(PLAN 606 · 요구 477).

자리 = `storage/_shared/sheets/<도엽>/`(모든 작업 폴더가 보는 정션 · 시놀로지로 데스크탑에도)
— 도엽 zip + 색인(그 도엽 하나 · B04 받기 함수에 이 폴더를 창고로 줌) + `layers/<겹>_<판>.geojson`.
조각 하나 = 도엽 도곽으로 자른 경위도 GeoJSON(자릿수 7) — 이웃 조각과 겹침 없음.
- 등고선 · 계곡선(25 m 배수) = 전국 등고선 파일에서 자름 · 그 도엽에 없으면(섬 등) 도엽 zip 등고선.
- 세류(하천중심선) · 물 면 셋 · 도로중심선 · 건물 · 지명 = 도엽 zip.
도엽 zip = 창고에 없으면 프로젝트 `map_sheets` · 공용 지형 창고(607)에서 복사(원본 안 지움) · 그래도
없으면 브이월드 받기. 받기 실패(로그인 · 망)는 저장 안 함(다음에 다시) · 검색 0(빈 도엽) = 빈 조각.
B04 함수는 부르기만(`prune_sheets` 안 부름 · 요구 491).
"""

from __future__ import annotations

import hashlib
import json
import logging
import math
import re
from pathlib import Path
from typing import Any

import numpy as np
from pyproj import Transformer

from B04_PreProcess.B04_PreProcess_Engine_MapSheet import (
    SHEET5K_SIZE_DEG,
    latlon_to_sheet5k,
    sheet5k_to_bounds,
)
from common_util.common_util_json import LONLAT_DIGITS, round_floats
from common_util.common_util_terrain_store import NATIONAL_GPKG, TERRAIN_ROOT, _Lock
from config.config_system import STORAGE_BASE_DIR

logger = logging.getLogger(__name__)

SHEETS_ROOT = Path(STORAGE_BASE_DIR) / "_shared" / "sheets"
TILE_VERSION = 1
CONTOUR, INDEX = "등고선", "계곡선"
INDEX_STEP_M = 25
ZIP_LAYERS = ("하천중심선", "도로중심선", "건물", "지명", "실폭하천", "수계면E003", "수계면E005")
TILE_NAMES = (CONTOUR, INDEX, *ZIP_LAYERS)
# 보이는 도엽 수 — 이하 = 다 · 넘고 INDEX_MAX 이하 = 계곡선만 · 넘음 = 선 겹 숨김(목록 안 줌)
FULL_MAX, INDEX_MAX = 12, 30
SHEET_NO = re.compile(r"^\d{8}$")
_EMPTY = {"type": "FeatureCollection", "features": []}

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


def sheets_in_box(bbox: Box) -> tuple[int, list[str]]:
    """경위도 상자에 닿는 도엽 (수 · 번호 — 가운데부터). 수가 `INDEX_MAX` 넘으면 번호 없음."""
    lon0, lat0, lon1, lat1 = bbox
    step = SHEET5K_SIZE_DEG
    i0, i1 = math.floor(min(lat0, lat1) / step), math.floor(max(lat0, lat1) / step)
    j0, j1 = math.floor(min(lon0, lon1) / step), math.floor(max(lon0, lon1) / step)
    count = (i1 - i0 + 1) * (j1 - j0 + 1)
    if count > INDEX_MAX:
        return count, []
    mid_lat, mid_lon = (lat0 + lat1) / 2, (lon0 + lon1) / 2
    cells = [(i, j) for i in range(i0, i1 + 1) for j in range(j0, j1 + 1)]
    cells.sort(
        key=lambda c: ((c[0] + 0.5) * step - mid_lat) ** 2 + ((c[1] + 0.5) * step - mid_lon) ** 2
    )
    return count, [latlon_to_sheet5k((i + 0.5) * step, (j + 0.5) * step) for i, j in cells]


def _gpkg_tag() -> str | None:
    try:
        stat = NATIONAL_GPKG.stat()
    except OSError:
        return None
    return hashlib.sha1(f"{stat.st_size}:{stat.st_mtime_ns}".encode()).hexdigest()[:8]


def _tile_file(folder: Path, name: str) -> Path | None:
    """지금 판의 조각 파일 자리 — 등고 둘은 전국 파일 지문을 이름에(바뀌면 다시) ·
    전국 파일 없는 PC 는 있는 조각을 믿음 · 그것도 없으면 None(zip 길)."""
    layers = folder / "layers"
    if name not in (CONTOUR, INDEX):
        return layers / f"{name}_v{TILE_VERSION}.geojson"
    tag = _gpkg_tag()
    if tag:
        return layers / f"{name}_v{TILE_VERSION}_{tag}.geojson"
    found = sorted(layers.glob(f"{name}_v{TILE_VERSION}_*.geojson"))
    return found[-1] if found else layers / f"{name}_v{TILE_VERSION}_zip.geojson"


def _write(path: Path, collection: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    part = path.with_suffix(".part")
    text = json.dumps(
        round_floats(collection, LONLAT_DIGITS), ensure_ascii=False, separators=(",", ":")
    )
    part.write_text(text, encoding="utf-8")
    part.replace(path)


def _clipped(frame: Any, rect: Box) -> dict[str, Any]:
    """경위도 GeoDataFrame → 도곽으로 자른 FeatureCollection(빈 형상 뺌 · 날짜는 글로)."""
    import shapely

    frame = frame.copy()
    frame["geometry"] = shapely.clip_by_rect(frame.geometry.values, *rect)
    frame = frame[~frame.geometry.is_empty]
    if frame.empty:
        return dict(_EMPTY)
    return json.loads(frame.to_json(drop_id=True, default=str))


def _project_zips(sheet_no: str) -> list[Path]:
    """옮겨 담을 후보 — 공용 지형 창고(607) · 프로젝트 `map_sheets`(새것부터)."""
    found = list(
        Path(STORAGE_BASE_DIR).glob(f"*/*/*/B04_PreProcess/processed/map_sheets/{sheet_no}.zip")
    )
    found.sort(key=lambda p: p.stat().st_mtime, reverse=True)
    terrain = TERRAIN_ROOT / sheet_no / f"{sheet_no}.zip"
    return ([terrain] if terrain.is_file() else []) + found


def ensure_zip(sheet_no: str) -> Path | None:
    """창고 도엽 zip — 없으면 옮겨 담기 → 받기. 빈 도엽(검색 0) = None · 받기 실패 = 오류."""
    from B04_PreProcess.B04_PreProcess_Engine_SheetStore import (
        ensure_sheets,
        get_sheet_path,
        ingest_zip,
    )

    folder = SHEETS_ROOT / sheet_no
    path = get_sheet_path(folder, sheet_no)
    if path is not None:
        return path
    with _Lock(folder / ".zip.lock"):
        path = get_sheet_path(folder, sheet_no)
        if path is not None:
            return path
        for source in _project_zips(sheet_no):
            try:
                ingest_zip(folder, source, sheet_no)
                logger.info("도엽 조각 zip 옮겨 담음 %s ← %s", sheet_no, source)
                return get_sheet_path(folder, sheet_no)
            except Exception as error:  # noqa: BLE001 — 깨진 사본이면 다음 후보
                logger.warning("도엽 zip 옮겨 담기 실패 %s (%s): %s", sheet_no, source, error)
        result = ensure_sheets(folder, [sheet_no])
        if sheet_no in result["available"]:
            return Path(result["available"][sheet_no])
        reason = result["failed"].get(sheet_no, "")
        if reason.startswith("LookupError"):
            return None
        raise RuntimeError(f"도엽 {sheet_no} 받기 실패: {reason}")


def _zip_layer(sheet_no: str, name: str) -> Any | None:
    """창고 zip 의 겹 하나 → 경위도 GeoDataFrame(없음 = None)."""
    from B04_PreProcess.B04_PreProcess_Engine_SheetParser import SHEET_LAYERS, _read_layer_from_zip
    from B04_PreProcess.B04_PreProcess_Engine_SheetStore import _load_index

    path = ensure_zip(sheet_no)
    if path is None:
        return None
    entry = _load_index(SHEETS_ROOT / sheet_no)["sheets"][sheet_no]
    prefix, code = SHEET_LAYERS[name]
    return _read_layer_from_zip(path, prefix, code, entry["crs"])


def _gpkg_contours(rect: Box) -> Any | None:
    """전국 등고선에서 도곽 상자에 닿는 선 → 경위도 GeoDataFrame(`등고수치`) · 없음 = None."""
    if not NATIONAL_GPKG.is_file():
        return None
    import pyogrio

    lon0, lat0, lon1, lat1 = rect
    t = np.linspace(0.0, 1.0, 11)
    lon = np.concatenate(
        [lon0 + (lon1 - lon0) * t, lon0 + (lon1 - lon0) * t, np.full(11, lon0), np.full(11, lon1)]
    )
    lat = np.concatenate(
        [np.full(11, lat0), np.full(11, lat1), lat0 + (lat1 - lat0) * t, lat0 + (lat1 - lat0) * t]
    )
    x, y = Transformer.from_crs("EPSG:4326", "EPSG:5179", always_xy=True).transform(lon, lat)
    frame = pyogrio.read_dataframe(
        NATIONAL_GPKG,
        bbox=(x.min() - 5, y.min() - 5, x.max() + 5, y.max() + 5),
        columns=["CTRLN_HG"],
    )
    if frame.empty:
        return None
    frame = frame.rename(columns={"CTRLN_HG": "등고수치"}).to_crs("EPSG:4326")
    return frame[["등고수치", "geometry"]]


def _build_contours(sheet_no: str, rect: Box) -> None:
    frame = _gpkg_contours(rect)
    if frame is None:
        from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import ELEVATION_KEYS

        frame = _zip_layer(sheet_no, CONTOUR)
        column = next((k for k in ELEVATION_KEYS if frame is not None and k in frame.columns), None)
        frame = (
            None
            if column is None
            else frame.rename(columns={column: "등고수치"})[["등고수치", "geometry"]]
        )
    full = dict(_EMPTY) if frame is None else _clipped(frame, rect)
    index = {
        **full,
        "features": [
            f
            for f in full["features"]
            if float(f["properties"].get("등고수치") or 0) % INDEX_STEP_M == 0
        ],
    }
    folder = SHEETS_ROOT / sheet_no
    _write(_tile_file(folder, INDEX), index)  # 계곡선 먼저 — 등고선 파일이 있으면 둘 다 있음
    _write(_tile_file(folder, CONTOUR), full)


def ensure_tile(sheet_no: str, name: str) -> Path:
    """도엽 조각 파일 — 있으면 그대로 · 없으면 만듦(받기 실패 = RuntimeError · 저장 안 함)."""
    if not SHEET_NO.match(sheet_no) or name not in TILE_NAMES:
        raise KeyError(f"{sheet_no}/{name}")
    folder = SHEETS_ROOT / sheet_no
    path = _tile_file(folder, name)
    if path.is_file():
        return path
    with _Lock(folder / f".{'contour' if name in (CONTOUR, INDEX) else name}.lock"):
        path = _tile_file(folder, name)
        if path.is_file():
            return path
        lon0, lat0, lon1, lat1 = sheet5k_to_bounds(sheet_no)
        rect = (lon0, lat0, lon1, lat1)
        if name in (CONTOUR, INDEX):
            _build_contours(sheet_no, rect)
            path = _tile_file(folder, name)
        else:
            frame = _zip_layer(sheet_no, name)
            _write(path, dict(_EMPTY) if frame is None else _clipped(frame, rect))
        return path


def collect_project_zips() -> dict[str, int]:
    """한 번 — 프로젝트마다 받아 둔 도엽 zip 을 공용 창고로 복사(원본 안 지움 · 있으면 건너뜀)."""
    nos = {
        p.stem
        for p in Path(STORAGE_BASE_DIR).glob("*/*/*/B04_PreProcess/processed/map_sheets/*.zip")
    }
    moved = failed = 0
    for no in sorted(n for n in nos if SHEET_NO.match(n)):
        try:
            moved += ensure_zip(no) is not None
        except Exception as error:  # noqa: BLE001
            failed += 1
            logger.warning("도엽 zip 모으기 실패 %s: %s", no, error)
    return {"sheets": len(nos), "stored": moved, "failed": failed}