Aislo/B09_Estimation/B09_Estimation_Rates.py

"""B09 원가 요율(PLAN 145-4 · 166-1) — M01 `요율_조달청제비율.json` 을 읽는 유일한 곳.

rate_candidates  줄마다 후보 글 한 줄(출처 풍선)
rate_table       짝 셈(`B09_Estimation_Calc_Total`)이 자동으로 고르는 압축 표 — 총괄 줄 키마다
                 {rt, pct} 단일 · {price, pcts} 추정가격 구간 · {direct, days, grid} 직접공사비 × 공기 ·
                 {target, pcts, bases} 안전 대상액 구간 · {years} 연도 · {kinds} 환경 공종
                 구간 = 위 끝 목록(그 값 미만 · 마지막 = 끝없음) · 공종 = 토목(임도) 고정
"""

from __future__ import annotations

import json
import math
from functools import lru_cache
from pathlib import Path
from typing import Any

FILE = Path(__file__).resolve().parents[1] / "resources" / "master_data" / "요율_조달청제비율.json"


def _percents(value: Any) -> list[float]:
    """값 안 모든 rate_percent — 구간 · 연도 · 공종 갈래 통째로."""
    if isinstance(value, dict):
        own = value.get("rate_percent")
        found = [float(own)] if isinstance(own, (int, float)) else []
        return found + [
            p for key, v in value.items() if key != "rate_percent" for p in _percents(v)
        ]
    if isinstance(value, list):
        return [p for item in value for p in _percents(item)]
    return []


@lru_cache(maxsize=1)
def _rows() -> tuple[dict[str, Any], ...]:
    try:
        return tuple(json.loads(FILE.read_text(encoding="utf-8")).get("줄") or [])
    except (OSError, ValueError):
        return ()


@lru_cache(maxsize=1)
def rate_candidates() -> dict[str, str]:
    """{RT 키: 「M01 RT… 3.56%」 · 「… 1.01~1.57%(구간 · 고르는 값 사용자)」} — 파일 없으면 {}."""
    rows = _rows()
    out: dict[str, str] = {}
    for row in rows:
        percents = sorted(set(_percents(row.get("값"))))
        if not percents:
            text = "구간 식(고르는 값 사용자)"
        elif len(percents) == 1:
            text = f"{percents[0]:g}%"
        else:
            text = f"{percents[0]:g}~{percents[-1]:g}%(구간 · 고르는 값 사용자)"
        out[str(row.get("키"))] = f"M01 {row.get('키')} {text}"
    return out


_UNIT = {"million": 10**6, "billion": 10**9, "days": 1}


def _upper(label: Any) -> float | None:
    """구간 이름 → 위 끝(그 값 미만) — 「gte_」 = 끝없음(inf) · 일수는 끝 포함이라 +1 ·
    수 없는 이름(below_official_threshold) = None. `billion` = 십억(M01 basis 글)."""
    words = str(label).split("_")
    if words[0] == "gte":
        return math.inf
    for i in range(len(words) - 1, 0, -1):
        if words[i] in _UNIT and words[i - 1].isdigit():
            return int(words[i - 1]) * _UNIT[words[i]] + (words[i] == "days")
    return None


def _steps(rows: list[dict[str, Any]], field: str) -> tuple[list[float], list[dict[str, Any]]]:
    """구간 줄 → (위 끝 목록 · 끝없음 뺌, 줄 목록) — 위 끝 차례."""
    keyed = sorted(
        ((u, r) for r in rows if (u := _upper(r.get(field))) is not None), key=lambda p: p[0]
    )
    return [u for u, _ in keyed[:-1]], [r for _, r in keyed]


def _grid(rows: list[dict[str, Any]]) -> dict[str, Any]:
    """간접노무비 · 기타경비 — 직접공사비 × 공기 × 토목."""
    civil = [r for r in rows if r.get("work_type") == "civil"]
    d_up = sorted({_upper(r["direct_cost_bracket"]) for r in civil})
    t_up = sorted({_upper(r["duration_bracket"]) for r in civil})
    grid: list[list[float | None]] = [[None] * len(t_up) for _ in d_up]
    for r in civil:
        i = d_up.index(_upper(r["direct_cost_bracket"]))
        grid[i][t_up.index(_upper(r["duration_bracket"]))] = r["rate_percent"]
    return {"direct": d_up[:-1], "days": t_up[:-1], "grid": grid, "note": "토목"}


def _priced(rows: list[dict[str, Any]], field: str, note: str = "") -> dict[str, Any]:
    bounds, picked = _steps(rows, field)
    return {"price": bounds, "pcts": [r["rate_percent"] for r in picked], "note": note}


@lru_cache(maxsize=1)
def rate_table() -> dict[str, dict[str, Any]]:
    """총괄 줄 키 → 압축 표(머리 글) — 파일 · 줄이 없으면 그 줄 없음(자동 없음 = 빈칸)."""
    v = {str(r.get("키")): r.get("값") or {} for r in _rows()}
    out: dict[str, dict[str, Any]] = {}

    def put(key: str, rt: str, build: Any) -> None:
        if rt in v:
            try:
                out[key] = {"rt": rt, **build(v[rt])}
            except (KeyError, TypeError, ValueError, IndexError):
                pass  # M01 꼴이 바뀌면 그 줄만 자동 없음

    for key, rt in (("accident", "RT000001"), ("health", "RT000003"), ("care", "RT000005")):
        put(key, rt, lambda x: {"pct": x["rate_percent"]})
    put("vat", "RT000019", lambda x: {"pct": x["rate_percent"]})
    put(
        "machine",
        "RT000015",
        lambda x: {
            "pct": next(
                r["rate_percent"]
                for r in x["general_construction"]
                if r["work_type"] == "civil_general"
            ),
            "note": "종합공사 토목 일반",
        },
    )
    put("employment", "RT000002", lambda x: _priced(x["brackets"], "estimated_amount_bracket"))
    put(
        "overhead",
        "RT000017",
        lambda x: _priced(
            x["civil_landscape_industrial"], "estimated_price_bracket", "종합(토목) 표"
        ),
    )
    put("profit", "RT000018", lambda x: _priced(x["brackets"], "estimated_price_bracket"))
    put("indirect_labor", "RT000008", lambda x: _grid(x["brackets"]))
    put("other", "RT000009", lambda x: _grid(x["brackets"]))

    def safety(x: dict[str, Any]) -> dict[str, Any]:
        # ponytail: 관리자 선임 대상(추정 1000억 이상) 구간은 안 고름 — 임도 규모 밖
        civil = [
            r
            for r in x["brackets"]
            if r["work_type"] == "civil" and "at_or_above" not in r["target_amount_bracket"]
        ]
        bounds, picked = _steps(civil, "target_amount_bracket")
        return {
            "target": bounds,
            "pcts": [r["rate_percent"] for r in picked],
            "bases": [r.get("base_amount_krw", 0) for r in picked],
            "note": "토목",
        }

    put("safety", "RT000006", safety)
    put(
        "pension",
        "RT000004",
        lambda x: {"years": {str(r["year"]): r["rate_percent"] for r in x["annual_rates"]}},
    )
    put(
        "environment",
        "RT000010",
        lambda x: {
            "kinds": {
                "road" if r["work_type"] == "road" else "other": r["rate_percent"]
                for r in x["forest_road_candidates"]
            }
        },
    )
    return out