Aislo/B09_Estimation/B09_Estimation_CostSheet.py

"""B09 원가계산서 화면 자료 — STmate 「제잡비 계산」(wM_KanJub) **일반형식** 서식 차례 (PLAN 12장).

⚠ 화면만 닮는다 — **값은 우리 엔진**(`Engine_Cost.calculate_cost`)이 낸다. 여기는
  ① 엔진 줄을 STmate 서식 차례·표기(나./ㄱ./1))로 늘어놓고
  ② 기준 입력판 칸(번호·제목은 DFM `TWM_KANJUB` 그대로)의 선택지를 요율 데이터에서 뽑고
  ③ 프로젝트 저장값을 `CostInput` 으로 옮기는 일만 한다. 계산을 두 벌로 짜지 않는다.
⚠ 형식은 「일반」만 — 수공·실적일반·실적수공은 항목·밑수·요율이 달라(35번 문서) 차례로 세움.
"""

from __future__ import annotations

from decimal import Decimal, InvalidOperation
from typing import Any

from B09_Estimation.B09_Estimation_Engine_Cost import CostInput, CostResult
from B09_Estimation.B09_Estimation_Engine_Cost_Options import (
    CUT_BASES,
    CUT_UNITS_KRW,
    OVERHEAD_CLASSES,
    VAT_MODES,
)
from B09_Estimation.B09_Estimation_Rates import RateDataset

_ZERO = Decimal(0)

#: 저장 자리 — `estimation` 구획 안 한 칸.
SETTINGS_KEY = "cost_sheet"

#: 기준 입력판 — DFM 번호·제목 그대로. 엔진이 받는 칸만 연다(나머지 번호는 형식이 서면 채움).
#:   (칸 이름, DFM 제목, 요율 변수 경로, 한글 이름표)
WORK_TYPE_FIELDS: tuple[tuple[str, str], ...] = (
    ("work_type_indirect_labor", "2.공사의종류(주)"),
    ("work_type_safety", "7.산업안전보건관리비"),
    ("environment_work_type", "8.환 경 보 전 비"),
    ("equipment_guarantee_work_type", " 9.건설기계대여금 지급수수료"),
)
AMOUNT_FIELDS: tuple[tuple[str, str], ...] = (
    ("owner_supplied_material_krw", "6.관/사급내용 — 관급자재대(순자재대)"),
    ("procurement_fee_krw", "6.관/사급내용 — 조달수수료"),
    ("indirect_material_krw", "- 간 접 재 료 비"),
    ("profit_adjustment_krw", "이윤 보정액"),
    ("overhead_managed_material_krw", "관리품목 자재대(일반관리비 밑수)"),
    ("tax_exempt_material_krw", "면세품 금액(산림조합-면세품)"),
)
#: 10·11·12 — 선택지가 요율 데이터가 아니라 **엔진 규칙**(`Engine_Cost_Options`)에서 옴.
CHOICE_FIELDS: tuple[tuple[str, str], ...] = (
    ("overhead_class", "10.일반관리비"),
    ("cut_basis", "11.공급가절사기준"),
    ("cut_unit_krw", "11.절사 단위"),
    ("vat_mode", "12.부 가 가 치 세"),
)
#: DFM `12.부 가 가 치 세` 선택지 표기 그대로.
VAT_MODE_LABELS = {
    "supply": "(0) 공급가액의 10%",
    "material": "(1) 재료비의 10%",
    "none": "(2) 없음",
    "forest_coop": "(3) 산림조합 형식",
    "forest_coop_exempt": "(4) 산림조합-면세품",
}


def choice_options() -> dict[str, list[dict[str, str]]]:
    """10·11·12 선택지. 절사 「안 함」이 첫째 = 기본(이윤을 자동으로 안 깎음)."""
    return {
        "overhead_class": [{"value": k, "label": v} for k, v in OVERHEAD_CLASSES.items()],
        "cut_basis": [
            {"value": "", "label": "절사 안 함(이윤보정액 입력)"},
            *({"value": k, "label": f"{v}에서 조정"} for k, v in CUT_BASES.items()),
        ],
        "cut_unit_krw": [{"value": str(u), "label": f"{u:,}원 미만"} for u in CUT_UNITS_KRW],
        "vat_mode": [{"value": k, "label": VAT_MODE_LABELS[k]} for k in VAT_MODES],
    }


