feat(b08): 계곡 통과 시설도 원단위 전개로 — 관 유입·유출부 기슭막이·집수정·독립 기슭막이·물넘이 (A1)

- 전개가 structures.json 만 읽던 구멍 — pipe_points.json 도 읽음(관 자체는 관 줄이 셈)
- 관 부설(품셈 12-11 m당)에 기슭막이 몫 없음 → 기슭막이는 전개 한 곳에서만 · 사유에 적음
- 안 적힌 벽 칸은 등록부 기본값(횡단도 그림과 같은 값) + 「기본값으로 섰음」 사유
- 구조물 목록에 남은 관 지점 종류 옛 저장분은 안 셈(두 번 방지)
- 기슭막이 공종 = 형태로 돌쌓기 찰·메 코드 · 독립 기슭막이 한쪽 설치에 두 칸이 다르면 사유만
- 산출식 없는 세월교가 「터파기 줄 없음」 거짓 사유를 안 내게 · 구조물도 제원 저장이 관 지점 시설엔 안 먹힘을 알림

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016VBGFXB9AbJBwXP19z75Qq

B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py

@@ -378,6 +378,11 @@ def _structure_rows(
                     "큰돌쌓기 쌓기 방식이 아직 입력되지 않았습니다 — 구조물 상세 입력에서 "
                     "메쌓기·찰쌓기 중 하나를 고르면 공종이 정해집니다"
                 )
+        if entry.get("class_from") == "form":
+            # 기슭막이 — **형태**가 공종을 가름(돌쌓기 찰·메). 그 밖 형태는 전개 사유가 막음.
+            form = str((structure.get("options") or {}).get("form") or "").strip()
+            code = (entry.get("form_codes") or {}).get(form)
+            class_basis = f"형태 「{form}」 → {code}" if code else ""
         if code and entry.get("class_from") == "back_length":
             class_key, class_basis = masonry_class(structure.get("options") or {})
 

B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py

