fix(B05): 크로스체크 5건 반영 — 마이그레이션 문자열 판별·서버 검증·기본값 원칙·404·STALE 정합

외부 AI 교차검증 미통과 지적을 전부 수정한다.

1. 마이그레이션: 확정 저장분은 structure 문자열뿐(structureType 없음) —
   라벨 파싱 판별 추가(기성막이/대피로 X.Xm/관종 D직경). 명시 필드가
   라벨 파싱보다 우선.
2. 서버 검증 강화(_validate_types 확장): 레지스트리 배치형태 대조,
   미정의 옵션 거절, number 옵션 유한·0 이상, select 선택지 검사,
   required 옵션 누락 거절, structure_id 중복 거절, 노선 연장 범위
   검증(라우터가 get_latest_route로 총연장 주입, 없으면 생략).
3. 기본값 원칙: 법정 명시값(별표2 측구 30cm·대피소 5/15m 등)·사용자
   기확정값(골막이)만 default 유지. 옹벽·돌쌓기 높이, 사토장·토취장
   면적/용량, 포장·쇄석 두께 등 미확정 수치는 default 제거 + required
   (화면 placeholder "필수 입력"+빈 값 추가 차단, 서버도 거절).
4. 404: get_project_storage_relative_path는 없는 프로젝트에서
   LookupError를 던짐 — _project_root에서 잡아 None, 라우터 예외
   사다리에도 LookupError→404 분기.
5. STALE 정합: _invalidate_downstream이 성공 여부 반환 —
   invalidated_downstream은 실제 전파 성공 시에만 true.

