Files
Aislo/B08_Quantity/B08_Quantity_Engine_Handoff.py
T
eomsangdonandClaude Opus 5 4deaf0239c feat(B08): 인계 줄마다 막힌 까닭을 실음 + 계약 칸 대조 시험
서브 화면에 「왜 안 서는지」가 안 뜨던 줄 셋(돌쌓기·흙막이·큰돌쌓기)을 고침.

- `blocked_reason` · `blocked_kind` 를 줄마다 실음. 갈래는 셋 —
  `input_missing`(입력하면 풀림) / `unit_data_missing`(자료 없음) /
  `formula_missing`(산출식 없음). 받는 쪽이 「사용자가 입력하면 풀리는 것」과
  「우리가 만들어야 하는 것」을 화면에서 갈라야 함(8-27 표와 같은 축).
- 문구는 `B08_Quantity_Wording` 것을 그대로 씀 — 두 벌로 짜면 갈림.
- 토공·운반 줄은 막힐 자리가 없어도 **칸은 둠** — 계약이 한 모양이어야 함.

⚠ 계약 칸 대조 시험 추가(`test_b08_handoff_contract.py`).
오늘 이 병이 계약 양쪽에서 하나씩 났음 — 우리 쪽은 레지스트리 키와 엔진이 읽는
키가 달라 저장값이 안 닿았고, 받는 쪽은 우리가 보낸 `composite_parts`·
`quantity_gross` 를 안 읽고 버렸음. **「보내는 쪽은 보냈는데 받는 쪽이 안 읽는」
자리는 양쪽 다 조용함.** 키를 늘릴 때 계약에 안 적고 늘리면 받는 쪽이 영영 모름.
자재 줄에 공종코드가 없는 것도 계약으로 못 박음.

검증 — 전체 638 passed. 실물 프로젝트(`5601e828`)에서 네 줄 다 사유가 실림.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-08 04:35:31 +09:00

