fix(532-1): E01 복사본 읽기 — 횡단 경로 null · 파일 없음 · 옛 꼴도 B06 이름 규칙으로 찾음 · 없으면 그 측점만 건너뜀 · 생성 복사 경고(요구 351)

- Source.stations · cross_section: 경로 칸 null · 파일 없음 · 절대 경로 · 옛 정수 m 이름이면 find_cross_file 로 찾음 · 그래도 없으면 「횡단 없음」(stations 에서 빠짐 · cross_section KeyError) · missing_stations 추가
- copy_stability: 같은 검사 → manifest missing_cross_chainages · warnings · E01 생성 응답 warnings
- 실측: 검증 프로젝트 14 복사 모두 stations 성공 · 검증_브레인232 null 1 행(359 m) 살림

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

E01_StabRegister/E01_StabRegister_Router.py

@@ -72,8 +72,8 @@ async def post_stab_project(
     request: Request,
     payload: CreateStabRequest,
     session: dict[str, Any] = Depends(require_company),
-) -> dict[str, str]:
-    """안정성평가 프로젝트 생성 — 원본 B06 미완료면 400."""
+) -> dict[str, Any]:
+    """안정성평가 프로젝트 생성 — 원본 B06 미완료면 400 · 횡단 없는 측점은 warnings."""
     company_id = session.get("company_id")
     rows = await _sources(company_id, payload.source_project_id) if company_id else []
     if not rows:
