fix(B04): .cpg 없는 도엽 dbf = cp949 로 읽음 · 병합 캐시 열쇠에 파서 판 번호 (PLAN 561-1)

- 도엽 파서 — zip 안 .cpg 가 없으면 encoding cp949 (latin-1 로 읽어 「등고수치」 칸 이름이 깨지던 흠)
- 병합 열쇠 {"sheets", "parser"} — B04 · D03 둘 다 · 옛 깨진 병합본은 다음 받기 때 다시 지음
- 새 지킴 시험 test_561_1 — .cpg 없는 cp949 견본 → 등고수치 읽힘
- 재현(4250c7bc tmp 사본) — 2026 구간 지형면 0 셀 → 201,088 셀 · 등고수치 빈 선 5,280 → 0

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-27 19:58:56 +09:00
co-authored by Claude Opus 5.5
parent 6d4964ae55
commit 70d3a69847
4 changed files with 90 additions and 5 deletions
+7 -2
View File
@@ -526,9 +526,14 @@ def download_geodata(
# 선정에서 빠진 zip은 쓰이지 않으므로 정리한다(다른 지역 잔재 포함).
prune_sheets(sheet_store, sheet_grid)
# 확보된 도엽을 레이어별 병합 GeoJSON으로 산출 — 열쇠(도엽 번호 목록)가 같으면 스킵
from B04_PreProcess.B04_PreProcess_Engine_SheetParser import PARSER_VERSION
# 확보된 도엽을 레이어별 병합 GeoJSON으로 산출 — 열쇠(도엽 번호 목록 · 파서 판)가 같으면 스킵
merge_key_path = processed_dir / "sheet_merge.key.json"
merge_key = {"sheets": sorted(str(item) for item in sheet_result["available"])}
merge_key = {
"sheets": sorted(str(item) for item in sheet_result["available"]),
"parser": PARSER_VERSION,
}
if sheet_result["available"] and (
rebuild
or not any(processed_dir.glob("도엽_*.geojson"))
@@ -37,6 +37,10 @@ SHEET_LAYERS: dict[str, tuple[str, str]] = {
# 산출 파일 접두사 (기존 processed/의 국가 GIS geojson과 구분)
_OUTPUT_PREFIX = "도엽"
# 파서 판 번호 — 병합 캐시 열쇠에 넣음 · 읽는 법이 바뀌면 올려 옛 병합본을 다시 지음
PARSER_VERSION = 2
# .cpg 없는 도엽 dbf 의 글자 인코딩(국토지리정보원 배포본)
DBF_FALLBACK_ENCODING = "cp949"
def _read_layer_from_zip(
@@ -57,7 +61,9 @@ def _read_layer_from_zip(
with tempfile.TemporaryDirectory() as td:
for n in wanted:
(Path(td) / Path(n).name).write_bytes(zf.read(n))
gdf = gpd.read_file(Path(td) / f"{member_base}.shp")
has_cpg = any(Path(n).suffix.lower() == ".cpg" for n in wanted)
options = {} if has_cpg else {"encoding": DBF_FALLBACK_ENCODING}
gdf = gpd.read_file(Path(td) / f"{member_base}.shp", **options)
if gdf.empty:
return None
+2 -2
View File
@@ -26,7 +26,7 @@ from B04_PreProcess.B04_PreProcess_Engine_MapSheet import (
sheet5k_to_bounds,
sheets_for_points,
)
from B04_PreProcess.B04_PreProcess_Engine_SheetParser import parse_and_merge_sheets
from B04_PreProcess.B04_PreProcess_Engine_SheetParser import PARSER_VERSION, parse_and_merge_sheets
from B04_PreProcess.B04_PreProcess_Engine_SheetStore import (
ensure_sheets,
get_project_map_sheets_dir,
@@ -114,7 +114,7 @@ def _merge_sheets(processed: Path, store: Path, available: list[str]) -> list[st
key_path = processed / "sheet_merge.key.json"
old = (_read_json(key_path) or {}).get("sheets") or []
sheets = sorted({*available, *(s for s in old if get_sheet_path(store, s) is not None)})
key = {"sheets": sheets}
key = {"sheets": sheets, "parser": PARSER_VERSION}
if sheets and (_read_json(key_path) != key or not any(processed.glob("도엽_*.geojson"))):
parse_and_merge_sheets(store, sheets, processed)
atomic_write_json(key_path, key)
@@ -0,0 +1,74 @@
"""PLAN 561-1 — `.cpg` 없는 도엽 dbf = cp949 로 읽음 · 병합 열쇠에 파서 판 번호.
./venv/Scripts/python.exe -m pytest resources/tester/test_561_1_sheet_encoding.py -q
`.cpg` 없는 dbf 를 그냥 읽으면 latin-1 → 칸 이름 「등고수치」가 깨져 등고 높이가 빔
(D 프로젝트 4250c7bc 「등고 라인 94단을 격자에 굽어 0셀」).
"""
from __future__ import annotations
import sys
import zipfile
from pathlib import Path
import geopandas as gpd
from shapely.geometry import LineString
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import B04_PreProcess.B04_PreProcess_Engine_SheetParser as parser # noqa: E402
BASE = "N3L_F0010000"
def _sheet_zip(tmp_path: Path, with_cpg: bool) -> Path:
"""등고선 두 줄짜리 도엽 zip — dbf 는 cp949 · `.cpg` 는 고름."""
shp_dir = tmp_path / ("cpg" if with_cpg else "nocpg")
shp_dir.mkdir()
gdf = gpd.GeoDataFrame(
{"등고수치": [100.0, 105.0], "구분": ["주곡선", "주곡선"]},
geometry=[LineString([(0, 0), (10, 0)]), LineString([(0, 5), (10, 5)])],
crs="EPSG:5186",
)
gdf.to_file(shp_dir / f"{BASE}.shp", encoding="cp949")
cpg = shp_dir / f"{BASE}.cpg"
if with_cpg:
cpg.write_text("CP949", encoding="ascii")
else:
cpg.unlink(missing_ok=True)
zip_path = tmp_path / f"{shp_dir.name}.zip"
with zipfile.ZipFile(zip_path, "w") as zf:
for f in shp_dir.iterdir():
zf.write(f, f.name)
return zip_path
def _read(zip_path: Path) -> gpd.GeoDataFrame:
gdf = parser._read_layer_from_zip(zip_path, "N3L", "F0010000", "EPSG:5186")
assert gdf is not None
return gdf
def test_no_cpg_reads_cp949(tmp_path):
gdf = _read(_sheet_zip(tmp_path, with_cpg=False))
assert "등고수치" in gdf.columns
assert sorted(gdf["등고수치"].tolist()) == [100.0, 105.0]
assert set(gdf["구분"]) == {"주곡선"}
def test_with_cpg_unchanged(tmp_path):
gdf = _read(_sheet_zip(tmp_path, with_cpg=True))
assert sorted(gdf["등고수치"].tolist()) == [100.0, 105.0]
def test_merge_keys_carry_parser_version():
assert parser.PARSER_VERSION >= 2
for rel in (
"B04_PreProcess/B04_PreProcess_Engine.py",
"D03_PlanAnalysis/D03_PlanAnalysis_Terrain.py",
):
text = (ROOT / rel).read_text(encoding="utf-8")
assert '"parser": PARSER_VERSION' in text, rel