@@ -243,7 +243,8 @@ def structure_earthwork_rows(
                 backfill += amount
             elif name == "잔토처리":
                 spoil += amount
-        if not any(
+        # 성분이 아예 없는 구조물(세월교 등 산출식 없음)은 제 줄이 이미 사유로 막힘 — 여기 안 적음.
+        if structure.get("components") and not any(
             str(component.get("name") or "") == "터파기"
             and float(component.get("amount") or 0.0) > 0
             for component in structure.get("components") or []

B08_Quantity/B08_Quantity_Engine_Pipe.py

@@ -16,9 +16,9 @@
   기슭막이가 같은 파일에 있다. `pipe_diameter_mm` 유무로 가르면 **관경 미지정 관을 놓친다**
   (실측: `5601e828` 11점 중 2점이 `facility: ford_bridge`).
 
-⚠ **유출·유입부 기슭막이는 여기서 세지 않는다.**
-  관 옵션(`outlet_revet_*`)이 정본이고 구조물 목록에서는 빠졌다(2026-08-28 이관).
-  구조물 쪽으로 또 세면 이중계상이다.
+⚠ **유출·유입부 기슭막이는 관 줄에 넣지 않는다 — `facility_structures` 가 원단위 전개로 한 번 셈.**
+  관 옵션(`outlet_revet_*`)이 정본이다(2026-08-28 이관). 관 부설(품셈 12-11 m당: 관·기초콘크리트·
+  거푸집)에 기슭막이 몫이 없어 겹치지 않는다. 2026-09-14 까지는 **어느 쪽도 안 세고 있었다**(A1).
 
 ⚠ **터파기·되메우기를 관 줄에 붙이지 않는다.**
   관 부설과 굴착이 각각 오면 **같은 굴착을 두 번** 센다(B09 ㉡ 가드와 같은 자리).
@@ -192,3 +192,136 @@ def build_rows(
         "ready_count": sum(1 for row in rows if row["in_bill"]),
         "length_total_m": round(sum(row["quantity"] for row in rows if row["in_bill"]), 3),
     }
+
+
+INLET_BASIN = "집수정"
+#: 기슭막이 벽 칸 — 관·독립 기슭막이 모두 `{쪽}_revet_{칸}` · 독립의 옛 저장분은 `{칸}`(B06 `_revet_side`).
+REVET_FIELDS = (
+    ("form", "형태"),
+    ("height_m", "높이"),
+    ("length_m", "길이"),
+    ("before_m", ""),
+    ("after_m", ""),
+)
+#: (저장 채널, 관에 딸린 줄 이름 꼬리, 독립 기슭막이 줄 이름 꼬리)
+REVET_ROLES = (
+    ("inlet", "유입부 기슭막이", "유입 칸 벽"),
+    ("outlet", "유출부 기슭막이", "유출 칸 벽"),
+)
+NOTE_DEFAULT_WALL = (
+    "⚠ 시설 지점에 {filled} 을 안 적어 등록부 기본값으로 섰음 — 횡단도 그림과 같은 값"
+)
+NOTE_SIDE_UNKNOWN = (
+    "⚠ 설치 측 「{side}」인데 두 칸(유입·유출) 값이 달라 어느 칸이 그쪽인지 서버가 못 가림"
+    "(횡단 지형이 정함) — 값을 안 세움 · 두 칸을 같게 적으면 섬"
+)
+
+
+def _blank(value: Any) -> bool:
+    return value is None or value == ""
+
+
+def _num_or_zero(value: Any) -> float:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def _revet_values(
+    options: dict[str, Any], role: str, defaults: dict[str, Any], legacy: bool
+) -> tuple[dict[str, Any], list[str]]:
+    """벽 한쪽 제원과 등록부 기본값으로 채운 칸 — B06 횡단 그림(`_side_spec`)과 같은 채움."""
+    values: dict[str, Any] = {}
+    filled: list[str] = []
+    for name, label in REVET_FIELDS:
+        value = options.get(f"{role}_revet_{name}")
+        if _blank(value) and legacy:
+            value = options.get(name)
+        if _blank(value):
+            value = defaults.get(name if legacy else f"{role}_revet_{name}")
+            if label and value is not None:
+                filled.append(f"{label} {value}")
+        values[name] = value
+    return values, filled
+
+
+def facility_structures(points: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """계곡 통과 시설(`pipe_points.json`) → 원단위 전개가 읽는 구조물 줄 (A1, 2026-09-14).
+
+    ⚠ 관 자체는 안 냄 — `build_rows` 가 관 연장으로 셈.
+    ⚠ 안 적힌 벽 칸은 **등록부 기본값** — 횡단 그림이 쓰는 값과 같고, 그 사실을 줄 사유로 붙임.
+    ⚠ 집수정·기슭막이 터파기·되메우기는 전개 성분(`destination: earthwork`)이라 토공집계로만 감.
+    """
+    from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
+    from B08_Quantity.B08_Quantity_Engine_StructureSummary import pipe_row_id
+    from B08_Quantity.B08_Quantity_Engine_UnitQuantity import attachments_of, child_row
+
+    types = structure_type_map()
+
+    def defaults(type_id: str) -> dict[str, Any]:
+        found = types.get(type_id)
+        return {option.key: option.default for option in found.options} if found else {}
+
+    rows: list[dict[str, Any]] = []
+    for point in points or []:
+        facility = str(point.get("facility") or FACILITY_PIPE)
+        chainage = float(point.get("chainage_m") or 0.0)
+        options = dict(point.get("options") or {})
+        base = {
+            "structure_id": pipe_row_id(point),
+            "type_id": facility,
+            "chainage_m": chainage,
+            "start_m": point.get("start_m") if point.get("start_m") is not None else chainage,
+            "end_m": point.get("end_m") if point.get("end_m") is not None else chainage,
+            "options": options,
+        }
+        if facility not in (FACILITY_PIPE, "revetment"):
+            rows.append(base)  # BOX암거·물넘이·세월교 — 전개식·관측값이 없으면 사유로 섬
+            continue
+        legacy = facility == "revetment"  # 독립 기슭막이 — 벽 칸 밖의 제원(뒷길이 등)도 벽이 씀
+        wall_defaults = defaults(facility)
+        roles = list(REVET_ROLES)
+        inlet_notes: list[str] = []
+        if not legacy:
+            parent = {**base, "type_id": "pipe"}
+            children = [r for r in attachments_of(parent) if r["type_id"] != "pipe_inlet_basin"]
+            inlet = str(options.get("inlet_type") or wall_defaults.get("inlet_type") or "")
+            if inlet == INLET_BASIN:
+                # 형식을 안 골랐어도 줄은 세움 — 관측표가 「형식을 골라야 섬」 사유를 냄.
+                children.append(child_row(parent, "pipe_inlet_basin", "유입부 집수정"))
+                roles = roles[1:]
+            elif options.get("inlet_basin_form"):
+                inlet_notes.append(
+                    f"⚠ 집수정 형식({options['inlet_basin_form']})이 적혀 있으나 유입구 구조가 "
+                    f"「{inlet}」라 집수정은 안 셈 — 유입구를 「집수정」으로 바꾸면 섬"
+                )
+            rows.extend(children)
+        walls = []
+        for role, pipe_label, own_label in roles:
+            values, filled = _revet_values(options, role, wall_defaults, legacy)
+            notes = list(inlet_notes) if role == "inlet" else []
+            if filled:
+                notes.append(NOTE_DEFAULT_WALL.format(filled=" · ".join(filled)))
+            walls.append([role, own_label if legacy else pipe_label, values, notes, False])
+        side = str(options.get("side") or "") if legacy else ""
+        if side in ("좌", "우"):
+            # 좌·우가 유입/유출 어느 칸인지는 **횡단 지형**이 정함 — 두 칸이 같을 때만 한 벽으로 셈.
+            same = walls[0][2] == walls[1][2]
+            walls = [[walls[0][0], f"{side} 벽", walls[0][2], walls[0][3], not same]]
+            if not same:
+                walls[0][3].append(NOTE_SIDE_UNKNOWN.format(side=side))
+        foundation = options.get("foundation" if legacy else "revet_foundation")
+        kept = {k: v for k, v in options.items() if not k.startswith(("inlet_", "outlet_"))}
+        for role, label, values, notes, withheld in walls:
+            before, after = _num_or_zero(values["before_m"]), _num_or_zero(values["after_m"])
+            row = child_row(base, "revetment", label, f"{role}_revet")
+            row.update(
+                start_m=chainage - before,
+                end_m=chainage + after,
+                options={**(kept if legacy else {}), **values, "foundation": foundation},
+                notes=notes,
+                withheld=withheld,
+            )
+            rows.append(row)
+    return rows

B08_Quantity/B08_Quantity_Engine_UnitQuantity.py

@@ -198,24 +198,26 @@ def wing_wall_double_count(options: dict[str, Any]) -> str | None:
     return WING_WALL_DOUBLE_COUNT if options.get("inlet_basin_form") else None
 
 
+def child_row(structure: dict[str, Any], type_id: str, label: str, suffix: str = "") -> dict:
+    """딸린 줄 한 벌 — 이름에 「어디에 딸렸나」가 남음(`expand` 가 부모 이름 · 꼬리로 붙임)."""
+    return {
+        **structure,
+        "structure_id": f"{structure.get('structure_id')}-{suffix or type_id}",
+        "type_id": type_id,
+        "attachment_of": structure.get("structure_id"),
+        "attachment_parent_type": structure.get("type_id"),
+        "attachment_label": label,
+    }
+
+
 def attachments_of(structure: dict[str, Any]) -> list[dict[str, Any]]:
     """구조물에 딸린 **별도 줄**을 만든다. 제원은 원본을 그대로 물려준다(치수 두 벌 금지)."""
-    rows: list[dict[str, Any]] = []
     options = structure.get("options") or {}
-    for type_id, gate_key, label in ATTACHMENTS.get(str(structure.get("type_id") or ""), ()):
-        if not options.get(gate_key):
-            continue  # 그 부속이 없는 배치다 — 빈 줄을 만들지 않는다
-        rows.append(
-            {
-                **structure,
-                "structure_id": f"{structure.get('structure_id')}-{type_id}",
-                "type_id": type_id,
-                "attachment_of": structure.get("structure_id"),
-                "attachment_parent_type": structure.get("type_id"),
-                "attachment_label": label,
-            }
-        )
-    return rows
+    return [
+        child_row(structure, type_id, label)
+        for type_id, gate_key, label in ATTACHMENTS.get(str(structure.get("type_id") or ""), ())
+        if options.get(gate_key)  # 그 부속이 없는 배치다 — 빈 줄을 만들지 않는다
+    ]
 
 
 def _observed_components(
@@ -288,6 +290,10 @@ def expand(
         end_m=end if structure.get("end_m") is not None else None,
         options=dict(options),
     )
+    # 줄을 만든 쪽이 붙인 사유(관 지점 시설의 기본값 · 못 가른 자리) — `withheld` 면 값을 안 세움.
+    result.notes.extend(str(note) for note in structure.get("notes") or [])
+    if structure.get("withheld"):
+        return result
     withheld = EXPANDER_WITHHELD.get(type_id)
     if withheld:
         result.notes.append(f"전개식 미확보 — {withheld}")

B08_Quantity/B08_Quantity_Router_Material.py

@@ -58,7 +58,17 @@ router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"])
 def _collect_structures(
     project_root: str,
 ) -> tuple[list[dict[str, Any]], dict[str, str], list[str]]:
-    """전개 대상 구조물·타입명·건너뛴 사유를 함께 낸다."""
+    """전개 대상 구조물·타입명·건너뛴 사유를 함께 낸다.
+
+    ⭐ 정본 둘을 다 읽음(A1, 2026-09-14) — `structures.json` + `pipe_points.json`(계곡 통과 시설).
+      앞서 뒤엣것을 안 읽어 집수정·날개벽·관 유입/유출 기슭막이·독립 기슭막이·물넘이포장이
+      원단위·인계·내역에 한 줄도 안 섰음.
+    ⚠ 관 지점 종류(`managed_by`)가 `structures.json` 에 옛 저장분으로 남아 있어도 안 셈 —
+      관 지점 정본이 주인이라 두 번 세지 않음(구조물 집계표와 같은 규칙).
+    """
+    from B08_Quantity.B08_Quantity_Engine_Pipe import facility_structures
+    from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
+
     _revision, items = load_structures(project_root)
     types = structure_type_map()
     targets: list[dict[str, Any]] = []
@@ -72,6 +82,11 @@ def _collect_structures(
             skipped.append(f"{type_id}: 레지스트리에 없는 타입")
             continue
         names[type_id] = definition.name
+        if definition.managed_by:
+            skipped.append(
+                f"{definition.name}: 관 지점 정본이 주인 — 구조물 목록 옛 저장분은 안 셈"
+            )
+            continue
         if definition.design_owner:
             skipped.append(
                 f"{definition.name}: {definition.design_owner} 가 이미 셈 — 중복 계상 방지"
@@ -86,6 +101,12 @@ def _collect_structures(
             #   여기서 빼지 않으면 **같은 시설이 두 줄로** 나간다.
             continue
         targets.append(payload)
+    points = read_pipe_points_file(pipe_points_path_in(Path(project_root)))
+    for row in facility_structures([point.as_dict() for point in points]):
+        for type_id in (row["type_id"], row.get("attachment_parent_type")):
+            if type_id in types:
+                names[type_id] = types[type_id].name
+        targets.append(row)
     return targets, names, sorted(set(skipped))
 
 

B08_Quantity/B08_Quantity_Router_StructureSheet.py

@@ -259,6 +259,13 @@ async def put_structure_sheet_spec(
     try:
         revision, stored = await asyncio.to_thread(load_structures, project_root)
         updated, changed = apply_spec(stored, member_ids, spec)
+        outside = member_ids - {str(item.structure_id) for item in stored}
+        if outside:
+            # 관 지점 시설(A1)은 `pipe_points.json` 이 정본 — 조용히 안 먹히지 않게 알림.
+            notes.append(
+                f"관 지점 시설 {len(outside)}개소(배수관 기슭막이 등)는 이 화면에서 제원을 못 적음"
+                " — B05 배수 시설·구조물 집계표에서 고칠 것"
+            )
         new_revision = await asyncio.to_thread(
             save_structures, project_root, updated, base_revision=payload.base_revision
         )