eomsangdonandClaude Opus 5
96c304ac35
- 반영률: `quantity`(적용 후) · `quantity_gross`(적용 전) · `application_ratio_pct` 세 칸을 함께 실음. 곱하기는 B08 한 곳에서만 하고 B09 는 산출근거에만 적음. 100 % 인 줄도 율을 적음 — 칸이 비면 받는 쪽이 적용 여부를 단정 못 함. `verify_ratio_math()` 가 세 값의 정합을 재고 `build_handoff()` 가 실제로 부름. - 할증 상태를 세 갈래로: `applied` / `not_applied` / `rate_unavailable`. 두 갈래면 「율을 못 찾아 안 붙인 것」이 「붙였다」로 나가 나중에 진짜 율이 들어왔을 때 한 번 더 붙음. 옛 깃발은 실제로 붙었을 때만 참으로 둠. - 관급구분 `unknown` 이 세 번째 계약값임을 주석에 명시. B09 는 관급자재대에도 도급 재료비에도 넣지 않고 missing 으로 뺌. - 운반 줄이 실제로 실리는지 시험으로 못 박음(무대 in_bill=false 포함 4줄). ⚠ 반영률 누락 지적은 사실이 아니었음 — 실서버 확인 결과 네 반영률이 전부 100 % 이고, 성토면다짐·층따기가 같은 사면 면적에서 나와 값이 겹친 것임. 곱셈은 정상 동작 중. 다만 세 칸 계약은 이중 적용을 막으므로 그대로 반영. 검증 — 인계 40건 통과, 전체 523 passed. 실서버에서 세 칸 정합·할증 상태 `rate_unavailable`·검산 경고 0 확인. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
432 lines
19 KiB
Python
432 lines
19 KiB
Python
"""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()` 가 세 값이 서로 맞는지 실제로 재 본다.
|
||
|
||
⚠ `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"),
|
||
"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,
|
||
"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,
|
||
"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"])
|