feat(b09): 운반 줄 단가연결 = 거리이정표 연결 · 연결 줄 모음 한 곳(ref_rows) (251-5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LLCk2y5xE8YGVgjbTntWsY

B09_Estimation/B09_Estimation_Bill_Items.py

@@ -20,6 +20,7 @@ from B08_Quantity.B08_Quantity_Material_Calc import _structure_row
 WHY_EARTH = "조합 밖 토공 산근 — 연결 없음"
 WHY_READY = "준비공 대가 — 연결 없음"
 WHY_HAUL = "운반 산근 — 연결 없음(운반거리 원천 없음)"
+WHY_LOGIC = "새 산근 — [저장] 뒤 단가(서버 셈)"
 WHY_STRUCTURE = "대가 호표 — 연결 없음(일위대가 145-2)"
 
 #: 토공 항목 — (공종, 엑셀 명칭, 단위, 토공집계표 (구분, 종류, 규격 | None) | None)
@@ -104,11 +105,16 @@ def structure_items(summary: dict[str, Any] | None) -> list[dict[str, Any]]:
     return out
 
 
-def haul_items(materials: dict[str, Any] | None) -> list[dict[str, Any]]:
-    """운반공 줄 — B08 운반 묶음(251-1 · 장비 둘 + 총괄자재 묶음 · 옛 키 haul:1~6 그대로)."""
+def haul_items(
+    materials: dict[str, Any] | None, refs: dict[str, Any] | None = None
+) -> list[dict[str, Any]]:
+    """운반공 줄 — B08 운반 묶음(251-1 · 장비 둘 + 총괄자재 묶음 · 옛 키 haul:1~6 그대로) ·
+    연결 = 거리이정표 단가연결(251-5 · `refs{줄 키: ref}` · 설계내역서 연결 덮개가 이김)."""
     out = []
     for g in haul_groups(materials):
         qty = g["qty"]
+        ref = (refs or {}).get(bill_key(g["key"]))
+        logic = (ref or {}).get("tab") == "logic"
         out.append(
             {
                 "key": bill_key(g["key"]),
@@ -118,8 +124,8 @@ def haul_items(materials: dict[str, Any] | None) -> list[dict[str, Any]]:
                 "unit": g["unit"],
                 "qty": qty,
                 "note": "B08 총괄자재" if qty is not None else "",
-                "ref": None,
-                "why": WHY_HAUL,
+                "ref": ref,
+                "why": WHY_LOGIC if logic else WHY_HAUL,
             }
         )
     return out
@@ -130,7 +136,7 @@ def bill_items(b08: dict[str, Any]) -> dict[str, Any]:
         "items": [
             *earth_items(b08.get("earthwork")),
             *structure_items(b08.get("summary")),
-            *haul_items(b08.get("materials")),
+            *haul_items(b08.get("materials"), b08.get("haul_refs")),
         ]
     }
 

B09_Estimation/B09_Estimation_Bill_Library.py

@@ -20,6 +20,7 @@ from pathlib import Path
 from typing import Any
 
 from B09_Estimation.B09_Estimation_Store import folder as store_folder
+from B09_Estimation.B09_Estimation_Store import ref_rows
 from common_util.common_util_json import atomic_write_json
 from M02_MasterTemplete import M02_Labor_Ref as refs
 from M02_MasterTemplete.M02_Item_Snapshot import MANIFEST_NAME, build_snapshot, now_text, origin_of
@@ -110,15 +111,11 @@ def copies(project_root: str | Path) -> list[tuple[Path, dict[str, Any]]]:
 
 
 def used_folders(user: dict[str, Any]) -> set[str]:
-    """설계내역서 덮개 줄이 잇는 서고 사본 폴더."""
-    rows = ((user or {}).get("bill") or {}).get("rows") or []
+    """설계내역서 덮개 줄 · 거리이정표 운반 연결(251-5)이 잇는 서고 사본 폴더."""
     return {
         str(ref["lib"])
-        for row in rows
-        if isinstance(row, dict)
-        and isinstance(ref := row.get("ref"), dict)
-        and ref.get("tab") == "dg"
-        and ref.get("lib")
+        for row in ref_rows(user)
+        if isinstance(ref := row.get("ref"), dict) and ref.get("tab") == "dg" and ref.get("lib")
     }
 
 

B09_Estimation/B09_Estimation_Calc_Bill.py

