258 lines
10 KiB
Python
258 lines
10 KiB
Python
# B04_wf1_Surface_Engine_SheetStore.py
|
|
# 1:5,000 수치지형도 도엽 zip 전역 영구저장소 관리.
|
|
#
|
|
# 조사 결과(2026-07-26 실측):
|
|
# - 도엽번호 산출: 브이월드 오픈 API(api.vworld.kr)에는 색인 레이어가 없어
|
|
# 자체 알고리즘(B04_wf1_Surface_Engine_MapSheet) 사용 — 보유 도엽 10매 정합 확인.
|
|
# - 도엽 검색: 지도서비스 내부 API `map.vworld.kr/ws3dmap/getDisitalMapList.do`
|
|
# (type=M02, searchNum=도엽번호, scale, version) — 무로그인 조회 가능.
|
|
# Ver2.0 = zip(SHP), Ver1.0 = dxf.
|
|
# - 도엽 다운로드: `map.vworld.kr/dtkmap/digitalDownload.do?DS_SQ=..&FILE_SQ=..`
|
|
# 서버가 로그인 세션 강제(무로그인 시 알림 HTML 159바이트). 로그인 브라우저의
|
|
# 세션 쿠키를 VWORLD_SESSION_COOKIE로 제공하면 자동 다운로드 시도 가능.
|
|
# - 확보된 zip은 resources/map_sheets/{도엽번호}.zip 으로 영구 보관한다.
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import json
|
|
import re
|
|
import shutil
|
|
import tempfile
|
|
import urllib.parse
|
|
import urllib.request
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from config.config_system import (
|
|
MAP_SHEETS_DIR,
|
|
MAP_SHEETS_INDEX_PATH,
|
|
VWORLD_SESSION_COOKIE,
|
|
)
|
|
|
|
from .B04_wf1_Surface_Engine_MapSheet import latlon_to_sheet5k, sheet5k_to_bounds
|
|
|
|
# 도곽선 레이어 코드 (수치지형도 v2.0 도엽본)
|
|
_SHEET_FRAME_CODE = "A0010000"
|
|
_SHEET_NO_RE = re.compile(r"(\d{8})")
|
|
|
|
# 도엽본 좌표계 후보: 동부원점(5187)이 기본, 오라벨 zip은 중부원점(5186) 선언 사례 실측됨
|
|
_CRS_CANDIDATES = ("EPSG:5187", "EPSG:5186", "EPSG:5185", "EPSG:5188")
|
|
|
|
VWORLD_DATASET_URL = "https://www.vworld.kr/dtmk/dtmk_ntads_s002.do?dsId=30205"
|
|
VWORLD_MAP_URL = "https://map.vworld.kr/map/dtkmap.do"
|
|
_SEARCH_API = "https://map.vworld.kr/ws3dmap/getDisitalMapList.do"
|
|
_DOWNLOAD_API = "https://map.vworld.kr/dtkmap/digitalDownload.do"
|
|
|
|
|
|
def search_sheet_files(sheet_no: str, version: int = 2, scale: int = 5000) -> list[dict]:
|
|
"""브이월드 지도서비스 도엽검색 API. 무로그인. version: 1=Ver1.0(dxf), 2=Ver2.0(zip), 3=둘 다.
|
|
|
|
반환 행: ds_sq, file_no, file_nm, file_extsn, file_size, dtm_id, dtm_nm,
|
|
scale_nm, open_isscntm(좌표계), last_upt_dat
|
|
"""
|
|
params = urllib.parse.urlencode(
|
|
{
|
|
"type": "M02",
|
|
"searchNum": str(sheet_no),
|
|
"scale": str(scale),
|
|
"cod": "1",
|
|
"version": str(version),
|
|
}
|
|
)
|
|
req = urllib.request.Request(f"{_SEARCH_API}?{params}")
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
data = json.loads(resp.read().decode("utf-8"))
|
|
return data.get("RSLT_DATA") or []
|
|
|
|
|
|
def download_sheet(sheet_no: str, session_cookie: str = "", version: int = 2) -> dict:
|
|
"""도엽 파일 다운로드 후 영구저장소 등록. 로그인 세션 쿠키 필수.
|
|
|
|
session_cookie: 로그인된 map.vworld.kr 브라우저의 Cookie 헤더 문자열
|
|
(예: "JSESSIONID=..; SSCSID=.."). 생략 시 .env의 VWORLD_SESSION_COOKIE 사용.
|
|
세션은 서버 유휴 타임아웃으로 만료됨 — 만료 시 PermissionError 발생, 쿠키 갱신 필요.
|
|
"""
|
|
session_cookie = session_cookie or VWORLD_SESSION_COOKIE
|
|
if not session_cookie:
|
|
raise PermissionError(
|
|
f"도엽 {sheet_no}: 세션 쿠키 없음 — .env VWORLD_SESSION_COOKIE 설정 또는 수동 다운로드 필요"
|
|
)
|
|
rows = search_sheet_files(sheet_no, version=version)
|
|
if not rows:
|
|
raise LookupError(f"도엽 {sheet_no}: 브이월드 검색 결과 없음")
|
|
row = rows[0]
|
|
|
|
params = urllib.parse.urlencode({"DS_SQ": row["ds_sq"], "FILE_SQ": row["file_no"]})
|
|
req = urllib.request.Request(
|
|
f"{_DOWNLOAD_API}?{params}",
|
|
headers={"Cookie": session_cookie, "Referer": VWORLD_MAP_URL},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=300) as resp:
|
|
payload = resp.read()
|
|
|
|
if payload[:2] != b"PK":
|
|
# zip 시그니처 아님 → 로그인 만료 등으로 알림 HTML이 온 경우
|
|
snippet = payload[:200].decode("utf-8", errors="replace")
|
|
raise PermissionError(f"도엽 {sheet_no}: 다운로드 실패(로그인 세션 확인 필요): {snippet}")
|
|
|
|
Path(MAP_SHEETS_DIR).mkdir(parents=True, exist_ok=True)
|
|
tmp_path = Path(MAP_SHEETS_DIR) / f"_dl_{row['file_nm']}"
|
|
tmp_path.write_bytes(payload)
|
|
try:
|
|
entry = ingest_zip(tmp_path, sheet_no=str(sheet_no))
|
|
entry["source"] = "vworld 지도서비스 자동 다운로드"
|
|
index = _load_index()
|
|
index["sheets"][str(sheet_no)] = entry
|
|
_save_index(index)
|
|
return entry
|
|
finally:
|
|
tmp_path.unlink(missing_ok=True)
|
|
|
|
|
|
def _load_index() -> dict:
|
|
p = Path(MAP_SHEETS_INDEX_PATH)
|
|
if p.exists():
|
|
return json.loads(p.read_text(encoding="utf-8"))
|
|
return {"sheets": {}}
|
|
|
|
|
|
def _save_index(index: dict) -> None:
|
|
Path(MAP_SHEETS_DIR).mkdir(parents=True, exist_ok=True)
|
|
Path(MAP_SHEETS_INDEX_PATH).write_text(
|
|
json.dumps(index, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
|
|
|
|
def get_sheet_path(sheet_no: str) -> Path | None:
|
|
"""영구저장소에 보관된 도엽 zip 경로. 없으면 None."""
|
|
entry = _load_index()["sheets"].get(str(sheet_no))
|
|
if not entry:
|
|
return None
|
|
path = Path(MAP_SHEETS_DIR) / entry["file"]
|
|
return path if path.exists() else None
|
|
|
|
|
|
def missing_sheets(sheet_nos: list[str]) -> list[str]:
|
|
"""요청 도엽 중 영구저장소에 없는 번호 목록."""
|
|
index = _load_index()["sheets"]
|
|
return [
|
|
s
|
|
for s in sheet_nos
|
|
if str(s) not in index or not (Path(MAP_SHEETS_DIR) / index[str(s)]["file"]).exists()
|
|
]
|
|
|
|
|
|
def acquisition_guidance(missing: list[str]) -> dict:
|
|
"""미보유 도엽 취득 안내 (수동 다운로드 폴백)."""
|
|
return {
|
|
"missing": missing,
|
|
"vworld_dataset_url": VWORLD_DATASET_URL,
|
|
"note": (
|
|
"브이월드 데이터센터(수치지형도 v2.0 1:5,000)는 로그인 후 시군구 단위 zip, "
|
|
"국토정보플랫폼은 도엽 단위 zip((B010)수치지도_도엽번호_*.zip)을 제공한다. "
|
|
"다운로드한 zip을 업로드하면 도곽 검증 후 영구저장소에 등록된다."
|
|
),
|
|
}
|
|
|
|
|
|
def _read_frame_bounds_wgs84(zip_path: Path) -> tuple[tuple[float, float, float, float], str]:
|
|
"""zip 내 도곽선(A0010000) SHP bounds를 WGS84로 반환. (bounds, 사용된 CRS).
|
|
|
|
선언된 PRJ가 오라벨(중부/동부원점 뒤바뀜)인 실측 사례가 있어, 파일명 도엽번호
|
|
기준 예측 도곽과 대조해 정합한 CRS를 자동 판별한다.
|
|
"""
|
|
import geopandas as gpd
|
|
|
|
with zipfile.ZipFile(zip_path) as zf:
|
|
names = zf.namelist()
|
|
shp = [n for n in names if n.upper().endswith(f"{_SHEET_FRAME_CODE}.SHP")]
|
|
if not shp:
|
|
raise ValueError(f"{zip_path.name}: 도곽선({_SHEET_FRAME_CODE}) SHP 없음")
|
|
base = shp[0][:-4]
|
|
members = {}
|
|
for ext in (".shp", ".shx", ".dbf", ".prj", ".cpg"):
|
|
for n in names:
|
|
if n.lower() == (base + ext).lower():
|
|
members[ext] = zf.read(n)
|
|
|
|
m = _SHEET_NO_RE.search(zip_path.name)
|
|
expected = sheet5k_to_bounds(m.group(1)) if m else None
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
for ext, data in members.items():
|
|
(Path(td) / f"frame{ext}").write_bytes(data)
|
|
shp_path = Path(td) / "frame.shp"
|
|
|
|
for crs in _CRS_CANDIDATES:
|
|
gdf = gpd.read_file(shp_path)
|
|
gdf = gdf.set_crs(crs, allow_override=True)
|
|
b = tuple(gdf.to_crs("EPSG:4326").total_bounds)
|
|
if expected is None:
|
|
return b, crs
|
|
tol = 0.002
|
|
if (
|
|
b[0] >= expected[0] - tol
|
|
and b[1] >= expected[1] - tol
|
|
and b[2] <= expected[2] + tol
|
|
and b[3] <= expected[3] + tol
|
|
):
|
|
return b, crs
|
|
raise ValueError(f"{zip_path.name}: 어떤 좌표계 후보로도 도엽번호 도곽과 정합하지 않음")
|
|
|
|
|
|
def ingest_zip(src_path: str | Path, sheet_no: str | None = None) -> dict:
|
|
"""도엽 zip을 검증 후 영구저장소에 등록한다.
|
|
|
|
- 도엽번호: 인자 우선, 없으면 파일명에서 8자리 추출.
|
|
- 도곽선 실측 bounds가 도엽번호 예측 도곽 안에 들어야 등록(해안 도엽은 부분 도곽 허용).
|
|
- PRJ 오라벨은 CRS 후보 대조로 자동 교정하고 crs 필드에 기록.
|
|
"""
|
|
src = Path(src_path)
|
|
if not src.exists():
|
|
raise FileNotFoundError(str(src))
|
|
|
|
if sheet_no is None:
|
|
m = _SHEET_NO_RE.search(src.name)
|
|
if not m:
|
|
raise ValueError(f"파일명에서 도엽번호(8자리)를 찾을 수 없음: {src.name}")
|
|
sheet_no = m.group(1)
|
|
sheet_no = str(sheet_no)
|
|
|
|
bounds, crs = _read_frame_bounds_wgs84(src)
|
|
center_no = latlon_to_sheet5k((bounds[1] + bounds[3]) / 2, (bounds[0] + bounds[2]) / 2)
|
|
if center_no != sheet_no:
|
|
raise ValueError(f"도곽 실측 중심의 도엽번호({center_no})가 요청 번호({sheet_no})와 불일치")
|
|
|
|
Path(MAP_SHEETS_DIR).mkdir(parents=True, exist_ok=True)
|
|
dest_name = f"{sheet_no}.zip"
|
|
dest = Path(MAP_SHEETS_DIR) / dest_name
|
|
if src.resolve() != dest.resolve():
|
|
shutil.copy2(src, dest)
|
|
|
|
index = _load_index()
|
|
index["sheets"][sheet_no] = {
|
|
"file": dest_name,
|
|
"original_name": src.name,
|
|
"crs": crs,
|
|
"bounds_wgs84": [round(v, 8) for v in bounds],
|
|
"acquired": datetime.date.today().isoformat(),
|
|
"source": "vworld/ngii 수동 다운로드",
|
|
}
|
|
_save_index(index)
|
|
return index["sheets"][sheet_no]
|
|
|
|
|
|
def ensure_sheets(sheet_nos: list[str]) -> dict:
|
|
"""도엽 목록의 확보 상태 요약: 보유 경로 + 미보유 취득 안내."""
|
|
available = {}
|
|
for s in sheet_nos:
|
|
p = get_sheet_path(s)
|
|
if p:
|
|
available[str(s)] = str(p)
|
|
missing = [str(s) for s in sheet_nos if str(s) not in available]
|
|
result = {"available": available, "missing": missing}
|
|
if missing:
|
|
result["guidance"] = acquisition_guidance(missing)
|
|
return result
|