Files
Aislo/B08_Quantity/B08_Quantity_Engine_Handoff.py
T
eomsangdonandClaude Opus 5 513769796f feat(B08): 콘크리트 구조물 관측 원단위 + 품셈 몫 표시
⑩ 콘크리트 구조물. 품셈에 구조물별 표준 물량표가 없어(구조물_수량.md ·
배수공_수량.md §2) 실무 설계원본 관측값을 데이터 파일로 둠.

- `resources/data_structure_unit/` — 반중력식옹벽 H=2.0(m당) · 돌집수정 ㄷ/ㄴ형 ·
  집수정 Ø800 · 콘크리트포장 T=20cm. 줄마다 출처를 실음.
- 규격 정확 일치로만 씀. 보간하지 않음 — H=1.6 을 H=2.0 에서 줄여 쓰면 틀림
  (기초·벽 두께는 높이에 비례하지 않음). 안 맞으면 「원단위 미확보」로 드러내고
  표에 있는 규격을 함께 알림.
- BOX암거·세월교는 관측값도 없고 저장 제원에 두께가 없어 전개식도 못 세움.
  `not_found` 에 사유와 함께 남기고 미확보로 냄 — 두께를 지어내면 콘크리트·
  거푸집·철근으로 번져 나감.
- 성분마다 `basis_kind`(derived/observed)를 달아 화면에 「치수 전개 / 실무 관측」
  으로 보임. 두 근거가 한 표에 섞이므로 안 적으면 되짚을 수 없음.
- 배수관의 유입부 집수정을 별도 줄로 세움. 품셈도 관부설과 집수정을 다른
  공종으로 둠 — 한 줄로 합치면 어느 쪽 물량인지 못 가름.

품셈 몫 표시 (조율 창 제보 교차 확인)
- `인력(10%)`·`장비(90%)` 처럼 딱지가 비율을 달고 오는 표 25건 확인. 형태 판정은
  정상(requirement)이었으나 **장비 몫이 시공능력 공식이라 값이 아님**을 아무도
  알 수 없었음. `resource_shares`·`partial_ratio` 를 실어 「단가가 일부만 선 것」이
  정상으로 흘러가지 않게 함. 공식 기호는 첫 표에만 있고 뒤 표가 물려받으므로
  기호 유무가 아니라 몫 유무로 판정 — 기호로 세면 절반을 놓침(9-13-2).

반영률 계약 갱신 — breakdown 을 늘 실음(`application_ratio_breakdown` ·
`quantity_breakdown`). `application_ratio_pct` 는 두 율이 같을 때만 채우는 편의값.

검증 — 관측 원단위 17건 · 인계 43건 · 품셈 23건 통과, 전체 548 passed. tsc 오류 0.

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

443 lines
20 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_"
#: 줄이 어디서 왔나 — 되짚을 때 쓴다.
ORIGIN_EARTHWORK = "earthwork"
ORIGIN_STRUCTURE = "structure"
ORIGIN_SLOPE = "slope"
ORIGIN_HAUL = "haul"
#: 암 시공법 → 매핑표의 지반 이름. 품셈이 **긁어내기와 터뜨리기를 다른 공종**으로 두기 때문에
#: 갈래 이름(연암·보통암…)만으로는 공종을 못 고른다(2026-09-07 일감 9 실서버에서 드러남).
METHOD_TO_GROUND = {"ripping": "리핑암", "blasting": "발파암"}
NOTE_METHOD_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)
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 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 {},
)
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": "",
# 합계 줄과 무대 줄은 값은 내되 내역에 안 선다.
"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,
"station_from": None,
"station_to": None,
"spec_detail": "",
"in_bill": in_bill,
"in_bill_reason": str(entry.get("reason") or ""),
"origin": ORIGIN_HAUL,
}
)
return rows, unmatched
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")
if code is None:
unmatched.append(f"구조물({type_id})")
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"),
"spec_detail": _spec_detail(structure),
"in_bill": True,
"in_bill_reason": "",
"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 row.get("in_bill") and not row.get("work_item_code"):
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"])