diff --git a/E01_StabRegister/E01_StabRegister_Router.py b/E01_StabRegister/E01_StabRegister_Router.py index a8f553d30..c356b8d75 100644 --- a/E01_StabRegister/E01_StabRegister_Router.py +++ b/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]: diff --git a/E01_StabRegister/E01_StabRegister_Source.py b/E01_StabRegister/E01_StabRegister_Source.py index ba4895d65..c8e5d9261 100644 --- a/E01_StabRegister/E01_StabRegister_Source.py +++ b/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}") diff --git a/common_util/common_util_project_copy.py b/common_util/common_util_project_copy.py index 12783a998..4cfe2e059 100644 --- a/common_util/common_util_project_copy.py +++ b/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, diff --git a/resources/tester/test_532_1_source_robust.py b/resources/tester/test_532_1_source_robust.py new file mode 100644 index 000000000..1e238ba99 --- /dev/null +++ b/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]