diff --git a/B09_Estimation/B09_Estimation_BillOfQuantities.py b/B09_Estimation/B09_Estimation_BillOfQuantities.py index 2f2fe682..e8fed13c 100644 --- a/B09_Estimation/B09_Estimation_BillOfQuantities.py +++ b/B09_Estimation/B09_Estimation_BillOfQuantities.py @@ -47,6 +47,7 @@ _ZERO = Decimal(0) # ⬇ 인계 읽기는 `_Input` 으로 옮겼다(2026-09-15 700줄 분리) — 쓰던 이름이 그대로 살게 다시 내보낸다. from B09_Estimation.B09_Estimation_BillOfQuantities_Input import ( # noqa: E402,F401 + BLOCKED_LINK_UNDECIDED, BLOCKED_UNCONFIRMED, SUPPLY_OWNER, SUPPLY_UNKNOWN, @@ -56,6 +57,7 @@ from B09_Estimation.B09_Estimation_BillOfQuantities_Input import ( # noqa: E402 _decimal, parse_handoff, ) +from common_util.common_util_work_item_link import undecided_links # noqa: E402 #: 총 절취량으로 셀 공종 — 9-3 토사깎기 · 9-4 암절취 · 9-5 발파암(인계 대응표 「흙깎기」 셋). #: ⚠ 판에 묶인 절 번호임(명세 17장) — 판이 바뀌면 대응표와 함께 고칠 것. @@ -75,6 +77,7 @@ _BLOCKED_LABELS = { "input_missing": "입력이 필요합니다", "unit_data_missing": "원단위가 없습니다(우리가 만들 것)", "formula_missing": "전개식이 없습니다(우리가 만들 것)", + BLOCKED_LINK_UNDECIDED: "연결고리 미판정 — 금액에 안 들어감", } @@ -353,6 +356,18 @@ def build_bill( } ) continue + if item.blocked_kind == BLOCKED_LINK_UNDECIDED: + # 산림·건설 어느 품을 쓸지 연결고리 표가 미판정 — 근거 없이 고르지 않음(브레인 · 코덱스 ②). + result.missing.append( + { + "name": item.display_name, + "unit": item.unit, + "quantity": str(item.quantity), + "reason": f"{_BLOCKED_LABELS[BLOCKED_LINK_UNDECIDED]} — {item.blocked_reason}", + "blocked_kind": BLOCKED_LINK_UNDECIDED, + } + ) + continue if (item.composite_parts or item.composite_not_ready) and not item.work_item_code: # 묶음 줄 — 품셈에 그 공종이 없어 **조각을 합쳐** 한 줄로 세운다 # (옹벽 = 타설 + 거푸집 + 철근 + 잡석). 「코드 없음」으로 세면 안 된다. @@ -577,6 +592,8 @@ def bill_summary(result: BillResult) -> dict[str, Any]: "direct_labor_krw": str(result.direct_labor_krw), "direct_expense_krw": str(result.direct_expense_krw), "notes": result.notes, + # 연결고리 표 미판정 행 — 열쇠가 비어 줄에 못 닿는 것도 드러냄(조용히 안 버림) + "link_undecided": undecided_links(), } diff --git a/B09_Estimation/B09_Estimation_BillOfQuantities_Input.py b/B09_Estimation/B09_Estimation_BillOfQuantities_Input.py index a48cfd2b..23112e39 100644 --- a/B09_Estimation/B09_Estimation_BillOfQuantities_Input.py +++ b/B09_Estimation/B09_Estimation_BillOfQuantities_Input.py @@ -14,7 +14,11 @@ from dataclasses import dataclass from decimal import Decimal from typing import Any -from common_util.common_util_work_item_key import work_item_key +from common_util.common_util_work_item_key import work_item_code_of, work_item_key +from common_util.common_util_work_item_link import choose + +#: 연결고리 표 미판정 — 산림·건설 어느 품을 쓸지 근거가 없어 **안 고른** 줄(금액에 안 듦). +BLOCKED_LINK_UNDECIDED = "link_undecided" _ZERO = Decimal(0) @@ -121,6 +125,23 @@ def _decimal(value: Any, default: Decimal | None = _ZERO) -> Decimal | None: return Decimal(str(value)) +def _linked(row: dict[str, Any]) -> dict[str, Any]: + """열쇠 길목 + **연결고리 표** — 산림·건설 어느 품을 쓸지 표가 정함(고시 총칙 7 · 표 `policy`). + + 둘 다 있는 공종이면 표가 적은 우선 품셈 열쇠·목차 코드로 바꿈 · 미판정이면 막음(사유 그대로). + ⚠ 이미 B08 이 막아 보낸 줄은 그 까닭을 덮지 않음. + """ + edition = row.get("pum_edition") or None + key = row.get("work_item_key") or work_item_key(row.get("work_item_code"), edition) + choice = choose(key) + out = {"work_item_key": key or "", "work_item_code": row.get("work_item_code")} + if choice.blocked and not row.get("blocked_reason"): + out.update(blocked_reason=choice.blocked, blocked_kind=BLOCKED_LINK_UNDECIDED) + elif choice.key and choice.key != key: + out.update(work_item_key=choice.key, work_item_code=work_item_code_of(choice.key, edition)) + return out + + def parse_handoff(payload: dict[str, Any]) -> tuple[list[HandoffWorkItem], list[HandoffMaterial]]: """인계 응답을 우리 자료형으로 옮긴다. **모르는 칸을 채우지 않는다.**""" if "work_items" not in payload or "materials" not in payload: @@ -128,7 +149,7 @@ def parse_handoff(payload: dict[str, Any]) -> tuple[list[HandoffWorkItem], list[ work_items = [ HandoffWorkItem( - work_item_code=row.get("work_item_code"), + work_item_code=linked["work_item_code"], name=row.get("name", ""), spec=row.get("spec") or "", unit=row.get("unit") or "", @@ -147,19 +168,18 @@ def parse_handoff(payload: dict[str, Any]) -> tuple[list[HandoffWorkItem], list[ composite_parts=tuple(row.get("composite_parts") or ()), composite_not_ready=tuple(row.get("composite_not_ready") or ()), structure_kind=row.get("structure_kind") or "", - blocked_reason=row.get("blocked_reason") or "", - blocked_kind=row.get("blocked_kind") or "", + blocked_reason=row.get("blocked_reason") or linked.get("blocked_reason") or "", + blocked_kind=row.get("blocked_kind") or linked.get("blocked_kind") or "", variant_axis=row.get("variant_axis") or "", variant_value=str(row.get("variant_value") or ""), 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 "", + # 옛 인계(열쇠 없음)도 같은 길목으로 채움 · 연결고리 표가 고른 열쇠 + work_item_key=linked["work_item_key"], ) for row in payload["work_items"] + for linked in (_linked(row),) ] materials = [ HandoffMaterial( diff --git a/Z01_MasterData/Z01_MasterData_WorkItems.py b/Z01_MasterData/Z01_MasterData_WorkItems.py index 63487f1d..213586f6 100644 --- a/Z01_MasterData/Z01_MasterData_WorkItems.py +++ b/Z01_MasterData/Z01_MasterData_WorkItems.py @@ -99,6 +99,8 @@ def base_rows(kind: str) -> list[dict[str, Any]]: "step_weight": weight, # 「임도가 쓰는 것」 — 연결고리 표가 확정으로 이은 줄(판정은 그 표 · 감추지 않고 표시만) "link": LINKED if _key(item) in linked_keys() else "", + # 줄 구실(총칙·공종) — 이름은 axis_policy.json · 감추지 않고 표시(거르기 상자로 고름) + "axis_role": _role_names().get(str(item.get("axis_role") or ""), ""), "effective_date": d.get("pum_edition") or d.get("effective_date"), } ) diff --git a/common_util/common_util_work_item_key.py b/common_util/common_util_work_item_key.py index 7d61f3da..94ec26dd 100644 --- a/common_util/common_util_work_item_key.py +++ b/common_util/common_util_work_item_key.py @@ -56,6 +56,20 @@ def work_item_key(code: str | None, pum_edition: str | None = None) -> str | Non return found.pop() if len(found) == 1 else None +def work_item_code_of(key: str | None, pum_edition: str | None = None) -> str | None: + """불변 열쇠 → 그 판의 목차 코드(되짚기). 한 곳에서만 찾아질 때만.""" + if not key: + return None + if _STABLE_CODE.match(key): + return key + found = { + code + for (edition, code), each in _keys().items() + if each == key and (pum_edition is None or edition == pum_edition) + } + return found.pop() if len(found) == 1 else None + + def attach_keys(rows: Iterable[dict[str, Any]]) -> list[str]: """인계 줄마다 `work_item_key` 를 더함(코드 없는 줄은 `None` — 칸은 늘 있음 · 인계 계약) · 코드는 있는데 열쇠를 못 찾은 코드 목록을 돌려줌.""" diff --git a/common_util/common_util_work_item_link.py b/common_util/common_util_work_item_link.py new file mode 100644 index 00000000..e949ecda --- /dev/null +++ b/common_util/common_util_work_item_link.py @@ -0,0 +1,87 @@ +"""연결고리 표(`work_item_link`)를 읽는 **한 자리** — 산림·건설 어느 품셈 품을 쓸지 **표가 정함** +(2026-09-17 브레인 · 코덱스 크로스체크 ② 「연결고리 미소비」). + +고시 총칙 7 「본 품셈에 명시되지 않은 품은 타 부문 표준품셈 적용 · 유사 공종은 본 품셈 우선」 — +그 규칙은 표의 `policy`(`both_precedence` · `unconfirmed_action`)에 있고 코드는 **읽기만** 함(「산림 우선」을 코드에 안 박음). + + 확정 · both → `both_precedence` 쪽 열쇠(지금 표는 forest) + 확정 · 한쪽만 → 그 열쇠 그대로 + 미판정 → **안 고름** · 사유를 붙여 막음(`unconfirmed_action: do_not_link`) + 표에 없음 → 제 품셈 그대로(품셈에 명시된 품) +⚠ 미판정 행 가운데 열쇠가 빈 행(여러 절 묶음)은 줄에 못 닿음 — `undecided_links()` 로 목록만 드러냄. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from functools import lru_cache +from pathlib import Path +from typing import Any + +FOLDER = Path(__file__).resolve().parent.parent / "resources" / "data_work_item_link" +CONFIRMED = "확정" +#: 표 `policy.both_precedence` 낱말 → 그 행의 열쇠 칸 +_KEY_COLUMN = {"forest": "forest_key", "construction": "const_key"} + + +@dataclass(frozen=True) +class Choice: + """쓸 열쇠 · 막힌 까닭(있으면 금액을 안 세움).""" + + key: str | None + blocked: str = "" + + +@lru_cache(maxsize=4) +def _read(path: str, _mtime_ns: int) -> dict[str, Any]: + return json.loads(Path(path).read_text(encoding="utf-8")) + + +def _doc() -> dict[str, Any]: + found = sorted(FOLDER.glob("work_item_link_*.json")) if FOLDER.is_dir() else [] + return _read(str(found[-1]), found[-1].stat().st_mtime_ns) if found else {} + + +def choose(key: str | None) -> Choice: + """열쇠 하나(FW-·CW-) → 실제로 쓸 열쇠. 미판정 행에 걸리면 안 고르고 막음.""" + if not key: + return Choice(key) + doc = _doc() + policy = doc.get("policy") or {} + rows = [ + link + for link in doc.get("links") or [] + if key in (link.get("forest_key"), link.get("const_key")) + ] + for link in rows: + if link.get("status") != CONFIRMED: + return Choice( + None, + "연결고리 미판정 — 산림·건설 어느 품을 쓸지 근거가 없어 고르지 않음" + f"({link.get('evidence') or '근거 없음'})", + ) + for link in rows: + if link.get("branch") == "both": + column = _KEY_COLUMN.get(str(policy.get("both_precedence") or "")) + chosen = link.get(column) if column else None + if not chosen: + return Choice( + None, "연결고리 표에 둘 다 있는 줄의 우선 품셈이 안 적혔음 — 고르지 않음" + ) + return Choice(chosen) + return Choice(key) + + +def undecided_links() -> list[dict[str, str]]: + """미판정 행 — 열쇠가 비어 줄에 못 닿는 행도 드러냄(조용히 안 버림).""" + return [ + { + "branch": str(link.get("branch") or ""), + "forest_key": str(link.get("forest_key") or ""), + "const_key": str(link.get("const_key") or ""), + "evidence": str(link.get("evidence") or ""), + } + for link in _doc().get("links") or [] + if link.get("status") != CONFIRMED + ] diff --git a/resources/data_master_labels/labels_2026-01-01.json b/resources/data_master_labels/labels_2026-01-01.json index 9aae9769..127bfcd1 100644 --- a/resources/data_master_labels/labels_2026-01-01.json +++ b/resources/data_master_labels/labels_2026-01-01.json @@ -1421,6 +1421,13 @@ "visible": true, "note": "「연결됨」 = 연결고리 표(`work_item_link`)가 확정으로 이은 줄 — 임도가 쓰는 공종. 판정은 그 표 · 감추지 않고 표시만." }, + { + "key": "axis_role", + "name_ko": "줄 구실", + "unit": "", + "visible": true, + "note": "총칙·공종 — 잣대는 `axis_policy.json`(데스크탑 서브). 총칙도 지우지 않고 표시만(원가가 값표를 읽음)." + }, { "key": "effective_date", "name_ko": "적용 시작일", @@ -1491,6 +1498,13 @@ "visible": true, "note": "「연결됨」 = 연결고리 표(`work_item_link`)가 확정으로 이은 줄 — 임도가 쓰는 공종. 판정은 그 표 · 감추지 않고 표시만." }, + { + "key": "axis_role", + "name_ko": "줄 구실", + "unit": "", + "visible": true, + "note": "총칙·공종 — 잣대는 `axis_policy.json`(데스크탑 서브). 총칙도 지우지 않고 표시만(원가가 값표를 읽음)." + }, { "key": "effective_date", "name_ko": "적용 시작일", @@ -1944,7 +1958,7 @@ }, "counts": { "merged_tables": 17, - "merged_columns": 174, + "merged_columns": 176, "files": 41, "tables": 100, "columns": 532, diff --git a/resources/tester/test_work_item_link_choice.py b/resources/tester/test_work_item_link_choice.py new file mode 100644 index 00000000..99d560b6 --- /dev/null +++ b/resources/tester/test_work_item_link_choice.py @@ -0,0 +1,138 @@ +"""연결고리 표 소비 — 산림·건설 어느 품을 쓸지 **표가 정함** (2026-09-17 브레인 · 코덱스 크로스체크 ②). + +고시 총칙 7 「유사 공종은 본 품셈 우선」 이 규칙으로만 있고 아무도 안 읽던 자리. 못박는 것: + ① 둘 다 있는 공종은 표의 `both_precedence` 쪽 — 표를 바꾸면 고르는 쪽도 바뀜(코드에 안 박힘) + ② 미판정은 안 고르고 막음 · 사유 그대로 · 금액에 안 듦 + ③ 지금 인계 코드는 한 줄도 판정이 안 달라짐 = **금액 불변**(이행 전 금액은 `test_work_item_key_gate` 가 못박음) +""" + +from __future__ import annotations + +import json +import re +from pathlib import Path + +import pytest + +from B09_Estimation.B09_Estimation_BillOfQuantities import bill_summary, build_bill +from B09_Estimation.B09_Estimation_BillOfQuantities_Input import ( + BLOCKED_LINK_UNDECIDED, + parse_handoff, +) +from B09_Estimation.B09_Estimation_UnitPrice import build_unit_prices +from common_util import common_util_work_item_link as link +from common_util.common_util_work_item_key import MASTER_DIR, work_item_code_of, work_item_key + +MAPPING = MASTER_DIR.parent / "data_work_item_mapping" / "work_item_mapping_2026-01-01.json" + + +def _table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, links: list, precedence="forest" +) -> None: + folder = tmp_path / "data_work_item_link" + folder.mkdir(parents=True) + doc = { + "policy": {"both_precedence": precedence, "unconfirmed_action": "do_not_link"}, + "links": links, + } + (folder / "work_item_link_2026-01-01.json").write_text( + json.dumps(doc, ensure_ascii=False), encoding="utf-8" + ) + monkeypatch.setattr(link, "FOLDER", folder) + + +def test_지금_인계_코드는_판정이_안_달라짐_금액_불변() -> None: + codes = set( + re.findall(r'"((?:FP|AX-WK|AX-ST)-[0-9A-Za-z-]+)"', MAPPING.read_text(encoding="utf-8")) + ) + changed = [] + for code in sorted(codes): + key = work_item_key(code, "2026-01-01") + choice = link.choose(key) + if choice.key != key or choice.blocked: + changed.append((code, key, choice)) + assert changed == [] # 달라지면 그동안 잘못 고르고 있었다는 뜻 — 브레인에 먼저 알릴 것 + + +def test_둘_다_있으면_표가_적은_쪽() -> None: + """실제 표 — 산림 9-7-1 무근콘크리트 깨기 ↔ 건설 8-2-13 대형브레이커(E 0.35 ↔ 0.45).""" + assert link.choose("CW-00330").key == "FW-00260" + assert link.choose("FW-00260").key == "FW-00260" + assert link.choose("CW-00844").key == "CW-00844" # 건설에만(모르타르 배합) — 그대로 + assert link.choose("FW-00249").key == "FW-00249" # 표에 없음 — 제 품셈 + + +def test_우선_품셈은_코드가_아니라_표가_정함( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + rows = [{"forest_key": "FW-00260", "const_key": "CW-00330", "branch": "both", "status": "확정"}] + _table(tmp_path, monkeypatch, rows, precedence="construction") + assert link.choose("FW-00260").key == "CW-00330" + _table(tmp_path / "b", monkeypatch, rows, precedence="") + assert link.choose("FW-00260").key is None and link.choose("FW-00260").blocked + + +def test_미판정은_안_고르고_막음_금액에_안_듦( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + payload = { + "work_items": [ + {"work_item_code": "FP-09-03-02", "name": "토사깍기", "unit": "㎥", "quantity": 100}, + {"work_item_code": "FP-09-12-01", "name": "측구터파기", "unit": "㎥", "quantity": 7}, + ], + "materials": [], + } + build = build_unit_prices() + before = build_bill(json.loads(json.dumps(payload)), build=build) + ditch = work_item_key("FP-09-12-01", "2026-01-01") + rows = [ + { + "forest_key": ditch, + "const_key": "", + "branch": "both", + "status": "미판정", + "evidence": "시험 — 측구 터파기 건설 대응 미확인", + } + ] + _table(tmp_path, monkeypatch, rows) + items, _ = parse_handoff(json.loads(json.dumps(payload))) + assert ( + items[1].blocked_kind == BLOCKED_LINK_UNDECIDED and "시험 — 측구" in items[1].blocked_reason + ) + after = build_bill(json.loads(json.dumps(payload)), build=build) + blocked = [m for m in after.missing if m.get("blocked_kind") == BLOCKED_LINK_UNDECIDED] + assert [m["name"] for m in blocked] == ["측구터파기"] + ditch_amount = sum( + r.amount_krw or 0 for r in before.rows if not r.is_group and r.name == "측구터파기" + ) + assert ditch_amount > 0 and after.body_total_krw == before.body_total_krw - ditch_amount + assert bill_summary(after)["link_undecided"][0]["evidence"].startswith("시험") + + +def test_건설_열쇠로_온_줄은_표가_고른_산림_열쇠와_목차_코드로() -> None: + const_code = work_item_code_of("CW-00330") + assert const_code and const_code.startswith("CP-") + items, _ = parse_handoff( + { + "work_items": [ + { + "work_item_code": const_code, + "name": "대형브레이커", + "unit": "㎥", + "quantity": 1, + "pum_edition": "2026-01-01", + } + ], + "materials": [], + } + ) + assert (items[0].work_item_key, items[0].work_item_code) == ( + "FW-00260", + work_item_code_of("FW-00260"), + ) + assert items[0].work_item_code.startswith("FP-") + + +def test_미판정_행은_열쇠가_비어도_목록으로_드러남() -> None: + undecided = link.undecided_links() + assert undecided and all(row["evidence"] for row in undecided) diff --git a/resources/tester/test_z01_work_items.py b/resources/tester/test_z01_work_items.py index c42dca97..5141e552 100644 --- a/resources/tester/test_z01_work_items.py +++ b/resources/tester/test_z01_work_items.py @@ -146,6 +146,7 @@ def test_거르기는_자료에_있는_가름만_서버가_판정(client: TestCl ) assert rules["total"] > 0 assert all(r["name"].startswith("공통부문 › 적용기준") for r in rules["rows"]) + assert {r["axis_role"] for r in rules["rows"]} == {"총칙"} # 표시 칸 — 감추지 않고 보임 assert [f["key"] for f in _get(client, "work_item_forest", size=1)["filters"]] == [ "link",