auto: 2026-07-26 18:15 (ESD_LAPTOP)
This commit is contained in:
@@ -1,14 +1,16 @@
|
||||
# B04_wf1_Surface_Engine_SheetStore.py
|
||||
# 1:5,000 수치지형도 도엽 zip 전역 영구저장소 관리.
|
||||
#
|
||||
# 조사 결과(2026-07-26):
|
||||
# - 브이월드 오픈 API에는 도엽 색인·수치지형도 레이어가 없어 도엽번호 산출은
|
||||
# 자체 알고리즘(B04_wf1_Surface_Engine_MapSheet) 사용 — 보유 도엽 10매 실측 정합 확인.
|
||||
# - 브이월드 데이터센터 수치지형도 v2.0(1:5000, dsId=30205)은 시군구 단위 zip이며
|
||||
# 다운로드(GET /dtmk/downloadResourceFile.do?ds_id=30205&fileNo=..)는 OIDC SSO
|
||||
# 로그인 세션 필수(비로그인 시 200/0바이트) → 서버 자동 다운로드 불가.
|
||||
# - 따라서 취득은 사용자 수동 다운로드 + 업로드(인제스트) 폴백으로 확정하고,
|
||||
# 확보된 zip은 resources/map_sheets/{도엽번호}.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
|
||||
|
||||
@@ -17,6 +19,8 @@ import json
|
||||
import re
|
||||
import shutil
|
||||
import tempfile
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
@@ -32,6 +36,68 @@ _SHEET_NO_RE = re.compile(r"(\d{8})")
|
||||
_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=..").
|
||||
"""
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user