diff --git a/B04_wf1_Surface/B04_wf1_Surface_Engine_SheetStore.py b/B04_wf1_Surface/B04_wf1_Surface_Engine_SheetStore.py index 569eca19..8766e65d 100644 --- a/B04_wf1_Surface/B04_wf1_Surface_Engine_SheetStore.py +++ b/B04_wf1_Surface/B04_wf1_Surface_Engine_SheetStore.py @@ -1,5 +1,7 @@ # B04_wf1_Surface_Engine_SheetStore.py -# 1:5,000 수치지형도 도엽 zip 전역 영구저장소 관리. +# 1:5,000 수치지형도 도엽 zip — 프로젝트 영구저장소 관리. +# 저장 위치: {프로젝트 저장소}/B04_wf1_Surface/processed/map_sheets/{도엽번호}.zip +# (위성지도·GIS 벡터 등 주변데이터와 동일한 프로젝트 processed 계층) # # 조사 결과(2026-07-26 실측): # - 도엽번호 산출: 브이월드 오픈 API(api.vworld.kr)에는 색인 레이어가 없어 @@ -8,9 +10,7 @@ # (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 으로 영구 보관한다. +# 로그인 세션 필요. 세션 만료 시 id/pw 자동 재로그인(Base64 POST, 캡차 없음 실측). from __future__ import annotations @@ -27,8 +27,8 @@ import zipfile from pathlib import Path from config.config_system import ( - MAP_SHEETS_DIR, - MAP_SHEETS_INDEX_PATH, + MAP_SHEETS_DIRNAME, + MAP_SHEETS_INDEX_FILENAME, VWORLD_LOGIN_ID, VWORLD_LOGIN_PW, VWORLD_SESSION_COOKIE, @@ -43,7 +43,6 @@ _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" @@ -53,6 +52,20 @@ _LOGIN_API = "https://map.vworld.kr/map/im_usrlogin_a004.do" _session_cookie_cache: str = "" +def get_project_map_sheets_dir(project_storage_path: str | Path) -> Path: + """프로젝트 영구저장소 내 도엽 저장 폴더 경로. + + project_storage_path: config_system.get_project_storage_path() 결과 + (storage/{회사}/{사용자}/{프로젝트ID}). + """ + return Path(project_storage_path) / "B04_wf1_Surface" / "processed" / MAP_SHEETS_DIRNAME + + +# ───────────────────────────────────────────────────────────────────────── +# 브이월드 접근 (로그인 / 검색 / 다운로드) +# ───────────────────────────────────────────────────────────────────────── + + def vworld_login() -> str: """id/pw로 브이월드 지도서비스 로그인, Cookie 헤더 문자열 반환. @@ -127,17 +140,17 @@ def _fetch_sheet_payload(row: dict, session_cookie: str) -> bytes: return payload -def download_sheet(sheet_no: str, session_cookie: str = "", version: int = 2) -> dict: - """도엽 파일 다운로드 후 영구저장소 등록. 세션 만료 시 id/pw로 자동 재로그인 1회 재시도. +def download_sheet(store_dir: str | Path, sheet_no: str, version: int = 2) -> dict: + """도엽 다운로드 후 프로젝트 영구저장소 등록. 세션 만료 시 자동 재로그인 1회 재시도. - 쿠키 우선순위: 인자 > 프로세스 캐시 > .env VWORLD_SESSION_COOKIE > 자동 로그인. + 쿠키 우선순위: 프로세스 캐시 > .env VWORLD_SESSION_COOKIE > 자동 로그인. """ rows = search_sheet_files(sheet_no, version=version) if not rows: raise LookupError(f"도엽 {sheet_no}: 브이월드 검색 결과 없음") row = rows[0] - cookie = session_cookie or _session_cookie_cache or VWORLD_SESSION_COOKIE + cookie = _session_cookie_cache or VWORLD_SESSION_COOKIE payload = None if cookie: try: @@ -151,64 +164,116 @@ def download_sheet(sheet_no: str, session_cookie: str = "", version: int = 2) -> except PermissionError as e: raise PermissionError(f"도엽 {sheet_no}: 재로그인 후에도 실패: {e}") from e - Path(MAP_SHEETS_DIR).mkdir(parents=True, exist_ok=True) - tmp_path = Path(MAP_SHEETS_DIR) / f"_dl_{row['file_nm']}" + store = Path(store_dir) + store.mkdir(parents=True, exist_ok=True) + tmp_path = store / f"_dl_{row['file_nm']}" tmp_path.write_bytes(payload) try: - entry = ingest_zip(tmp_path, sheet_no=str(sheet_no)) + entry = ingest_zip(store, tmp_path, sheet_no=str(sheet_no)) entry["source"] = "vworld 지도서비스 자동 다운로드" - index = _load_index() + index = _load_index(store) index["sheets"][str(sheet_no)] = entry - _save_index(index) + _save_index(store, index) return entry finally: tmp_path.unlink(missing_ok=True) -def _load_index() -> dict: - p = Path(MAP_SHEETS_INDEX_PATH) +def ensure_sheets(store_dir: str | Path, sheet_nos: list[str]) -> dict: + """도엽 목록 확보: 없는 도엽은 자동 다운로드. {available, failed} 반환.""" + store = Path(store_dir) + available: dict[str, str] = {} + failed: dict[str, str] = {} + for s in sheet_nos: + s = str(s) + path = get_sheet_path(store, s) + if path is None: + try: + download_sheet(store, s) + path = get_sheet_path(store, s) + except Exception as e: + failed[s] = f"{type(e).__name__}: {e}" + continue + available[s] = str(path) + return {"available": available, "failed": failed} + + +# ───────────────────────────────────────────────────────────────────────── +# 영구저장소 인덱스 / 조회 / 인제스트 +# ───────────────────────────────────────────────────────────────────────── + + +def _load_index(store_dir: Path) -> dict: + p = store_dir / MAP_SHEETS_INDEX_FILENAME 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( +def _save_index(store_dir: Path, index: dict) -> None: + store_dir.mkdir(parents=True, exist_ok=True) + (store_dir / MAP_SHEETS_INDEX_FILENAME).write_text( json.dumps(index, ensure_ascii=False, indent=2), encoding="utf-8" ) -def get_sheet_path(sheet_no: str) -> Path | None: +def get_sheet_path(store_dir: str | Path, sheet_no: str) -> Path | None: """영구저장소에 보관된 도엽 zip 경로. 없으면 None.""" - entry = _load_index()["sheets"].get(str(sheet_no)) + store = Path(store_dir) + entry = _load_index(store)["sheets"].get(str(sheet_no)) if not entry: return None - path = Path(MAP_SHEETS_DIR) / entry["file"] + path = store / entry["file"] return path if path.exists() else None -def missing_sheets(sheet_nos: list[str]) -> list[str]: +def missing_sheets(store_dir: str | Path, 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() - ] + store = Path(store_dir) + return [s for s in sheet_nos if get_sheet_path(store, str(s)) is None] -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 ingest_zip(store_dir: str | Path, src_path: str | Path, sheet_no: str | None = None) -> dict: + """도엽 zip을 검증 후 영구저장소에 등록한다 (수동 다운로드 폴백 겸용). + + - 도엽번호: 인자 우선, 없으면 파일명에서 8자리 추출. + - 도곽선 실측 bounds가 도엽번호 예측 도곽 안에 들어야 등록(해안 도엽은 부분 도곽 허용). + - PRJ 오라벨은 CRS 후보 대조로 자동 교정하고 crs 필드에 기록. + """ + store = Path(store_dir) + 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})와 불일치") + + store.mkdir(parents=True, exist_ok=True) + dest_name = f"{sheet_no}.zip" + dest = store / dest_name + if src.resolve() != dest.resolve(): + shutil.copy2(src, dest) + + index = _load_index(store) + 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": "수동 다운로드 인제스트", } + _save_index(store, index) + return index["sheets"][sheet_no] def _read_frame_bounds_wgs84(zip_path: Path) -> tuple[tuple[float, float, float, float], str]: @@ -254,59 +319,3 @@ def _read_frame_bounds_wgs84(zip_path: Path) -> tuple[tuple[float, float, float, ): 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 diff --git a/config/config_system.py b/config/config_system.py index bcdc5708..9a5bd69f 100644 --- a/config/config_system.py +++ b/config/config_system.py @@ -430,9 +430,10 @@ FOREST_ROAD_PROFILE_ALIGNMENT = { # ───────────────────────────────────────────────────────────────────────── STORAGE_BASE_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "storage") -# 1:5,000 수치지형도 도엽 전역 영구저장소 (국가 데이터, 프로젝트 간 공유 — 재다운로드 금지) -MAP_SHEETS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "resources", "map_sheets") -MAP_SHEETS_INDEX_PATH = os.path.join(MAP_SHEETS_DIR, "map_sheets_index.json") +# 1:5,000 수치지형도 도엽 저장 폴더명 — 프로젝트 영구저장소 하위 +# (storage/{회사}/{사용자}/{프로젝트ID}/B04_wf1_Surface/processed/map_sheets/) +MAP_SHEETS_DIRNAME = "map_sheets" +MAP_SHEETS_INDEX_FILENAME = "map_sheets_index.json" # 브이월드 지도서비스 도엽 자동 다운로드 — 세션 만료 시 id/pw로 자동 재로그인 (.env) VWORLD_LOGIN_ID = os.getenv("VWORLD_LOGIN_ID", "") diff --git a/scratch/test_sheetstore_ingest.py b/scratch/test_sheetstore_ingest.py index 100dfd6b..35bd8616 100644 --- a/scratch/test_sheetstore_ingest.py +++ b/scratch/test_sheetstore_ingest.py @@ -1,7 +1,7 @@ -"""도엽 영구저장소(SheetStore) 인제스트 실테스트. +"""도엽 프로젝트 영구저장소(SheetStore) 실테스트. -Downloads의 보유 도엽 zip 전량을 resources/map_sheets/에 등록하고, -CRS 오라벨 zip(36902002.zip) 자동 교정과 ensure_sheets 3x3 흐름을 확인한다. +- 저장 위치: {프로젝트 저장소}/B04_wf1_Surface/processed/map_sheets/ +- ensure_sheets: 보유 도엽은 즉시 반환, 미보유는 자동 다운로드(세션 만료 시 자동 재로그인) """ import sys @@ -13,34 +13,30 @@ sys.path.insert(0, str(PROJECT_ROOT)) from B04_wf1_Surface.B04_wf1_Surface_Engine_MapSheet import neighbors_3x3 from B04_wf1_Surface.B04_wf1_Surface_Engine_SheetStore import ( ensure_sheets, + get_project_map_sheets_dir, get_sheet_path, - ingest_zip, ) -DOWNLOADS = Path.home() / "Downloads" +TEST_PROJECT = PROJECT_ROOT / "storage/1/3/acb9170b-9ac8-49b3-82a0-51cfa32bb42d" def main(): - zips = sorted(DOWNLOADS.glob("*수치지도_*.zip")) + [ - p for p in (DOWNLOADS / "36906042.zip", DOWNLOADS / "36902002.zip") if p.exists() - ] - for zp in zips: - try: - entry = ingest_zip(zp) - print(f"OK {zp.name} -> {entry['file']} crs={entry['crs']}") - except Exception as e: - print(f"FAIL {zp.name}: {e}") + store = get_project_map_sheets_dir(TEST_PROJECT) + print("영구저장소:", store) - print("\n--- ensure_sheets: 36906042 중심 3x3 ---") - result = ensure_sheets(neighbors_3x3("36906042")) - print("보유:", sorted(result["available"].keys())) - print("미보유:", result["missing"]) - if "guidance" in result: - print("안내 URL:", result["guidance"]["vworld_dataset_url"]) + print("\n--- ensure_sheets: 36906042 중심 3x3 (전량 보유 — 다운로드 없이 즉시) ---") + result = ensure_sheets(store, neighbors_3x3("36906042")) + print("확보:", sorted(result["available"].keys())) + print("실패:", result["failed"]) + + print("\n--- ensure_sheets: 미보유 1매 자동 다운로드 (36906024) ---") + result = ensure_sheets(store, ["36906024"]) + print("확보:", sorted(result["available"].keys())) + print("실패:", result["failed"]) print("\n--- get_sheet_path ---") - print("36906042:", get_sheet_path("36906042")) - print("99999999:", get_sheet_path("99999999")) + print("36906042:", get_sheet_path(store, "36906042")) + print("99999999:", get_sheet_path(store, "99999999")) if __name__ == "__main__":