Files
Aislo/B08_Quantity/B08_Quantity_Engine_Handoff.py
T
eomsangdonandClaude Opus 5 8e78940b7a feat(B08): 거푸집 사용횟수 + 공종코드 잇기
- 거푸집 사용횟수는 **관측값이 아니라 법**임. 품셈 1-7-1 이 구조물 종류별로
  정해 둠(옹벽 3회 · 보호공 기초 6회). 원문 문구를 데이터에 싣고 우리 구조물이
  어느 예시에 걸리는지 적음. 걸리는 예시가 없으면 「사용횟수 미확보」.
- ⚠ 횟수별 재료 환산(품셈 12-4 합판 3회 46.1 %)은 **하지 않음**. 그 비율은
  일위대가 재료비에 걸리는 값이라 B08 이 곱하면 B09 와 겹쳐 두 번 줌.
  B08 이 내는 것은 접촉 면적 그대로 + 몇 회짜리인가까지. 시험으로 못 박음.
- 동바리는 슬래브를 떠받칠 때 쓰는 것이라 지금 서는 구조물(옹벽·집수정)은
  대상이 아님. 0 으로 적지 않고 「대상 없음 + 사유」로 냄.
- 거푸집 이름은 정확 일치로만 봄 — 부분일치면 「거푸집씻기」(공사용수)가 걸림.

공종코드 잇기
- 집수정 → FP-12-15, 물넘이포장 → FP-12-06 로 이음.
- 옹벽은 **품셈 12장에 그 이름의 공종이 없음**. 빈 코드로 두면 「매핑을 못 찾은
  줄」과 구별이 안 되므로 `composite` 로 묶음(타설+거푸집+철근+기초잡석)을 적음.
  일위대가 조립은 B09 몫이고 B08 은 물량과 묶음만 넘김.

검증 — 거푸집 8건 · 인계 45건 통과, 전체 558 passed. tsc 오류 0.

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

462 lines
21 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)
composite: 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 {},
)
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")
composite = mapping.composite_for(type_id) if code is None else None
if code is None and composite 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),
# 품셈에 그 이름의 공종이 없어 여러 공종을 묶는 자리 — 빈 코드와 구별한다.
"composite_parts": (composite or {}).get("parts"),
"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"])