#: 공종 이름표 — 조달청 제비율 적용기준(현행 2026-04-13) 표기.
WORK_TYPE_LABELS: dict[str, str] = {
    "civil": "토목",
    "landscape": "조경",
    "industrial_facilities_civil": "산업설비(토목)",
    "building": "건축공사",
    "heavy_construction": "중건설공사",
    "special_construction": "특수건설공사",
    "civil_road": "토목 — 도로(교량·터널·활주로 등)",
    "civil_plant": "토목 — 플랜트(발전소·쓰레기소각장 등)",
    "civil_subway": "토목 — 지하철",
    "civil_railway": "토목 — 철도",
    "civil_water_and_sewer": "토목 — 상하수도",
    "civil_port": "토목 — 항만",
    "civil_port_with_silt_screen": "토목 — 항만(오탁방지막·준설토방지막 설치)",
    "civil_dam": "토목 — 댐",
    "civil_land_development": "토목 — 택지개발",
    "civil_other": "토목 — 그 밖의 토목공사(하천 등)",
    "building_housing_redevelopment": "건축 — 주택(재개발·재건축)",
    "building_new_housing": "건축 — 주택(신축)",
    "building_other": "건축 — 그 밖의 건축",
    "civil_general": "일반건설 — 토목",
    "industrial_facilities": "일반건설 — 산업설비",
}
#: 산업안전보건관리비 공종은 「토목공사」 표기(간접노무비 「토목」과 키가 같아 따로 둠).
SAFETY_LABELS = {"civil": "토목공사"}


def _unique(values: list[str]) -> list[str]:
    return list(dict.fromkeys(values))


def field_options(dataset: RateDataset) -> dict[str, list[dict[str, str]]]:
    """기준 입력판 고르개 선택지 — **요율 데이터에 있는 공종만**(코드에 목록을 박지 않음)."""
    variables = dataset.variables
    sources = {
        "work_type_indirect_labor": [
            row["work_type"] for row in variables["rate_indirect_labor"]["brackets"]
        ],
        "work_type_safety": [row["work_type"] for row in variables["rate_safety_pct"]["brackets"]],
        "environment_work_type": [
            row["work_type"] for row in variables["rate_environment"]["all_work_types"]
        ],
        "equipment_guarantee_work_type": [
            row["work_type"]
            for row in variables["rate_equipment_payment_guarantee"]["general_construction"]
        ],
    }
    options: dict[str, list[dict[str, str]]] = {}
    for key, values in sources.items():
        labels = {**WORK_TYPE_LABELS, **(SAFETY_LABELS if key == "work_type_safety" else {})}
        options[key] = [{"value": v, "label": labels.get(v, v)} for v in _unique(values)]
    return {**options, **choice_options()}


def _decimal(value: Any) -> Decimal:
    try:
        number = Decimal(str(value))
    except (InvalidOperation, ValueError):
        return _ZERO
    return number if number >= 0 else _ZERO


def clean_settings(values: dict[str, Any], dataset: RateDataset) -> dict[str, Any]:
    """저장할 값 — 데이터에 없는 공종·음수·빈칸은 버린다(버린 칸은 엔진 기본값이 섬)."""
    options = field_options(dataset)
    cleaned: dict[str, Any] = {}
    for key, _ in (*WORK_TYPE_FIELDS, *CHOICE_FIELDS):
        value = str(values.get(key) or "")
        if value and value in {option["value"] for option in options[key]}:
            cleaned[key] = int(value) if key == "cut_unit_krw" else value
    for key, _ in AMOUNT_FIELDS:
        if values.get(key) not in (None, ""):
            cleaned[key] = str(_decimal(values.get(key)))
    try:
        days = int(values.get("duration_days"))
    except (TypeError, ValueError):
        days = 0
    if days > 0:
        cleaned["duration_days"] = days
    return cleaned


def cost_input(
    direct: dict[str, Decimal],
    stored: dict[str, Any],
    waste_krw: Decimal = _ZERO,
    waste_separate_order: bool = False,
) -> CostInput:
    """내역서 직접비 + 저장값 → 엔진 입력. 저장 안 한 칸은 **엔진 기본값**."""
    kwargs: dict[str, Any] = {
        key: stored[key] for key, _ in (*WORK_TYPE_FIELDS, *CHOICE_FIELDS) if stored.get(key)
    }
    kwargs.update(
        {key: _decimal(stored[key]) for key, _ in AMOUNT_FIELDS if stored.get(key) is not None}
    )
    if stored.get("duration_days"):
        kwargs["duration_days"] = int(stored["duration_days"])
    return CostInput(
        direct_material_krw=direct["material"],
        direct_labor_krw=direct["labor"],
        direct_expense_krw=direct["expense"],
        waste_disposal_krw=waste_krw,
        waste_separate_order=waste_separate_order,
        **kwargs,
    )


