feat(배수유역): 342-3 노선 · 관 자리 지문 — 노선이 분석 때와 다르면 세부 유역 안 나누고 「낡음」

- 분석 응답 · 저장분(00_watershed_response.json)에 route_signature · pipe_signature
- 세부 유역(GET · POST · DELETE · PUT pipe-points): 분석 지문 ≠ 지금 노선(옛 저장분 = 지문 없음) → 관 자리만 · basins 없음 · stale 까닭 글 · 관 목록도 없으면 409
- PUT 낡음 = 관 자리 · 속성만 정본 · 세부 유역 산출물(04_detailed_basins) 안 씀
- 응답에 지금 route_signature · pipe_signature · B05 배수유역 칸 = 낡음 글 표시(DetailBasinResponse 칸 셋)
- 시험 test_342_3_drainage_stale

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XWh5wNNycMkbwMpACz1rte

B04_PreProcess/B04_PreProcess_Api_Fetch.ts

@@ -553,6 +553,11 @@ export interface DetailBasinResponse {
   upstream_lonlat: Array<Array<[number, number]>>;
   /** [저장] 때 못 만든 구조물 조합 사본 폴더(142-2 · `*` = 통째 실패). */
   snapshot_failed?: string[];
+  /** 지문(342-3) — 지금 노선 · 관 자리. */
+  route_signature?: string;
+  pipe_signature?: string;
+  /** 분석이 낡음(노선이 분석 때와 다름) — 까닭 글 · 세부 유역 없음 · 관 자리만. */
+  stale?: string | null;
 }
 
 let lastNote = { text: "", at: 0 };

B04_PreProcess/B04_PreProcess_Router_Basins.py

@@ -28,15 +28,21 @@ from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
 from B06_Section.B06_Section_Engine_Culvert import pipe_wall_min_height_m
 from common_util.common_util_auth import get_optional_session
 from common_util.common_util_drainage_context import DrainageContext, load_drainage_context