@@ -416,7 +416,7 @@ def compute_bill(
         if price_of is not None and item.get("owners"):
             parts = split_by_block(item, price_of)
         else:
-            parts = [(item, ref_price(item.get("ref"), books, hopyo))]
+            parts = [(item, ref_price(item.get("ref"), books, hopyo, item["key"]))]
         for part, prices in parts:
             items.append(_item_row(*_covered(part, prices, cover.get(part["key"]), books, hopyo)))
     for item in _material_items(inputs, base["compare"]):

B09_Estimation/B09_Estimation_Calc_Bill.ts

@@ -386,7 +386,7 @@ export function computeBill(
     const parts: Array<[Row, Row | null]> =
       priceOf !== null && item.owners && Object.keys(item.owners).length
         ? splitByBlock(item, priceOf)
-        : [[item, refPrice(item.ref ?? null, book, hopyo)]];
+        : [[item, refPrice(item.ref ?? null, book, hopyo, item.key)]];
     for (const [part, prices] of parts)
       items.push(itemRow(...covered(part, prices, cover[part.key], book, hopyo)));
   }

B09_Estimation/B09_Estimation_Router.py

@@ -27,9 +27,11 @@ from uuid import UUID
 from fastapi import APIRouter, Body, Depends, HTTPException
 from pydantic import BaseModel
 
+from B08_Quantity.B08_Quantity_Haul import bill_key
 from B08_Quantity.B08_Quantity_Router_Earthwork import earthwork_table
 from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
 from B08_Quantity.B08_Quantity_Router_Material import material_sheets
+from B08_Quantity.B08_Quantity_Router_Milestone import read_data as read_milestone
 from B09_Estimation import B09_Estimation_Bill_Library as library
 from B09_Estimation import B09_Estimation_Prices as prices
 from B09_Estimation import B09_Estimation_Sangeun as sangeun
@@ -103,10 +105,25 @@ def compute_twice(root: Path, inputs: dict[str, Any], user: dict[str, Any]) -> t
     return sheets, compute_summary(sheets, hopyo_lines(sheets))
 
 
+def haul_refs(root: Path) -> dict[str, Any]:
+    """거리이정표 운반 상자 단가연결(251-5) → `{설계내역서 줄 키: ref}` — 못 읽으면 빈 묶음."""
+    try:
+        routes = read_milestone(root).get("routes") or []
+    except Exception:  # noqa: BLE001 — 거리이정표 없음 = 연결 없음
+        return {}
+    return {bill_key(r["haul"]): r["ref"] for r in routes if r.get("haul") and r.get("ref")}
+
+
+def calc_user(root: Path) -> dict[str, Any]:
+    """셈 · 수요 · 사본 정리용 사용자 값 — 저장 값 + 거리이정표 운반 연결(화면엔 안 보냄)."""
+    return {**store.read_user(root), "haul_refs": haul_refs(root)}
+
+
 async def _demand(root: Path) -> list[str]:
-    """단가 수요 키 — 연결된 사본 + 설계내역서 손 줄 로직(166-2)."""
+    """단가 수요 키 — 연결된 사본 + 설계내역서 손 줄 로직(166-2) +
+    거리이정표 운반 연결(251-5)."""
     keys = await asyncio.to_thread(linked_resource_keys, root)
-    user = await asyncio.to_thread(store.read_user, root)
+    user = await asyncio.to_thread(calc_user, root)
     hand = await asyncio.to_thread(sangeun.hand_resource_keys, user)
     return [*keys, *(k for k in hand if k not in keys)]
 
@@ -121,11 +138,12 @@ async def payload(project_id: UUID, root: Path) -> dict[str, Any]:
         )
     book = await asyncio.to_thread(store.read_book, root)
     inputs = await asyncio.to_thread(prices.build_inputs, book, keys, b08["materials"])
-    inputs["bill"] = bill_items(b08)
+    inputs["bill"] = bill_items({**b08, "haul_refs": await asyncio.to_thread(haul_refs, root)})
     inputs["hopyo"] = hopyo_prices(root)
     inputs["daega"] = await asyncio.to_thread(daega_inputs, root)
     user = await asyncio.to_thread(store.read_user, root)
-    sheets, summary = await asyncio.to_thread(compute_twice, root, inputs, user)
+    calc = {**user, "haul_refs": await asyncio.to_thread(haul_refs, root)}
+    sheets, summary = await asyncio.to_thread(compute_twice, root, inputs, calc)
     return {
         "inputs": inputs,
         "user": user,
@@ -151,7 +169,8 @@ async def estimation(project_id: UUID) -> dict[str, Any]:
 async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) -> dict[str, Any]:
     root = await _project_root(project_id)
     user = await asyncio.to_thread(store.write_user, root, body.get("user"))
-    await asyncio.to_thread(library.prune, root, user)  # 204-2 줄이 안 잇는 서고 사본
+    calc = {**user, "haul_refs": await asyncio.to_thread(haul_refs, root)}
+    await asyncio.to_thread(library.prune, root, calc)  # 204-2 줄이 안 잇는 서고 사본
     keys = await _demand(root)
     book = await asyncio.to_thread(store.read_book, root)
     book = await asyncio.to_thread(prices.extend_book, book, keys)

B09_Estimation/B09_Estimation_Sangeun.py

@@ -38,6 +38,7 @@ from pathlib import Path
 from typing import Any
 
 from B09_Estimation.B09_Estimation_Calc_Base import gap, num, round_half, rounddown, total
+from B09_Estimation.B09_Estimation_Store import ref_rows
 from M01_MasterData import M01_MasterData_Store as m01
 from M01_MasterData import M01_MasterData_Store_Combo as combo_store
 from M02_MasterTemplete import M02_Item_Prices as item_prices
