Aislo/B03_FileInput/B03_FileInput_Engine_Shapefile.py

"""계획노선 shapefile(.shp) 판독 — 기하·속성·좌표계.

GDAL/geopandas를 쓰지 않고 ESRI Shapefile 규격을 직접 읽는다. 업로드 **직후**
메타데이터를 내야 하는데 그 시점엔 형제 파일(.shx/.dbf/.prj)이 아직 다 도착하지
않았을 수 있고, GDAL은 한 짝이라도 비면 열기 자체를 실패하기 때문이다. 규격 파싱은
.shp 하나만으로 기하를 읽어 낸다.

좌표계는 EPSG 코드로 가리지 않는다 — 짝 PRJ 원문을 `crs_input_from_prj()`에 넘겨
"EPSG:n" 또는 **원문 WKT**를 그대로 변환기 입력으로 쓴다(2026-08-31 사용자 확정).
"""

import logging
import struct
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)

_HEADER_BYTES = 100
_FILE_CODE = 9994

# 규격상 폴리라인 계열만 계획노선으로 받는다. Z/M 변형도 XY는 같은 자리에 있다.
_POLYLINE_TYPES = {3: "PolyLine", 13: "PolyLineZ", 23: "PolyLineM"}
_SHAPE_TYPE_NAMES = {
    0: "Null",
    1: "Point",
    3: "PolyLine",
    5: "Polygon",
    8: "MultiPoint",
    11: "PointZ",
    13: "PolyLineZ",
    15: "PolygonZ",
    18: "MultiPointZ",
    21: "PointM",
    23: "PolyLineM",
    25: "PolygonM",
    28: "MultiPointM",
    31: "MultiPatch",
}


def _normalize_codepage(text: str) -> str | None:
    """.cpg 내용을 파이썬 인코딩 이름으로 바꾼다.

    실물은 `949` 한 줄만 들어 있다(2026-08-31 원청 자료) — `CP949`도 `EUC-KR`도
    아니라서 그대로 넘기면 LookupError가 난다.
    """
    value = (text or "").strip()
    if not value:
        return None
    if value.isdigit():
        return f"cp{value}"
    upper = value.upper().replace("-", "").replace("_", "")
    if upper in {"ANSI", "OEM", "SYSTEM"}:
        return "cp949"
    return value


def shapefile_encoding(path: Path) -> str:
    """짝 .cpg에 적힌 인코딩. 없으면 국내 자료 관례대로 CP949."""
    cpg_path = path.with_suffix(".cpg")
    if cpg_path.exists():
        try:
            encoding = _normalize_codepage(cpg_path.read_text(encoding="ascii", errors="ignore"))
            if encoding:
                "".encode(encoding)  # 이름이 실재하는지 확인 — 없으면 LookupError
                return encoding
        except (OSError, LookupError):
            logger.warning("shapefile .cpg 인코딩을 해석하지 못했습니다: %s", cpg_path.name)
    return "cp949"


def shapefile_crs_input(path: Path) -> str | None:
    """짝 .prj를 `Transformer.from_crs` 입력 문자열로 정규화해 돌려준다."""
    prj_path = path.with_suffix(".prj")
    if not prj_path.exists():
        return None
    from common_util.common_util_crs import crs_input_from_prj

    return crs_input_from_prj(prj_path.read_text(encoding="utf-8", errors="ignore"))


def shapefile_epsg_label(path: Path) -> int | None:
    """짝 .prj의 EPSG **라벨**. 변환에는 쓰지 않는다 — 메타 표시·로그용."""
    prj_path = path.with_suffix(".prj")
    if not prj_path.exists():
        return None
    from pyproj import CRS

    from common_util.common_util_crs import identify_epsg

    text = prj_path.read_text(encoding="utf-8", errors="ignore")
    try:
        return identify_epsg(CRS.from_wkt(text), text)
    except Exception:  # pragma: no cover — WKT 불량은 라벨 없음으로 흘린다
        return None


def _read_header(blob: bytes) -> dict[str, Any]:
    if len(blob) < _HEADER_BYTES:
        raise ValueError("shapefile 헤더가 100바이트에 못 미칩니다.")
    file_code = struct.unpack(">i", blob[0:4])[0]
    if file_code != _FILE_CODE:
        raise ValueError("shapefile 파일 코드가 규격(9994)과 다릅니다.")
    declared_bytes = struct.unpack(">i", blob[24:28])[0] * 2
    shape_type = struct.unpack("<i", blob[32:36])[0]
    x_min, y_min, x_max, y_max = struct.unpack("<4d", blob[36:68])
    z_min, z_max = struct.unpack("<2d", blob[68:84])
    return {
        "shape_type": shape_type,
        "shape_type_name": _SHAPE_TYPE_NAMES.get(shape_type, f"Unknown({shape_type})"),
        "declared_bytes": declared_bytes,
        "bounds": {
            "x_min": x_min,
            "x_max": x_max,
            "y_min": y_min,
            "y_max": y_max,
            "z_min": z_min,
            "z_max": z_max,
        },
    }


