feat(B09): 원가 단가 수요 = 연결된 사본이 부른 자원(143-1 원천 · 144)

- B09_Estimation_Linked = linked_snapshots 사본마다 담은 로직을 사본 단가 묶음 · 입력값 · 고른자재로 돌려 부른 LB · EQ · MT · MO 줄만 모음(묶음 통째 · 고르기 후보 안 씀 · 쓰기 없음)
- 자리만이던 수요 함수를 바꿈 · 입력 빈 사본은 부른 줄 없음(빈 입력 자동값은 142-1 몫)
- 시험 = test_144_b09_skeleton 더함(기계경비 로직 → EQ000004 · MO000002 · LB000048)

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

B09_Estimation/B09_Estimation_Linked.py

@@ -0,0 +1,71 @@
+"""B09 연결된 사본 → 원가 수요 자원 키(PLAN 144 · 143-1 원천) — 서버 단독 · 읽기만.
+
+연결된 사본 = `M02_Item_Snapshot.linked_snapshots`(지정 ∩ 같은 항목 사본 · B08 · B09 공용) ·
+사본마다 `일위대가.json` 담은 로직을 그 사본 단가 묶음(`단가.json`) · 입력값 · 고른자재로 돌리며
+실제로 부른 자원 줄(인력 LB · 기계 EQ · 자재 MT · 유가 MO)만 모음 — 단가 묶음 통째(고르기 후보 전부)는
+안 씀 · 값은 쓰지 않음(단가는 원가 한 벌 `price_book.json` 이 맡음).
+구조물 조합 사본 없으면 그 구조물은 없음 · 일위대가 없는 사본은 자원 없음.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from M02_MasterTemplete import M02_Item_Prices as item_prices
+from M02_MasterTemplete.M02_Item_Snapshot import (
+    PICKS_FILE,
+    VALUES_FILE,
+    linked_snapshots,
+    structures_dir,
+)
+from M02_MasterTemplete.M02_Item_Source import LABOR_FILE, PRICE_FILE, read_json
+
+#: 원가 단가 목록에 오르는 자원 표
+RESOURCE_TABLES = ("LB", "EQ", "MT", "MO")
+
+mf = item_prices.mf
+
+
+def used_keys(folder: Path) -> list[str]:
+    """사본 하나가 담은 로직을 돌릴 때 부른 자원 키 — 부른 차례 · 겹침 없음."""
+    labor = read_json(folder / LABOR_FILE)
+    if not isinstance(labor, dict):
+        return []
+    snapshot = item_prices.load_prices(read_json(folder / PRICE_FILE) or {})
+    values = read_json(folder / VALUES_FILE) or {}
+    picks = read_json(folder / PICKS_FILE) or {}
+    seen: list[str] = []
+    base_get = snapshot.get
+
+    def get(ref: str) -> dict[str, Any]:
+        row = base_get(ref)
+        key = str(row.get("키") or "")
+        if key[:2] in RESOURCE_TABLES and key not in seen:
+            seen.append(key)
+        return row
+
+    snapshot.get = get  # 이 사본 묶음 한 벌에만 — 마스터 캐시는 안 건드림
+    for key in item_prices.mcb.logic_keys(labor.get("줄") or {}):
+        if not key:
+            continue
+        given = {name: (one or {}).get("값") for name, one in (values.get(key) or {}).items()}
+        try:
+            with mf.mp.choosing((picks or {}).get(key)):
+                mf.run(snapshot, key, given)
+        except (mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError):
+            continue  # 못 돈 로직 — 부른 데까지만
+    return seen
+
+
+def linked_resource_keys(project_root: str | Path) -> list[str]:
+    """연결된 사본 전부가 부른 자원 키 — 구조물 id 차례 · 겹침 없음."""
+    base = structures_dir(project_root)
+    out: list[str] = []
+    for _owner, linked in sorted(linked_snapshots(project_root).items()):
+        if "일위대가" not in linked.get("부분", []):
+            continue
+        for key in used_keys(base / linked["폴더"]):
+            if key not in out:
+                out.append(key)
+    return out

B09_Estimation/B09_Estimation_Prices.py

@@ -4,8 +4,7 @@
 [단가 갱신] 때만 새로 · 대가 · 산근도 이 단가로 셈(한 직종 · 한 자재 = 한 단가).
 
 줄 = 수요 따라(사용자 승인 ① · 10-9) — 엑셀 목록을 기본으로 깔지 않음:
-    연결된 사본이 쓰는 자원 키(`linked_resource_keys` — 143-1 「연결된 사본」 원천 · 145-1 · 2 호표가
-    들어오면 그 결과를 돌려줌 · 그전엔 빈 목록 = 빈 표 「연결 없음」)
+    연결된 사본이 부른 자원 키(`B09_Estimation_Linked` — 143-1 「연결된 사본」 원천 · 없으면 빈 표)
     + 기계가 부르는 운전사(조종원직종 → 인력 원문번호) · 연료(주연료종류 → 유가 「종류:전국평균」)
     + B08 총괄자재수량집계표 자재(M01 짝 없음 → 출처 빈칸 · 적용 단가는 사용자 입력)
 운전사 시간 계수 = M01 CC000336(시간환산 · 상여계수 · 휴지계수 분자 · 분모 곱).
@@ -30,15 +29,6 @@ FUEL_REGION = "전국평균"
 MT_SOURCES = ("관급", "물가자료", "유통물가", "물가정보", "거래가격")
 
 
-def linked_resource_keys(project_root: str | Path) -> list[str]:
-    """연결된 사본(구조물 조합 사본)이 쓰는 자원 키(LB · EQ · MT · MO) — 자리만.
-
-    143-1 「연결된 사본」 원천 함수 · 145-1 · 2 산근 · 대가 호표가 들어오면 그 자원 키를 돌려줌 ·
-    그전엔 빈 목록(구조물 조합 사본 없으면 그 구조물은 없음 · 연결 안 된 값은 비움).
-    """
-    return []
-
-
 def _f(value: Any) -> float | None:
     if isinstance(value, Decimal):
         return float(value)

B09_Estimation/B09_Estimation_Router.py

@@ -22,6 +22,7 @@ from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
 from B08_Quantity.B08_Quantity_Router_Material import material_sheets
 from B09_Estimation import B09_Estimation_Prices as prices
 from B09_Estimation import B09_Estimation_Store as store
+from B09_Estimation.B09_Estimation_Linked import linked_resource_keys
 from B09_Estimation.B09_Estimation_Calc_Base import compute_base
 
 router = APIRouter(prefix="/api/projects", tags=["B09 Estimation"])
@@ -41,7 +42,7 @@ async def _b08(project_id: UUID, notes: list[str]) -> dict[str, Any]:
 async def payload(project_id: UUID, root: Path) -> dict[str, Any]:
     notes: list[str] = []
     b08 = await _b08(project_id, notes)
-    keys = prices.linked_resource_keys(root)
+    keys = await asyncio.to_thread(linked_resource_keys, root)
     if not keys:
         notes.append(
             "연결된 구조물 조합 사본 없음 — 노무 · 경비 · 중기 단가 줄은 사본이 연결되면 생김"
@@ -68,7 +69,7 @@ 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)
     await asyncio.to_thread(store.write_user, root, body.get("user"))
-    keys = prices.linked_resource_keys(root)
+    keys = await asyncio.to_thread(linked_resource_keys, root)
     book = await asyncio.to_thread(store.read_book, root)
     book = await asyncio.to_thread(prices.extend_book, book, keys)
     await asyncio.to_thread(store.write_book, root, book)
@@ -78,7 +79,7 @@ async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) ->
 @router.post("/{project_id}/estimation/prices")
 async def refresh_prices(project_id: UUID) -> dict[str, Any]:
     root = await _project_root(project_id)
-    keys = prices.linked_resource_keys(root)
+    keys = await asyncio.to_thread(linked_resource_keys, root)
     book = await asyncio.to_thread(prices.refresh_book, keys)
     await asyncio.to_thread(store.write_book, root, book)
     return await payload(project_id, root)

resources/tester/test_144_b09_skeleton.py

@@ -112,3 +112,22 @@ def test_연결_없으면_빈_표와_연결_없음(tmp_path, monkeypatch):
     assert any("연결된 구조물 조합 사본 없음" in n for n in body["notes"])
     assert body["book"] is None  # GET 은 아무것도 쓰지 않음
     assert not (tmp_path / "B09_Estimation").exists()
+
+
+def test_연결된_사본이_부른_자원만(tmp_path):
+    """사본 단가 묶음 통째(고르기 후보 기계 100여)가 아니라 로직이 실제로 부른 줄만."""
+    from B09_Estimation.B09_Estimation_Linked import used_keys
+    from M02_MasterTemplete import M02_Item_Prices as item_prices
+
+    bundle = item_prices.to_json(item_prices.collect(["GC000268"]))  # 기계 시간당 기계경비
+    files = {
+        "단가.json": bundle,
+        "일위대가.json": {"줄": {"종류": "일위대가", "담은로직": [{"로직": "GC000268"}]}},
+        "입력값.json": {"GC000268": {"기계": {"값": "0101-0019"}, "지역": {"값": "전국평균"}}},
+    }
+    for name, data in files.items():
+        (tmp_path / name).write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+    assert used_keys(tmp_path) == ["EQ000004", "MO000002", "LB000048"]
+    (tmp_path / "입력값.json").write_text("{}", encoding="utf-8")
+    assert used_keys(tmp_path) == []  # 입력 빈 로직은 못 돎 — 부른 줄 없음
+    assert used_keys(tmp_path / "없음") == []