fix(공용 지형): 전국 0차곡 이웃 도엽 받기 실패 36 가름 · 면 도곽 · zip 없는 도엽 · 어긋난 도곽 = 빈 지형 (PLAN 628-3)

- 도곽 겹 N3A_H0010000(면) 도 읽음 — 섬 · 해안 작은 zip 은 선 도곽 · 도로 면 없이 면 도곽뿐
- Ver2.0 이 NGI · NDA 뿐인 도엽 = LookupError(zip 없음) — 옛 길은 NGI 글을 zip 으로 받다 로그인 실패로 오판
- 공용 지형 받기: ValueError(도곽이 번호 자리와 어긋남 = 제주 · 울릉 번호 체계) 도 빈 지형 · 로그인 · 망 실패는 그대로 실패
- 시험 test_628_3_sheet_fetch_fallback 새로

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-31 20:58:41 +09:00
co-authored by Claude Opus 5.5
parent 216e7c565b
commit a5179e4dfe
3 changed files with 97 additions and 6 deletions
@@ -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
+6 -2
View File
@@ -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
@@ -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)