Aislo/B08_Quantity/B08_Quantity_Engine_Handoff_Boundaries.py

"""B08 → B09 인계 — **이중계상 경계를 갈 곳(`destination`) 칸으로 판정** (명세 6장 · 2026-09-13).

주석 경고뿐이던 규칙 셋을 수로 걸리는 검사로 세운다. 원문은 옛 계획서 8-7절(㉠·㉢)과 명세 6장.

    ① 구조물 터파기·되메우기는 **토공집계로만** — 갈 곳이 `earthwork` 인 성분을 다른 공종 줄이
       또 집으면 같은 물량이 두 공종 코드에 붙음. 터파기·되메우기가 `earthwork` 밖으로 가도 오류
    ② 배합 성분(시멘트·모래·자갈)은 **B09 일위대가만 쪼갬** — 자재총괄에 뜨면 오류(㉢)
    ③ 자재 할증은 **자재총괄 한 곳** — 원단위표가 이미 할증을 붙였으면 오류(㉠).
       자재총괄 합계의 할증 한 번·일위대가 재료비의 할증 전은 B09 가드가 금액 자리에서 봄

⚠ 새 얼개를 만들지 않는다 — 성분마다 붙은 `destination` 과, 인계 줄이 성분을
   **이름으로 집는 자리**(구조물 줄 `billing_component` · 묶음 조각 `from_components` ·
   콘크리트 타설 · 기초잡석)만 대조한다.
⚠ 여기는 목록만 낸다. 멈추는 것은 B09 `build_bill`(`DoubleCountError`) — 가드가 모인 한 자리.
"""

from __future__ import annotations

from typing import Any

from B08_Quantity.B08_Quantity_Engine_Handoff_Mapping import WorkItemMapping
from B08_Quantity.B08_Quantity_Engine_Handoff_Rows import PLACING_TARGET_NAMES
from B08_Quantity.B08_Quantity_Engine_Handoff_Trench import RUBBLE_GROUP
from B08_Quantity.B08_Quantity_Engine_MaterialSummary import verify_single_surcharge
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import MIX_COMPONENTS

#: 토공집계로만 가야 하는 성분 — 전개·양식·관측 원단위가 같은 이름을 쓴다.
EARTHWORK_ONLY = ("터파기", "되메우기")
#: 공종 줄이 값으로 매기면 안 되는 갈 곳 — 이미 다른 자리가 세거나(토공·사토 공제) 참고값이다.
NOT_BILLED_BY_WORK_ITEM = {
    "earthwork": "토공집계",
    "haul_deduction": "유토곡선 사토 공제",
    "reference": "참고 줄",
}


def _pickers(structure: dict[str, Any], mapping: WorkItemMapping) -> list[tuple[str, set[str]]]:
    """이 구조물의 성분을 **이름으로 집는** 공종 줄 — (자리, 이름들). 집는 조건은 각 빌더와 같다."""
    type_id = str(structure.get("type_id") or "")
    entry = mapping.for_structure(type_id) or {}
    composite = mapping.composite_for(type_id)
    found: list[tuple[str, set[str]]] = []
    if entry.get("billing_component"):
        found.append(("구조물 줄", {str(entry["billing_component"])}))
    if composite and not entry.get("work_item_code"):
        for part in composite.get("parts") or []:
            if isinstance(part, dict):
                label = f"묶음 조각 「{part.get('name') or part.get('code')}」"
                found.append((label, {str(name) for name in part.get("from_components") or []}))
    if not composite:
        # 묶음 구조물은 타설·기초잡석 줄이 건너뛴다(조각이 이미 셈) — 빌더와 같은 조건.
        found.append(("콘크리트 타설 줄", set(PLACING_TARGET_NAMES)))
        found.append(("기초잡석 줄", {RUBBLE_GROUP}))
    return found


