Aislo/B09_Estimation/B09_Estimation_Rates.py

"""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", {}),
    )


@lru_cache(maxsize=8)
def load_rate_dataset_from_path(path: str) -> RateDataset:
    """매니페스트 밖의 요율 파일을 읽는다 — **옛 연도 재현 검산 전용**.

    정본 요율은 `load_rate_dataset` 으로만 읽는다. 이 함수는 「2024년 값으로 돌리면
    그때 서류가 재현되는가」를 시험하려고 두는 것이고, 지문이 없으므로 결과에
    `sha256=""` 로 남아 **정본이 아님이 드러난다**.
    """
    with open(path, encoding="utf-8") as handle:
        payload = json.load(handle)
    return RateDataset(
        dataset_id=payload.get("dataset_id", ""),
        effective_date=payload.get("effective_date", ""),
        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,
    prefer_suffix: 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 len(candidates) > 1 and prefer_suffix is not None and amount_field is not None:
        # 같은 금액 구간이 공종으로 갈리는 표가 있다(하도급보증의 `…_integrated_civil…`).
        # 부르는 쪽이 공종을 대야 하며, 조용한 기본값이 아니다.
        narrowed = [r for r in candidates if str(r.get(amount_field, "")).endswith(prefer_suffix)]
        if narrowed:
            candidates = narrowed

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