From 8b7325a28e0881e8e39ee2d123e2250fbf6387cd Mon Sep 17 00:00:00 2001 From: umsangdon Date: Mon, 7 Sep 2026 19:49:00 +0900 Subject: [PATCH] =?UTF-8?q?feat(B09):=20=EC=9B=90=EA=B0=80=EA=B3=84?= =?UTF-8?q?=EC=82=B0=20=EC=97=94=EC=A7=84=C2=B7=EC=9A=94=EC=9C=A8=20?= =?UTF-8?q?=EB=A1=9C=EB=8D=94=20=EB=BC=88=EB=8C=80=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - `B09_Estimation_Rates.py` — 요율 데이터(`resources/data_cost_input_value/rates_*.json`) 로더 + 구간 조회. 요율을 코드에 안 박음. 구간 라벨의 `billion` = 십억 원. 구간 판정 실패 시 기본값으로 안 때우고 `RateLookupError` 로 멈춤. - `B09_Estimation_Engine_Cost.py` — ⑤ 공사원가계산서 계산. 순공사비를 받아 법정경비·일반관리비·이윤·부가세를 얹음. 수량·단가와 무관하게 홀로 돎. - 지킨 것 (PLAN 8-9·8-10, 실무 원가계산서 재현으로 확인된 것만) · 모든 줄 원 단위 버림 · 밑수가 항목마다 갈림(직노 / 직노+간노 / 건강보험료 / …) · 안전관리비 A·B 두 값 중 작은 쪽 — 대상액이 구간 경계를 넘으면 A 가 더 커짐 · 이윤 수동 조정액은 설계자 명시 입력일 때만 · 비목 목록을 코드에 안 박음 - 검사 16건 별도(`tmp/tests/test_b09_cost_engine.py`, git 밖) 전건 통과. 실무 실측 두 방향(울진 A 채택 · 거창 B 채택)을 고정값으로 씀. - 기존 파일 무수정 · DB 미사용(기준자료는 파일, PLAN 8-5·9-2). Co-Authored-By: Claude Opus 5 (1M context) --- B09_Estimation/B09_Estimation_Engine_Cost.py | 679 +++++++++++++++++++ B09_Estimation/B09_Estimation_Rates.py | 245 +++++++ 2 files changed, 924 insertions(+) create mode 100644 B09_Estimation/B09_Estimation_Engine_Cost.py create mode 100644 B09_Estimation/B09_Estimation_Rates.py diff --git a/B09_Estimation/B09_Estimation_Engine_Cost.py b/B09_Estimation/B09_Estimation_Engine_Cost.py new file mode 100644 index 00000000..e5ca8445 --- /dev/null +++ b/B09_Estimation/B09_Estimation_Engine_Cost.py @@ -0,0 +1,679 @@ +"""B09 원가계산 — ⑤ 공사원가계산서 엔진. + +순공사비(직접재료비·직접노무비·직접경비)를 받아 법정경비·일반관리비·이윤·부가세를 얹어 +**공사원가계산서 한 장**을 만든다. 수량·단가와 무관하게 홀로 도는 계산이다 (PLAN 9-5). + +지켜야 할 것 (PLAN 8-9·8-10 — 실무 원가계산서 재현으로 확인된 것만) + 1. **모든 줄은 원 단위 버림**(ROUNDDOWN). 반올림이 아니다. + 2. **밑수가 항목마다 갈린다** — 직노 / 직노+간노 / 건강보험료 / 재료비+직노+관급항 / … + 하나로 뭉치면 틀린다. + 3. **안전관리비 = A·B 두 값을 다 내고 작은 쪽**(고용노동부 고시 제2025-11호). + ⚠ **A 가 항상 작지 않다** — 관급을 넣어 대상액이 구간 경계를 넘으면 뒤집힌다 + (실증: 울진 A 채택 / 거창 B 채택). + 4. **이윤 수동 조정액** — 실무는 도급공사비 끝수를 맞추려 이윤을 깎는다. 법에 없는 + 관행이므로 **설계자가 명시로 넣을 때만** 적용하고 프로그램이 스스로 깎지 않는다. + 5. **비목 목록을 코드에 박지 않는다** — 공사마다 있는 줄이 다르다(퇴직공제·폐기물처리 등). + 6. 요율은 전부 `B09_Estimation_Rates` 를 거쳐 데이터에서 읽는다. 코드에 숫자가 없다. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal + +from B09_Estimation.B09_Estimation_Rates import ( + RateDataset, + RateLookupError, + base_amount, + flat_rate, + load_rate_dataset, + pension_rate_percent, + rate_percent, + select_bracket, +) + +_ZERO = Decimal(0) +_HUNDRED = Decimal(100) +_VAT_DIVISOR = Decimal("1.1") + +#: 기본으로 켜는 법정경비 비목. 공사마다 다르므로 `CostInput.enabled_items` 로 갈아끼운다. +DEFAULT_STATUTORY_ITEMS: tuple[str, ...] = ( + "industrial_accident_insurance", + "employment_insurance", + "health_insurance", + "long_term_care_insurance", + "national_pension", + "safety_management_cost", + "other_expense", + "environment_preservation", + "retirement_mutual_aid", +) + +#: 켤 수 있으나 기본은 끄는 비목 (공사·발주처에 따라 등장). +OPTIONAL_STATUTORY_ITEMS: tuple[str, ...] = ( + "wage_claim_contribution", + "asbestos_contribution", + "equipment_payment_guarantee", + "subcontract_payment_guarantee", + "performance_guarantee_fee", +) + + +def floor_won(value: Decimal) -> Decimal: + """원 단위 버림 — 원가계산서 모든 줄의 기본 처리.""" + return value.quantize(Decimal(1), rounding=ROUND_FLOOR) + + +def ceil_thousand(value: Decimal) -> Decimal: + """천원 올림 — 관급자재대 표기.""" + return (value / 1000).quantize(Decimal(1), rounding=ROUND_CEILING) * 1000 + + +@dataclass +class CostInput: + """원가계산 입력. + + 금액은 전부 원 단위 `Decimal`. 요율·구간 판정에 쓰는 조건이 함께 들어온다. + """ + + direct_material_krw: Decimal + direct_labor_krw: Decimal + direct_expense_krw: Decimal + indirect_material_krw: Decimal = _ZERO + + #: 구간 판정용 공종·기간. `work_type` 은 요율 데이터의 값을 그대로 쓴다. + work_type_indirect_labor: str = "civil" + work_type_safety: str = "civil" + duration_days: int = 183 + pension_year: int = 2026 + + #: 관급자재 — 순자재대와 조달수수료를 나눠 받는다(순환 정의 방지, 원가계산_체계 §1). + owner_supplied_material_krw: Decimal = _ZERO + procurement_fee_krw: Decimal = _ZERO + include_fee_in_owner_material_total: bool = True + + #: 안전관리비 대상액에 들어가는 **도급자설치 관급금액**. None 이면 관급 전액을 쓴다. + owner_supplied_for_safety_krw: Decimal | None = None + #: 위 금액이 부가세 포함인가 — 포함이면 1.1 로 나눠 부가세를 뺀다(규정: 부가세 제외 기준). + owner_supplied_includes_vat: bool = True + + #: 규모 구간 판정에 쓸 금액. None 이면 순공사원가를 쓴다(추정가격 순환 회피). + estimated_price_krw: Decimal | None = None + + #: 이윤 수동 조정액 — 설계자가 명시로 넣을 때만. 프로그램이 스스로 채우지 않는다. + profit_adjustment_krw: Decimal = _ZERO + + #: 환경보전비 공종(요율 데이터 `rate_environment.all_work_types` 의 값). + environment_work_type: str = "civil_road" + #: 건설기계대여대금 지급보증 공종. + equipment_guarantee_work_type: str = "civil_general" + + enabled_items: tuple[str, ...] = DEFAULT_STATUTORY_ITEMS + + #: 요율 데이터 파일명. 연도를 갈아끼우는 자리. + rate_file_name: str = "rates_2026.json" + + +@dataclass +class CostLine: + """원가계산서 한 줄 — 화면이 「밑수 · 율 · 금액」 셋을 다 보이므로 셋을 다 든다.""" + + key: str + name: str + base_label: str + base_amount_krw: Decimal + rate_percent: Decimal | None + flat_amount_krw: Decimal + amount_krw: Decimal + note: str = "" + + +@dataclass +class CostResult: + lines: list[CostLine] = field(default_factory=list) + totals: dict[str, Decimal] = field(default_factory=dict) + rate_version: dict[str, str] = field(default_factory=dict) + notes: list[str] = field(default_factory=list) + + def line(self, key: str) -> CostLine: + for item in self.lines: + if item.key == key: + return item + raise KeyError(f"원가계산서에 없는 줄입니다: {key}") + + def amount(self, key: str) -> Decimal: + return self.line(key).amount_krw + + +def _line( + result: CostResult, + *, + key: str, + name: str, + base_label: str, + base: Decimal, + percent: Decimal | None = None, + flat: Decimal = _ZERO, + amount: Decimal | None = None, + note: str = "", +) -> Decimal: + """줄 하나를 계산해 결과에 담고 금액을 돌려준다. 금액은 항상 원 단위 버림.""" + if amount is None: + computed = base * (percent or _ZERO) / _HUNDRED + flat + amount = floor_won(computed) + result.lines.append( + CostLine( + key=key, + name=name, + base_label=base_label, + base_amount_krw=base, + rate_percent=percent, + flat_amount_krw=flat, + amount_krw=amount, + note=note, + ) + ) + return amount + + +def _safety_management_cost( + result: CostResult, + dataset: RateDataset, + data: CostInput, + *, + material_cost: Decimal, +) -> Decimal: + """산업안전보건관리비 — A·B 두 값을 다 내고 **작은 쪽**을 채택한다. + + A) (재료비 + 직접노무비 + 도급자설치 관급금액) × 요율 + 기초액 + B) ((재료비 + 직접노무비) × 요율 + 기초액) × 1.2 + 두 대상액이 **다른 구간에 떨어질 수 있어** A 가 항상 작지는 않다. + """ + variable = dataset.variable("rate_safety_pct") + brackets = variable["brackets"] + + owner_supplied = data.owner_supplied_for_safety_krw + if owner_supplied is None: + owner_supplied = data.owner_supplied_material_krw + if data.owner_supplied_includes_vat: + owner_supplied = owner_supplied / _VAT_DIVISOR + + base_with = material_cost + data.direct_labor_krw + owner_supplied + base_without = material_cost + data.direct_labor_krw + + def evaluate( + base: Decimal, *, multiplier: Decimal, label: str + ) -> tuple[Decimal, Decimal, Decimal]: + row = select_bracket( + brackets, + amount_field="target_amount_bracket", + amount=base, + equals={"work_type": data.work_type_safety}, + label=label, + ) + percent = rate_percent(row, label=label) + flat = base_amount(row) + amount = floor_won((base * percent / _HUNDRED + flat) * multiplier) + return amount, percent, flat + + amount_a, percent_a, flat_a = evaluate( + base_with, multiplier=Decimal(1), label="안전관리비 A(관급 포함)" + ) + amount_b, percent_b, flat_b = evaluate( + base_without, multiplier=Decimal("1.2"), label="안전관리비 B(관급 제외 × 1.2)" + ) + + adopted = "A" if amount_a <= amount_b else "B" + + _line( + result, + key="safety_management_cost_a", + name="산업안전보건관리비 A(관급 포함)", + base_label="재료비+직접노무비+도급자설치 관급금액(부가세 제외)", + base=base_with, + percent=percent_a, + flat=flat_a, + amount=amount_a, + note="채택" if adopted == "A" else "미채택", + ) + _line( + result, + key="safety_management_cost_b", + name="산업안전보건관리비 B(관급 제외 × 1.2)", + base_label="(재료비+직접노무비) × 요율 + 기초액, 그 값의 1.2배", + base=base_without, + percent=percent_b, + flat=flat_b, + amount=amount_b, + note="채택" if adopted == "B" else "미채택", + ) + + adopted_amount = min(amount_a, amount_b) + return _line( + result, + key="safety_management_cost", + name="산업안전보건관리비", + base_label=f"A·B 중 작은 금액 (채택 = {adopted})", + base=base_with if adopted == "A" else base_without, + percent=percent_a if adopted == "A" else percent_b, + amount=adopted_amount, + note="고용노동부 고시 제2025-11호 — 둘 중 작은 금액", + ) + + +def _statutory_expenses( + result: CostResult, + dataset: RateDataset, + data: CostInput, + *, + material_cost: Decimal, + total_labor_cost: Decimal, + direct_construction_cost: Decimal, +) -> Decimal: + """법정경비 묶음. `enabled_items` 에 든 줄만 계산한다.""" + enabled = set(data.enabled_items) + total = _ZERO + + if "industrial_accident_insurance" in enabled: + total += _line( + result, + key="industrial_accident_insurance", + name="산재보험료", + base_label="노무비(직접+간접)", + base=total_labor_cost, + percent=flat_rate(dataset, "rate_sanjae"), + ) + + if "employment_insurance" in enabled: + variable = dataset.variable("rate_goyong") + # 고용보험료는 등급(1~7)이 추정가격으로 갈린다. 임도는 대개 고시 기준금액 미만이라 + # 숫자 구간에 안 걸리므로 잔여 구간을 이름으로 지정한다(등급 7·그 이하 모두 1.01 %). + row = select_bracket( + variable["brackets"], + amount_field="estimated_amount_bracket", + amount=_scale_reference(data, direct_construction_cost), + residual_label="below_official_threshold", + label="고용보험료", + ) + total += _line( + result, + key="employment_insurance", + name="고용보험료", + base_label="노무비(직접+간접)", + base=total_labor_cost, + percent=rate_percent(row, label="고용보험료"), + ) + + health_amount = _ZERO + if "health_insurance" in enabled: + health_amount = _line( + result, + key="health_insurance", + name="국민건강보험료", + base_label="직접노무비", + base=data.direct_labor_krw, + percent=flat_rate(dataset, "rate_health"), + ) + total += health_amount + + if "long_term_care_insurance" in enabled: + if "health_insurance" not in enabled: + raise RateLookupError( + "노인장기요양보험료는 건강보험료를 밑수로 씁니다 — 건강보험료를 켜야 합니다" + ) + total += _line( + result, + key="long_term_care_insurance", + name="노인장기요양보험료", + base_label="국민건강보험료", + base=health_amount, + percent=flat_rate(dataset, "rate_care"), + ) + + if "national_pension" in enabled: + total += _line( + result, + key="national_pension", + name="국민연금보험료", + base_label="직접노무비", + base=data.direct_labor_krw, + percent=pension_rate_percent(dataset, data.pension_year), + ) + + if "safety_management_cost" in enabled: + total += _safety_management_cost(result, dataset, data, material_cost=material_cost) + + if "other_expense" in enabled: + variable = dataset.variable("rate_other_expense") + row = select_bracket( + variable["brackets"], + amount_field="direct_cost_bracket", + amount=direct_construction_cost, + duration_days=data.duration_days, + equals={"work_type": data.work_type_indirect_labor}, + label="기타경비", + ) + total += _line( + result, + key="other_expense", + name="기타경비", + base_label="재료비+노무비(직접+간접)", + base=material_cost + total_labor_cost, + percent=rate_percent(row, label="기타경비"), + ) + + if "environment_preservation" in enabled: + variable = dataset.variable("rate_environment") + threshold = Decimal(str(variable.get("minimum_estimated_amount_krw", 0))) + if _scale_reference(data, direct_construction_cost) >= threshold: + row = next( + ( + r + for r in variable["all_work_types"] + if r.get("work_type") == data.environment_work_type + ), + None, + ) + if row is None: + raise RateLookupError( + f"환경보전비: 공종을 못 찾았습니다 — {data.environment_work_type}" + ) + total += _line( + result, + key="environment_preservation", + name="환경보전비", + base_label="직접공사비", + base=direct_construction_cost, + percent=rate_percent(row, label="환경보전비"), + note=( + "⚠ 임도 공종 채택값 미확정 — 지식DB " + "`rate_environment.forest_road_selection_status: pending`" + ), + ) + + if "retirement_mutual_aid" in enabled: + variable = dataset.variable("rate_retirement_mutual_aid") + threshold = Decimal(str(variable.get("minimum_estimated_amount_krw", 0))) + if _scale_reference(data, direct_construction_cost) >= threshold: + total += _line( + result, + key="retirement_mutual_aid", + name="퇴직공제부금비", + base_label="직접노무비", + base=data.direct_labor_krw, + percent=Decimal(str(variable["rate_percent"])), + ) + + if "wage_claim_contribution" in enabled: + total += _line( + result, + key="wage_claim_contribution", + name="임금채권보장기금 부담금", + base_label="노무비(직접+간접)", + base=total_labor_cost, + percent=flat_rate(dataset, "rate_wage_claim_contribution"), + ) + + if "asbestos_contribution" in enabled: + total += _line( + result, + key="asbestos_contribution", + name="석면피해구제 분담금", + base_label="노무비(직접+간접)", + base=total_labor_cost, + percent=flat_rate(dataset, "rate_asbestos_contribution"), + ) + + if "equipment_payment_guarantee" in enabled: + variable = dataset.variable("rate_equipment_payment_guarantee") + row = next( + ( + r + for r in variable["general_construction"] + variable["specialty_construction"] + if r.get("work_type") == data.equipment_guarantee_work_type + ), + None, + ) + if row is None: + raise RateLookupError( + "건설기계대여대금 지급보증: 공종을 못 찾았습니다 — " + f"{data.equipment_guarantee_work_type}" + ) + total += _line( + result, + key="equipment_payment_guarantee", + name="건설기계대여대금 지급보증수수료", + base_label="직접공사비", + base=direct_construction_cost, + percent=rate_percent(row, label="건설기계대여대금 지급보증수수료"), + ) + + if "subcontract_payment_guarantee" in enabled: + variable = dataset.variable("rate_subcontract_payment_guarantee") + row = select_bracket( + variable["brackets"], + amount_field="estimated_price_bracket", + amount=_scale_reference(data, direct_construction_cost), + label="하도급대금 지급보증수수료", + ) + total += _line( + result, + key="subcontract_payment_guarantee", + name="하도급대금 지급보증수수료", + base_label="직접공사비", + base=direct_construction_cost, + percent=rate_percent(row, label="하도급대금 지급보증수수료"), + ) + + return total + + +def _scale_reference(data: CostInput, direct_construction_cost: Decimal) -> Decimal: + """규모 구간 판정 기준액. + + 조달청 제비율표는 「추정가격」으로 구간을 나누지만, 추정가격은 원가 계산 결과에 + 딸려 나오므로 그대로 쓰면 순환이 된다. 설계자가 추정가격을 명시하면 그 값을, + 없으면 **직접공사비**를 기준으로 쓴다. + """ + if data.estimated_price_krw is not None: + return data.estimated_price_krw + return direct_construction_cost + + +def calculate_cost(data: CostInput) -> CostResult: + """공사원가계산서 한 장을 계산한다.""" + dataset = load_rate_dataset(data.rate_file_name) + result = CostResult(rate_version=dataset.version_stamp) + + material_cost = data.direct_material_krw + data.indirect_material_krw + _line( + result, + key="material_cost", + name="재료비", + base_label="직접재료비+간접재료비", + base=material_cost, + amount=material_cost, + ) + + direct_construction_cost = material_cost + data.direct_labor_krw + data.direct_expense_krw + + indirect_labor_row = select_bracket( + dataset.variable("rate_indirect_labor")["brackets"], + amount_field="direct_cost_bracket", + amount=direct_construction_cost, + duration_days=data.duration_days, + equals={"work_type": data.work_type_indirect_labor}, + label="간접노무비", + ) + indirect_labor = _line( + result, + key="indirect_labor_cost", + name="간접노무비", + base_label="직접노무비", + base=data.direct_labor_krw, + percent=rate_percent(indirect_labor_row, label="간접노무비"), + ) + total_labor_cost = data.direct_labor_krw + indirect_labor + _line( + result, + key="labor_cost", + name="노무비", + base_label="직접노무비+간접노무비", + base=total_labor_cost, + amount=total_labor_cost, + ) + + statutory = _statutory_expenses( + result, + dataset, + data, + material_cost=material_cost, + total_labor_cost=total_labor_cost, + direct_construction_cost=direct_construction_cost, + ) + expense_total = data.direct_expense_krw + statutory + _line( + result, + key="expense", + name="경비", + base_label="직접경비(산출경비)+법정경비", + base=expense_total, + amount=expense_total, + ) + + net_construction_cost = material_cost + total_labor_cost + expense_total + _line( + result, + key="net_construction_cost", + name="순공사원가", + base_label="재료비+노무비+경비", + base=net_construction_cost, + amount=net_construction_cost, + ) + + scale = _scale_reference(data, direct_construction_cost) + overhead_row = select_bracket( + dataset.variable("rate_overhead")["civil_landscape_industrial"], + amount_field="estimated_price_bracket", + amount=scale, + label="일반관리비", + ) + overhead = _line( + result, + key="general_overhead", + name="일반관리비", + base_label="순공사원가", + base=net_construction_cost, + percent=rate_percent(overhead_row, label="일반관리비"), + ) + + profit_row = select_bracket( + dataset.variable("rate_profit")["brackets"], + amount_field="estimated_price_bracket", + amount=scale, + label="이윤", + ) + profit_base = total_labor_cost + expense_total + overhead + profit_before = floor_won(profit_base * rate_percent(profit_row, label="이윤") / _HUNDRED) + _line( + result, + key="profit_before_adjustment", + name="이윤(조정 전)", + base_label="노무비+경비+일반관리비 (재료비 제외)", + base=profit_base, + percent=rate_percent(profit_row, label="이윤"), + amount=profit_before, + ) + if data.profit_adjustment_krw: + _line( + result, + key="profit_adjustment", + name="이윤 조정액", + base_label="설계자 명시 입력", + base=_ZERO, + amount=-data.profit_adjustment_krw, + note="도급공사비 끝수 맞춤 — 법정 항목 아님", + ) + profit = profit_before - data.profit_adjustment_krw + _line( + result, + key="profit", + name="이윤", + base_label="조정 전 이윤 − 조정액", + base=profit_base, + amount=profit, + ) + + total_cost = net_construction_cost + overhead + profit + _line( + result, + key="total_cost", + name="총원가", + base_label="순공사원가+일반관리비+이윤", + base=total_cost, + amount=total_cost, + ) + + vat = _line( + result, + key="vat", + name="부가가치세", + base_label="총원가", + base=total_cost, + percent=flat_rate(dataset, "rate_vat"), + ) + contract_amount = total_cost + vat + _line( + result, + key="contract_amount", + name="도급공사비", + base_label="총원가+부가가치세", + base=contract_amount, + amount=contract_amount, + ) + + owner_total = _ZERO + if data.owner_supplied_material_krw: + raw = data.owner_supplied_material_krw + if data.include_fee_in_owner_material_total: + raw = raw + data.procurement_fee_krw + owner_total = ceil_thousand(raw) + _line( + result, + key="owner_supplied_material_total", + name="관급자재대", + base_label=( + "순자재대+조달수수료 (천원 올림)" + if data.include_fee_in_owner_material_total + else "순자재대 (천원 올림)" + ), + base=raw, + amount=owner_total, + note="총원가 밖 별도 표기", + ) + + grand_total = contract_amount + owner_total + _line( + result, + key="grand_total", + name="총공사비", + base_label="도급공사비+관급자재대", + base=grand_total, + amount=grand_total, + ) + + result.totals = { + "material_cost": material_cost, + "labor_cost": total_labor_cost, + "expense": expense_total, + "direct_construction_cost": direct_construction_cost, + "net_construction_cost": net_construction_cost, + "general_overhead": overhead, + "profit": profit, + "total_cost": total_cost, + "vat": vat, + "contract_amount": contract_amount, + "owner_supplied_material_total": owner_total, + "grand_total": grand_total, + } + return result diff --git a/B09_Estimation/B09_Estimation_Rates.py b/B09_Estimation/B09_Estimation_Rates.py new file mode 100644 index 00000000..b1c0155b --- /dev/null +++ b/B09_Estimation/B09_Estimation_Rates.py @@ -0,0 +1,245 @@ +"""B09 원가계산 — 요율 데이터 로더·구간 조회. + +요율은 **코드에 박지 않는다**. `resources/data_cost_input_value/rates_*.json` 이 정본이고 +이 모듈은 그 파일을 읽어 구간을 골라 주는 일만 한다 (PLAN 9-2·8-10 ★법대로). + +핵심 규칙 (PLAN 8-9·8-10 — 실무 원가계산서 재현으로 확인): + - 요율표는 **한 벌**이다. 안전관리비 A/B 는 요율이 두 벌인 것이 아니라 + **같은 표를 대상액 두 개로 각각 조회**하는 것이다. + - 구간 라벨의 `billion` 은 **십억 원(10^9)**, `million` 은 **백만 원(10^6)** 이다. + `lt_5_billion` = 50억 미만. (2026-09-07 값 파일 대조로 확정) + - 판정 실패는 **조용히 넘기지 않는다** — 기본값으로 때우면 금액이 조용히 틀린다. +""" + +from __future__ import annotations + +import json +import os +import re +from dataclasses import dataclass +from decimal import Decimal +from functools import lru_cache +from typing import Any + +# 구간 라벨의 단위 접미사 → 원(KRW) 배수. +_UNIT_MULTIPLIER: dict[str, int] = { + "million": 1_000_000, + "billion": 1_000_000_000, +} + +_RESOURCE_SUBPATH = ("resources", "data_cost_input_value") + +# 라벨 문법 — 숫자 구간만 해석한다. 그 밖(`turnkey_or_alternative` 등)은 명시 선택자로 고른다. +_RE_LT = re.compile(r"^lt_(\d+(?:\.\d+)?)_(million|billion)$") +_RE_GTE = re.compile(r"^gte_(\d+(?:\.\d+)?)_(million|billion)(?:_(.+))?$") +_RE_RANGE_ONE_UNIT = re.compile(r"^(\d+(?:\.\d+)?)_to_(\d+(?:\.\d+)?)_(million|billion)$") +_RE_RANGE_TWO_UNIT = re.compile( + r"^(\d+(?:\.\d+)?)_(million|billion)_to_(\d+(?:\.\d+)?)_(million|billion)$" +) +_RE_DAYS_LTE = re.compile(r"^lte_(\d+)_days$") +_RE_DAYS_GTE = re.compile(r"^gte_(\d+)_days$") +_RE_DAYS_RANGE = re.compile(r"^(\d+)_to_(\d+)_days$") + + +class RateLookupError(LookupError): + """요율 구간을 못 고른 경우. 기본값으로 때우지 않고 여기서 멈춘다.""" + + +@dataclass(frozen=True) +class RateDataset: + """요율 데이터셋 한 벌 — 재현성 표기용 신원(9-2)을 함께 든다.""" + + dataset_id: str + effective_date: str + sha256: str + variables: dict[str, Any] + + def variable(self, name: str) -> Any: + try: + return self.variables[name] + except KeyError as exc: # pragma: no cover - 데이터 파손 시에만 + raise RateLookupError(f"요율 항목이 데이터셋에 없습니다: {name}") from exc + + @property + def version_stamp(self) -> dict[str, str]: + """내역서·화면에 남길 「어느 판으로 계산했나」 표기.""" + return { + "dataset_id": self.dataset_id, + "effective_date": self.effective_date, + "sha256": self.sha256, + } + + +def _project_root() -> str: + return os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + + +def _dataset_dir() -> str: + return os.path.join(_project_root(), *_RESOURCE_SUBPATH) + + +def _manifest_entry(file_name: str) -> dict[str, Any]: + manifest_path = os.path.join(_dataset_dir(), "_manifest.json") + with open(manifest_path, encoding="utf-8") as handle: + manifest = json.load(handle) + for entry in manifest.get("files", []): + if entry.get("file") == file_name: + return entry + raise RateLookupError(f"매니페스트에 없는 요율 파일입니다: {file_name}") + + +@lru_cache(maxsize=8) +def load_rate_dataset(file_name: str = "rates_2026.json") -> RateDataset: + """요율 파일 한 벌을 읽는다. 매니페스트의 지문·적용일을 함께 실어 재현성을 남긴다.""" + entry = _manifest_entry(file_name) + with open(os.path.join(_dataset_dir(), file_name), encoding="utf-8") as handle: + payload = json.load(handle) + return RateDataset( + dataset_id=payload.get("dataset_id", entry.get("dataset_id", "")), + effective_date=payload.get("effective_date", entry.get("effective_date", "")), + sha256=entry.get("sha256", ""), + variables=payload.get("variables", {}), + ) + + +def _bracket_bounds(label: str) -> tuple[Decimal, Decimal] | None: + """금액 구간 라벨 → [하한, 상한). 숫자 구간이 아니면 None.""" + match = _RE_LT.match(label) + if match: + return Decimal(0), Decimal(match.group(1)) * _UNIT_MULTIPLIER[match.group(2)] + + match = _RE_RANGE_TWO_UNIT.match(label) + if match: + low = Decimal(match.group(1)) * _UNIT_MULTIPLIER[match.group(2)] + high = Decimal(match.group(3)) * _UNIT_MULTIPLIER[match.group(4)] + return low, high + + match = _RE_RANGE_ONE_UNIT.match(label) + if match: + unit = _UNIT_MULTIPLIER[match.group(3)] + return Decimal(match.group(1)) * unit, Decimal(match.group(2)) * unit + + match = _RE_GTE.match(label) + if match: + return Decimal(match.group(1)) * _UNIT_MULTIPLIER[match.group(2)], Decimal("Infinity") + + return None + + +def _duration_bounds(label: str) -> tuple[int, int] | None: + """공사기간 구간 라벨 → [하한일, 상한일]. 숫자 구간이 아니면 None.""" + match = _RE_DAYS_LTE.match(label) + if match: + return 0, int(match.group(1)) + + match = _RE_DAYS_RANGE.match(label) + if match: + return int(match.group(1)), int(match.group(2)) + + match = _RE_DAYS_GTE.match(label) + if match: + return int(match.group(1)), 10**9 + + return None + + +def _amount_matches(label: str, amount: Decimal) -> bool: + bounds = _bracket_bounds(label) + if bounds is None: + return False + low, high = bounds + return low <= amount < high + + +def _duration_matches(label: str, days: int) -> bool: + bounds = _duration_bounds(label) + if bounds is None: + return False + low, high = bounds + return low <= days <= high + + +def select_bracket( + brackets: list[dict[str, Any]], + *, + amount_field: str | None = None, + amount: Decimal | None = None, + duration_days: int | None = None, + duration_field: str = "duration_bracket", + equals: dict[str, Any] | None = None, + residual_label: str | None = None, + label: str, +) -> dict[str, Any]: + """구간 목록에서 한 행을 고른다. 못 고르면 `RateLookupError` — 기본값으로 안 때운다. + + `equals` 는 `work_type` 처럼 값이 그대로 맞아야 하는 열이다. + `residual_label` 은 숫자 구간이 아닌 **잔여 구간** 라벨이다(예: 고용보험료의 + `below_official_threshold`). 숫자 구간이 하나도 안 맞을 때만 쓰며, **부르는 쪽이 + 이름을 대야** 한다 — 조용한 기본값이 아니다. + """ + candidates = list(brackets) + + if equals: + for key, expected in equals.items(): + candidates = [row for row in candidates if row.get(key) == expected] + + if amount_field is not None and amount is not None: + candidates = [ + row for row in candidates if _amount_matches(str(row.get(amount_field, "")), amount) + ] + + if duration_days is not None: + candidates = [ + row + for row in candidates + if _duration_matches(str(row.get(duration_field, "")), duration_days) + ] + + if not candidates and residual_label is not None and amount_field is not None: + candidates = [row for row in brackets if row.get(amount_field) == residual_label] + if equals: + for key, expected in equals.items(): + candidates = [row for row in candidates if row.get(key) == expected] + + if not candidates: + raise RateLookupError( + f"{label}: 조건에 맞는 요율 구간이 없습니다 " + f"(금액={amount}, 기간={duration_days}일, 조건={equals})" + ) + if len(candidates) > 1: + raise RateLookupError( + f"{label}: 요율 구간이 {len(candidates)}개 겹칩니다 — 데이터 점검 필요 " + f"({[row.get(amount_field) for row in candidates]})" + ) + return candidates[0] + + +def rate_percent(row: dict[str, Any], *, label: str) -> Decimal: + if "rate_percent" not in row: + raise RateLookupError(f"{label}: 고른 구간에 요율이 없습니다 ({row})") + return Decimal(str(row["rate_percent"])) + + +def base_amount(row: dict[str, Any]) -> Decimal: + """구간에 딸린 기초액(안전관리비 등). 없으면 0.""" + return Decimal(str(row.get("base_amount_krw", 0))) + + +def flat_rate(dataset: RateDataset, name: str) -> Decimal: + """구간이 없는 단일 요율(산재·건강·요양·부가세 등).""" + variable = dataset.variable(name) + if "rate_percent" not in variable: + raise RateLookupError(f"{name}: 단일 요율이 아닙니다 — 구간 조회가 필요합니다") + return Decimal(str(variable["rate_percent"])) + + +def pension_rate_percent(dataset: RateDataset, year: int) -> Decimal: + """국민연금 — 연도별 특례 스케줄(2026 = 4.75 %, 2033~ 본칙 6.5 %).""" + variable = dataset.variable("rate_pension") + for row in variable.get("annual_rates", []): + if int(row.get("year", 0)) == year: + return Decimal(str(row["rate_percent"])) + fallback = variable.get("rate_from_2033_percent") + if fallback is None: + raise RateLookupError(f"rate_pension: {year}년 요율이 없습니다") + return Decimal(str(fallback))