783 lines
38 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""B08 → B09 인계 (일감 9 · PLAN 8-2 · 2026-09-07 3자 조율 확정).
**두 벌로 낸다. 한 벌로 합치지 않는다.**
① `work_items` = **작업 공종** 축. B09 ④예산내역서가 되는 줄이다. 공종코드가 붙는다.
② `materials` = **자재** 축. B09 자재대가 되는 줄이다. **공종코드가 붙지 않는다.**
⚠ 자재 줄에 공종코드를 붙이지 않는 까닭
내역 줄의 실체는 `돌쌓기(찰) H=1.5 · 70m` 이지 그 전개인 콘크리트 0.31㎥ 가 아니다
(8-2 이중계상 함정). 자재에 공종코드를 붙이면 **자재가 내역 줄로 오해될 자리**가 생긴다.
자재를 카탈로그 키에 잇는 것은 **B09 자원 축의 일**이다(8-7).
⚠ 할증 전/후는 **자재 쪽에만** 있다
작업 공종에는 할증이 없다. 자재는 `net_amount`(전)·`total_amount`(후)를 둘 다 넘긴다 —
하나만 넘기면 B09 가 어느 쪽인지 몰라 역산한다.
⚠ `in_bill` 을 반드시 싣는다 (㉡)
무대(소운반 20m)처럼 **값은 내되 내역에 안 서는** 줄이 있다. B09 가 이 깃발을 안 보면
④예산내역서에 무대가 서서 운반비가 두 번 붙는다. 빼고 넘기지 않는 까닭은,
빠진 줄과 제외된 줄을 나중에 구별할 수 없기 때문이다.
⚠ **반영률은 B08 한 곳에서만 곱한다** (2026-09-07 3자 계약)
`quantity` 는 **곱한 뒤** 값이고 `quantity_gross` 는 곱하기 전 값이며 `application_ratio_pct`
는 쓴 율이다. **셋을 함께 싣는 까닭**은 받는 쪽이 「이미 곱해졌나」를 단정할 수 있어야
하기 때문이다 — 율만 보내면 B09 가 또 곱해 값이 두 배가 된다. 100 % 인 줄도 `100.0` 을
적고, `None` 은 **반영률 개념이 없는 줄**에만 쓴다.
`verify_ratio_math()` 가 세 값이 서로 맞는지 실제로 재 본다.
성·절토면이 갈리는 줄은 **늘** `application_ratio_breakdown`(갈래별 율)과
`quantity_breakdown`(갈래별 물량)을 싣는다. `application_ratio_pct` 는 두 율이 같을 때만
채우는 **편의값**이다 — 「같을 때만 pct, 다를 때만 breakdown」으로 두면 받는 쪽에 갈래가
둘 생기고 그게 한쪽만 고쳐지는 자리가 된다(2026-09-07 3자 계약).
⚠ `ground_class_set` 을 함께 싣는다 (2026-09-07 서브 이견 채택)
값이 「연암」이어도 **그 프로젝트가 몇 갈래 세트를 쓰는지**를 알아야 ④예산내역서에서 줄을
세울 수 있다(울진 2 · 거창 5 · 오솔길 1). 설정 파일을 안 봐도 **인계본만으로 ④가 서게** 한다.
⚠ 못 이은 줄은 **빈 코드로 두지 않는다**
`unmatched_work_items` 로 낸다. 빈칸이면 「코드가 없는 줄」과 「매핑을 못 찾은 줄」이
구별되지 않고, 조용히 없어진 줄은 아무도 못 찾는다.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Iterable
from common_util.common_util_quantity_spread import spread_by_unit
DATASET_DIR = Path(__file__).resolve().parents[1] / "resources" / "data_work_item_mapping"
DATASET_PREFIX = "work_item_mapping_"
#: 철근 갈래표 — **품셈 12-3 [주]① 원문**이 구조물 예시로 갈라 둔 것이라 사람이 고르는 값이
#: 아니다(거푸집 사용횟수 1-7-1 과 같은 자리). 원문 예시에 안 걸리면 지어내지 않는다.
REBAR_DIR = Path(__file__).resolve().parents[1] / "resources" / "data_rebar"
REBAR_PREFIX = "rebar_complexity_"
#: 돌쌓기 규격 갈래표 — 저장 제원 값으로 자동 판정한다(사람이 고르는 값이 아니다).
MASONRY_DIR = Path(__file__).resolve().parents[1] / "resources" / "data_masonry"
MASONRY_PREFIX = "masonry_class_"
def load_masonry_table(path: Path | None = None) -> dict[str, Any]:
"""돌쌓기 갈래표. 파일이 없으면 빈 표 — 갈래가 안 붙고 그대로 드러난다."""
target = path
if target is None:
files = sorted(MASONRY_DIR.glob(MASONRY_PREFIX + "*.json")) if MASONRY_DIR.is_dir() else []
target = files[-1] if files else None
if target is None or not target.is_file():
return {}
return json.loads(target.read_text(encoding="utf-8"))
def masonry_class(
options: dict[str, Any], table: dict[str, Any] | None = None
) -> tuple[str | None, str]:
"""돌쌓기 갈래 — 저장 뒷길이로 **「…㎝ 이하」 구간**을 고른다.
⚠ 저장 선택지(25·30·35·45·55·60·75)와 단가 갈래(35·55·75 이하)는 축이 다르다.
**저장값 이상인 첫 경계**를 고르는 것이 「이하」 구간의 뜻이다.
"""
found = table if table is not None else load_masonry_table()
spec = (found or {}).get("back_length") or {}
from B08_Quantity.B08_Quantity_Wording import option_missing
option_key = str(spec.get("option_key") or "back_len_cm")
raw = options.get(option_key)
if raw is None:
# ⚠ 「없다」만 말하지 않는다 — **어디서 채우면 단가가 붙는지**까지.
# 이름은 부르는 쪽(`unmatched`)이 이미 앞에 붙이므로 여기서는 칸 이름만 말한다.
return None, option_missing(option_key) + " (단가 갈래를 못 고름)"
try:
value = float(raw)
except (TypeError, ValueError):
return None, f"뒷길이 값을 못 읽음({raw!r})"
for row in spec.get("classes") or []:
if value <= float(row["max_cm"]):
return str(row["key"]), f"뒷길이 {value:g}㎝ → 품셈 13-4 「{row['key']}」 구간"
return None, f"뒷길이 {value:g}㎝ 를 덮는 갈래가 표에 없음"
from B08_Quantity.B08_Quantity_Wording import type_label as wording_type_label
def normalize_kind_key(label: str) -> str:
"""갈래 키 — **내부 공백만** 지운다 (2026-09-07 두 창 확정).
원문 표는 「보 통」처럼 자간 공백이 들어 있어 그대로 쓰면 양쪽이 안 맞는다.
⚠ **다른 글자는 손대지 않는다** — 정규화를 넓히면 오늘 아홉 번 겪은 그 병을
여기서 새로 만든다. 원문 문구는 버리지 않고 `label` 로 함께 싣는다.
"""
return "".join(str(label).split())
#: 줄이 어디서 왔나 — 되짚을 때 쓴다.
ORIGIN_EARTHWORK = "earthwork"
ORIGIN_STRUCTURE = "structure"
ORIGIN_SLOPE = "slope"
ORIGIN_HAUL = "haul"
#: 암 시공법 → 매핑표의 지반 이름. 품셈이 **긁어내기와 터뜨리기를 다른 공종**으로 두기 때문에
#: 갈래 이름(연암·보통암…)만으로는 공종을 못 고른다(2026-09-07 일감 9 실서버에서 드러남).
METHOD_TO_GROUND = {"ripping": "리핑암", "blasting": "발파암"}
NOTE_METHOD_MISSING = "시공법 미지정으로 공종을 못 고름"
#: 철근으로 보는 성분 이름 조각. **정확한 낱말이 아니라 앞머리**로 본다 —
#: 「이형철근 D13」·「원형철근」처럼 규격이 뒤에 붙기 때문이다. `철근콘크리트`는 성분 이름이
#: 아니라 공종 이름이라 성분 목록에는 안 온다.
REBAR_PREFIXES = ("이형철근", "원형철근", "철근")
#: 줄이 왜 막혔나 — **받는 쪽이 「사용자가 입력하면 풀리는 것」과 「우리가 만들어야 하는 것」을
#: 화면에서 갈라야** 한다(2026-09-07 3자 확정). 8-27 표에서 이미 가른 그 축이다.
BLOCKED_INPUT_MISSING = "input_missing" # 저장 제원 칸이 비어 있음 — 입력하면 풀림
BLOCKED_UNIT_DATA_MISSING = "unit_data_missing" # 원단위·표준 물량 자료가 없음
BLOCKED_FORMULA_MISSING = "formula_missing" # 수량 산출식 자체가 없음
#: 사면 계열 — 토공집계에 함께 실리지만 출처가 사면적이다.
SLOPE_GROUPS = frozenset({"성토면다짐", "초류종자살포", "지장목제거", "층따기", "면고르기"})
#: 집계 합계 줄 — 내역 줄이 아니라 검산용이다.
SUBTOTAL_GROUPS = frozenset({"보정량계"})
def _latest_dataset_path(directory: Path | None = None) -> Path | None:
folder = directory or DATASET_DIR
if not folder.is_dir():
return None
files = sorted(folder.glob(DATASET_PREFIX + "*.json"))
return files[-1] if files else None
@dataclass
class WorkItemMapping:
"""수량 줄 → 공종 마스터 코드. 못 찾으면 `None` 을 돌려주고 부른 쪽이 목록에 남긴다."""
effective_date: str = ""
earthwork: list[dict[str, Any]] = field(default_factory=list)
haul: list[dict[str, Any]] = field(default_factory=list)
structure: list[dict[str, Any]] = field(default_factory=list)
pending_user: dict[str, Any] = field(default_factory=dict)
composite: dict[str, Any] = field(default_factory=dict)
concrete_placing: dict[str, Any] = field(default_factory=dict)
unit_conversion: dict[str, Any] = field(default_factory=dict)
def for_earthwork(self, group: str, ground: str | None) -> dict[str, Any] | None: # noqa: D401
"""공종+지반유형으로 찾는다. 지반이 갈리지 않는 공종은 `ground` 없는 줄이 받는다."""
exact = [
row
for row in self.earthwork
if row.get("group") == group and row.get("ground") == ground
]
if exact:
return exact[0]
# 지반을 안 가르는 공종(성토·층따기 등)은 `ground` 칸이 없는 줄로 맞춘다.
loose = [row for row in self.earthwork if row.get("group") == group and "ground" not in row]
return loose[0] if loose else None
def for_haul(self, equipment: str) -> dict[str, Any] | None:
for row in self.haul:
if row.get("equipment") == equipment:
return row
return None
def for_structure(self, type_id: str) -> dict[str, Any] | None:
for row in self.structure:
if row.get("type_id") == type_id:
return row
return None
def composite_for(self, type_id: str) -> dict[str, Any] | None:
"""품셈에 그 이름의 공종이 없어 **여러 공종을 묶는** 자리인가.
빈 코드로 두면 「매핑을 못 찾은 줄」과 구별이 안 된다 — 묶음을 적어 구별한다.
"""
for row in self.composite.get("items") or []:
if row.get("type_id") == type_id:
return row
return None
def load_mapping(path: Path | None = None) -> WorkItemMapping:
"""매핑표를 읽는다. 파일이 없으면 **빈 표** — 전 줄이 `unmatched` 로 드러난다."""
target = path or _latest_dataset_path()
if target is None or not target.is_file():
return WorkItemMapping()
payload = json.loads(target.read_text(encoding="utf-8"))
return WorkItemMapping(
effective_date=str(payload.get("effective_date") or ""),
earthwork=list(payload.get("earthwork") or []),
haul=list(payload.get("haul") or []),
structure=list(payload.get("structure") or []),
pending_user=payload.get("pending_user") or {},
composite=payload.get("composite") or {},
concrete_placing=payload.get("concrete_placing") or {},
unit_conversion=(payload.get("composite") or {}).get("unit_conversion") or {},
)
def composite_quantities(
structure: dict[str, Any],
composite: dict[str, Any],
mapping: WorkItemMapping,
) -> tuple[list[dict[str, Any]], list[str]]:
"""묶음 조각마다 **부위별 수량**을 채운다. (조각 목록, 못 채운 사유).
⚠ 조각은 **원단위 성분 이름으로** 찾는다. 이름이 어긋나면 물량이 조용히 0 이 되므로
못 찾으면 그 조각을 `not_ready` 로 남기고 사유를 적는다 — 0 을 적지 않는다.
⚠ **단위를 반드시 맞춘다.** 철근 단가는 `원/ton` 인데 원단위는 `㎏` 이다.
안 맞추면 **1000배 틀린다** — 밑수에서 겪은 것과 같은 자리다.
"""
# ⚠ 원단위 자체가 없으면 조각을 늘어놓지 않는다 — 같은 사유가 다섯 번 반복되면
# **진짜 사유가 묻힌다**(화면에서 실제로 그렇게 보였다). 한 줄로 말한다.
if not (structure.get("components") or []):
note = "; ".join(structure.get("notes") or []) or "구조물 원단위가 없음"
return [], [{"code": None, "reason": note}]
amounts: dict[str, tuple[float, str]] = {}
for component in structure.get("components") or []:
name = str(component.get("name") or "").strip()
amounts[name] = (float(component.get("amount") or 0.0), str(component.get("unit") or ""))
kg_to_ton = float((mapping.unit_conversion or {}).get("kg_to_ton") or 0.001)
parts: list[dict[str, Any]] = []
missing: list[str] = []
for spec in composite.get("parts") or []:
if not isinstance(spec, dict): # 옛 모양(코드 문자열)은 그대로 흘린다
parts.append({"code": str(spec)})
continue
sources = list(spec.get("from_components") or [])
found = [name for name in sources if name in amounts]
total = sum(amounts[name][0] for name in found)
if spec.get("unit_from") == "kg" and spec.get("unit") == "ton":
total *= kg_to_ton
kinds = {
component.get("basis_kind")
for component in structure.get("components") or []
if str(component.get("name") or "").strip() in found
}
suffix = spec.get("kind_suffix")
entry: dict[str, Any] = {
"code": spec.get("code"),
"name": spec.get("name"),
"unit": spec.get("unit"),
"quantity": total if found else None,
# 조각마다 근거를 단다 — 치수 전개와 관측값이 한 묶음에 섞인다.
"basis_kind": next(iter(kinds)) if len(kinds) == 1 else (sorted(kinds) or None),
"from_components": sources,
}
if spec.get("incomplete_note"):
# ⚠ 물량은 섰으나 **일부 몫이 빠진** 조각 — 「못 채움」과 달리 값은 있다.
# 화면·인계 둘 다 그 사실을 알아야 「다 섰다」로 오해하지 않는다.
entry["incomplete_note"] = spec["incomplete_note"]
if suffix == "euroform_type":
kind, why = euroform_type(str(structure.get("type_id") or ""))
entry["kind"] = normalize_kind_key(kind) if kind else None
entry["kind_label"] = kind # 원문 문구 그대로
entry["kind_basis"] = why
if kind:
entry["code"] = f"{spec.get('code')}#{normalize_kind_key(kind)}"
else:
entry["not_ready"] = True
entry["why"] = why
missing.append({"code": spec.get("code"), "reason": why})
if suffix == "rebar_complexity":
# 갈래는 원문이 정한다 — 화면·인계에 이름과 근거를 함께 실어 사람이 검증하게 한다.
complexity, why = rebar_complexity(
str(structure.get("type_id") or ""), structure.get("options") or {}
)
entry["kind"] = normalize_kind_key(complexity) if complexity else None
entry["kind_label"] = complexity # 원문 문구 그대로(자간 공백 포함)
entry["kind_basis"] = why
if complexity:
entry["code"] = f"{spec.get('code')}#{normalize_kind_key(complexity)}"
else:
entry["not_ready"] = True
entry["why"] = why
missing.append({"code": spec.get("code"), "reason": why})
if spec.get("not_ready") or not found:
entry["not_ready"] = True
entry["why"] = str(spec.get("why") or "원단위에 해당 성분이 없음")
# ⚠ 「단가 없음」과 「물량 없음」을 받는 쪽이 갈라야 하므로 **구조로** 낸다.
missing.append({"code": spec.get("code"), "reason": entry["why"]})
parts.append(entry)
return parts, missing
def structure_kind(structure: dict[str, Any]) -> str:
"""콘크리트 구조물 종류 — **원단위에 철근이 있나 없나로 판정한다.**
사람이 고르는 값이 아니다(2026-09-07 3자 확정). 옹벽 관측 원단위에 `D13`·`D16` 이
실려 있으므로 철근구조물로 자동으로 선다. 소형구조물 판정 기준은 아직 없다.
"""
for component in structure.get("components") or []:
name = str(component.get("name") or "").strip()
if any(name.startswith(prefix) for prefix in REBAR_PREFIXES):
return "철근구조물"
return "무근구조물"
def placing_code(mapping: WorkItemMapping, method: str | None) -> tuple[str | None, bool]:
"""(타설 공종코드, 기본값을 쓴 것인가). 모르는 방식이면 기본으로 떨어지되 그 사실을 알린다."""
table = mapping.concrete_placing or {}
codes = table.get("method_codes") or {}
default = str(table.get("default_method") or "")
if method in codes:
return codes[method], False
return codes.get(default), True
def load_rebar_table(path: Path | None = None) -> dict[str, Any]:
"""철근 갈래표를 읽는다. 파일이 없으면 **빈 표** — 전부 「갈래 미확보」로 드러난다."""
target = path
if target is None:
files = sorted(REBAR_DIR.glob(REBAR_PREFIX + "*.json")) if REBAR_DIR.is_dir() else []
target = files[-1] if files else None
if target is None or not target.is_file():
return {}
return json.loads(target.read_text(encoding="utf-8"))
def rebar_complexity(
type_id: str, options: dict[str, Any], table: dict[str, Any] | None = None
) -> tuple[str | None, str]:
"""(철근 갈래, 근거). **원문 예시에 걸리는 것만** 정하고 안 걸리면 `(None, 사유)`.
품셈 12-3 [주]① — 「간단: 측구·간단한 기초·**중력식 옹벽** / 보통: 수문·**반중력식 옹벽**·
교대 / 복잡: 교량 슬래브·암거·우물통·**부벽식 옹벽** / 매우복잡: 구주식 교대·교각…」.
사람에게 묻지 않는다 — **판정할 수 있는 것을 물으면 그것이 곧 미결이 된다.**
"""
found = table if table is not None else load_rebar_table()
form = options.get("form")
fallback: dict[str, Any] | None = None
for row in found.get("form_map") or []:
if row.get("type_id") != type_id:
continue
if "form" not in row:
fallback = row
continue
if row.get("form") == form:
if row.get("class"):
return str(row["class"]), f"품셈 12-3 [주]① 「{row.get('matched')}」"
return None, str(row.get("why") or "원문 예시에 없음")
if fallback and fallback.get("class"):
return str(fallback["class"]), f"품셈 12-3 [주]① 「{fallback.get('matched')}」"
from B08_Quantity.B08_Quantity_Wording import type_label
detail = f"({form})" if form else "(형식이 아직 입력되지 않음)"
return None, (
f"{type_label(type_id)} {detail} 는 품셈 12-3 [주]① 예시에 없어 "
"철근 갈래를 정하지 못했습니다 — 임의로 고르지 않습니다"
)
def euroform_type(type_id: str, table: dict[str, Any] | None = None) -> tuple[str | None, str]:
"""유로폼 설치·해체 유형 — **품셈 12-38-3 [주]④ 원문**이 시설 예시로 갈라 둔다.
「보통: 측구, 수로, **옹벽**, 일반적인 벽체, 박스」. 거푸집 사용횟수(1-7-1)·철근 갈래
(12-3 [주]①)에 이어 **네 번째** 같은 자리다 — 미결로 올리기 전에 원문부터 뒤진다.
⚠ 12-38-1 「사용횟수」와 헷갈리지 말 것 — 그쪽은 유로폼(강재)의 **잔존율**이고
1-7-1 의 소모성 거푸집 전용 횟수와도 다른 자리다.
"""
from B08_Quantity.B08_Quantity_Engine_Formwork import load_formwork_table
found = table if table is not None else load_formwork_table().euroform_type
for row in (found or {}).get("type_map") or []:
if row.get("type_id") == type_id and row.get("class"):
return str(row["class"]), f"품셈 12-38-3 [주]④ 「{row.get('matched')}」"
from B08_Quantity.B08_Quantity_Wording import type_label
return None, (
f"{type_label(type_id)} 는 품셈 12-38-3 [주]④ 예시에 없어 유로폼 유형을 "
"정하지 못했습니다 — 임의로 고르지 않습니다"
)
def _spec_detail(structure: dict[str, Any]) -> str:
"""규격 표기 — 저장된 제원에서 만든다. 없는 값은 적지 않는다."""
parts: list[str] = []
height = structure.get("height_m")
length = structure.get("length_m")
if height:
parts.append(f"H={height:g}")
if length:
parts.append(f"L={length:g}m")
return "·".join(parts)
def _mapping_ground(ground: str | None, methods: dict[str, str | None]) -> tuple[str | None, str]:
"""갈래 이름을 **매핑표가 아는 이름**으로 바꾼다.
「토사」는 그대로 가고, 암 갈래는 **시공법이 정해져야** 리핑암·발파암으로 간다.
안 정했으면 `(None, 사유)` — 찍지 않는다. 잘못 찍으면 공종이 조용히 틀린다.
"""
if ground is None or ground == "토사":
return ground, ""
method = methods.get(ground)
mapped = METHOD_TO_GROUND.get(method or "")
if mapped:
return mapped, ""
return None, NOTE_METHOD_MISSING
def _earthwork_rows(
summary_table: dict[str, Any],
mapping: WorkItemMapping,
methods: dict[str, str | None],
) -> tuple[list[dict[str, Any]], list[str]]:
"""토공집계표 줄을 내역 줄로 옮긴다.
⚠ 「보정량계」 같은 합계 줄은 **내역 줄이 아니다** — 빼지 않고 `in_bill: False` 로 넘긴다.
빼 버리면 B09 가 검산할 때 합이 안 맞는 까닭을 알 수 없다.
"""
rows: list[dict[str, Any]] = []
unmatched: list[str] = []
for row in summary_table.get("rows") or []:
group = str(row.get("group") or "")
if not group:
continue
ground = row.get("item") or None
origin = ORIGIN_SLOPE if group in SLOPE_GROUPS else ORIGIN_EARTHWORK
is_subtotal = group in SUBTOTAL_GROUPS
lookup_ground, method_note = _mapping_ground(ground, methods)
entry = mapping.for_earthwork(group, lookup_ground) if method_note == "" else None
code = (entry or {}).get("work_item_code")
if code is None and not is_subtotal:
label = f"{group}({ground})" if ground else group
unmatched.append(f"{label}{method_note}" if method_note else label)
rows.append(
{
"work_item_code": code,
"name": group,
"spec": str(row.get("spec") or ""),
"unit": str(row.get("unit") or "㎥"),
"quantity": float(row.get("amount") or 0.0),
# 반영률 — 곱하기는 여기가 끝이다. 받는 쪽은 적기만 한다.
"quantity_gross": row.get("amount_gross"),
"application_ratio_pct": row.get("application_ratio_pct"),
# 율이 부분마다 다른 줄 — 받는 쪽이 문장을 안 뜯게 칸으로 준다.
"application_ratio_breakdown": row.get("application_ratio_breakdown"),
"quantity_breakdown": row.get("quantity_breakdown"),
"ground_class": ground,
"haul_distance_m": None,
"haul_equipment": None,
"station_from": None,
"station_to": None,
"spec_detail": "",
"composite_parts": None,
"structure_kind": None,
"spec_class": None,
"spec_class_basis": "",
# 토공 줄은 막힐 자리가 없다 — 그래도 **칸은 둔다**(계약이 한 모양이어야 한다).
"blocked_kind": None,
"blocked_reason": "",
"composite_not_ready": None,
# 합계 줄과 무대 줄은 값은 내되 내역에 안 선다.
"in_bill": bool(row.get("in_bill", True)) and not is_subtotal,
"excavation_method": methods.get(ground) if ground else None,
"in_bill_reason": "집계 합계 줄 — 검산용"
if is_subtotal
else str(row.get("note") or ""),
"origin": origin,
}
)
return rows, unmatched
def _haul_rows(
haul_table: dict[str, Any], mapping: WorkItemMapping
) -> tuple[list[dict[str, Any]], list[str]]:
"""운반 줄 — 가중평균 거리가 붙는다. 무대는 `in_bill: False` 로 함께 넘긴다(㉡)."""
rows: list[dict[str, Any]] = []
unmatched: list[str] = []
for row in haul_table.get("rows") or []:
equipment = str(row.get("equipment") or "")
entry = mapping.for_haul(equipment) or {}
code = entry.get("work_item_code")
in_bill = bool(row.get("in_bill", True)) and entry.get("in_bill", True)
if code is None and in_bill:
unmatched.append(f"운반({equipment})")
rows.append(
{
"work_item_code": code,
"name": f"{equipment} 운반",
"spec": str(row.get("ground") or ""),
"unit": "㎥",
"quantity": float(row.get("volume_m3") or 0.0),
# 운반에는 반영률 개념이 없다 — 그래서 `None` 이다(0 이 아니다).
"quantity_gross": None,
"application_ratio_pct": None,
"application_ratio_breakdown": None,
"quantity_breakdown": None,
"ground_class": row.get("ground") or None,
"haul_distance_m": float(row.get("average_distance_m") or 0.0),
"haul_equipment": equipment,
"excavation_method": None,
"station_from": None,
"station_to": None,
"spec_detail": "",
"composite_parts": None,
"structure_kind": None,
"spec_class": None,
"spec_class_basis": "",
"blocked_kind": None,
"blocked_reason": "",
"composite_not_ready": None,
"in_bill": in_bill,
"in_bill_reason": str(entry.get("reason") or ""),
"origin": ORIGIN_HAUL,
}
)
return rows, unmatched
def blocked_of(structure: dict[str, Any], class_basis: str = "") -> tuple[str | None, str]:
"""(막힌 갈래, 사유). 안 막혔으면 `(None, "")`.
⚠ 사유 문구는 **`B08_Quantity_Wording` 것을 그대로** 쓴다 — 두 벌로 짜면 갈린다.
전개 알림(`notes`)에 이미 사람 말로 적혀 있으므로 그것을 그대로 옮긴다.
"""
notes = [str(note) for note in structure.get("notes") or []]
if structure.get("components"):
# 물량은 섰는데 **단가 갈래**를 못 고른 자리(돌쌓기 뒷길이 등).
if class_basis and "입력되지 않았습니다" in class_basis:
return BLOCKED_INPUT_MISSING, class_basis
return None, ""
for note in notes:
if "입력되지 않았습니다" in note:
return BLOCKED_INPUT_MISSING, note
if "자료에 없습니다" in note or "표준 물량 자료" in note:
return BLOCKED_UNIT_DATA_MISSING, note
if "산출식이 아직 없습니다" in note:
return BLOCKED_FORMULA_MISSING, note
if notes:
return BLOCKED_UNIT_DATA_MISSING, notes[0]
return None, ""
def _structure_rows(
unit_quantity_table: dict[str, Any], mapping: WorkItemMapping
) -> tuple[list[dict[str, Any]], list[str]]:
"""구조물 줄 — **작업 공종 하나**로 선다.
⚠ 전개 성분(터파기·야면석·모르터)은 여기 오지 않는다. 구조물 한 기가 내역 한 줄이고,
그 전개는 토공 합산·자재총괄·일위대가로 갈린다(8-2 이중계상 함정).
"""
rows: list[dict[str, Any]] = []
unmatched: list[str] = []
for structure in unit_quantity_table.get("structures") or []:
type_id = str(structure.get("type_id") or "")
entry = mapping.for_structure(type_id) or {}
code = entry.get("work_item_code")
# 돌쌓기는 **뒷길이 갈래**로 단가가 갈린다 — 저장 제원에서 자동으로 고른다.
class_key: str | None = None
class_basis = ""
if code and entry.get("class_from") == "back_length":
class_key, class_basis = masonry_class(structure.get("options") or {})
if class_key:
code = f"{code}#{normalize_kind_key(class_key)}"
composite = mapping.composite_for(type_id) if code is None else None
kind = structure_kind(structure) if composite else None
parts: list[dict[str, Any]] | None = None
parts_missing: list[dict[str, Any]] = []
if composite:
parts, parts_missing = composite_quantities(structure, composite, mapping)
blocked_kind, blocked_reason = blocked_of(structure, class_basis)
if code is None and composite is None:
unmatched.append(f"{wording_type_label(type_id)} — 품셈 공종을 아직 못 이었습니다")
elif entry.get("class_from") == "back_length" and class_key is None:
unmatched.append(f"{wording_type_label(type_id)}{class_basis}")
length = float(structure.get("length_m") or 0.0)
rows.append(
{
"work_item_code": code,
"name": str(structure.get("name") or type_id),
"spec": _spec_detail(structure),
"unit": "m",
"quantity": length,
"quantity_gross": None,
"application_ratio_pct": None,
"application_ratio_breakdown": None,
"quantity_breakdown": None,
"ground_class": None,
"haul_distance_m": None,
"haul_equipment": None,
"station_from": structure.get("start_m"),
"station_to": structure.get("end_m"),
"excavation_method": None,
"spec_detail": _spec_detail(structure),
# 품셈에 그 이름의 공종이 없어 여러 공종을 묶는 자리 — 빈 코드와 구별한다.
"composite_parts": parts,
# 철근이 있나 없나로 자동 판정 — 사람이 고르는 값이 아니다.
"structure_kind": kind,
# ⚠ 줄마다 **왜 막혔는지**를 싣는다 — 안 실으면 받는 쪽 화면이 빈다.
"blocked_kind": blocked_kind,
"blocked_reason": blocked_reason,
# 규격 갈래(뒷길이 …㎝ 이하) — 못 고르면 사유가 남는다.
"spec_class": class_key,
"spec_class_basis": class_basis,
# ⚠ 물량을 못 채운 조각 — 0 으로 적지 않고 사유와 함께 드러낸다.
"composite_not_ready": parts_missing or None,
"in_bill": True,
"in_bill_reason": (composite or {}).get("why", ""),
"origin": ORIGIN_STRUCTURE,
}
)
return rows, unmatched
def _material_rows(material_table: dict[str, Any]) -> list[dict[str, Any]]:
"""자재 줄 — **공종코드를 붙이지 않는다.** 자재 축은 B09 카탈로그가 잇는다(8-7)."""
rows: list[dict[str, Any]] = []
for row in material_table.get("rows") or []:
rows.append(
{
"material_name": row.get("name"),
"spec": row.get("spec") or "",
"unit": row.get("unit"),
"net_amount": row.get("net_amount"),
"total_amount": row.get("total_amount"),
"surcharge_pct": row.get("surcharge_pct"),
"surcharge_note": row.get("note") or "",
"supply_type": row.get("supply"),
"install_by": row.get("install_by"),
"source_structure": row.get("sources") or [],
}
)
return rows
def build_handoff(
*,
summary_table: dict[str, Any] | None = None,
haul_table: dict[str, Any] | None = None,
unit_quantity_table: dict[str, Any] | None = None,
material_table: dict[str, Any] | None = None,
mapping: WorkItemMapping | None = None,
ground_class_set: str | None = None,
ground_classes: list[str] | None = None,
ground_methods: dict[str, str | None] | None = None,
) -> dict[str, Any]:
"""B09 가 그대로 받는 모양. 없는 표는 건너뛰되 **빈 표와 구별해 적는다**."""
table = mapping or load_mapping()
work_items: list[dict[str, Any]] = []
unmatched: list[str] = []
methods = {key: value for key, value in (ground_methods or {}).items() if value}
if summary_table:
rows, misses = _earthwork_rows(summary_table, table, methods)
work_items.extend(rows)
unmatched.extend(misses)
if haul_table:
rows, misses = _haul_rows(haul_table, table)
work_items.extend(rows)
unmatched.extend(misses)
if unit_quantity_table:
rows, misses = _structure_rows(unit_quantity_table, table)
work_items.extend(rows)
unmatched.extend(misses)
materials = _material_rows(material_table or {})
result: dict[str, Any] = {
"work_items": work_items,
"materials": materials,
# 갈래 세트 — 「연암」이 몇 갈래 중 하나인지 알아야 ④가 선다.
"ground_class_set": ground_class_set,
"ground_classes": list(ground_classes or []),
"ground_methods": dict(methods),
# 시공법을 안 정해 공종을 못 고른 갈래 — 화면이 이 목록으로 안내를 띄운다.
"missing_method_classes": sorted(
{
str(row.get("ground_class"))
for row in work_items
if row.get("ground_class")
and row.get("ground_class") != "토사"
and row.get("work_item_code") is None
and row.get("origin") == ORIGIN_EARTHWORK
}
),
# 자재 쪽에만 할증이 있다 — 작업 공종에는 없다.
# ⚠ 세 갈래로 그대로 나른다(`applied`·`not_applied`·`rate_unavailable`).
# 「율이 없어 못 붙인 것」을 「붙였다」로 말하면 B09 가 나중에 한 번 더 붙인다.
"surcharge_status": (material_table or {}).get("surcharge_status"),
"surcharge_applied_to_materials": bool((material_table or {}).get("surcharge_applied")),
"unmatched_work_items": sorted(set(unmatched)),
"mapping_pending_user": table.pending_user,
"mapping_edition": table.effective_date,
"quantity_spread": spread_by_unit(
[row for row in work_items if row["in_bill"]], value_key="quantity"
),
"material_spread": spread_by_unit(
[{"unit": row["unit"], "q": row["total_amount"]} for row in materials], value_key="q"
),
"bill_row_count": sum(1 for row in work_items if row["in_bill"]),
"excluded_row_count": sum(1 for row in work_items if not row["in_bill"]),
}
# ⚠ 검사는 **실제로 부른다** — 만들어 두고 안 부르면 없는 것과 같다.
result["ratio_math_warnings"] = verify_ratio_math(result)
return result
def verify_no_code_on_materials(handoff: dict[str, Any]) -> list[str]:
"""⚠ 자재 줄에 공종코드가 섞이면 알린다.
자재에 공종코드가 붙으면 **내역 줄로 오해될 자리**가 생기고 그게 곧 이중계상이다.
축이 둘이라는 것은 주석이 아니라 검사로 지켜야 한다.
"""
found: list[str] = []
for row in handoff.get("materials") or []:
if row.get("work_item_code"):
found.append(str(row.get("material_name")))
return found
def verify_ratio_math(handoff: dict[str, Any], *, tolerance: float = 1e-6) -> list[str]:
"""⚠ `quantity == quantity_gross × 율/100` 이 실제로 맞는지 재 본다.
세 칸을 실어 두고 **서로 어긋나면** 받는 쪽이 어느 값을 믿을지 알 수 없다.
「만들어 두고 안 부르면 없는 것과 같다」를 피하려고 `build_handoff()` 가 직접 부른다.
"""
found: list[str] = []
for row in handoff.get("work_items") or []:
gross = row.get("quantity_gross")
ratio = row.get("application_ratio_pct")
if gross is None or ratio is None:
continue
expected = float(gross) * float(ratio) / 100.0
actual = float(row.get("quantity") or 0.0)
if abs(expected - actual) > max(tolerance, abs(expected) * 1e-9):
found.append(f"{row.get('name')}: {actual:g}{gross:g} × {ratio:g} %")
return found
def verify_bill_flags(handoff: dict[str, Any]) -> list[str]:
"""⚠ 코드가 없는데 내역에 서는 줄이 있으면 알린다.
빈 코드로 내역에 세우면 B09 가 단가를 못 붙인 채 0원 줄을 만든다.
"""
found: list[str] = []
for row in handoff.get("work_items") or []:
if not row.get("in_bill") or row.get("work_item_code"):
continue
# 묶음으로 서는 줄은 코드가 없어도 정상이다 — 무엇으로 묶이는지 적혀 있다.
if row.get("composite_parts"):
continue
found.append(str(row.get("name")))
return found
def summarize(handoff: dict[str, Any]) -> dict[str, Any]:
"""화면 안내용 한 줄 요약 — 넘긴 줄과 못 이은 줄을 함께 보인다."""
return {
"bill_rows": handoff.get("bill_row_count", 0),
"excluded_rows": handoff.get("excluded_row_count", 0),
"materials": len(handoff.get("materials") or []),
"unmatched": handoff.get("unmatched_work_items") or [],
}
def iter_bill_rows(handoff: dict[str, Any]) -> Iterable[dict[str, Any]]:
"""내역에 서는 줄만 — B09 ④예산내역서가 쓰는 입구."""
return (row for row in handoff.get("work_items") or [] if row["in_bill"])