fix(B05): 레거시 기성막이 측점을 관 지점 정본으로 이관

기슭막이가 관 지점 시설로 옮겨간 뒤(f916aa52) 이관 경로만 옛 스펙에 남아
`POST /route/structures/migrate` 가 400 으로 통째 실패했음 — 저장소가
`managed_by` 타입을 거절하는데 이관은 계속 구조물로 만들었기 때문. 같은 요청에
실린 대피소·기타까지 하나도 이관되지 않았음. 호출부가 B05 진입 경로라 레거시
프로젝트는 진입할 때마다 실패했음.

- B05_Profile_Structures_Migration: 결과를 MigrationPlan(구조물 / 관 시설)으로 나눔.
  managed_by 타입은 레지스트리 기본값(사용자 확정분)을 승계한 PipePoint 로 만듦 —
  구간은 기준측점 ∓ 전/후
- B05_Profile_Structures_Router: 관 시설분을 pipe_points.json 에 덧붙임. 관 지점
  파일이 없으면 그 몫만 미루고 이력도 남기지 않음(원천 측점이 남아 재이관됨)
- common_util_drainage_pipes: read_pipe_points_file · append_pipe_points_file ·
  pipe_points_path_in 신설 — 저장 당시 노선 지문 보존
- 관 시설에서 투영된 "기슭막이" 라벨이 되이관돼 "기타" 로 굳던 경로도 함께 막음
- B06_Section_Engine_Revetment 삭제 — 호출자 0건(벽은 관 세트가 만듦)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-01 20:18:33 +09:00
co-authored by Claude Fable 5
parent bc9f947d0e
commit 1c95290a18
4 changed files with 191 additions and 104 deletions
+61
View File
@@ -33,6 +33,7 @@ from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
from common_util.common_util_json import atomic_write_json
from common_util.common_util_route_geometry import RouteVertex, interpolate_vertex
from config.config_system import (
DRAINAGE_CACHE_DIRNAME,
DRAINAGE_DETAIL_FILENAME,
DRAINAGE_EDITS_DIRNAME,
DRAINAGE_PIPE_POINTS_FILENAME,
@@ -113,6 +114,21 @@ def pipe_points_path(stored_path: str) -> Path:
return edits_dir(stored_path) / DRAINAGE_PIPE_POINTS_FILENAME
def pipe_points_path_in(project_root: Path) -> Path:
"""`pipe_points_path`와 같은 자리를 **프로젝트 실경로**로 가리킨다.
B05는 스토리지 상대경로가 아니라 실경로를 쥐고 있어 `resolve_stored_project_path`를
다시 태울 수 없다(절대경로를 거절한다).
"""
return (
project_root
/ "B04_PreProcess"
/ DRAINAGE_CACHE_DIRNAME
/ DRAINAGE_EDITS_DIRNAME
/ DRAINAGE_PIPE_POINTS_FILENAME
)
def detail_basins_path(stored_path: str) -> Path:
return drainage_dir(stored_path) / DRAINAGE_DETAIL_FILENAME
@@ -330,6 +346,51 @@ def save_pipe_points(
return len(points)
def read_pipe_points_file(path: Path) -> list[PipePoint]:
"""저장분을 **노선 지문과 무관하게** 그대로 읽는다. 파일이 없거나 깨졌으면 빈 목록.
지문 대조는 "이 노선에 그려도 되는가"를 가리는 것이고, 여기서 알고 싶은 것은
"그 자리에 이미 시설이 있는가"다(이관 중복 방지). 두 물음이 달라 읽기도 다르다.
"""
if not path.is_file():
return []
try:
with path.open("r", encoding="utf-8") as file:
document = json.load(file)
except (OSError, json.JSONDecodeError):
logger.warning("배수유역: 관 지점 파일을 읽지 못했습니다 (%s).", path)
return []
return parse_pipe_points(document.get("points"))
def append_pipe_points_file(path: Path, points: list[PipePoint]) -> int | None:
"""관 지점 정본에 시설을 덧붙인다. 덧붙인 개수, 파일이 없으면 None.
저장 당시 노선 지문과 이미 있는 관은 그대로 둔다 — 여기서 지문을 새로 만들면
읽는 쪽이 노선이 바뀐 것으로 보고 저장분을 통째로 버린다(`load_pipe_points_file`).
지문을 알 수 없는 상황(파일 없음 = B04 배수유역 산출물 없음)에서는 쓰지 않고
None으로 알린다 — 지어낸 지문으로 쓰면 다음 읽기에서 사라진다.
"""
if not path.is_file():
return None
try:
with path.open("r", encoding="utf-8") as file:
document = json.load(file)
except (OSError, json.JSONDecodeError):
logger.warning("배수유역: 관 지점 파일을 읽지 못해 덧붙이지 못했습니다 (%s).", path)
return None
stored = list(document.get("points") or [])
atomic_write_json(
path,
{
"route_signature": str(document.get("route_signature") or ""),
"points": [*stored, *(point.as_dict() for point in points)],
},
)
logger.info("배수유역: 관 지점 %d개를 덧붙였습니다 (%s).", len(points), path.name)
return len(points)
def clear_pipe_points(stored_path: str) -> bool:
"""저장된 관 지점과 그 파생물을 지운다. 하나라도 지웠으면 True.