pytest 42건(tmp/tests) 통과 · tsc 0 · ruff 통과.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-16 22:58:43 +09:00
co-authored by Claude Fable 5
parent 61108f540d
commit 4af0dbdb8a
7 changed files with 1106 additions and 100 deletions
@@ -25,6 +25,8 @@ export interface StructureOptionField {
choices: string[];
unit: string | null;
default: string | number | null;
/** 미확정 항목(기본값 없음) — 사용자가 값을 넣어야 저장된다. */
required?: boolean;
}
export interface StructureType {
File diff suppressed because it is too large Load Diff
@@ -12,6 +12,7 @@
최소 구간을 임시로 주고, 사용자가 화면에서 종점을 조정한다 — 값을 지어내는 것보다 낫다.
"""
import re
from typing import Any, Iterable
from B05_Profile.B05_Profile_Structures_Schema import StructureInstance, structure_type_map
@@ -26,15 +27,50 @@ LEGACY_TYPE_MAP = {
# 종점을 모르는 구간형 항목에 주는 임시 길이(m). 화면에서 조정하라는 표시값이다.
DEFAULT_INTERVAL_LENGTH_M = 15.0
# 확정 저장분은 표시 라벨 문자열뿐이다(예: "파형강관 D800", "대피로 2.0m").
# 라벨 생성 규칙은 B05_Profile_UI_IrregularStations.ts의 structureLabel()이 정본.
_PIPE_LABEL = re.compile(r"^(배관|.*관)\s+D\d+", re.UNICODE)
_ESCAPE_LABEL = re.compile(r"^대피로(?:\s+([\d.]+)\s*m)?", re.UNICODE)
def _is_pipe(entry: dict[str, Any]) -> bool:
label = str(entry.get("structure", "")).strip()
return (
entry.get("origin") == "pipe"
or entry.get("structureType") == PIPE_STRUCTURE_NAME
or str(entry.get("structure", "")).strip() == PIPE_STRUCTURE_NAME
or label == PIPE_STRUCTURE_NAME
or bool(_PIPE_LABEL.match(label))
)
def _legacy_type_of(entry: dict[str, Any]) -> str | None:
"""기존 종류 이름을 알아낸다 — structureType이 없으면 라벨 문자열에서 판별한다.
확정 저장 형식(`{chainage_m, structure}`)에는 structureType이 없다(크로스체크
지적 1, 2026-08-16). 라벨은 structureLabel()이 만든 고정 형식이라 되짚을 수 있다.
"""
explicit = entry.get("structureType")
if explicit in LEGACY_TYPE_MAP:
return str(explicit)
label = str(entry.get("structure", "")).strip()
if label == "기성막이":
return "기성막이"
if _ESCAPE_LABEL.match(label):
return "대피로"
return None
def _escape_width_of(entry: dict[str, Any]) -> float | None:
"""대피로 폭 — 명시 필드가 우선, 없으면 라벨("대피로 2.0m")에서 파싱한다."""
width = entry.get("escapeWidthM")
if width is not None:
return float(width)
match = _ESCAPE_LABEL.match(str(entry.get("structure", "")).strip())
if match and match.group(1):
return float(match.group(1))
return None
def migrate_irregular_stations(entries: Iterable[dict[str, Any]]) -> list[StructureInstance]:
"""기존 비정규 측점 목록을 구조물 인스턴스로 옮긴다(같은 입력이면 같은 결과)."""
types = structure_type_map()
@@ -46,8 +82,8 @@ def migrate_irregular_stations(entries: Iterable[dict[str, Any]]) -> list[Struct
continue
chainage = float(entry.get("chainage_m", 0.0))
legacy_type = entry.get("structureType")
type_id = LEGACY_TYPE_MAP.get(legacy_type, "etc")
legacy_type = _legacy_type_of(entry)
type_id = LEGACY_TYPE_MAP.get(legacy_type or "", "etc")
definition = types[type_id]
key = (type_id, round(chainage, 3))
@@ -75,8 +111,8 @@ def migrate_irregular_stations(entries: Iterable[dict[str, Any]]) -> list[Struct
def _options_for(type_id: str, entry: dict[str, Any]) -> dict[str, Any]:
if type_id == "refuge":
width = entry.get("escapeWidthM")
return {"width_m": float(width)} if width is not None else {}
width = _escape_width_of(entry)
return {"width_m": width} if width is not None else {}
if type_id == "etc":
name = entry.get("customName") or entry.get("structure") or "기타 구조물"
return {"name": str(name).strip()}
@@ -62,10 +62,16 @@ def save_structures(
structures: Iterable[StructureInstance],
*,
base_revision: int,
max_chainage_m: float | None = None,
) -> int:
"""구조물 목록을 정본에 덮어쓰고 새 판번호를 돌려준다."""
"""구조물 목록을 정본에 덮어쓰고 새 판번호를 돌려준다.
`max_chainage_m`는 노선 총연장(m) — 주어지면 범위 밖 배치를 거절한다.
"""
items = list(structures)
_validate_types(items)
_validate_unique_ids(items)
_validate_range(items, max_chainage_m)
current_revision, _ = load_structures(project_root)
if current_revision != base_revision:
@@ -119,7 +125,11 @@ def _design_fingerprint(items: Iterable[StructureInstance]) -> set[str]:
def _validate_types(items: list[StructureInstance]) -> None:
"""레지스트리에 없는 타입, 다른 정본이 관리하는 타입(배관)을 걸러낸다."""
"""레지스트리 대조 검증 — 타입 존재·관리 주체·배치형태·옵션까지 서버가 지킨다.
화면만 믿으면 조작된 요청(placement 불일치·음수 제원·미정의 옵션)이 정본에
들어간다(2026-08-16 크로스체크 지적 2). 정본에 닿는 마지막 관문은 여기다.
"""
types = structure_type_map()
for item in items:
definition = types.get(item.type_id)
@@ -130,3 +140,53 @@ def _validate_types(items: list[StructureInstance]) -> None:
f"{definition.name}은(는) {definition.managed_by} 정본이 관리합니다 — "
"구조물 목록에 저장할 수 없습니다."
)
if item.placement != definition.placement:
raise ValueError(
f"{definition.name}의 배치형태는 {definition.placement}인데 "
f"{item.placement}로 보냈습니다."
)
_validate_options(item, definition)
def _validate_options(item: StructureInstance, definition) -> None:
allowed = {option.key: option for option in definition.options}
for key, value in item.options.items():
option = allowed.get(key)
if option is None:
raise ValueError(f"{definition.name}에 정의되지 않은 옵션입니다: {key}")
if option.input == "number":
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{definition.name}{option.label}은(는) 숫자여야 합니다.")
if not (value == value and abs(value) != float("inf")) or value < 0:
raise ValueError(f"{definition.name}{option.label}은(는) 0 이상이어야 합니다.")
elif option.input == "select" and option.choices and str(value) not in option.choices:
raise ValueError(f"{definition.name}{option.label} 값이 선택지에 없습니다: {value}")
for option in definition.options:
if option.required and item.options.get(option.key) in (None, ""):
raise ValueError(
f"{definition.name}{option.label}은(는) 필수 입력입니다 — "
"미확정 항목이라 기본값이 없습니다."
)
def _validate_unique_ids(items: list[StructureInstance]) -> None:
seen: set[str] = set()
for item in items:
if not item.structure_id:
continue
if item.structure_id in seen:
raise ValueError(f"구조물 식별자가 중복되었습니다: {item.structure_id}")
seen.add(item.structure_id)
def _validate_range(items: list[StructureInstance], max_chainage_m: float | None) -> None:
"""노선 연장을 알 때만 범위를 지킨다 — 연장 밖 배치는 도면·수량 어디에도 못 실린다."""
if max_chainage_m is None:
return
limit = max_chainage_m + 1e-6
for item in items:
positions = [item.chainage_m, item.start_m, item.end_m]
if any(value is not None and value > limit for value in positions):
raise ValueError(
f"구조물 위치가 노선 연장({max_chainage_m:.1f} m)을 벗어났습니다: {item.type_id}"
)
+44 -9
View File
@@ -16,6 +16,7 @@ from fastapi import APIRouter
from fastapi.responses import JSONResponse
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_Profile.B05_Profile_Repository import get_latest_route
from B05_Profile.B05_Profile_Structures_Repository import (
StructureRevisionConflict,
load_structures,
@@ -56,19 +57,41 @@ async def read_structure_types() -> StructureTypesResponse:
async def _project_root(project_id: UUID) -> str | None:
"""프로젝트 저장 경로 — 없는 프로젝트는 None (호출부가 404로 답한다).
`get_project_storage_relative_path`는 없는 프로젝트에서 LookupError를 **던진다**
(None 반환이 아님) — 잡지 않으면 500으로 샌다(2026-08-16 크로스체크 지적 5).
"""
pool = get_db_pool()
async with pool.acquire() as connection:
stored_path = await get_project_storage_relative_path(connection, project_id)
try:
async with pool.acquire() as connection:
stored_path = await get_project_storage_relative_path(connection, project_id)
except LookupError:
return None
if not stored_path:
return None
return str(Path(resolve_stored_project_path(stored_path)))
async def _invalidate_downstream(project_id: UUID) -> None:
async def _route_length(project_id: UUID) -> float | None:
"""최신 노선 총연장(m). 노선이 없거나 조회 실패면 None — 범위 검증만 생략된다."""
try:
pool = get_db_pool()
async with pool.acquire() as connection:
latest = await get_latest_route(connection, project_id)
length = latest.get("total_length_m") if latest else None
return float(length) if length else None
except Exception:
logger.exception("B05 노선 연장 조회 실패: project_id=%s", project_id)
return None
async def _invalidate_downstream(project_id: UUID) -> bool:
"""구조물이 바뀌었으니 B06(stage 3) 이후의 완료 단계를 STALE로 되돌린다.
실패해도 저장은 이미 끝났다 — 무효화를 못 했다고 저장을 되돌리면 정본과 화면이
어긋난다. 로그만 남기고 넘어간다.
어긋난다. 대신 성공 여부를 돌려줘 응답이 사실만 말하게 한다(성공한 척 금지 —
2026-08-16 크로스체크 지적 5).
"""
try:
pool = get_db_pool()
@@ -83,8 +106,10 @@ async def _invalidate_downstream(project_id: UUID) -> None:
(str(project_id), ROUTE_STAGE_NO),
)
await connection.commit()
return True
except Exception:
logger.exception("B05 구조물 변경 후속 단계 무효화 실패: project_id=%s", project_id)
return False
@router.get("/{project_id}/route/structures", response_model=StructureListResponse)
@@ -98,6 +123,9 @@ async def read_structures(project_id: UUID) -> StructureListResponse | JSONRespo
return StructureListResponse(
project_id=str(project_id), revision=revision, structures=structures
)
except LookupError:
# 저장 경로 조회가 예외로 알려온 "프로젝트 없음" — 500이 아니라 404다.
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
except Exception:
logger.exception("B05 구조물 조회 실패: project_id=%s", project_id)
return JSONResponse(
@@ -116,18 +144,25 @@ async def write_structures(
if root is None:
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
_, previous = load_structures(root)
revision = save_structures(root, payload.structures, base_revision=payload.base_revision)
revision = save_structures(
root,
payload.structures,
base_revision=payload.base_revision,
max_chainage_m=await _route_length(project_id),
)
# 설계에 영향을 주는 변경일 때만 B06 이후를 STALE로 돌린다 — 메모만 고쳐도
# 횡단·수량을 다시 돌리게 만들지 않기 위함이다.
invalidated = requires_downstream_invalidation(previous, payload.structures)
if invalidated:
await _invalidate_downstream(project_id)
# 횡단·수량을 다시 돌리게 만들지 않기 위함이다. 응답 플래그는 실제로 STALE
# 전파가 **성공했을 때만** true(실패를 성공처럼 알리지 않는다).
needs_invalidation = requires_downstream_invalidation(previous, payload.structures)
invalidated = needs_invalidation and await _invalidate_downstream(project_id)
return StructureSaveResponse(
project_id=str(project_id),
revision=revision,
count=len(payload.structures),
invalidated_downstream=invalidated,
)
except LookupError:
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
except StructureRevisionConflict as conflict:
return JSONResponse(
status_code=409,
@@ -37,6 +37,9 @@ class StructureOptionField(BaseModel):
choices: list[str] = Field(default_factory=list)
unit: str | None = None
default: Any = None
# 미확정 수치(기본값 없음)는 사용자가 직접 넣어야 저장된다 — 지식DB 원칙:
# 기본값 선정은 사용자 협의 영역, 임의값 자동 저장 금지(2026-08-16 크로스체크 반영).
required: bool = False
class StructureType(BaseModel):
+20 -1
View File
@@ -160,7 +160,13 @@ export function createStructuresSection(callbacks: StructuresCallbacks): Structu
let types: StructureType[] = [];
let structures: StructureInstance[] = [];
let editingId: string | null = null;
let optionInputs: Array<{ key: string; read: () => string | number }> = [];
let optionInputs: Array<{
key: string;
required: boolean;
input: HTMLInputElement | HTMLSelectElement;
read: () => string | number;
isEmpty: () => boolean;
}> = [];
function typeMap(): Map<string, StructureType> {
return new Map(types.map((type) => [type.type_id, type]));
@@ -201,6 +207,8 @@ export function createStructuresSection(callbacks: StructuresCallbacks): Structu
} else if (option.input === "number") {
input = numberInput("0.1", "0");
input.value = String(preset ?? "");
// 미확정 항목(기본값 없음)은 비워 두면 저장이 거절된다 — 칸에서 미리 알린다.
if (option.required) (input as HTMLInputElement).placeholder = "필수 입력";
} else {
input = document.createElement("input");
(input as HTMLInputElement).type = "text";
@@ -210,7 +218,10 @@ export function createStructuresSection(callbacks: StructuresCallbacks): Structu
optionRow.append(field(label, input));
optionInputs.push({
key: option.key,
required: !!option.required,
input,
read: () => (option.input === "number" ? Number(input.value) || 0 : input.value),
isEmpty: () => input.value.trim() === "",
});
});
}
@@ -303,6 +314,8 @@ export function createStructuresSection(callbacks: StructuresCallbacks): Structu
function readOptions(): Record<string, string | number> {
const values: Record<string, string | number> = {};
optionInputs.forEach((input) => {
// 빈 칸은 아예 넣지 않는다 — 빈 숫자를 0으로 저장하면 "0으로 확정"과 구분이 안 된다.
if (input.isEmpty()) return;
values[input.key] = input.read();
});
return values;
@@ -330,6 +343,12 @@ export function createStructuresSection(callbacks: StructuresCallbacks): Structu
return;
}
}
// 필수 옵션(미확정 기본값 없음)이 비어 있으면 추가하지 않는다 — 서버도 거절한다.
const missing = optionInputs.find((entry) => entry.required && entry.isEmpty());
if (missing) {
missing.input.focus();
return;
}
const base = {
type_id: type.type_id,