#: 경비 세목 — STmate 일반형식 서식 차례·표기. 키는 우리 엔진 줄.
EXPENSE_ORDER: tuple[tuple[str, str], ...] = (
    ("industrial_accident_insurance", "산재보험료"),
    ("employment_insurance", "고용보험료"),
    ("health_insurance", "건강보험료"),
    ("long_term_care_insurance", "장기요양보험료"),
    ("national_pension", "연금보험료"),
    ("retirement_mutual_aid", "퇴직공제부금비"),
    ("safety_management_cost", "산업안전보건관리비"),
    ("other_expense", "기타 경비"),
    ("performance_guarantee_fee", "공사이행 수수료"),
    ("subcontract_payment_guarantee", "하도대금 수수료"),
    ("environment_preservation", "환 경 보 전 비"),
    ("equipment_payment_guarantee", "건설기계 대여금"),
    ("asbestos_contribution", "석면분담금"),
    ("wage_claim_contribution", "임금채권부담금"),
    ("waste_disposal", "폐기물처리비"),
)


def _row(
    mark: str, label: str, amount: Decimal, level: int, key: str, result: CostResult | None = None
) -> dict[str, Any]:
    line = result.line(key) if result is not None and result.has(key) else None
    return {
        "key": key,
        "mark": mark,
        "label": label,
        "level": level,
        "formula": line.formula_text if line else "",
        "rate_percent": None
        if line is None or line.rate_percent is None
        else str(line.rate_percent),
        "amount_krw": str(amount),
        "note": line.note if line else "",
        "total": level == 0,
    }


def sheet_rows(result: CostResult, data: CostInput) -> list[dict[str, Any]]:
    """일반형식 서식 줄 — 나.순공사원가 … 바.총공사비. 엔진에 없는 줄은 안 세움."""
    amount = result.amount
    rows = [
        _row(
            "나.",
            "순공사원가(ㄱ+ㄴ+ㄷ)",
            amount("net_construction_cost"),
            0,
            "net_construction_cost",
            result,
        ),
        _row("ㄱ.", "재 료 비", amount("material_cost"), 1, "material_cost", result),
        _row("ㄴ.", "노 무 비", amount("labor_cost"), 1, "labor_cost", result),
        _row("1)", "직접노무비", data.direct_labor_krw, 2, "direct_labor"),
        _row("2)", "간접노무비", amount("indirect_labor_cost"), 2, "indirect_labor_cost", result),
        _row("ㄷ.", "경 비", amount("expense"), 1, "expense", result),
        _row("1)", "내역서경비", data.direct_expense_krw, 2, "direct_expense"),
    ]
    number = 2
    for key, label in EXPENSE_ORDER:
        if not result.has(key) or (key == "waste_disposal" and data.waste_separate_order):
            continue
        rows.append(_row(f"{number})", label, amount(key), 2, key, result))
        if key == "safety_management_cost":
            # STmate 도 두 비교 줄을 나란히 둠 — 작은 쪽 채택.
            for sub in ("safety_management_cost_a", "safety_management_cost_b"):
                if result.has(sub):
                    rows.append(_row("", result.line(sub).name, amount(sub), 3, sub, result))
        number += 1
    overhead = amount("general_overhead")
    rows.append(_row("5)", "일반관리비", overhead, 1, "general_overhead", result))
    rows.append(
        {
            **_row("다.", "소 계", amount("net_construction_cost") + overhead, 0, "subtotal"),
            "formula": "나.순공사원가 + 5)일반관리비",
        }
    )
    if result.has("profit_adjustment"):
        rows.append(
            _row(
                "",
                "이윤(보정 전)",
                amount("profit_before_adjustment"),
                2,
                "profit_before_adjustment",
                result,
            )
        )
        rows.append(
            _row("", "이윤 보정액", amount("profit_adjustment"), 2, "profit_adjustment", result)
        )
    profit_row = _row("6)", "이윤", amount("profit"), 1, "profit", result)
    if result.has("profit_before_adjustment"):
        before = result.line("profit_before_adjustment")
        profit_row.update(formula=before.formula_text, rate_percent=str(before.rate_percent))
    rows.append(profit_row)
    rows.append(_row("라.", "공급가액", amount("total_cost"), 0, "total_cost", result))
    rows.append(_row("", "부가가치세", amount("vat"), 1, "vat", result))
    rows.append(_row("마.", "도급공사비", amount("contract_amount"), 0, "contract_amount", result))
    if result.has("owner_supplied_material_total"):
        rows.append(
            _row(
                "",
                "관급자재대",
                amount("owner_supplied_material_total"),
                1,
                "owner_supplied_material_total",
                result,
            )
        )
    if data.waste_separate_order and result.has("waste_disposal"):
        rows.append(
            _row(
                "", "폐기물처리비(분리발주)", amount("waste_disposal"), 1, "waste_disposal", result
            )
        )
    rows.append(_row("바.", "총공사비", amount("grand_total"), 0, "grand_total", result))
    return rows


def status_line(result: CostResult, data: CostInput) -> dict[str, str]:
    """아래 상태줄 — DFM `직재: / 직노: / 산경: / 이윤:` 차례."""
    return {
        "직재": str(data.direct_material_krw),
        "직노": str(data.direct_labor_krw),
        "산경": str(data.direct_expense_krw),
        "이윤": str(result.amount("profit")),
    }