Merge remote-tracking branch 'origin/dev' into main_laptop_1

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
         )

resources/data_work_item_mapping/work_item_mapping_2026-01-01.json

@@ -220,6 +220,21 @@
       "work_item_code": "FP-12-15",
       "master_name": "집수정"
     },
+    {
+      "type_id": "revetment",
+      "work_item_code": null,
+      "class_from": "form",
+      "form_codes": {
+        "돌쌓기(찰)": "FP-13-04-05",
+        "돌쌓기(메)": "FP-13-04-02"
+      },
+      "class_note": "기슭막이는 **형태가 돌쌓기면 돌쌓기 식**(`_UnitQuantity_Revetment`) — 공종도 돌쌓기 찰·메와 같음(실무 정본 탭 「돌기슭막이(H=2.0m, 찰쌓기, 기초유)」). 그 밖 형태(콘크리트·돌망태·통나무·바자)는 전개식이 없어 사유로 막힘. 관 유입·유출부 기슭막이와 독립 기슭막이가 이 줄로 섬(A1, 2026-09-14).",
+      "secondary_axes": [
+        "stone_kind"
+      ],
+      "billing_component": "돌쌓기",
+      "variant_axis": "back_len_cm"
+    },
     {
       "type_id": "ford_pavement",
       "work_item_code": "FP-12-06",
@@ -414,7 +429,7 @@
     "diameter_option_key": "pipe_diameter_mm",
     "variant_axis": "pipe_diameter_mm",
     "facility_rule": "⚠ `pipe_points.json` 은 **계곡 통과 시설 전부의 정본**이다(배관·BOX암거·물넘이·세월교·독립 기슭막이). `facility` 가 `pipe` 인 점만 배관이다 — `pipe_diameter_mm` 유무로 가르면 **관경 미지정 관을 놓친다**(2026-09-08 랩탑 창).",
-    "revetment_note": "⚠ 유출·유입부 기슭막이는 **관 옵션(`inlet_revet_*`·`outlet_revet_*`)이 정본**이다. 레지스트리 `revetment` 타입이 `managed_by: pipe_points` 라 구조물 목록에서 빠졌으므로 **구조물 쪽으로 또 세지 않는다**(2026-08-28 이관).",
+    "revetment_note": "⚠ 유출·유입부 기슭막이는 **관 옵션(`inlet_revet_*`·`outlet_revet_*`)이 정본**이다. 레지스트리 `revetment` 타입이 `managed_by: pipe_points` 라 구조물 목록에서 빠졌다(2026-08-28 이관). ⭐ **관 줄에는 안 넣고 원단위 전개가 한 번 셈** — B08 이 관 지점을 읽어 「배수관 · 유입부/유출부 기슭막이」 줄로 세움(`Engine_Pipe.facility_structures`, A1 2026-09-14). 관 부설(품셈 12-11 m당: 관·기초콘크리트·거푸집)에 기슭막이 몫이 없어 겹치지 않음. 그전에는 **어느 쪽도 안 세고 있었음**.",
     "not_ready": {
       "흄관 밑수 두 벌": "`FP-12-11-02` 는 밑수가 「1 m」와 「1 개소」 두 벌이다(표가 둘). B09 가 `#갈래` 로 두 표를 각각 세우므로 B08 은 `variant_value` 로 어느 쪽인지 보내면 된다.",
       "터파기·되메우기": "⚠ 관 부설과 터파기·되메우기가 각각 오면 **같은 굴착을 두 번 셀 수 있다**(B09 ㉡ 가드). 관 줄에는 지금 터파기를 붙이지 않는다."

resources/tester/test_b08_facility_structures.py

@@ -0,0 +1,86 @@
+"""계곡 통과 시설(`pipe_points.json`)이 원단위 전개·인계에 선다 (A1, 2026-09-14).
+
+⚠ 앞서 전개가 `structures.json` 만 읽어 관 유입·유출부 기슭막이·집수정·독립 기슭막이가
+  원단위·내역에 한 줄도 안 섰음. 관 줄(품셈 12-11 m당)에는 기슭막이 몫이 없어 **한 번만** 셈.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from B05_Profile.B05_Profile_Structures_Schema import StructureInstance  # noqa: E402
+from B08_Quantity import B08_Quantity_Router_Material as material  # noqa: E402
+from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff  # noqa: E402
+from B08_Quantity.B08_Quantity_Engine_Pipe import build_rows, facility_structures  # noqa: E402
+from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table  # noqa: E402
+
+NAMES = {"pipe": "배수관", "revetment": "기슭막이", "ford_bridge": "세월교"}
+
+
+def _table(points: list[dict]) -> dict:
+    return build_table(facility_structures(points), NAMES)
+
+
+def test_관_하나에_기슭막이_둘이_기본값_사유와_함께_서고_관_줄에는_없다() -> None:
+    point = {"chainage_m": 100.0, "options": {"pipe_diameter_mm": 1000}}
+    table = _table([point])
+    names = [s["name"] for s in table["structures"]]
+    assert names == ["배수관 · 유입부 기슭막이", "배수관 · 유출부 기슭막이"]
+    inlet, outlet = table["structures"]
+    assert inlet["options"]["form"] == "돌쌓기(찰)" and outlet["options"]["form"] == "돌쌓기(메)"
+    assert all(s["components"] and s["length_m"] == 10 for s in table["structures"])
+    assert "등록부 기본값으로 섰음" in inlet["notes"][0]
+    # 관 줄은 관 연장만 — 기슭막이가 거기 섞이지 않음(두 번 안 셈).
+    pipe = build_rows([point], [{"chainage_m": 100.0, "design": {"pipe_length_m": 8}}])
+    assert [row["quantity"] for row in pipe["rows"]] == [8.0]
+
+    work = {row["name"]: row for row in build_handoff(unit_quantity_table=table)["work_items"]}
+    assert work["배수관 · 유입부 기슭막이"]["work_item_code"] == "FP-13-04-05"
+    assert work["배수관 · 유출부 기슭막이"]["work_item_code"] == "FP-13-04-02"
+    assert (
+        work["배수관 · 유입부 기슭막이"]["in_bill"]
+        and work["배수관 · 유입부 기슭막이"]["unit"] == "㎡"
+    )
+
+
+def test_유입구가_집수정이면_집수정_줄이_사유로_서고_유입_기슭막이는_없다() -> None:
+    table = _table([{"chainage_m": 5.0, "options": {"inlet_type": "집수정"}}])
+    basin, outlet = table["structures"]
+    assert basin["type_id"] == "pipe_inlet_basin" and not basin["components"] and basin["notes"]
+    assert outlet["name"] == "배수관 · 유출부 기슭막이"
+
+
+def test_유입구가_기슭막이인데_집수정_형식이_남아_있으면_집수정은_안_세고_알린다() -> None:
+    options = {"inlet_type": "기슭막이", "inlet_basin_form": "□형(기본형)"}
+    inlet = _table([{"chainage_m": 5.0, "options": options}])["structures"][0]
+    assert inlet["type_id"] == "revetment" and "집수정은 안 셈" in inlet["notes"][0]
+
+
+def test_독립_기슭막이_한쪽이면_두_칸이_같을_때만_세고_다르면_사유() -> None:
+    same = {"facility": "revetment", "chainage_m": 50.0, "options": {"side": "좌"}}
+    rows = _table([same])["structures"]
+    assert [r["name"] for r in rows] == ["기슭막이 · 좌 벽"] and rows[0]["components"]
+    differ = {**same, "options": {"side": "우", "inlet_revet_height_m": 3.0}}
+    row = _table([differ])["structures"][0]
+    assert not row["components"] and "서버가 못 가림" in row["notes"][-1]
+
+
+def test_산출식_없는_세월교는_사유로_서고_터파기_빠짐_줄을_안_만든다() -> None:
+    table = _table([{"facility": "ford_bridge", "chainage_m": 30.0, "options": {}}])
+    assert "산출식이 아직 없습니다" in table["structures"][0]["notes"][0]
+    rows = build_handoff(unit_quantity_table=table)["work_items"]
+    assert not any("터파기가 안 선 구조물" in str(row.get("spec_detail")) for row in rows)
+
+
+def test_구조물_목록의_관_지점_종류_옛_저장분은_안_셈(monkeypatch, tmp_path) -> None:
+    old = StructureInstance(
+        structure_id="old", type_id="revetment", placement="point", chainage_m=10.0
+    )
+    monkeypatch.setattr(material, "load_structures", lambda root: (1, [old]))
+    targets, _names, skipped = material._collect_structures(str(tmp_path))
+    assert targets == [] and any("관 지점 정본이 주인" in note for note in skipped)