@@ -477,7 +478,7 @@ LISTED = ("no", "key", "logic", "name", "spec", "spec_raw", "unit", "sum", *PART
 def hand_sources(user: dict[str, Any] | None) -> list[tuple[str, str, dict[str, Any]]]:
     """설계내역서 손 줄 중 M01 로직 연결 → [(줄 키, 쓰는 곳 글, 산근 원천)]."""
     out = []
-    for row in ((user or {}).get("bill") or {}).get("rows") or []:
+    for row in ref_rows(user):  # 251-5 거리이정표 운반 연결 포함
         ref = row.get("ref") or {}
         if ref.get("tab") != "logic" or not ref.get("code"):
             continue

B09_Estimation/B09_Estimation_Store.py

@@ -94,6 +94,21 @@ def _text(value: Any) -> str:
     return str(value)[:200] if isinstance(value, (str, int, float)) else ""
 
 
+def ref_rows(user: dict[str, Any] | None) -> list[dict[str, Any]]:
+    """연결 있는 줄 모음 — 설계내역서 덮개 줄 + 거리이정표 운반 연결
+    (251-5 · `haul_refs{줄 키: ref}` · 서버 셈 때만 붙임) ·
+    같은 줄 키에 사용자 연결 덮개가 있으면 그것만."""
+    rows = [r for r in ((user or {}).get("bill") or {}).get("rows") or [] if isinstance(r, dict)]
+    covered = {r.get("key") for r in rows if "ref" in r}
+    haul = (user or {}).get("haul_refs") or {}
+    extra = [
+        {"key": key, "name": "", "ref": ref}
+        for key, ref in haul.items()
+        if key not in covered and isinstance(ref, dict)
+    ]
+    return [*rows, *extra]
+
+
 def _clean_ref(ref: Any) -> dict[str, Any] | None:
     """손 줄 연결 — dg · sg = 호표 키 · logic = M01 로직(입력값 · 켜쌓기율 · 고름) ·
     dg `lib` = 서고 사본 폴더(204-2 · [저장] 때 안 쓰는 사본 지움)."""

resources/tester/test_251_1_haul_groups.py

@@ -76,3 +76,34 @@ def test_251_2_서버_모델_읽기_길():
     assert r.haul == "sand" and r.ref == {"tab": "logic", "code": "GF000284"}
     src = (ROOT / "B08_Quantity" / "B08_Quantity_Router_Milestone.py").read_text(encoding="utf-8")
     assert "with_haul_routes(read_data(root), groups)" in src
+
+
+def test_251_5_B09_운반_줄_연결_거리이정표():
+    from B09_Estimation.B09_Estimation_Bill_Library import used_folders
+    from B09_Estimation.B09_Estimation_Sangeun import hand_sources
+    from B09_Estimation.B09_Estimation_Store import ref_rows
+
+    logic = {"tab": "logic", "code": "GF000284", "name": "덤프 운반", "given": {"L": 5}}
+    lib = {"tab": "dg", "key": "DG:x", "lib": "precombo@system@-@SC-1"}
+    items = {i["key"]: i for i in haul_items(_mats(ROWS), {"haul:3": logic, "haul:remicon": lib})}
+    assert items["haul:3"]["ref"] == logic and items["haul:3"]["why"].startswith("새 산근")
+    assert items["haul:remicon"]["ref"] == lib and items["haul:6"]["ref"] is None
+    # 셈 · 수요 · 사본 정리 = 설계내역서 덮개 + 운반 연결 · 같은 줄 덮개 연결이 이김
+    user = {"bill": {"v": 2, "rows": [{"key": "haul:remicon", "ref": None}]},
+            "haul_refs": {"haul:3": logic, "haul:remicon": lib}}  # fmt: skip
+    assert [r["key"] for r in ref_rows(user)] == ["haul:remicon", "haul:3"]
+    assert [k for k, _u, _s in hand_sources(user)] == ["haul:3"]
+    assert used_folders(user) == set()  # 덮개가 연결 풀었음 = 사본 안 지킴
+    assert used_folders({"haul_refs": {"haul:remicon": lib}}) == {"precombo@system@-@SC-1"}
+
+
+def test_251_5_자동_줄_단가_줄_키_짝():
+    from B09_Estimation.B09_Estimation_Calc_Bill import ref_price
+
+    books = {"sg_list": [{"key": "SG:9", "no": 9, "sum": 100}], "sg_hand": {"haul:3": "SG:9"}}
+    got = ref_price({"tab": "logic", "code": "GF000284"}, books, {}, "haul:3")
+    assert got and got["key"] == "SG:9" and got["no"] == "산근 9호표"
+    py = (ROOT / "B09_Estimation" / "B09_Estimation_Calc_Bill.py").read_text(encoding="utf-8")
+    ts = (ROOT / "B09_Estimation" / "B09_Estimation_Calc_Bill.ts").read_text(encoding="utf-8")
+    assert 'ref_price(item.get("ref"), books, hopyo, item["key"])' in py
+    assert "refPrice(item.ref ?? null, book, hopyo, item.key)" in ts