def _read_polyline_record(blob: bytes, offset: int, content_bytes: int) -> list[list[tuple]]:
    """폴리라인 레코드 하나를 파트별 정점 목록으로 푼다."""
    record_type = struct.unpack("<i", blob[offset : offset + 4])[0]
    if record_type == 0:  # Null shape — 규격상 건너뛴다
        return []
    if record_type not in _POLYLINE_TYPES:
        raise ValueError(f"계획노선은 폴리라인이어야 합니다(받은 형상: {record_type}).")

    cursor = offset + 4 + 32  # 형상종류 + 레코드 bbox
    part_count, point_count = struct.unpack("<2i", blob[cursor : cursor + 8])
    cursor += 8
    parts = list(struct.unpack(f"<{part_count}i", blob[cursor : cursor + part_count * 4]))
    cursor += part_count * 4
    flat_xy = struct.unpack(f"<{point_count * 2}d", blob[cursor : cursor + point_count * 16])
    cursor += point_count * 16

    zs: tuple[float, ...] = ()
    if record_type == 13:
        cursor += 16  # Z 범위
        end = cursor + point_count * 8
        if end - offset <= content_bytes:
            zs = struct.unpack(f"<{point_count}d", blob[cursor:end])

    points = [
        (flat_xy[index * 2], flat_xy[index * 2 + 1], zs[index] if zs else 0.0)
        for index in range(point_count)
    ]
    boundaries = [*parts, point_count]
    return [points[boundaries[i] : boundaries[i + 1]] for i in range(part_count)]


def read_shapefile_parts(path: str | Path) -> list[list[tuple]]:
    """.shp의 모든 폴리라인 파트를 (x, y, z) 정점 목록으로 읽는다."""
    source = Path(path)
    blob = source.read_bytes()
    header = _read_header(blob)
    if header["shape_type"] not in _POLYLINE_TYPES and header["shape_type"] != 0:
        raise ValueError(
            f"계획노선 shapefile은 폴리라인이어야 합니다(받은 형상: {header['shape_type_name']})."
        )

    limit = min(len(blob), header["declared_bytes"] or len(blob))
    parts: list[list[tuple]] = []
    offset = _HEADER_BYTES
    while offset + 8 <= limit:
        content_bytes = struct.unpack(">i", blob[offset + 4 : offset + 8])[0] * 2
        if content_bytes <= 0:
            break
        parts.extend(_read_polyline_record(blob, offset + 8, content_bytes))
        offset += 8 + content_bytes
    return [part for part in parts if len(part) >= 2]


def read_shapefile_attributes(path: str | Path) -> dict[str, str]:
    """짝 .dbf 첫 레코드의 속성. 없거나 못 읽으면 빈 dict."""
    dbf_path = Path(path).with_suffix(".dbf")
    if not dbf_path.exists():
        return {}
    encoding = shapefile_encoding(Path(path))
    try:
        blob = dbf_path.read_bytes()
        header_bytes, record_bytes = struct.unpack("<2H", blob[8:12])
        fields: list[tuple[str, int]] = []
        cursor = 32
        while cursor < header_bytes - 1 and blob[cursor] != 0x0D:
            descriptor = blob[cursor : cursor + 32]
            name = descriptor[0:11].split(b"\x00")[0].decode(encoding, errors="replace").strip()
            fields.append((name, descriptor[16]))
            cursor += 32
        record = blob[header_bytes : header_bytes + record_bytes]
        if not record:
            return {}
        values: dict[str, str] = {}
        position = 1  # 첫 바이트는 삭제 표시
        for name, width in fields:
            raw = record[position : position + width]
            values[name] = raw.decode(encoding, errors="replace").strip()
            position += width
        return values
    except (OSError, struct.error, ValueError):
        logger.warning("shapefile .dbf 속성을 읽지 못했습니다: %s", dbf_path.name)
        return {}


def _route_name_from_attributes(attributes: dict[str, str], fallback: str) -> str:
    for key in ("대상지", "노선명", "route_name", "NAME", "name"):
        value = attributes.get(key)
        if value:
            return value
    return fallback


def analyze_shapefile_metadata(path: str | Path) -> dict[str, Any]:
    """계획노선 shapefile의 B03 메타데이터를 만든다."""
    source = Path(path)
    header = _read_header(source.read_bytes()[:_HEADER_BYTES])
    parts = read_shapefile_parts(source)
    attributes = read_shapefile_attributes(source)
    point_count = sum(len(part) for part in parts)
    missing = [
        extension for extension in (".shx", ".dbf") if not source.with_suffix(extension).exists()
    ]

    return {
        "file": source.name,
        "extension": "shp",
        "size_bytes": source.stat().st_size,
        "purpose": "planned_route",
        "route_name": _route_name_from_attributes(attributes, source.stem),
        "shape_type": header["shape_type_name"],
        "part_count": len(parts),
        "point_count": point_count,
        "epsg": shapefile_epsg_label(source),
        "crs_input": shapefile_crs_input(source),
        "encoding": shapefile_encoding(source),
        "attributes": attributes,
        "missing_members": missing,
        "bounds": header["bounds"],
        "start_point": list(parts[0][0]) if parts else None,
        "end_point": list(parts[-1][-1]) if parts else None,
    }