diff --git a/B04_PreProcess/B04_PreProcess_Engine_SheetStore.py b/B04_PreProcess/B04_PreProcess_Engine_SheetStore.py index 5aab57a21..9c8bc3354 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_SheetStore.py +++ b/B04_PreProcess/B04_PreProcess_Engine_SheetStore.py @@ -39,8 +39,10 @@ from .B04_PreProcess_Engine_MapSheet import latlon_to_sheet5k, sheet5k_to_bounds logger = logging.getLogger(__name__) -# 도곽선 레이어 코드 (수치지형도 v2.0 도엽본) — 도곽선 N3L_H0010000 이 없는 zip 은 옛 A0010000(도로 면)로 대신 +# 도곽선 레이어 코드 (수치지형도 v2.0 도엽본) — 선 N3L_H0010000 · 면 N3A_H0010000(섬 · 해안 작은 도엽) +# · 둘 다 없는 zip 은 옛 A0010000(도로 면)로 대신 _SHEET_FRAME_CODE = "N3L_H0010000" +_SHEET_FRAME_AREA_CODE = "N3A_H0010000" _SHEET_FRAME_FALLBACK_CODE = "A0010000" _SHEET_NO_RE = re.compile(r"(\d{8})") _SHARED_DIRNAME = "_shared" # storage/_shared 아래 = 여러 프로젝트가 같이 쓰는 창고 @@ -153,7 +155,11 @@ def download_sheet(store_dir: str | Path, sheet_no: str, version: int = 2) -> di rows = search_sheet_files(sheet_no, version=version) if not rows: raise LookupError(f"도엽 {sheet_no}: 브이월드 검색 결과 없음") - row = rows[0] + # Ver2.0 이 zip 아닌 NGI · NDA 뿐인 도엽(작은 섬) — 받을 zip 없음 + row = next((r for r in rows if str(r.get("file_nm", "")).lower().endswith(".zip")), None) + if row is None: + kinds = ", ".join(sorted({str(r.get("file_extsn", "")) for r in rows})) + raise LookupError(f"도엽 {sheet_no}: Ver2.0 zip 없음({kinds})") cookie = _session_cookie_cache or VWORLD_SESSION_COOKIE payload = None @@ -318,7 +324,7 @@ def ingest_zip(store_dir: str | Path, src_path: str | Path, sheet_no: str | None def _read_frame_bounds_wgs84(zip_path: Path) -> tuple[tuple[float, float, float, float], str]: - """zip 내 도곽선(N3L_H0010000 · 없으면 A0010000) SHP bounds를 WGS84로 반환. (bounds, 사용된 CRS). + """zip 내 도곽선(N3L_H0010000 · N3A_H0010000 · 없으면 A0010000) SHP bounds를 WGS84로 반환. (bounds, 사용된 CRS). 선언된 PRJ가 오라벨(중부/동부원점 뒤바뀜)인 실측 사례가 있어, 파일명 도엽번호 기준 예측 도곽과 대조해 정합한 CRS를 자동 판별한다. @@ -328,7 +334,7 @@ def _read_frame_bounds_wgs84(zip_path: Path) -> tuple[tuple[float, float, float, with zipfile.ZipFile(zip_path) as zf: names = zf.namelist() shp: list[str] = [] - for code in (_SHEET_FRAME_CODE, _SHEET_FRAME_FALLBACK_CODE): + for code in (_SHEET_FRAME_CODE, _SHEET_FRAME_AREA_CODE, _SHEET_FRAME_FALLBACK_CODE): shp = [n for n in names if n.upper().endswith(f"{code}.SHP")] if shp: break diff --git a/common_util/common_util_terrain_store.py b/common_util/common_util_terrain_store.py index b4302f407..acb7ce59b 100644 --- a/common_util/common_util_terrain_store.py +++ b/common_util/common_util_terrain_store.py @@ -273,8 +273,12 @@ def _fetch_zips(sheet_no: str, report: Callable[[int, str], None]) -> tuple[dict zips[no] = _sha256(TERRAIN_ROOT / no / f"{no}.zip") elif no == sheet_no: reason = result["failed"].get(no, "") - # 검색 결과 없음 = 바다 등 진짜 빈 도엽 · 그 밖(로그인 · 망)은 다음에 다시 - return zips, None if reason.startswith("LookupError") else reason + # 검색 결과 · zip 없음 = 바다 · 작은 섬 · zip 꼴 흠(ValueError — 도곽이 번호 자리와 어긋남 = + # 제주 · 울릉 등 다른 번호 체계) = 그 자리 등고선 없음 · 그 밖(로그인 · 망)은 다음에 다시 + if reason.startswith(("LookupError", "ValueError")): + logger.warning("도엽 %s zip 못 씀 — 빈 지형: %s", no, reason[:200]) + return zips, None + return zips, reason return zips, None diff --git a/resources/tester/test_628_3_sheet_fetch_fallback.py b/resources/tester/test_628_3_sheet_fetch_fallback.py new file mode 100644 index 000000000..8c2323fba --- /dev/null +++ b/resources/tester/test_628_3_sheet_fetch_fallback.py @@ -0,0 +1,81 @@ +"""628-3 — 전국 0차곡 이웃 도엽 받기 실패 셋: 면 도곽(N3A_H0010000) · zip 없는 도엽(NGI 뿐) · +도곽이 번호 자리와 어긋난 zip(제주 · 울릉) = 빈 지형 · 로그인 · 망 실패는 그대로 실패.""" + +import sys +import tempfile +import zipfile +from pathlib import Path + +import geopandas as gpd +import pytest +from shapely.geometry import box + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from B04_PreProcess import B04_PreProcess_Engine_SheetStore as sheet_store # noqa: E402 +from B04_PreProcess.B04_PreProcess_Engine_MapSheet import sheet5k_to_bounds # noqa: E402 +from common_util import common_util_terrain_store as store # noqa: E402 + +SHEET = "34616006" +CRS = "EPSG:5186" + + +def _make_zip(path: Path, code: str, bb: tuple[float, float, float, float]) -> None: + with tempfile.TemporaryDirectory() as td, zipfile.ZipFile(path, "w") as zf: + gdf = gpd.GeoDataFrame(geometry=[box(*bb)], crs="EPSG:4326").to_crs(CRS) + gdf.to_file(Path(td) / f"{code}.shp") + for f in Path(td).glob(f"{code}.*"): + zf.write(f, f.name) + + +def test_면_도곽만_있는_섬_도엽_받기_통과(tmp_path): + z = tmp_path / f"{SHEET}.zip" + _make_zip(z, "N3A_H0010000", sheet5k_to_bounds(SHEET)) + entry = sheet_store.ingest_zip(tmp_path / "store", z, sheet_no=SHEET) + for got, want in zip(entry["bounds_wgs84"], sheet5k_to_bounds(SHEET)): + assert got == pytest.approx(want, abs=1e-6) + + +def test_도곽이_번호_자리와_어긋나면_ValueError(tmp_path): + w, s, e, n = sheet5k_to_bounds(SHEET) + z = tmp_path / f"{SHEET}.zip" + _make_zip(z, "N3A_H0010000", (w, s - 0.1, e, n - 0.1)) # 제주식 0.1° 어긋남 + with pytest.raises(ValueError, match="정합하지 않음"): + sheet_store.ingest_zip(tmp_path / "store", z, sheet_no=SHEET) + + +def test_zip_없는_도엽은_LookupError_받기_안_함(tmp_path, monkeypatch): + rows = [ + {"ds_sq": 1, "file_no": 1, "file_nm": f"{SHEET}.nda", "file_extsn": "nda"}, + {"ds_sq": 1, "file_no": 2, "file_nm": f"{SHEET}.ngi", "file_extsn": "ngi"}, + ] + monkeypatch.setattr(sheet_store, "search_sheet_files", lambda no, version=2: rows) + + def no_fetch(*_a, **_k): + raise AssertionError("받기 안 함") + + monkeypatch.setattr(sheet_store, "_fetch_sheet_payload", no_fetch) + monkeypatch.setattr(sheet_store, "vworld_login", no_fetch) + with pytest.raises(LookupError, match="zip 없음"): + sheet_store.download_sheet(tmp_path, SHEET) + result = sheet_store.ensure_sheets(tmp_path, [SHEET]) + assert result["failed"][SHEET].startswith("LookupError") + + +@pytest.mark.parametrize( + ("reason", "empty"), + [ + ("LookupError: 검색 결과 없음", True), + ("LookupError: Ver2.0 zip 없음(nda, ngi)", True), + ("ValueError: _dl_x.zip: 어떤 좌표계 후보로도 도엽번호 도곽과 정합하지 않음", True), + ("PermissionError: 재로그인 후에도 실패", False), + ("URLError: timed out", False), + ], +) +def test_가운데_도엽_받기_실패_가름(monkeypatch, reason, empty): + monkeypatch.setattr( + sheet_store, "ensure_sheets", lambda d, nos: {"available": {}, "failed": {nos[0]: reason}} + ) + zips, failure = store._fetch_zips(SHEET, lambda p, s: None) + assert zips == {} + assert failure == (None if empty else reason)