Aislo/B08_Quantity/B08_Quantity_Engine_Handoff.py

"""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

from typing import Any, Iterable

from B08_Quantity.B08_Quantity_Engine_BasisUnit import verify_unit_matches_basis

# ⚠ 파일만 갈랐고 **계약은 그대로다** — 종전에 이 이름으로 가져다 쓰던 곳이 그대로 돌게
#   여기서 다시 내보낸다(2026-09-08 분리).
from B08_Quantity.B08_Quantity_Engine_Handoff_Mapping import (  # noqa: F401
    BLOCKED_FORMULA_MISSING,
    BLOCKED_INPUT_MISSING,
    BLOCKED_UNIT_DATA_MISSING,
    METHOD_TO_GROUND,
    NOTE_METHOD_MISSING,
    ORIGIN_EARTHWORK,
    ORIGIN_HAUL,
    ORIGIN_PIPE,
    ORIGIN_PREPARATION,
    ORIGIN_SLOPE,
    ORIGIN_STRUCTURE,
    REBAR_PREFIXES,
    SLOPE_GROUPS,
    SUBTOTAL_GROUPS,
    WorkItemMapping,
    composite_quantities,
    euroform_type,
    load_mapping,
    load_masonry_table,
    load_rebar_table,
    load_timber_table,
    masonry_class,
    normalize_kind_key,
    placing_code,
    rebar_complexity,
    structure_kind,
    timber_class,
)
from B08_Quantity.B08_Quantity_Engine_Handoff_Rows import (  # noqa: F401
    PLACING_TARGET_NAMES,
    _earthwork_rows,
    _haul_rows,
    _length_rows,
    _material_rows,
    _pipe_rows,
    _placing_rows,
    _preparation_rows,
    _structure_rows,
    blocked_of,
)
from common_util.common_util_quantity_spread import spread_by_unit


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,
    preparation_table: dict[str, Any] | None = None,
    length_table: list[dict[str, Any]] | None = None,
    pipe_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,
    concrete_placing_method: str | 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)

    # 준비공·사방공 — **못 내는 줄도 사유와 함께** 보낸다(빼면 빠진 줄이 안 보인다).
    # B군 종단배수 — 겹침을 합친 연장으로 종류별 한 줄(위 `_length_rows` 주석).
    work_items.extend(_length_rows(length_table or [], table))
    work_items.extend(_preparation_rows(preparation_table or {}))
    # 배수관 — 정본 셋(관 지점·측점 연장·매핑)을 이은 결과. 못 서는 줄도 사유와 함께 감.
    work_items.extend(_pipe_rows(pipe_table or {}))

    # 콘크리트 타설 — 품은 이 줄, 재료는 자재 쪽. 겹치지 않는다(위 `_placing_rows` 주석).
    placing_rows, placing_notes = _placing_rows(
        unit_quantity_table or {}, table, concrete_placing_method
    )
    work_items.extend(placing_rows)

    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"]),
    }
    # ⚠ 검사는 **실제로 부른다** — 만들어 두고 안 부르면 없는 것과 같다.
    # 2026-09-08 ㉘ 자기 감사: 아래 줄 하나만 이어져 있고 형제 둘은 **시험에서만** 불리고
    #   있었다. B09 에서 같은 병(가드 둘이 놀고 있음)을 지적해 놓고 내 쪽도 같았다.
    result["ratio_math_warnings"] = verify_ratio_math(result)
    result["material_code_warnings"] = verify_no_code_on_materials(result)
    result["bill_flag_warnings"] = verify_bill_flags(result)
    # ⚠ 보내는 단위가 **품셈 밑수**와 같은가 — 받는 쪽이 그대로 곱하는 자리다(2026-09-08).
    result["basis_unit_warnings"] = verify_unit_matches_basis(
        work_items, extra=table.declared_units()
    )
    # ⚠ 채집석 공제 — **양수 ㎥ 로 넘기기만** 한다. 빼는 자리는 유토곡선의 사토뿐이다
    #   (2026-09-09 세 창 확정 · 부호를 넘기면 두 번 뒤집힌다).
    result["collected_stone_deduction_m3"] = float(
        (unit_quantity_table or {}).get("collected_stone_deduction_m3") or 0.0
    )
    result["placing_notes"] = placing_notes
    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"])