From c2e3c4cda2bc7de22b715af7ef7d063f941a32d4 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 13 Sep 2026 22:20:26 +0900 Subject: [PATCH] =?UTF-8?q?feat(b09):=20=EA=B5=AC=EC=A1=B0=EB=AC=BC?= =?UTF-8?q?=EB=8F=84=20=EC=9D=BC=EC=9C=84=EB=8C=80=EA=B0=80=EB=A5=BC=20?= =?UTF-8?q?=EB=82=B4=EC=97=AD=EC=9C=BC=EB=A1=9C=20=EB=B0=9B=EB=8A=94=20?= =?UTF-8?q?=EB=AC=B8=20=E2=80=94=20=ED=98=B8=ED=91=9C=20B-AX-ST-<8hex>#?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 양식 일위대가로 셀 구조물도 장은 인계 줄 하나(코드 AX-ST · 갈래 = 제원 키 · 수량 연장 m) - 양식이 품은 줄(기초잡석)은 따로 인계 안 함 — 이중계상 막음 - B09 내역이 자기 단가표로 B08 일위대가 엔진을 돌려 금액 · 못 받거나 막히면 사유 - 수동 단가도 씀 — 내역 요약에 미확정 N건 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016VBGFXB9AbJBwXP19z75Qq --- B08_Quantity/B08_Quantity_Engine_Handoff.py | 11 +- .../B08_Quantity_Engine_Handoff_Rows.py | 16 +- .../B08_Quantity_Engine_Handoff_Trench.py | 12 +- .../B08_Quantity_Engine_StructurePriceLink.py | 163 ++++++++++++++++++ .../B08_Quantity_Engine_StructureUnitPrice.py | 16 +- B08_Quantity/B08_Quantity_Router_Material.py | 73 +++++++- .../B09_Estimation_BillOfQuantities.py | 27 ++- .../B09_Estimation_BillOfQuantities_Rows.py | 59 +++++++ B09_Estimation/B09_Estimation_Router.py | 12 +- .../tester/test_b09_structure_price_link.py | 128 ++++++++++++++ 10 files changed, 505 insertions(+), 12 deletions(-) create mode 100644 B08_Quantity/B08_Quantity_Engine_StructurePriceLink.py create mode 100644 resources/tester/test_b09_structure_price_link.py diff --git a/B08_Quantity/B08_Quantity_Engine_Handoff.py b/B08_Quantity/B08_Quantity_Engine_Handoff.py index 090e4b1a..0423f76e 100644 --- a/B08_Quantity/B08_Quantity_Engine_Handoff.py +++ b/B08_Quantity/B08_Quantity_Engine_Handoff.py @@ -112,8 +112,13 @@ def build_handoff( bench_cut_depth_m: float | None = None, structure_trench_water: str | None = None, stand_volume_class: str | None = None, + priced_sheets: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: - """B09 가 그대로 받는 모양. 없는 표는 건너뛰되 **빈 표와 구별해 적는다**.""" + """B09 가 그대로 받는 모양. 없는 표는 건너뛰되 **빈 표와 구별해 적는다**. + + `priced_sheets` — 양식 일위대가로 셀 구조물도 장(`StructurePriceLink.priced_sheets`). + 그 장의 구조물은 호표 `AX-ST` 줄 하나로 가고, 일위대가가 품은 줄은 따로 안 감. + """ table = mapping or load_mapping() work_items: list[dict[str, Any]] = [] unmatched: list[str] = [] @@ -135,7 +140,7 @@ def build_handoff( # 띠·이동에서만 운반이 만들어져 사토 잔량이 어디에도 안 실렸다. work_items.extend(spoil_haul_rows(haul_table, table)) if unit_quantity_table: - rows, misses = _structure_rows(unit_quantity_table, table) + rows, misses = _structure_rows(unit_quantity_table, table, priced_sheets) work_items.extend(rows) unmatched.extend(misses) # 구조물이 낸 터파기·되메우기·잔토 — **공종 축으로 올린다.** @@ -145,7 +150,7 @@ def build_handoff( work_items.extend(rows) unmatched.extend(misses) # 기초잡석 — 버림이 선 구조물에 함께 서는 공종(확정 3차 ②). 묶음 구조물은 제외한다. - rows, misses = rubble_base_rows(unit_quantity_table, table) + rows, misses = rubble_base_rows(unit_quantity_table, table, priced_sheets) work_items.extend(rows) unmatched.extend(misses) diff --git a/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py b/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py index 1e219791..b0fe9ba5 100644 --- a/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py +++ b/B08_Quantity/B08_Quantity_Engine_Handoff_Rows.py @@ -342,13 +342,19 @@ def _component_billing( def _structure_rows( - unit_quantity_table: dict[str, Any], mapping: WorkItemMapping + unit_quantity_table: dict[str, Any], + mapping: WorkItemMapping, + priced_sheets: list[dict[str, Any]] | None = None, ) -> tuple[list[dict[str, Any]], list[str]]: """구조물 줄 — **작업 공종 하나**로 선다. ⚠ 전개 성분(터파기·야면석·모르터)은 여기 오지 않는다. 구조물 한 기가 내역 한 줄이고, 그 전개는 토공 합산·자재총괄·일위대가로 갈린다(8-2 이중계상 함정). + ⭐ 양식 일위대가로 셀 장(`priced_sheets`)에 든 구조물은 **호표 `AX-ST` 줄**로 감(PLAN 6장 ②). """ + from B08_Quantity.B08_Quantity_Engine_StructurePriceLink import by_structure, template_row + + priced = by_structure(priced_sheets or []) rows: list[dict[str, Any]] = [] unmatched: list[str] = [] for structure in unit_quantity_table.get("structures") or []: @@ -400,7 +406,11 @@ def _structure_rows( {"axis": axis, "value": (structure.get("options") or {}).get(axis)} for axis in entry.get("secondary_axes") or [] ] - if code is None and composite is None: + # 양식 일위대가로 셀 구조물 — 막힘은 일위대가 표가 가림(전개 갈래 사유는 안 셈). + sheet_entry = priced.get(str(structure.get("structure_id"))) + if sheet_entry is not None: + pass + elif code is None and composite is None: unmatched.append(f"{wording_type_label(type_id)} — 품셈 공종을 아직 못 이었습니다") elif entry.get("class_from") in ("back_length", "bond") and class_key is None: unmatched.append(f"{wording_type_label(type_id)} — {class_basis}") @@ -476,6 +486,8 @@ def _structure_rows( "origin": ORIGIN_STRUCTURE, } ) + if sheet_entry is not None: + rows[-1] = template_row(rows[-1], structure, sheet_entry) return rows, unmatched diff --git a/B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py b/B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py index 1c1f3b2f..0e7ac65d 100644 --- a/B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py +++ b/B08_Quantity/B08_Quantity_Engine_Handoff_Trench.py @@ -149,18 +149,28 @@ def _row(**fields: Any) -> dict[str, Any]: def rubble_base_rows( - unit_quantity_table: dict[str, Any], mapping: WorkItemMapping + unit_quantity_table: dict[str, Any], + mapping: WorkItemMapping, + priced_sheets: list[dict[str, Any]] | None = None, ) -> tuple[list[dict[str, Any]], list[str]]: """기초잡석(품셈 12-25) — 구조물 전개가 낸 성분을 **한 줄로** 올린다. ⚠ **묶음으로 서는 구조물은 건너뛴다** — 옹벽 묶음에 이미 `FP-12-25` 조각이 있어 여기서 또 세우면 같은 잡석을 두 번 센다(타설 줄에서 이미 겪은 자리). + ⚠ **양식 일위대가가 기초잡석을 품은 구조물도 건너뛴다**(PLAN 6장 ② 브레인 판정) — + 호표 `AX-ST` 금액에 이미 들어 있음. """ + from B08_Quantity.B08_Quantity_Engine_StructurePriceLink import by_structure + + priced = by_structure(priced_sheets or []) total = 0.0 bases: list[str] = [] for structure in unit_quantity_table.get("structures") or []: if mapping.composite_for(str(structure.get("type_id") or "")): continue + sheet_entry = priced.get(str(structure.get("structure_id"))) + if sheet_entry is not None and RUBBLE_GROUP in sheet_entry["covered"]: + continue for component in structure.get("components") or []: if str(component.get("name") or "") != RUBBLE_GROUP: continue diff --git a/B08_Quantity/B08_Quantity_Engine_StructurePriceLink.py b/B08_Quantity/B08_Quantity_Engine_StructurePriceLink.py new file mode 100644 index 00000000..6f7aaede --- /dev/null +++ b/B08_Quantity/B08_Quantity_Engine_StructurePriceLink.py @@ -0,0 +1,163 @@ +"""구조물도 일위대가 → B09 내역 **받는 문** (PLAN 6장 첫 일감 ② · 2026-09-13 브레인 판정). + +⭐ 종전엔 구조물도가 세운 일위대가(`B-AX-ST-*`)가 인계로 **한 번도 안 나갔음** — 돌쌓기는 + `FP-13-04-05` ㎡ 단가만 가고 모르터·기초잡석은 따로거나 빠졌음. 사슬의 진짜 빈칸. + +한 벌로 두는 것 — 인계(B08)와 내역(B09)이 **같은 목록·같은 키**를 봄 + 양식 일위대가로 셀 장 → `priced_sheets` (장마다 코드 · 제원 키 · 구성원 · 품은 줄) + 인계 줄 → `template_row` (코드 AX-ST · 갈래 = 제원 키 · 수량 = 연장 m) + 내역 금액 → `price_sheets` (B09 단가표로 B08 일위대가 엔진을 돌림 — 코드 한 벌) + +⚠ 호표 코드 `B-AX-ST-<8hex>#<키>` — 키는 **사람이 읽는 제목이 아니라** 양식 제원 칸 값을 칸 이름 + 차례로 이은 것(연장 L 뺌) + `normalize_variant_key`. 제목 글자가 바뀌어도 코드는 안 바뀜. +⚠ 양식 일위대가가 품은 원단위 줄(돌쌓기·모르터·기초잡석)은 **따로 인계하지 않음** — `covered`. +⚠ 금액 정본은 B09 — 여기 `price_sheets` 는 B09 가 넘긴 단가표로만 셈. 수동 단가도 입력으로 씀. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterable +from typing import Any + +#: 인계 줄 갈래 축 이름 — 「구조물도 장 제원」. +VARIANT_AXIS = "structure_sheet" + + +def _key_value(value: Any) -> str: + if value is None: + return "" + if isinstance(value, (int, float)) and not isinstance(value, bool): + return format(float(value), ".6g") + return str(value).strip() + + +def sheet_variant_key(template: dict[str, Any], sheet: dict[str, Any]) -> str: + """장의 **제원 키** — 양식 칸 이름 차례로 `이름=값` 을 `|` 로 이음(연장 칸 뺌: m당).""" + values = (sheet.get("formula_sheet") or {}).get("vars") or {} + specs = template.get("vars") or {} + return "|".join( + f"{name}={_key_value(values.get(name))}" + for name in sorted(specs) + if specs[name].get("source") != "length_m" + ) + + +def structure_price_code(code: str, key: str) -> str: + """호표 코드 `B-AX-ST-<8hex>#<정규화 키>` — 인계·내역이 이것 하나로 만남.""" + from B09_Estimation.B09_Estimation_UnitPrice import normalize_variant_key + + return f"B-{code}#{normalize_variant_key(key)}" + + +def priced_sheets( + sheets: Iterable[dict[str, Any]], + templates: dict[str, dict[str, Any]], + row_edits: dict[str, Any] | None = None, +) -> list[dict[str, Any]]: + """양식 일위대가로 셀 장 — 양식(고친 줄 조합 얹음)에 일위대가 줄과 코드가 있는 장만.""" + from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import with_rows + + found = [] + for sheet in sheets: + type_id = str(sheet.get("type_id") or "") + base = templates.get(type_id) + if not base or not base.get("code") or not sheet.get("formula_sheet"): + continue + template = with_rows(base, (row_edits or {}).get(type_id)) + rows = template["unit_price"]["rows"] + if not rows: + continue + names = {row.get("seq"): str(row.get("name") or "") for row in template.get("rows") or []} + found.append( + { + "type_id": type_id, + "code": str(template["code"]), + "name": str(template.get("name") or type_id), + "title": str(sheet.get("title") or ""), + "key": sheet_variant_key(template, sheet), + "structure_ids": [ + str(member["structure_id"]) + for member in sheet.get("members") or [] + if member.get("structure_id") + ], + # 일위대가가 품은 원단위 줄 이름 — 따로 인계하면 두 번 셈. + "covered": sorted( + {names[r["from_row"]] for r in rows if r.get("from_row") in names} + ), + "template": template, + "sheet": sheet, + } + ) + return found + + +def by_structure(entries: Iterable[dict[str, Any]]) -> dict[str, dict[str, Any]]: + return {sid: entry for entry in entries for sid in entry["structure_ids"]} + + +def template_row(row: dict[str, Any], structure: dict[str, Any], entry: dict[str, Any]) -> dict: + """전개 인계 줄(같은 칸 29개 계약)을 **양식 일위대가 줄**로 바꿔 씀 — 코드·갈래·수량·단위만.""" + length = float(structure.get("length_m") or 0.0) + in_bill = length > 0 + return { + **row, + "work_item_code": entry["code"], + "name": entry["name"], + "spec": entry["title"], + "spec_detail": entry["title"], + "unit": "m", + "quantity": length, + "composite_parts": None, + "composite_not_ready": None, + "structure_kind": None, + "variant_axis": VARIANT_AXIS, + "variant_value": entry["key"], + "secondary_axes": None, + "spec_class": None, + "spec_class_basis": f"구조물도 양식 일위대가 — 품은 줄: {'·'.join(entry['covered'])}", + # 막힘은 일위대가 표가 정함(B09 가 셈) — 전개 쪽 갈래 사유는 여기 안 실음. + "blocked_kind": None if in_bill else row.get("blocked_kind"), + "blocked_reason": "" if in_bill else row.get("blocked_reason") or "연장이 0 이라 안 세움", + "in_bill": in_bill, + "in_bill_reason": "" if in_bill else "연장이 0 이라 내역에 안 세움", + } + + +def price_sheets( + entries: Iterable[dict[str, Any]], + book: Any, + find_variant: Callable[[str, str], str | None], + library: Iterable[dict[str, Any]], + manual_all: dict[str, Any] | None = None, +) -> dict[str, dict[str, Any]]: + """B09 내역이 쓸 호표 금액 `{B-AX-ST-…#키: {money, blocked, unconfirmed, reasons}}`. + + `money` 는 **안 자른** 단위(m)당 3분할 — 자르기는 내역 줄 자리 규칙(B09)이 함. + """ + from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import unit_price_money + + prices: dict[str, dict[str, Any]] = {} + library = list(library) + for entry in entries: + assembled = unit_price_money( + entry["template"], + entry["sheet"], + book, + find_variant, + library, + (manual_all or {}).get(entry["type_id"]) or {}, + ) + if assembled is None: + continue + table, money = assembled + prices[structure_price_code(entry["code"], entry["key"])] = { + "money": money, + "blocked": table["blocked"], + "unconfirmed": table["unconfirmed"], + "reasons": [ + f"{row['name']}: {row['reason']}" + for row in table["rows"] + if row["reason"] and not row["skipped"] and "total" not in row + ], + } + return prices diff --git a/B08_Quantity/B08_Quantity_Engine_StructureUnitPrice.py b/B08_Quantity/B08_Quantity_Engine_StructureUnitPrice.py index 40d383ab..ab9e471a 100644 --- a/B08_Quantity/B08_Quantity_Engine_StructureUnitPrice.py +++ b/B08_Quantity/B08_Quantity_Engine_StructureUnitPrice.py @@ -286,12 +286,24 @@ def unit_price_table( `library` — 하위 구조물 일위대가(`B-AX-ST-*`)를 찾을 양식들(프로젝트에 박힌 것 + 프로그램 기본). `manual` — 이 프로젝트의 수동 단가 `{seq: {material, labor, expense, source, entered_at}}`. """ + assembled = unit_price_money(template, sheet, book, find_variant, library, manual) + return assembled[0] if assembled else None + + +def unit_price_money( + template: dict[str, Any], + sheet: dict[str, Any], + book: Any, + find_variant: Callable[[str, str], str | None], + library: Iterable[dict[str, Any]] = (), + manual: dict[str, dict[str, Any]] | None = None, +) -> tuple[dict[str, Any], Any] | None: + """`unit_price_table` 과 같되 **안 자른 단위당 3분할**도 — B09 내역 줄이 이 값에 수량을 곱함.""" ctx = _Context( book, find_variant, {str(t["code"]): t for t in library if t.get("code")}, manual or {} ) code = str(template.get("code") or "") - assembled = _assemble(template, sheet, ctx, 1, (code,) if code else ()) - return assembled[0] if assembled else None + return _assemble(template, sheet, ctx, 1, (code,) if code else ()) def with_rows(template: dict[str, Any], rows: list[dict[str, Any]] | None) -> dict[str, Any]: diff --git a/B08_Quantity/B08_Quantity_Router_Material.py b/B08_Quantity/B08_Quantity_Router_Material.py index 99070be4..acfb3708 100644 --- a/B08_Quantity/B08_Quantity_Router_Material.py +++ b/B08_Quantity/B08_Quantity_Router_Material.py @@ -405,10 +405,11 @@ async def get_handoff(project_id: UUID) -> JSONResponse: structures, names, skipped = _collect_structures(project_root) settings = quantity_settings(project_root) + modes = await _section_modes(project_id) unit_table = build_unit_table( structures, names, - await _section_modes(project_id), + modes, await _ground_types(project_id), settings.get("rubble_base_thickness_m"), structure_formulas=settings.get("structure_formula_overrides"), @@ -463,6 +464,8 @@ async def get_handoff(project_id: UUID) -> JSONResponse: structure_trench_water=settings.get("structure_trench_water"), # 임목축적 등급 — 지장목제거 뿌리뽑기(제근 9-21)의 품 갈래. 안 넣으면 B09 가 후보를 보임. stand_volume_class=settings.get("stand_volume_class"), + # 구조물도 양식 일위대가로 셀 장 — 그 구조물은 호표 `AX-ST` 줄 하나로(PLAN 6장 ②). + priced_sheets=_priced_sheets(project_root, unit_table, modes, settings), ) handoff["summary"] = summarize(handoff) handoff["skipped_structures"] = skipped @@ -470,6 +473,74 @@ async def get_handoff(project_id: UUID) -> JSONResponse: return JSONResponse(content=handoff) +def _priced_sheets( + project_root: str, + unit_table: dict[str, Any], + section_modes: dict[float, str] | None, + settings: dict[str, Any], +) -> list[dict[str, Any]]: + """양식 일위대가로 셀 구조물도 장 — 구조물도 탭과 **같은 접기**(`build_standard_sheets` → + `apply_templates`)로 세움. 인계와 내역 금액이 이 한 목록을 봄.""" + from B08_Quantity.B08_Quantity_Engine_StructurePriceLink import priced_sheets + from B08_Quantity.B08_Quantity_Engine_StructureSheet import build_standard_sheets + from B08_Quantity.B08_Quantity_Engine_StructureTemplate import apply_templates, template_of + from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import ROWS_KEY + + embedded = project_templates(project_root) + payload = build_standard_sheets(unit_table, section_modes) + apply_templates(payload, settings, embedded) + sheets = payload.get("sheets") or [] + templates = { + type_id: template + for type_id in {str(sheet.get("type_id") or "") for sheet in sheets} + if (template := template_of(type_id, embedded)) + } + return priced_sheets(sheets, templates, settings.get(ROWS_KEY)) + + +async def structure_bill_prices(project_id: UUID, build: Any) -> dict[str, dict[str, Any]]: + """B09 내역이 부름 — 구조물도 호표 금액. **B09 가 넘긴 단가표**로 B08 일위대가 엔진을 돌림. + + ⚠ 금액 정본은 B09 — 여기서 단가표를 따로 짓지 않음. 수동 단가(프로젝트 값)도 입력으로 씀. + ⚠ 인계(`get_handoff`)와 **같은 전개**를 한 번 더 돎 — 두 요청이라 한 벌을 못 나눔. + """ + # ponytail: 인계와 전개를 두 번 돎 — 느려지면 인계 응답에 장 목록을 실어 한 번으로. + from B08_Quantity.B08_Quantity_Engine_StructureLibrary import available_templates + from B08_Quantity.B08_Quantity_Engine_StructurePriceLink import price_sheets + from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import MANUAL_KEY, ROWS_KEY, with_rows + from B09_Estimation.B09_Estimation_UnitPrice import find_variant_code + + stored_path = await run_with_connection(get_project_storage_relative_path, project_id) + project_root = resolve_stored_project_path(stored_path) + structures, names, _skipped = _collect_structures(project_root) + settings = quantity_settings(project_root) + modes = await _section_modes(project_id) + unit_table = build_unit_table( + structures, + names, + modes, + await _ground_types(project_id), + settings.get("rubble_base_thickness_m"), + structure_formulas=settings.get("structure_formula_overrides"), + structure_templates=project_templates(project_root), + ) + entries = _priced_sheets(project_root, unit_table, modes, settings) + if not entries: + return {} + row_edits = settings.get(ROWS_KEY) or {} + library = [ + with_rows(template, row_edits.get(template.get("type_id"))) + for template in available_templates(project_root) + ] + return price_sheets( + entries, + build.book, + lambda code, value: find_variant_code(code, value, build), + library, + settings.get(MANUAL_KEY) or {}, + ) + + def _pipe_table( project_root: str, pipe_lengths: list[dict[str, Any]], diff --git a/B09_Estimation/B09_Estimation_BillOfQuantities.py b/B09_Estimation/B09_Estimation_BillOfQuantities.py index ee755457..affa698c 100644 --- a/B09_Estimation/B09_Estimation_BillOfQuantities.py +++ b/B09_Estimation/B09_Estimation_BillOfQuantities.py @@ -45,6 +45,8 @@ _ZERO = Decimal(0) #: 총 절취량으로 셀 공종 — 9-3 토사깎기 · 9-4 암절취 · 9-5 발파암(인계 대응표 「흙깎기」 셋). #: ⚠ 판에 묶인 절 번호임(명세 17장) — 판이 바뀌면 대응표와 함께 고칠 것. CUT_CODE_PREFIXES = ("FP-09-03", "FP-09-04", "FP-09-05") +#: 구조물도 양식 호표 코드 머리(명세 2장 자체 확장) — 금액은 B08 일위대가 엔진을 B09 단가표로 셈. +STRUCTURE_PRICE_PREFIX = "AX-ST-" #: 자재 공급 구분이 안 갈린 값. B08 이 실제로 이 값을 보낸다(2026-09-08 실물 확인). #: **관급자재대에도, 도급 재료비에도 넣지 않는다** — 어느 쪽에 넣어도 총액이 틀린다. @@ -228,6 +230,8 @@ class BillResult: price_basis: Any = None #: 자재대 표 — 사급·관급·미정 셋으로 갈린다(PLAN 8-7 「금액은 B09」). material_sheet: Any = None + #: 수동 단가로 선 자리 — 내역서 끝 「미확정 N건」(PLAN 확정 ⑦). 줄마다 `{name, count}`. + unconfirmed: list[dict[str, Any]] = field(default_factory=list) @property def direct_material_krw(self) -> Decimal: @@ -357,6 +361,7 @@ from B09_Estimation.B09_Estimation_BillOfQuantities_Rows import ( # noqa: E402 _excluded_row, _leaf_row, _material_row, + _structure_price_row, ) @@ -365,8 +370,13 @@ def build_bill( *, build: UnitPriceBuild | None = None, master: dict[str, Any] | None = None, + structure_prices: dict[str, dict[str, Any]] | None = None, ) -> BillResult: - """인계 응답 한 벌을 ④ 예산내역서 한 장으로 접는다.""" + """인계 응답 한 벌을 ④ 예산내역서 한 장으로 접는다. + + `structure_prices` — 구조물도 호표 금액 `{B-AX-ST-…#키: …}`(B08 `structure_bill_prices`, + **이 `build` 단가표로** 셈). 안 주면 호표 줄은 금액 없이 사유와 함께 섬. + """ work_items, materials = parse_handoff(payload) unit_prices = build or cached_build() index = _master_index(master or load_work_item_master()) @@ -377,6 +387,7 @@ def build_bill( used: list[tuple[tuple[int, ...], HandoffWorkItem, list[_MasterNode]]] = [] orphans: list[HandoffWorkItem] = [] composites: list[HandoffWorkItem] = [] + templated: list[HandoffWorkItem] = [] for item in work_items: if not item.in_bill: # ⚠ 코드 유무보다 **먼저** 가른다. 보정량계는 공종코드가 없어서가 아니라 @@ -397,6 +408,10 @@ def build_bill( } ) continue + if str(item.work_item_code or "").startswith(STRUCTURE_PRICE_PREFIX): + # 구조물도 호표(PLAN 6장 ②) — 공종 마스터에 없는 자체 확장 코드라 나무에 안 끼움. + templated.append(item) + continue if (item.composite_parts or item.composite_not_ready) and not item.work_item_code: # 묶음 줄 — 품셈에 그 공종이 없어 **조각을 합쳐** 한 줄로 세운다 # (옹벽 = 타설 + 거푸집 + 철근 + 잡석). 「코드 없음」으로 세면 안 된다. @@ -455,6 +470,14 @@ def build_bill( row_of[id(item)] = _composite_row(str(counters[""]), item, unit_prices, result) result.rows.append(row_of[id(item)]) + # ── 1-3) 구조물도 호표 줄 ───────────────────────────────────────────────── + for item in templated: + counters[""] = counters.get("", 0) + 1 + row_of[id(item)] = _structure_price_row( + str(counters[""]), item, structure_prices or {}, result + ) + result.rows.append(row_of[id(item)]) + # ── 2) 공종을 못 고른 줄 — 이름째 남긴다 ──────────────────────────────────── for item in orphans: # 구조물 줄은 사유가 다르다 — 품셈에 그 공종이 없어 **전개식(원단위)** 이 있어야 @@ -569,6 +592,8 @@ def bill_summary(result: BillResult) -> dict[str, Any]: "material_rows": len(result.material_rows), "material_sheet": result.material_sheet.as_dict() if result.material_sheet else None, "missing": result.missing, + "unconfirmed": result.unconfirmed, + "unconfirmed_count": sum(int(entry["count"]) for entry in result.unconfirmed), "body_total_krw": str(result.body_total_krw), "direct_material_krw": str(result.direct_material_krw), "direct_labor_krw": str(result.direct_labor_krw), diff --git a/B09_Estimation/B09_Estimation_BillOfQuantities_Rows.py b/B09_Estimation/B09_Estimation_BillOfQuantities_Rows.py index 350b0df9..582aa524 100644 --- a/B09_Estimation/B09_Estimation_BillOfQuantities_Rows.py +++ b/B09_Estimation/B09_Estimation_BillOfQuantities_Rows.py @@ -95,6 +95,65 @@ def _composite_row( return row +def _structure_price_row( + item_no: str, + item: HandoffWorkItem, + structure_prices: dict[str, dict], + result: BillResult, +) -> BillRow: + """구조물도 호표 줄(PLAN 6장 ②) — m당 금액은 B08 일위대가 엔진을 **이 내역의 단가표**로 돌린 값. + + ⚠ 일위대가에 막힌 줄이 있으면 **금액을 안 세움** — 절반짜리 단가가 제일 위험(묶음 줄과 같음). + ⚠ 수동 단가로 선 줄은 금액을 세우되 「미확정」으로 셈 — 구조물도 화면과 내역이 같은 값. + """ + from B08_Quantity.B08_Quantity_Engine_StructurePriceLink import structure_price_code + + ref = structure_price_code(str(item.work_item_code), item.variant_value) + row = BillRow( + item_no=item_no, + level=1, + code=item.work_item_code, + name=item.name, + spec=item.spec, + unit=item.unit, + quantity=item.quantity, + in_bill=item.in_bill, + ) + entry = structure_prices.get(ref) + reason = "" + if entry is None: + reason = f"구조물도 일위대가 {ref} 를 못 받음 — 구조물도 탭에서 그 장의 일위대가를 확인" + elif entry["blocked"]: + reason = f"구조물도 일위대가 미완 — 막힌 줄 {entry['blocked']}: " + "; ".join( + entry["reasons"][:3] + ) + if reason: + row.add_note("unit_price_krw", reason) + result.missing.append( + { + "name": row.name, + "code": ref, + "unit": row.unit, + "quantity": str(item.quantity), + "reason": reason, + } + ) + return row + + money = entry["money"] + line = money.scaled(item.quantity) + row.unit_price_krw = round_at(money.total, OutputPlace.UNIT_PRICE_ROW) + row.amount_krw = round_at(line.total, OutputPlace.BOQ_ROW) + row.material_krw = line.material + row.labor_krw = line.labor + row.expense_krw = line.expense + row.add_note("unit_price_krw", f"호표 {ref}") + if entry["unconfirmed"]: + row.add_note("unit_price_krw", f"⚠ 수동 단가 {entry['unconfirmed']}건 미확정") + result.unconfirmed.append({"name": row.name, "code": ref, "count": entry["unconfirmed"]}) + return row + + def _excluded_row(item: HandoffWorkItem) -> BillRow: """`in_bill=false` 줄. **수량만 보이고 단가·금액을 안 붙인다.** diff --git a/B09_Estimation/B09_Estimation_Router.py b/B09_Estimation/B09_Estimation_Router.py index 8b93f3fd..ed33fc40 100644 --- a/B09_Estimation/B09_Estimation_Router.py +++ b/B09_Estimation/B09_Estimation_Router.py @@ -753,8 +753,9 @@ async def get_bill(project_id: UUID) -> JSONResponse: ⚠ 단가가 없거나 밑수를 모르는 줄은 **0 으로 안 때우고** `missing` 으로 드러낸다 — 화면이 그 목록을 그대로 보인다. """ - from B08_Quantity.B08_Quantity_Router_Material import get_handoff + from B08_Quantity.B08_Quantity_Router_Material import get_handoff, structure_bill_prices + build = cached_build() try: response = await get_handoff(project_id) payload = json.loads(bytes(response.body).decode("utf-8")) @@ -767,9 +768,16 @@ async def get_bill(project_id: UUID) -> JSONResponse: if "work_items" not in payload: # B08 이 오류 응답을 준 경우 — 그 사유를 그대로 넘긴다(감추지 않는다). return JSONResponse(status_code=502, content={"status": "error", **payload}) + try: + # 구조물도 호표 금액 — **이 내역과 같은 단가표**로(PLAN 6장 ②). + structure_prices = await structure_bill_prices(project_id, build) + except Exception: + # 못 세면 호표 줄이 금액 없이 「못 받음」 사유로 섬 — 조용히 0 원 안 됨. + logger.exception("B09 내역서 — 구조물도 호표 금액 실패: project_id=%s", project_id) + structure_prices = {} try: - result = build_bill(payload) + result = build_bill(payload, build=build, structure_prices=structure_prices) except DoubleCountError as error: # 이중계상 감시에 걸린 경우 — 표를 그리지 않고 멈춘다. logger.warning("B09 내역서 이중계상 감지: project_id=%s, %s", project_id, error) diff --git a/resources/tester/test_b09_structure_price_link.py b/resources/tester/test_b09_structure_price_link.py new file mode 100644 index 00000000..bbeb6dde --- /dev/null +++ b/resources/tester/test_b09_structure_price_link.py @@ -0,0 +1,128 @@ +"""구조물도 일위대가 → B09 내역 받는 문 (2026-09-13, PLAN 6장 첫 일감 ② · 브레인 판정). + +겨누는 것 + ① 양식 일위대가로 셀 구조물은 인계 줄 하나 = 코드 AX-ST · 갈래 = 제원 키 · 수량 = 연장 m + ② 양식이 품은 기초잡석은 따로 인계 안 함(이중계상) — 양식 장이 없으면 종전대로 섬 + ③ 제원 키는 제목 글이 아니라 제원 값 — 연장이 달라도 같은 키 · 뒷길이가 다르면 다른 키 + ④ B09 내역 — 호표 금액 × 연장 · 금액 못 받으면 사유 · 수동 단가는 미확정 +""" + +from __future__ import annotations + +from decimal import Decimal +from functools import lru_cache + +from B05_Profile.B05_Profile_Structures_Schema import StructureInstance +from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff +from B08_Quantity.B08_Quantity_Engine_StructurePriceLink import ( + price_sheets, + priced_sheets, + structure_price_code, +) +from B08_Quantity.B08_Quantity_Engine_StructureSheet import build_standard_sheets +from B08_Quantity.B08_Quantity_Engine_StructureTemplate import apply_templates, load_template +from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table +from B09_Estimation.B09_Estimation_BillOfQuantities import bill_summary, build_bill +from B09_Estimation.B09_Estimation_Rounding import OutputPlace, round_at +from B09_Estimation.B09_Estimation_UnitPrice import build_unit_prices, find_variant_code + + +@lru_cache(maxsize=1) +def _build(): + return build_unit_prices() + + +def _wall(sid: str, start: float, end: float, back: int = 45) -> dict: + return StructureInstance.model_validate( + { + "structure_id": sid, + "type_id": "masonry_wet", + "placement": "interval", + "start_m": start, + "end_m": end, + "options": {"height_m": 2.5, "back_len_cm": back}, + } + ).model_dump() + + +def _chain(walls: list[dict]) -> tuple[dict, list[dict]]: + table = build_table(walls, {"masonry_wet": "돌쌓기(찰)"}, {}, {}, None) + payload = build_standard_sheets(table, {}) + apply_templates(payload, {}, {}) + entries = priced_sheets(payload["sheets"], {"masonry_wet": load_template("masonry_wet")}) + return table, entries + + +def _rows(handoff: dict, prefix: str) -> list[dict]: + return [ + r for r in handoff["work_items"] if str(r.get("work_item_code") or "").startswith(prefix) + ] + + +def test_양식_장_구조물은_호표_줄_하나로_가고_기초잡석은_따로_안_간다() -> None: + table, entries = _chain([_wall("a", 0, 10), _wall("b", 50, 56)]) + assert len(entries) == 1 and entries[0]["structure_ids"] == ["a", "b"] + assert "기초잡석" in entries[0]["covered"] and "모르터" in entries[0]["covered"] + + before = build_handoff(unit_quantity_table=table) + assert _rows(before, "FP-13-04-05") and _rows(before, "FP-12-25") # 종전 — 돌쌓기 ㎡ + 잡석 + + after = build_handoff(unit_quantity_table=table, priced_sheets=entries) + ax = _rows(after, "AX-ST-") + assert [(r["unit"], r["quantity"]) for r in ax] == [("m", 10.0), ("m", 6.0)] + assert {r["variant_axis"] for r in ax} == {"structure_sheet"} + assert not _rows(after, "FP-13-04-05") and not _rows(after, "FP-12-25") + assert set(ax[0]) == set(before["work_items"][0]) # 인계 칸 계약 그대로 + + +def test_제원_키는_제원_값이고_연장과_제목은_안_든다() -> None: + _table, entries = _chain([_wall("a", 0, 10), _wall("c", 20, 30, back=55)]) + keys = {e["structure_ids"][0]: e["key"] for e in entries} + assert "L3=45" in keys["a"] and "L3=55" in keys["c"] and "|L=" not in keys["a"] + assert keys["a"] != keys["c"] + assert structure_price_code("AX-ST-56e81a2c", "A = 1 | B~2").endswith("#A=1|B~2") + + +def _bill(manual: dict | None = None): + build = _build() + table, entries = _chain([_wall("a", 0, 10)]) + handoff = build_handoff(unit_quantity_table=table, priced_sheets=entries) + prices = price_sheets( + entries, + build.book, + lambda code, value: find_variant_code(code, value, build), + [load_template("masonry_wet")], + manual, + ) + return entries, prices, build_bill(handoff, build=build, structure_prices=prices) + + +def test_내역은_호표_금액에_연장을_곱한다() -> None: + entries, prices, bill = _bill() + ref = structure_price_code(entries[0]["code"], entries[0]["key"]) + entry = prices[ref] + row = next(r for r in bill.rows if r.code == entries[0]["code"]) + # 모르터 공종 코드가 선 뒤(랩탑 메인 AX-WK) 찰쌓기 호표 세 줄이 다 섬. + assert not entry["blocked"], entry["reasons"] + assert row.unit_price_krw == round_at(entry["money"].total, OutputPlace.UNIT_PRICE_ROW) + assert row.amount_krw == round_at(entry["money"].scaled(Decimal(10)).total, OutputPlace.BOQ_ROW) + assert row.quantity == Decimal(10) and f"호표 {ref}" in row.note + + +def test_금액을_못_받으면_사유로_서고_수동_단가는_미확정으로_센다() -> None: + build = _build() + table, entries = _chain([_wall("a", 0, 10)]) + handoff = build_handoff(unit_quantity_table=table, priced_sheets=entries) + missing = build_bill(handoff, build=build, structure_prices={}) + row = next(r for r in missing.rows if r.code == entries[0]["code"]) + assert row.amount_krw is None and "못 받음" in row.note + + # 모든 일위대가 줄에 수동 단가 — 막힘 없이 서고 미확정 3건. + manual = { + "masonry_wet": {str(s): {"material": 1000, "labor": 0, "expense": 0} for s in (1, 2, 3)} + } + _entries, _prices, bill = _bill(manual) + row = next(r for r in bill.rows if r.code == entries[0]["code"]) + assert row.amount_krw is not None and "미확정" in row.note + summary = bill_summary(bill) + assert summary["unconfirmed_count"] == 3 and summary["unconfirmed"][0]["count"] == 3