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

B08_Quantity/B08_Quantity_Engine_Handoff.py

@@ -249,6 +249,11 @@ def build_handoff(
     result["pum_edition"] = table.pum_edition or None
     for row in work_items:
         row.setdefault("pum_edition", table.pum_edition or None)
+    # 불변 열쇠 길목(2026-09-17 8-6 이행) — 줄마다 `work_item_key`(FW-·AX-) 를 더함.
+    #   `work_item_code` 는 사람이 보는 목차 번호로 그대로 · 못 찾은 코드는 목록으로(지어내지 않음).
+    from common_util.common_util_work_item_key import attach_keys
+
+    result["unkeyed_work_item_codes"] = attach_keys(work_items)
     return result
 
 

B09_Estimation/B09_Estimation_BillOfQuantities_Input.py

@@ -14,6 +14,8 @@ from dataclasses import dataclass
 from decimal import Decimal
 from typing import Any
 
+from common_util.common_util_work_item_key import work_item_key
+
 _ZERO = Decimal(0)
 
 #: 자재 공급 구분이 안 갈린 값. B08 이 실제로 이 값을 보낸다(2026-09-08 실물 확인).
@@ -78,6 +80,9 @@ class HandoffWorkItem:
     spec_class_basis: str = ""
     #: 이 줄의 공종 코드가 본 품셈 판 — 마스터 판과 다르면 값을 안 씀(명세 17장).
     pum_edition: str = ""
+    #: 불변 열쇠(FW-·AX-) — 목차 번호(`work_item_code`)가 개정으로 밀려도 안 바뀜(8-6 이행 길목).
+    #:   ⚠ 금액 셈은 아직 `work_item_code` 로 함 — 이 칸은 더하기만(630곳 안 고침).
+    work_item_key: str = ""
 
     @property
     def display_name(self) -> str:
@@ -149,6 +154,10 @@ def parse_handoff(payload: dict[str, Any]) -> tuple[list[HandoffWorkItem], list[
             spec_class=row.get("spec_class") or "",
             spec_class_basis=row.get("spec_class_basis") or "",
             pum_edition=str(row.get("pum_edition") or ""),
+            # 옛 인계(열쇠 없음)도 같은 길목으로 채움
+            work_item_key=row.get("work_item_key")
+            or work_item_key(row.get("work_item_code"), row.get("pum_edition") or None)
+            or "",
         )
         for row in payload["work_items"]
     ]

Z01_MasterData/Z01_MasterData_BasePrices.py

@@ -511,6 +511,7 @@ def table(
     q: str = "",
     sort: str = "",
     desc: bool = False,
+    filters: dict[str, str] | None = None,
 ) -> dict[str, Any]:
     all_rows = rows(kind)
     columns = _columns(kind, all_rows)
@@ -524,6 +525,8 @@ def table(
         if needle
         else all_rows
     )
+    if kind in work_items.KINDS:  # 서버가 준 거르기(부문·줄 구실) — 고른 값만 옴
+        hits = work_items.apply_filters(kind, hits, filters or {})
     if sort:  # ⚠ 거른 뒤에 세움 — 거꾸로 하면 차례가 깨짐(브레인 ③)
         if sort not in columns:
             raise SortError(f"없는 열로 세울 수 없음: {sort}")

Z01_MasterData/Z01_MasterData_Router.py

@@ -3,7 +3,7 @@
   GET /api/master-data/tree                                  갈래 → 파일 → 표
   GET /api/master-data/rows?file=&table=&page=&size=&q=      표 줄(쪽 나누기 · 검색)
   GET /api/master-data/base-prices                           표 목록(갈래·kind·이름·줄 수 · 공종 축 산림·건설 포함)
-  GET /api/master-data/base-prices/{kind}?page=&size=&q=&sort=&desc=  한 표(기초단가 다섯 · 품셈 기준 열)
+  GET /api/master-data/base-prices/{kind}?page=&size=&q=&sort=&desc=[&거르기=값]  한 표(기초단가 다섯 · 품셈 기준 열 · 공종 축 둘)
   PUT /api/master-data/base-prices/{kind}/{row_id}           {values:{열:값}} → 덮개에만 씀 · null = 되돌리기
   GET /api/master-data/overrides                             고친 것·원본 바뀐 것·주인 없는 것(서버 정렬)
 ⚠ 권한은 등록하는 쪽(`main.py` · 랩탑 서브)이 `dependencies=[verify_session, require_system_admin]` 로 붙임.
@@ -13,13 +13,14 @@ from __future__ import annotations
 
 from typing import Any, Literal
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
 from common_util.common_util_auth import verify_session
 from Z01_MasterData import Z01_MasterData_BasePrices as base_prices
 from Z01_MasterData import Z01_MasterData_Tables as tables
+from Z01_MasterData import Z01_MasterData_WorkItems as work_items
 
 router = APIRouter(prefix="/api/master-data", tags=["Z01 MasterData"])
 
@@ -58,14 +59,19 @@ def get_base_price_kinds() -> dict:
 @router.get("/base-prices/{kind}")
 def get_base_prices(
     kind: str,
+    request: Request,
     page: int = 1,
     size: int = tables.DEFAULT_PAGE_SIZE,
     q: str = "",
     sort: str = "",
     desc: int = 0,
 ) -> dict:
+    # 거르기 — 표가 준 `filters` 의 key 만 받음(공종 축 부문·줄 구실)
+    picked = {k: v for k, v in request.query_params.items() if k in work_items.FILTER_KEYS}
     try:
-        return base_prices.table(_kind(kind), page=page, size=size, q=q, sort=sort, desc=bool(desc))
+        return base_prices.table(
+            _kind(kind), page=page, size=size, q=q, sort=sort, desc=bool(desc), filters=picked
+        )
     except base_prices.SortError as error:
         raise HTTPException(status_code=400, detail=str(error)) from error
 

Z01_MasterData/Z01_MasterData_WorkItems.py

@@ -144,10 +144,68 @@ def notice(kind: str) -> list[str]:
     return lines
 
 
+#: 거르기 — **자료에 있는 가름만**(건설 부문 · 줄 구실). 어느 줄을 거를지는 서버 한 곳 · 화면은 고른 값만 보냄.
+#: ⚠ 「임도가 쓰는 것」 같은 쓰임 가름은 잣대가 아직 없어 안 냄(지어내지 않음).
+FILTER_KEYS = ("division", "axis_role")
+ALL = ""
+
+
+def _facets(kind: str) -> dict[str, dict[str, str]]:
+    found = doc(kind)
+    items = found[0].get("work_items") or [] if found else []
+    return {_key(i): {key: str(i.get(key) or "") for key in FILTER_KEYS} for i in items}
+
+
+def _role_names() -> dict[str, str]:
+    """줄 구실 이름 — `axis_policy.json`(데스크탑 서브 잣대) 한 곳에서."""
+    path = FOLDER / "axis_policy.json"
+    if not path.is_file():
+        return {}
+    roles = _read(str(path), path.stat().st_mtime_ns).get("roles") or {}
+    return {key: str(role.get("name_ko") or key) for key, role in roles.items()}
+
+
+def filters(kind: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """거르기 상자 — 값이 둘 이상 갈리는 가름만 · 기본은 「전부」(안 거름)."""
+    facets = _facets(kind)
+    labels = {"division": ("부문", {}), "axis_role": ("줄 구실", _role_names())}
+    out = []
+    for key in FILTER_KEYS:
+        counts: dict[str, int] = {}
+        for row in rows:
+            value = facets.get(row["@id"], {}).get(key, ALL)
+            counts[value] = counts.get(value, 0) + 1
+        values = [v for v in counts if v != ALL]
+        if len(values) < 2:
+            continue
+        label, names = labels[key]
+        out.append(
+            {
+                "key": key,
+                "label": label,
+                "options": [{"value": ALL, "label": "전부", "rows": len(rows)}]
+                + [{"value": v, "label": names.get(v, v), "rows": counts[v]} for v in values],
+                "default": ALL,
+            }
+        )
+    return out
+
+
+def apply_filters(
+    kind: str, rows: list[dict[str, Any]], picked: dict[str, str]
+) -> list[dict[str, Any]]:
+    facets = _facets(kind)
+    for key, value in picked.items():
+        if key in FILTER_KEYS and value != ALL:
+            rows = [r for r in rows if facets.get(r["@id"], {}).get(key) == value]
+    return rows
+
+
 def extra(kind: str) -> dict[str, Any]:
     """표 수준 덧붙임 — 계산 규칙 · 줄 수 · 갈래 편 수(쪽 나누기와 무관하게 전체)."""
     rows = base_rows(kind)
     return {
+        "filters": filters(kind, rows),
         "rules": rules(kind),
         "stats": {
             "work_items": len(rows),

common_util/common_util_work_item_key.py

@@ -0,0 +1,68 @@
+"""공종 불변 열쇠 **길목** — 목차 코드(`FP-`·`CP-`) → 불변 열쇠(`FW-`·`CW-`) (2026-09-17 PLAN_공종축 8-6 이행).
+
+코덱스 조사(FP- 참조 630곳 · 77파일)의 결론대로 **경계 두 곳에서만** 부름 — 630곳은 안 고침.
+  B08 인계 출구  `B08_Quantity_Engine_Handoff.build_handoff`  줄마다 `work_item_key` 를 더함
+  B09 인계 입구  `B09_Estimation_BillOfQuantities_Input.parse_handoff`  옛 인계(열쇠 없음)에도 채움
+`work_item_code` 는 **그대로 남김** — 사람이 보는 목차 번호(화면·로그에 보이는 것이 값짐).
+
+정본은 공종 마스터 파일의 줄(`work_item_code` ↔ `work_item_key` · 판 `pum_edition`) — 장부를 따로 안 읽음
+(마스터가 장부로 지은 결과라 두 벌이 안 됨).
+⚠ `AX-WK-`·`AX-ST-` 는 난수 8자리(명세 2장 ④ · 동등 비교만)라 **이미 불변** — 그대로 열쇠.
+⚠ 못 찾은 코드는 `None` — 지어내지 않음. 받는 쪽이 목록으로 드러냄(`unkeyed_work_item_codes`).
+⚠ 한 판 안에서 같은 목차 코드가 두 줄이면(목차 오기) 어느 열쇠인지 못 가름 → `None`.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from functools import lru_cache
+from pathlib import Path
+from typing import Any, Iterable
+
+MASTER_DIR = Path(__file__).resolve().parent.parent / "resources" / "data_work_item_master"
+#: 우리가 세운 공종·구조물 — 난수 코드가 곧 불변 열쇠
+_STABLE_CODE = re.compile(r"^AX-(?:WK|ST)-[0-9a-f]{8}$")
+
+
+@lru_cache(maxsize=8)
+def _read(path: str, _mtime_ns: int) -> dict[tuple[str, str], str | None]:
+    doc = json.loads(Path(path).read_text(encoding="utf-8"))
+    edition = str(doc.get("pum_edition") or doc.get("effective_date") or "")
+    keys: dict[tuple[str, str], str | None] = {}
+    for item in doc.get("work_items") or []:
+        slot = (edition, str(item.get("work_item_code") or ""))
+        keys[slot] = None if slot in keys else item.get("work_item_key")
+    return keys
+
+
+def _keys() -> dict[tuple[str, str], str | None]:
+    out: dict[tuple[str, str], str | None] = {}
+    for path in sorted(MASTER_DIR.glob("*work_item_master_*.json")):  # 산림 + 건설(`const_`)
+        out.update(_read(str(path), path.stat().st_mtime_ns))
+    return out
+
+
+def work_item_key(code: str | None, pum_edition: str | None = None) -> str | None:
+    """목차 코드 → 불변 열쇠. 판을 모르면 코드가 한 판에만 있을 때만 줌."""
+    if not code:
+        return None
+    if _STABLE_CODE.match(code):
+        return code
+    keys = _keys()
+    if pum_edition:
+        return keys.get((pum_edition, code))
+    found = {key for (_, each), key in keys.items() if each == code}
+    return found.pop() if len(found) == 1 else None
+
+
+def attach_keys(rows: Iterable[dict[str, Any]]) -> list[str]:
+    """인계 줄마다 `work_item_key` 를 더함(코드 없는 줄은 `None` — 칸은 늘 있음 · 인계 계약) ·
+    코드는 있는데 열쇠를 못 찾은 코드 목록을 돌려줌."""
+    missing: set[str] = set()
+    for row in rows:
+        code = row.get("work_item_code")
+        row["work_item_key"] = work_item_key(code, row.get("pum_edition"))
+        if code and row["work_item_key"] is None:
+            missing.add(str(code))
+    return sorted(missing)

resources/tester/test_b08_handoff_contract.py

@@ -53,6 +53,9 @@ WORK_ITEM_KEYS = {
     "origin",
     # 공종 코드가 본 품셈 판 — B09 가 마스터 판과 다르면 값을 안 씀(명세 17장 · 2026-09-13 추가).
     "pum_edition",
+    # 불변 열쇠(FW-·AX-) — 목차 번호가 개정으로 밀려도 안 바뀜(2026-09-17 8-6 이행 길목 ·
+    #   받는 쪽 B09 `parse_handoff` 가 읽음). 코드 없는 줄은 None.
+    "work_item_key",
 }
 
 #: **자재 줄**이 내보내는 칸. ⚠ 여기에 `work_item_code` 가 들어가면 안 된다(축이 다르다).

resources/tester/test_work_item_key_gate.py

@@ -0,0 +1,139 @@
+"""공종 불변 열쇠 **길목** — PLAN_공종축 8-6 이행(2026-09-17 브레인 · 코덱스 조사).
+
+잣대 = **금액 불변.** 열쇠를 더하는 일이지 값을 바꾸는 일이 아님 —
+같은 인계로 세운 내역서가 열쇠가 있든(새 인계) 없든(옛 인계) **한 원도** 안 달라야 함.
+길목은 경계 두 곳뿐(B08 인계 출구 · B09 인계 입구) · `work_item_code` 는 목차 번호로 그대로 남음.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+import re
+from functools import lru_cache
+
+from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff
+from B08_Quantity.B08_Quantity_Engine_MaterialSummary import build_table as build_material_table
+from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table as build_unit_table
+from B09_Estimation.B09_Estimation_BillOfQuantities import bill_summary, build_bill
+from B09_Estimation.B09_Estimation_BillOfQuantities_Input import parse_handoff
+from B09_Estimation.B09_Estimation_UnitPrice import build_unit_prices
+from common_util.common_util_work_item_key import MASTER_DIR, attach_keys, work_item_key
+
+MAPPING = MASTER_DIR.parent / "data_work_item_mapping" / "work_item_mapping_2026-01-01.json"
+
+
+def _strip_keys(value):
+    """옛 인계 — 길목이 더한 칸을 뗌."""
+    if isinstance(value, dict):
+        return {
+            k: _strip_keys(v)
+            for k, v in value.items()
+            if k not in ("work_item_key", "unkeyed_work_item_codes")
+        }
+    if isinstance(value, list):
+        return [_strip_keys(v) for v in value]
+    return value
+
+
+@lru_cache(maxsize=1)
+def _handoff() -> dict:
+    """토공·운반·구조물·자재가 다 선 인계 — 금액이 실제로 붙는 줄이 여럿."""
+    unit = build_unit_table(
+        [
+            {
+                "structure_id": "s1",
+                "type_id": "masonry_wet",
+                "start_m": 35.0,
+                "end_m": 45.0,
+                "options": {"height_m": 1.5, "length_m": 10.0},
+            }
+        ]
+    )
+    rows = [
+        {"group": "흙깎기", "item": "토사", "spec": "", "unit": "㎥", "amount": 1234.5},
+        {"group": "측구터파기", "item": "토사", "spec": "", "unit": "㎥", "amount": 77.0},
+        {"group": "층따기", "item": "", "spec": "", "unit": "㎥", "amount": 12.0},
+    ]
+    haul = {
+        "rows": [
+            {
+                "equipment": "dump_truck",
+                "ground": "토사",
+                "volume_m3": 500.0,
+                "average_distance_m": 1200.0,
+                "in_bill": True,
+            }
+        ]
+    }
+    return build_handoff(
+        summary_table={"rows": rows},
+        haul_table=haul,
+        unit_quantity_table=unit,
+        material_table=build_material_table(unit),
+    )
+
+
+def test_목차_코드는_불변_열쇠로_난수_코드는_그대로() -> None:
+    forest = json.loads(
+        (MASTER_DIR / "work_item_master_2026-01-01.json").read_text(encoding="utf-8")
+    )
+    for item in forest["work_items"]:
+        assert work_item_key(item["work_item_code"], forest["pum_edition"]) == item["work_item_key"]
+    assert work_item_key("FP-09-03-02") == work_item_key("FP-09-03-02", "2026-01-01")
+    assert re.fullmatch(r"CW-\d{5}", work_item_key("CP-01-03-02-01") or "")
+    assert work_item_key("AX-WK-c0842a0d") == "AX-WK-c0842a0d"  # 난수 8자리 — 이미 불변
+    assert work_item_key("AX-ST-0123abcd") == "AX-ST-0123abcd"
+    assert work_item_key("FP-09-03-02", "1999-01-01") is None  # 판이 다르면 안 줌
+    assert work_item_key("FP-99-99") is None and work_item_key(None) is None
+
+
+def test_매핑의_코드가_다_열쇠를_받음() -> None:
+    """B08 이 인계에 싣는 코드의 샘 — 하나라도 못 받으면 그 줄은 열쇠 없이 감."""
+    text = MAPPING.read_text(encoding="utf-8")
+    codes = set(re.findall(r'"((?:FP|CP|AX-WK|AX-ST)-[0-9A-Za-z-]+)"', text))
+    assert len(codes) > 50
+    assert [c for c in sorted(codes) if work_item_key(c, "2026-01-01") is None] == []
+
+
+def test_B08_인계_출구에서_열쇠가_붙고_목차_코드는_그대로() -> None:
+    handoff = _handoff()
+    coded = [row for row in handoff["work_items"] if row.get("work_item_code")]
+    assert len(coded) >= 5
+    for row in coded:
+        assert row["work_item_key"] == work_item_key(row["work_item_code"], row["pum_edition"])
+        assert row["work_item_code"].startswith(("FP-", "AX-"))  # 사람이 보는 번호는 남음
+    assert handoff["unkeyed_work_item_codes"] == []
+    assert all("work_item_key" not in row for row in handoff["materials"])  # 자재는 공종 축 아님
+
+
+def test_못_찾은_코드는_지어내지_않고_목록() -> None:
+    rows = [
+        {"work_item_code": "FP-99-99"},
+        {"work_item_code": None},
+        {"work_item_code": "FP-09-18"},
+    ]
+    assert attach_keys(rows) == ["FP-99-99"]
+    assert rows[0]["work_item_key"] is None and rows[1]["work_item_key"] is None
+    assert rows[2]["work_item_key"].startswith("FW-")
+
+
+def test_B09_입구는_옛_인계에도_같은_열쇠() -> None:
+    new, _ = parse_handoff(_handoff())
+    old, _ = parse_handoff(_strip_keys(copy.deepcopy(_handoff())))
+    assert [w.work_item_key for w in old] == [w.work_item_key for w in new]
+    assert any(w.work_item_key.startswith("FW-") for w in new)
+
+
+def test_금액_불변_열쇠가_있든_없든_내역서가_한_원도_안_다름() -> None:
+    """⭐ 이번 일감의 잣대(브레인) — 새 인계(열쇠 있음)와 옛 인계(열쇠 없음)로 세운 내역서가 같음."""
+    build = build_unit_prices()
+    new = build_bill(copy.deepcopy(_handoff()), build=build)
+    old = build_bill(_strip_keys(copy.deepcopy(_handoff())), build=build)
+    assert bill_summary(new) == bill_summary(old)
+    assert [row.as_dict() for row in new.rows] == [row.as_dict() for row in old.rows]
+    assert [row.as_dict() for row in new.material_rows] == [
+        row.as_dict() for row in old.material_rows
+    ]
+    priced = [row for row in new.rows if not row.is_group and row.amount_krw]
+    assert len(priced) >= 3 and new.body_total_krw > 0  # 금액이 실제로 선 줄로 잼

resources/tester/test_z01_work_items.py

@@ -113,3 +113,39 @@ def test_불변_열쇠가_오면_id_로_씀_자료_파일은_판_id_로_가림(
         json={"values": {"name": "바꿈"}},
     )
     assert res.status_code == 400 and "공종 축 뼈대" in res.json()["detail"]
+
+
+def test_거르기는_자료에_있는_가름만_서버가_판정(client: TestClient) -> None:
+    """화면(서브 b890fba0)이 `filters` 를 받으면 상자를 세우고 고른 값만 보냄 — 판정은 서버 한 곳.
+    건설 부문 다섯 · 줄 구실(이름은 axis_policy.json). ⚠ 「임도가 쓰는 것」은 잣대가 없어 안 냄."""
+    const = _get(client, "work_item_const", size=1)
+    by_key = {f["key"]: f for f in const["filters"]}
+    assert set(by_key) == {"division", "axis_role"}
+    division = by_key["division"]
+    assert division["default"] == ""
+    assert division["options"][0] == {"value": "", "label": "전부", "rows": const["total"]}
+    assert [o["value"] for o in division["options"][1:]] == [
+        "공통부문",
+        "토목부문",
+        "건축부문",
+        "기계설비부문",
+        "유지관리부문",
+    ]
+    assert sum(o["rows"] for o in division["options"][1:]) == const["total"]
+    assert {o["value"]: o["label"] for o in by_key["axis_role"]["options"]}[
+        "general_provision"
+    ] == "총칙"
+
+    civil = _get(client, "work_item_const", size=500, division="토목부문")
+    assert civil["total"] == next(
+        o["rows"] for o in division["options"] if o["value"] == "토목부문"
+    )
+    assert all(r["name"].startswith("토목부문 › ") for r in civil["rows"])
+    rules = _get(
+        client, "work_item_const", size=500, division="공통부문", axis_role="general_provision"
+    )
+    assert rules["total"] > 0
+    assert all(r["name"].startswith("공통부문 › 적용기준") for r in rules["rows"])
+
+    assert [f["key"] for f in _get(client, "work_item_forest", size=1)["filters"]] == ["axis_role"]
+    assert _get(client, "labor", size=1, division="토목부문")["total"] == 261  # 다른 표는 안 거름