def verify_double_count_boundaries(
    unit_quantity_table: dict[str, Any] | None,
    material_table: dict[str, Any] | None,
    mapping: WorkItemMapping,
    priced_sheets: list[dict[str, Any]] | None = None,
    work_items: list[dict[str, Any]] | None = None,
) -> list[str]:
    """어긴 자리를 사람 말로 — 비면 경계가 지켜진 것."""
    found: list[str] = []
    for structure in (unit_quantity_table or {}).get("structures") or []:
        label = str(structure.get("name") or structure.get("type_id") or "구조물")
        destinations: dict[str, set[str]] = {}
        for component in structure.get("components") or []:
            if float(component.get("amount") or 0.0) <= 0:
                continue
            name = str(component.get("name") or "").strip()
            destination = str(component.get("destination") or "")
            destinations.setdefault(name, set()).add(destination)
            if name in EARTHWORK_ONLY and destination != "earthwork":
                found.append(
                    f"① {label} 「{name}」 갈 곳이 {destination or '(없음)'} — "
                    "구조물 터파기·되메우기는 토공집계(earthwork)로만 감"
                )
        for where, names in _pickers(structure, mapping):
            for name in sorted(names):
                for destination in sorted(
                    destinations.get(name, set()) & set(NOT_BILLED_BY_WORK_ITEM)
                ):
                    found.append(
                        f"① {label} 「{name}」은 {NOT_BILLED_BY_WORK_ITEM[destination]}"
                        f"({destination})로 가는데 {where}이 또 셈 — 같은 물량이 두 자리에 붙음"
                    )
    for row in (material_table or {}).get("rows") or []:
        name = str(row.get("name") or "").strip()
        if name in MIX_COMPONENTS:
            found.append(
                f"② 자재총괄에 배합 성분 「{name}」 — 시멘트·모래·자갈은 B09 일위대가만 쪼갬(㉢)"
            )
    found.extend(f"③ {warning} (㉠)" for warning in verify_single_surcharge(unit_quantity_table))
    found.extend(_verify_conserved(unit_quantity_table, mapping, priced_sheets, work_items))
    return found


#: 여러 구조물의 성분을 **이름으로 모아 한 줄**로 세우는 인계 줄 — (줄 이름, 모으는 성분 이름들).
#: ⚠ 이 줄들은 구조물마다 「누구 것을 뺐나」를 안 남긴다 — 그래서 물량 보존으로 본다.
SUMMED_ROWS = (
    (RUBBLE_GROUP, frozenset({RUBBLE_GROUP})),
    ("콘크리트 타설", frozenset(PLACING_TARGET_NAMES)),
)
_TOLERANCE_M3 = 1e-6


def _verify_conserved(
    unit_quantity_table: dict[str, Any] | None,
    mapping: WorkItemMapping,
    priced_sheets: list[dict[str, Any]] | None,
    work_items: list[dict[str, Any]] | None,
) -> list[str]:
    """④ 양식 호표·묶음 조각이 품은 성분을 모음 줄이 또 셈 — **물량 보존**으로 판정(2026-09-13).

    모음 줄 + 호표가 품은 몫 + 묶음 조각 ≤ 원단위 성분 합. 넘치면 같은 물량을 두 자리에서 셈
    (검증 프로젝트 찰쌓기: 호표가 품은 기초잡석이 인계 줄로도 나가 8.3㎥ ← 실제 3.5㎥).
    ⚠ 빌더마다 「무엇을 건너뛰나」를 따라 짜지 않는다 — 건너뛰기를 빠뜨린 빌더를 잡는 검사다.
    """
    if not unit_quantity_table or work_items is None:
        return []
    covered = {
        sid: set(entry.get("covered") or [])
        for entry in priced_sheets or []
        for sid in entry.get("structure_ids") or []
    }
    found: list[str] = []
    for row_name, names in SUMMED_ROWS:
        available = held = 0.0
        for structure in unit_quantity_table.get("structures") or []:
            taken = covered.get(str(structure.get("structure_id")), set())
            for component in structure.get("components") or []:
                name = str(component.get("name") or "").strip()
                if name not in names or component.get("unit") != "㎥":
                    continue
                amount = float(component.get("amount") or 0.0)
                available += amount
                held += amount if name in taken else 0.0
        in_parts = sum(
            float(part.get("quantity") or 0.0)
            for row in work_items
            for part in row.get("composite_parts") or []
            if isinstance(part, dict)
            and part.get("from_components")
            and set(part["from_components"]) <= names
        )
        summed = sum(
            float(row.get("quantity") or 0.0)
            for row in work_items
            if row.get("name") == row_name and not row.get("composite_parts")
        )
        if summed + held + in_parts > available + _TOLERANCE_M3:
            found.append(
                f"④ 「{row_name}」 줄 {summed:g}㎥ + 양식 호표가 품은 {held:g}㎥ + 묶음 조각 "
                f"{in_parts:g}㎥ 가 원단위 합 {available:g}㎥ 를 넘음 — 호표·조각이 품은 성분을"
                " 모음 줄이 또 셈"
            )
    return found