diff --git a/B08_Quantity/B08_Quantity_Engine_Handoff.py b/B08_Quantity/B08_Quantity_Engine_Handoff.py index bcbe42cc..a5614914 100644 --- a/B08_Quantity/B08_Quantity_Engine_Handoff.py +++ b/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 diff --git a/B09_Estimation/B09_Estimation_BillOfQuantities_Input.py b/B09_Estimation/B09_Estimation_BillOfQuantities_Input.py index 1b2ae559..a48cfd2b 100644 --- a/B09_Estimation/B09_Estimation_BillOfQuantities_Input.py +++ b/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"] ] diff --git a/common_util/common_util_work_item_key.py b/common_util/common_util_work_item_key.py new file mode 100644 index 00000000..7d61f3da --- /dev/null +++ b/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) diff --git a/resources/tester/test_b08_handoff_contract.py b/resources/tester/test_b08_handoff_contract.py index 8f4d9d8a..56429d5c 100644 --- a/resources/tester/test_b08_handoff_contract.py +++ b/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` 가 들어가면 안 된다(축이 다르다). diff --git a/resources/tester/test_work_item_key_gate.py b/resources/tester/test_work_item_key_gate.py new file mode 100644 index 00000000..7df9eeb0 --- /dev/null +++ b/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 # 금액이 실제로 선 줄로 잼