@@ -102,7 +102,11 @@ async def post_stab_project(
         logger.exception("E01 복사 실패 — 만든 프로젝트를 지움: %s", project_id)
         await hard_delete_project(project_id, int(session["user_id"]), request)
         raise HTTPException(status_code=500, detail="원본 자료 복사에 실패했습니다.") from None
-    return {"project_id": project_id, "source_fingerprint": manifest["source_fingerprint"]}
+    return {
+        "project_id": project_id,
+        "source_fingerprint": manifest["source_fingerprint"],
+        "warnings": manifest.get("warnings", []),
+    }
 
 
 async def _copy_source(project_id: str, source_id: str) -> dict[str, Any]:

E01_StabRegister/E01_StabRegister_Source.py

@@ -18,6 +18,7 @@ from common_util.common_util_project_copy import (
     MANIFEST_NAME,
     STABILITY_DB_NAME,
     STABILITY_SOURCE_DIR,
+    resolve_cross_file,
 )
 from common_util.common_util_storage import resolve_stored_project_path
 from config.config_db import get_db_pool
@@ -69,16 +70,25 @@ def db_snapshot(root: Path) -> dict[str, Any]:
 
 
 def _cross_rows(root: Path) -> list[dict[str, Any]]:
-    rows = db_snapshot(root)["cross_sections"]
-    return sorted(rows, key=lambda row: (float(row["chainage_m"] or 0.0), row["sequence_num"] or 0))
+    rows = [row for row in db_snapshot(root)["cross_sections"] if row["chainage_m"] is not None]
+    return sorted(rows, key=lambda row: (float(row["chainage_m"]), row["sequence_num"] or 0))
+
+
+def _row_data(row: dict[str, Any]) -> dict[str, Any]:
+    data = row.get("data")
+    return (json.loads(data) if isinstance(data, str) else data) or {}
 
 
 def stations(root: Path) -> list[dict[str, Any]]:
-    """측점 목록(누가 차례) — chainage_m · label · 중심 좌표 · 방위각 · 오르막 쪽 · 지반 유형."""
+    """측점 목록(누가 차례) — chainage_m · label · 중심 좌표 · 방위각 · 오르막 쪽 · 지반 유형.
+
+    횡단 파일을 못 찾은 측점(`missing_stations`)은 「횡단 없음」으로 빠짐 — 전체는 안 깨짐."""
     out = []
     for row in _cross_rows(root):
-        data = json.loads(row["data"]) if isinstance(row["data"], str) else row["data"]
-        cross = _json(root / row["cross_section_file_path"])
+        path = resolve_cross_file(root, row)
+        if path is None:
+            continue
+        cross = _json(path)
         out.append(
             {
                 "chainage_m": float(row["chainage_m"]),
@@ -89,27 +99,39 @@ def stations(root: Path) -> list[dict[str, Any]]:
                 "center_z": cross.get("center_z"),
                 "azimuth_deg": cross.get("azimuth_deg"),
                 "uphill_side": cross.get("uphill_side"),
-                "ground_type": (data.get("design") or {}).get("ground_type"),
+                "ground_type": (_row_data(row).get("design") or {}).get("ground_type"),
             }
         )
     return out
 
 
+def missing_stations(root: Path) -> list[float]:
+    """횡단 없음으로 빠진 측점 누가거리 — 화면 알림용(개수 = len)."""
+    return [
+        float(row["chainage_m"])
+        for row in _cross_rows(root)
+        if resolve_cross_file(root, row) is None
+    ]
+
+
 def cross_section(root: Path, chainage_m: float) -> dict[str, Any]:
     """한 측점 횡단 — 지반선(samples: offset_m · x · y · z · valid) · 좌표틀 · B06 설계값
     (`design`: design_line · road_edges · section_mode · 절성토 경사 · 소단 · ground_type ·
-    rock_boundary_offset_m(암 경계 = 지반선을 그만큼 내린 선 · 암 측점만) …)."""
+    rock_boundary_offset_m(암 경계 = 지반선을 그만큼 내린 선 · 암 측점만) …).
+    측점이 없거나 횡단 없음이면 KeyError."""
     for row in _cross_rows(root):
         if abs(float(row["chainage_m"]) - float(chainage_m)) < 1e-6:
-            data = json.loads(row["data"]) if isinstance(row["data"], str) else row["data"]
-            cross = _json(root / row["cross_section_file_path"])
+            path = resolve_cross_file(root, row)
+            if path is None:
+                raise KeyError(f"횡단 없음: {chainage_m}")
+            cross = _json(path)
             return {
                 "chainage_m": float(row["chainage_m"]),
                 "label": cross.get("label"),
                 "frame": cross.get("frame"),
                 "uphill_side": cross.get("uphill_side"),
                 "samples": cross.get("samples") or [],
-                "design": data.get("design") or {},
+                "design": _row_data(row).get("design") or {},
             }
     raise KeyError(f"측점 없음: {chainage_m}")
 

common_util/common_util_project_copy.py

@@ -27,6 +27,7 @@ from uuid import uuid4
 
 import aiomysql
 
+from B05_Profile.B05_Profile_Engine_Sections_Names import find_cross_file
 from common_util.common_util_storage import resolve_stored_project_path
 from common_util.common_util_surface_confirmation import get_surface_confirmation_params
 
@@ -42,6 +43,19 @@ _STABILITY_TREES = (
     "B06_Section/cross_sections",
     "B04_PreProcess/drainage/edits",
 )
+_CROSS_DIR = "B06_Section/cross_sections"
+
+
+def resolve_cross_file(root: Path, row: dict[str, Any]) -> Path | None:
+    """횡단 행 → root 아래 횡단 파일 — 경로 칸이 없거나(null) · 파일이 없거나 · 절대 경로(옛 꼴)면
+    B06 이름 규칙(`find_cross_file` · 옛 정수 m 이름 포함)으로 찾음 · 그래도 없으면 None."""
+    relative = row.get("cross_section_file_path")
+    if relative and not Path(relative).is_absolute() and (root / relative).is_file():
+        return root / relative
+    if row.get("chainage_m") is None:
+        return None
+    found = find_cross_file(root / _CROSS_DIR, float(row["chainage_m"]))
+    return found if found.is_file() else None
 
 
 async def _rows(cursor, sql, args):
@@ -424,6 +438,11 @@ async def copy_stability(connection, source_id: str, new_id: str, project_root:
     digest.update(STABILITY_DB_NAME.encode("utf-8") + b"\0" + db_text.encode("utf-8"))
     (target / STABILITY_DB_NAME).write_text(db_text, encoding="utf-8")
     sizes[STABILITY_DB_NAME] = len(db_text.encode("utf-8"))
+    missing = sorted(
+        float(row["chainage_m"] or 0.0)
+        for row in dump["cross_sections"]
+        if resolve_cross_file(target, row) is None
+    )
     manifest = {
         "scope": "stability",
         "source_project_id": source_id,
@@ -437,6 +456,9 @@ async def copy_stability(connection, source_id: str, new_id: str, project_root:
             "longitudinal_sections": len(dump["longitudinal_sections"]),
         },
         "uuid_replaced": replaced,
+        #: 횡단 파일을 못 찾은 측점(누가거리) — E 는 그 측점을 「횡단 없음」으로 건너뜀
+        "missing_cross_chainages": missing,
+        "warnings": [f"횡단 없는 측점 {len(missing)} 개 — 해석에서 빠짐"] if missing else [],
         "total_bytes": sum(sizes.values()),
         "elapsed_s": round(time.perf_counter() - started, 2),
         "files": sizes,

resources/tester/test_532_1_source_robust.py

@@ -0,0 +1,121 @@
+"""532-1 덧 · 요구 351 — E01 복사본 읽기가 횡단 경로 null · 파일 없음 · 옛 꼴에서도 돎.
+
+- 경로 null · 경로 파일 없음 → B06 이름 규칙(`find_cross_file`)으로 찾음
+- 옛 꼴 — 절대 경로(원본 쪽을 읽지 않음) · 소수 측점의 옛 정수 m 이름 · data 없음
+- 어디에도 없으면 그 측점만 빠짐(`missing_stations`) · cross_section 은 KeyError
+- 생성 복사(`copy_stability`)가 같은 검사 → manifest `missing_cross_chainages` · `warnings`
+"""
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from common_util import common_util_project_copy as cp  # noqa: E402
+from E01_StabRegister import E01_StabRegister_Source as src  # noqa: E402
+
+CROSS = "B06_Section/cross_sections"
+
+
+def _cross(root: Path, name: str, label: str) -> None:
+    path = root / CROSS / name
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps({"label": label, "samples": [{"offset_m": 0}]}), encoding="utf-8")
+
+
+def _row(chainage, path, data=..., seq=0):
+    design = {"design": {"ground_type": "soil"}}
+    return {
+        "chainage_m": chainage,
+        "sequence_num": seq,
+        "cross_section_file_path": path,
+        "data": json.dumps(design) if data is ... else data,
+    }
+
+
+@pytest.fixture
+def root(tmp_path):
+    outside = tmp_path / "original" / CROSS / "cross_00040m.json"
+    outside.parent.mkdir(parents=True)
+    outside.write_text(json.dumps({"label": "원본"}), encoding="utf-8")
+    copy = tmp_path / "copy"
+    _cross(copy, "cross_00000m.json", "정상")
+    _cross(copy, "cross_00020m.json", "null")
+    _cross(copy, "cross_00030m.json", "경로 틀림")
+    _cross(copy, "cross_00040m.json", "절대")
+    _cross(copy, "cross_00050m.json", "옛 정수")  # 49.8 m 측점의 옛 이름
+    rows = [
+        _row(0.0, f"{CROSS}/cross_00000m.json"),
+        _row(20.0, None, seq=None),
+        _row(30.0, f"{CROSS}/cross_gone.json"),
+        _row(40.0, str(outside)),
+        _row(49.8, None, data=None),
+        _row(60.0, None),  # 어디에도 없음
+        _row(None, None),  # 누가거리 없는 행 — 측점 아님
+    ]
+    (copy / src.STABILITY_DB_NAME).write_text(
+        json.dumps({"cross_sections": rows}), encoding="utf-8"
+    )
+    (copy / src.MANIFEST_NAME).write_text("{}", encoding="utf-8")
+    return copy
+
+
+def test_stations_skip_only_missing(root):
+    got = src.stations(root)
+    assert [s["chainage_m"] for s in got] == [0.0, 20.0, 30.0, 40.0, 49.8]
+    assert [s["label"] for s in got] == ["정상", "null", "경로 틀림", "절대", "옛 정수"]
+    assert got[0]["ground_type"] == "soil" and got[4]["ground_type"] is None
+    assert src.missing_stations(root) == [60.0]
+
+
+def test_cross_section(root):
+    assert src.cross_section(root, 20.0)["label"] == "null"
+    assert src.cross_section(root, 40.0)["label"] == "절대"  # 원본 쪽 파일 아님
+    assert src.cross_section(root, 49.8)["design"] == {}
+    with pytest.raises(KeyError, match="횡단 없음"):
+        src.cross_section(root, 60.0)
+    with pytest.raises(KeyError, match="측점 없음"):
+        src.cross_section(root, 70.0)
+
+
+def test_normal_copy_has_no_missing(root):
+    db = json.loads((root / src.STABILITY_DB_NAME).read_text(encoding="utf-8"))
+    db["cross_sections"] = db["cross_sections"][:1]
+    (root / src.STABILITY_DB_NAME).write_text(json.dumps(db), encoding="utf-8")
+    assert [s["label"] for s in src.stations(root)] == ["정상"]
+    assert src.missing_stations(root) == []
+
+
+def test_copy_stability_warns(tmp_path, monkeypatch):
+    source = tmp_path / "source"
+    _cross(source, "cross_00000m.json", "a")
+    _cross(source, "cross_00020m.json", "b")
+    dump = {
+        "project": {"storage_path": "x"},
+        "surface_params": {"source_filter": "none"},
+        "route": None,
+        "cross_sections": [
+            _row(0.0, None),
+            _row(20.0, f"{CROSS}/cross_00020m.json"),
+            _row(35.5, None),
+        ],
+        "route_points": [],
+        "longitudinal_sections": [],
+    }
+
+    async def fake_rows(connection, source_id):
+        return dump
+
+    monkeypatch.setattr(cp, "_stability_rows", fake_rows)
+    monkeypatch.setattr(cp, "resolve_stored_project_path", lambda _: str(source))
+    manifest = asyncio.run(cp.copy_stability(None, "s" * 36, "e" * 36, tmp_path / "e"))
+    assert manifest["missing_cross_chainages"] == [35.5]
+    assert manifest["warnings"] == ["횡단 없는 측점 1 개 — 해석에서 빠짐"]
+    written = json.loads(
+        (tmp_path / "e" / cp.STABILITY_SOURCE_DIR / cp.MANIFEST_NAME).read_text("utf-8")
+    )
+    assert written["missing_cross_chainages"] == [35.5]