-from common_util.common_util_drainage_detail import DrainageDetail, build_detail
+from common_util.common_util_drainage_detail import (
+    DrainageDetail,
+    build_detail,
+    pipes_from_chainages,
+)
 from common_util.common_util_drainage_pipes import (
     PIPE_SOURCE_USER,
     PipePoint,
+    analysis_route_signature,
     carry_facility_attributes,
     carry_indices,
     clear_pipe_points,
     load_pipe_points,
     parse_pipe_points,
+    pipe_signature,
     route_signature,
     save_detail_basins,
     save_pipe_points,
@@ -54,6 +60,10 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/projects", tags=["B04 Surface Drainage Basins"])
 
 _NO_ANALYSIS = "배수유역 분석 결과가 없습니다. [유역 분석]을 먼저 실행하세요."
+# 노선 지문이 분석 때와 다름(옛 저장분은 지문 없음) — 옛 격자로 나누면 틀린 유역이 나온다(342-3).
+STALE_MESSAGE = (
+    "배수유역 낡음 — 노선이 분석 때와 다릅니다. 유역을 다시 분석해야 세부 유역을 나눕니다."
+)
 
 
 def _build(
@@ -72,7 +82,7 @@ def _build(
         return None, []
     if points is not None:
         _retag(detail.pipes, points)
-    rebuilt = [PipePoint(chainage_m=pipe.chainage_m, source=pipe.reason) for pipe in detail.pipes]
+    rebuilt = _rebuilt(detail)
     # 원본에 없던 자리 = 이번에 새로 놓인 관(자동 배치 · 간격 보충) — 등록부 기본값을 받을 관
     fresh = [
         pick is None
@@ -90,6 +100,21 @@ def _build(
     return detail, carried
 
 
+def _rebuilt(detail: DrainageDetail) -> list[PipePoint]:
+    return [PipePoint(chainage_m=pipe.chainage_m, source=pipe.reason) for pipe in detail.pipes]
+
+
+def _build_stale(context: DrainageContext, points: list[PipePoint]) -> DrainageDetail:
+    """낡은 분석 — 세부 유역 없이 관 자리만(지금 노선 위) · 시설 속성은 넘겨받은 그대로."""
+    to_lonlat = context.to_lonlat
+    detail = DrainageDetail(
+        route_lonlat=[list(to_lonlat(vertex.x, vertex.y)) for vertex in context.vertices],
+        pipes=pipes_from_chainages(context.vertices, [point.chainage_m for point in points]),
+    )
+    _retag(detail.pipes, points)
+    return detail
+
+
 # 관 보호공 · 유입 집수정 높이 칸 — 새 관 기본 = 관경 + 0.5(139-1) · 유입 흙막이만 + 0.7(211-1)
 PIPE_HEIGHT_KEYS = ("inlet_revet_height_m", "outlet_revet_height_m", "inlet_basin_height_m")
 
@@ -181,6 +206,7 @@ def _payload(
     detail: DrainageDetail,
     points: list[PipePoint],
     saved: bool,
+    stale: bool = False,
 ) -> dict[str, Any]:
     """관 목록과 세부유역을 화면 좌표(WGS84)로 정리한다.
 
@@ -188,7 +214,12 @@ def _payload(
     응답이 151KB → 100KB(압축까지 얹으면 22KB)가 된다(2026-09-06 실측). 화면이 그리는 값이고
     다시 계산에 넣지 않으므로 줄여도 결과가 안 갈린다.
     """
-    return round_floats(_raw_payload(project_id, context, detail, points, saved), LONLAT_DIGITS)
+    payload = _raw_payload(project_id, context, detail, points, saved)
+    # 지문(342-3) — 화면이 지금 노선 · 관 자리와 견줌 · 낡으면 까닭 글(세부 유역 없음)
+    payload["route_signature"] = route_signature(context.vertices)
+    payload["pipe_signature"] = pipe_signature([pipe.chainage_m for pipe in detail.pipes])
+    payload["stale"] = STALE_MESSAGE if stale else None
+    return round_floats(payload, LONLAT_DIGITS)
 
 
 def _raw_payload(
@@ -322,9 +353,12 @@ def _basin_features(
 
 async def _resolve(
     project_id: UUID, payload: dict[str, Any] | None, *, use_stored: bool
-) -> tuple[DrainageContext, DrainageDetail, list[PipePoint], bool] | JSONResponse:
+) -> tuple[DrainageContext, DrainageDetail, list[PipePoint], bool, bool] | JSONResponse:
     """요청 본문 → 저장분 → 자동 생성 순으로 관 목록을 정하고 세부유역까지 계산한다.
 
+    분석의 노선 지문이 지금 노선과 다르면(옛 저장분 = 지문 없음) 나누지 않고 관 자리만 낡음으로
+    돌려준다 — 관 목록도 없으면(자동 생성 자리) 409(342-3). 마지막 값 = 낡음.
+
     빈 목록(`[]`)은 「관 0개」로 그대로 쓴다 — 자동 생성은 요청에 칸이 없고 저장분도 없을(None) 때만(312).
     """
     context, reason = await load_drainage_context(project_id)
@@ -341,12 +375,21 @@ async def _resolve(
         load_pipe_points(context.stored_path, signature, context.vertices) if use_stored else None
     )
     points = requested if requested is not None else stored
+    saved = requested is None and stored is not None
+
+    if analysis_route_signature(context.stored_path) != signature:
+        if points is None:
+            return JSONResponse(
+                status_code=409, content={"status": "stale", "message": STALE_MESSAGE}
+            )
+        detail = _build_stale(context, points)
+        return context, detail, carry_facility_attributes(_rebuilt(detail), points), saved, True
 
     result = await asyncio.to_thread(_build, context.stored_path, context, points)
     detail, resolved = result
     if detail is None:
         return JSONResponse(status_code=404, content={"status": "error", "message": _NO_ANALYSIS})
-    return context, detail, resolved, requested is None and stored is not None
+    return context, detail, resolved, saved, False
 
 
 @router.get("/{project_id}/drainage/pipe-points", response_model=None)
@@ -355,8 +398,8 @@ async def get_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
     resolved = await _resolve(project_id, None, use_stored=True)
     if isinstance(resolved, JSONResponse):
         return resolved
-    context, detail, points, saved = resolved
-    return _payload(project_id, context, detail, points, saved)
+    context, detail, points, saved, stale = resolved
+    return _payload(project_id, context, detail, points, saved, stale)
 
 
 @router.post("/{project_id}/drainage/detail-basins", response_model=None)
@@ -371,8 +414,8 @@ async def post_detail_basins(
     resolved = await _resolve(project_id, payload, use_stored=False)
     if isinstance(resolved, JSONResponse):
         return resolved
-    context, detail, points, _ = resolved
-    return _payload(project_id, context, detail, points, saved=False)
+    context, detail, points, _, stale = resolved
+    return _payload(project_id, context, detail, points, False, stale)
 
 
 @router.delete("/{project_id}/drainage/pipe-points", response_model=None)
@@ -390,8 +433,8 @@ async def delete_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
     resolved = await _resolve(project_id, None, use_stored=False)
     if isinstance(resolved, JSONResponse):
         return resolved
-    context, detail, points, _ = resolved
-    return _payload(project_id, context, detail, points, saved=False)
+    context, detail, points, _, stale = resolved
+    return _payload(project_id, context, detail, points, False, stale)
 
 
 @router.put("/{project_id}/drainage/pipe-points", response_model=None)
@@ -404,19 +447,20 @@ async def put_pipe_points(
     resolved = await _resolve(project_id, payload, use_stored=True)
     if isinstance(resolved, JSONResponse):
         return resolved
-    context, detail, points, _ = resolved
+    context, detail, points, _, stale = resolved
 
     signature = route_signature(context.vertices)
     # 좌표를 같이 남긴다 — 다른 선(B05 최적 경로)으로 읽어도 그 자리에 되놓는다.
     saved = await asyncio.to_thread(
         save_pipe_points, context.stored_path, signature, points, context.vertices
     )
-    await asyncio.to_thread(
-        save_detail_basins,
-        context.stored_path,
-        _basin_features(context, detail, points),
-        context.crs,
-    )
+    if not stale:  # 낡은 분석 — 관 자리 · 속성만 정본으로 · 세부 유역 산출물은 안 씀
+        await asyncio.to_thread(
+            save_detail_basins,
+            context.stored_path,
+            _basin_features(context, detail, points),
+            context.crs,
+        )
     # 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
     from common_util.common_util_storage import resolve_stored_project_path
     from M02_MasterTemplete.M02_Item_Snapshot import importer, safe_sync
@@ -424,7 +468,7 @@ async def put_pipe_points(
     failed = await asyncio.to_thread(
         safe_sync, resolve_stored_project_path(context.stored_path), importer(session)
     )
-    result = _payload(project_id, context, detail, points, saved=True)
+    result = _payload(project_id, context, detail, points, True, stale)
     result["saved_count"] = saved
     result["snapshot_failed"] = failed  # 못 만든 구조물 조합 복사본(142-2) — [저장] 알림
     return result

B04_PreProcess/B04_PreProcess_Router_Watershed.py

@@ -43,6 +43,7 @@ from B04_PreProcess.B04_PreProcess_Router_Watershed_Output import (
 )
 from B05_Profile.B05_Profile_Repository import get_surface_crs_epsg
 from common_util.common_util_crs import resolve_project_crs
+from common_util.common_util_drainage_pipes import pipe_signature, route_signature
 from common_util.common_util_route_geometry import (
     StructureCandidate,
     find_planned_route_file,
@@ -419,6 +420,9 @@ async def get_primary_region(
         "schema_version": RESPONSE_SCHEMA_VERSION,
         "project_id": str(project_id),
         "route_source": prepared["route_source"],
+        # 지문(342-3) — 노선 · 기본 관 자리가 지금과 다르면 이 결과는 낡음(세부 유역 안 나눔).
+        "route_signature": route_signature(prepared["vertices"]),
+        "pipe_signature": pipe_signature([pipe.chainage_m for pipe in preview.pipes]),
         "radius_m": region.radius_m,
         # 채택된 상류 세류망 = 1차 영역의 기준선.
         "upstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.upstream],

B05_Profile/B05_Profile_UI_Drainage_Panel.ts

@@ -387,7 +387,7 @@ export function createDrainagePanel(callbacks: DrainagePanelCallbacks = {}): Dra
       if (seq !== requestSeq) return; // 뒤에 보낸 요청이 있음 — 그 응답을 기다림
       apply(response);
       status.hidden = basins.length > 0;
-      if (basins.length === 0) status.textContent = L("B05_Drainage_Status_NoBasin");
+      if (!basins.length) status.textContent = response.stale ?? L("B05_Drainage_Status_NoBasin");
       scheduleDraw();
     } catch (error) {
       if (seq !== requestSeq) return;

common_util/common_util_drainage_pipes.py

@@ -45,6 +45,7 @@ from config.config_system import (
     DRAINAGE_DETAIL_FILENAME,
     DRAINAGE_EDITS_DIRNAME,
     DRAINAGE_PIPE_POINTS_FILENAME,
+    DRAINAGE_RESPONSE_FILENAME,
 )
 
 logger = logging.getLogger(__name__)
@@ -165,6 +166,38 @@ def route_signature(vertices: list[RouteVertex]) -> str:
     return f"{len(vertices)}-{digest.hexdigest()[:16]}"
 
 
+def pipe_signature(chainages: list[float]) -> str:
+    """관 자리 지문 — 누가거리를 0.01 m 로 끊어 차례대로 해시한다(342-3)."""
+    digest = hashlib.sha1(usedforsecurity=False)
+    for chainage in sorted(chainages):
+        digest.update(f"{chainage:.2f};".encode())
+    return f"{len(chainages)}-{digest.hexdigest()[:16]}"
+
+
+# 배수유역 분석 저장분의 노선 지문 — 파일 자국(수정시각 · 크기)이 같으면 다시 읽지 않는다.
+_ANALYSIS_SIGNATURE: dict[str, tuple[tuple[float, int], str | None]] = {}
+
+
+def analysis_route_signature(stored_path: str) -> str | None:
+    """배수유역 분석(`00_watershed_response.json`)의 노선 지문 — 없으면 None(옛 저장분 = 낡음)."""
+    path = drainage_dir(stored_path) / DRAINAGE_RESPONSE_FILENAME
+    try:
+        info = path.stat()
+    except OSError:
+        return None
+    stamp = (info.st_mtime, info.st_size)
+    cached = _ANALYSIS_SIGNATURE.get(str(path))
+    if cached is not None and cached[0] == stamp:
+        return cached[1]
+    try:
+        signature = json.loads(path.read_text(encoding="utf-8")).get("route_signature")
+    except (OSError, ValueError):
+        signature = None
+    _ANALYSIS_SIGNATURE.clear()
+    _ANALYSIS_SIGNATURE[str(path)] = (stamp, signature)
+    return signature
+
+
 def fill_pipe_coordinates(points: list[PipePoint], vertices: list[RouteVertex]) -> list[PipePoint]:
     """좌표가 비어 있는 관에 그 누가거리의 노선 좌표를 채운다(제자리 수정)."""
     if not vertices:

resources/tester/test_342_3_drainage_stale.py

@@ -0,0 +1,91 @@
+"""342-3 배수유역 지문 · 낡음 — 분석 노선 지문이 지금 노선과 다르면 세부 유역을 안 나눔.
+
+- 지문: 관 자리 지문은 차례와 무관 · 분석 저장분 지문은 파일이 바뀔 때만 다시 읽음 · 없으면 None.
+- 낡음: 관 목록이 있으면 관 자리만(세부 유역 없음 · 시설 속성 그대로 · 까닭 글) · 없으면 409.
+- 같으면 그대로 나눔(_build).
+"""
+
+import asyncio
+import json
+import os
+from pathlib import Path
+from types import SimpleNamespace
+from uuid import uuid4
+
+from fastapi.responses import JSONResponse
+
+from B04_PreProcess import B04_PreProcess_Router_Basins as basins
+from common_util import common_util_drainage_pipes as pipes_mod
+from common_util.common_util_drainage_context import DrainageContext
+from common_util.common_util_drainage_pipes import PipePoint, pipe_signature, route_signature
+from common_util.common_util_route_geometry import RouteVertex
+
+VERTICES = [RouteVertex(0.0, 0.0, 100.0, 0.0), RouteVertex(300.0, 0.0, 90.0, 300.0)]
+
+
+def _context() -> DrainageContext:
+    return DrainageContext(stored_path="x", project_root=Path("."), vertices=list(VERTICES))
+
+
+def test_pipe_signature_ignores_order_and_noise():
+    assert pipe_signature([10.0, 120.0]) == pipe_signature([120.0, 10.001])
+    assert pipe_signature([10.0, 120.0]) != pipe_signature([10.0, 121.0])
+
+
+def test_analysis_signature_reads_saved_and_follows_file(tmp_path, monkeypatch):
+    monkeypatch.setattr(pipes_mod, "drainage_dir", lambda stored: tmp_path)
+    assert pipes_mod.analysis_route_signature("x") is None  # 저장분 없음
+    path = tmp_path / "00_watershed_response.json"
+    path.write_text(json.dumps({"status": "success"}), "utf-8")
+    assert pipes_mod.analysis_route_signature("x") is None  # 옛 저장분 = 지문 없음
+    path.write_text(json.dumps({"route_signature": "2-abc"}), "utf-8")
+    os.utime(path, (1, 1))
+    assert pipes_mod.analysis_route_signature("x") == "2-abc"
+
+
+def _patch(monkeypatch, analysis_signature, stored):
+    async def load(project_id):
+        return _context(), ""
+
+    monkeypatch.setattr(basins, "load_drainage_context", load)
+    monkeypatch.setattr(basins, "analysis_route_signature", lambda stored_path: analysis_signature)
+    monkeypatch.setattr(basins, "load_pipe_points", lambda *args: stored)
+    built = SimpleNamespace(called=False)
+
+    def build(stored_path, context, points):
+        built.called = True
+        return SimpleNamespace(pipes=[], basins=["유역"]), points or []
+
+    monkeypatch.setattr(basins, "_build", build)
+    return built
+
+
+def test_stale_keeps_pipes_without_basins(monkeypatch):
+    stored = [
+        PipePoint(chainage_m=120.0, source="stream", facility="ford_bridge"),
+        PipePoint(chainage_m=40.0, source="user"),
+    ]
+    built = _patch(monkeypatch, None, stored)  # 옛 저장분
+    context, detail, points, saved, stale = asyncio.run(
+        basins._resolve(uuid4(), None, use_stored=True)
+    )
+    assert stale and saved and not built.called
+    assert detail.basins == []
+    assert [point.chainage_m for point in points] == [40.0, 120.0]
+    assert points[1].facility == "ford_bridge"
+    payload = basins._payload(uuid4(), context, detail, points, saved, stale)
+    assert payload["stale"] == basins.STALE_MESSAGE
+    assert payload["route_signature"] == route_signature(VERTICES)
+    assert payload["pipe_signature"] == pipe_signature([40.0, 120.0])
+
+
+def test_stale_without_pipes_is_409(monkeypatch):
+    _patch(monkeypatch, "다른-노선", None)
+    result = asyncio.run(basins._resolve(uuid4(), None, use_stored=True))
+    assert isinstance(result, JSONResponse) and result.status_code == 409
+
+
+def test_same_signature_splits(monkeypatch):
+    built = _patch(monkeypatch, route_signature(VERTICES), None)
+    *_, stale = asyncio.run(basins._resolve(uuid4(), None, use_stored=True))
+    assert built.called and not stale