Aislo/B09_Estimation/B09_Estimation_MaterialPrices.py

"""B09 원가계산 — **자재 수동 단가** 한 벌 (PLAN 1장 Ⓐ · 2026-09-14 브레인 판정).

    저장 자리   산출 조건 `estimation.material_prices`
                {키: {price_krw, source, entered_at, name, spec, unit}}
    키          자원 코드가 있으면 **코드**(`AR-M-…`) · 코드 없는 자재만 「이름 규격」
                (명세 2장 「잇기는 코드로」 · 11장 「규격은 조인 키」)
                ⚠ 이름·규격·단위는 **보이기용** — 맞추는 데 안 씀. 이름이 고쳐져도 단가가 안 끊김
    얹는 자리   조립 때 자재 제목(M)을 세움 — 6번 슬롯 = 수동 단가 · 쪽수 칸 = 출처
                → 자원 축 자재 줄이 그 제목에 붙어 일위대가 재료비가 섬
    넣은 날     값·출처가 같으면 다시 저장해도 안 바뀜(구조물도 수동 단가 `save_manual` 과 같은 꼴)
    미확정      수동 단가가 닿은 내역 줄마다 「미확정 N건」(구조물도 수동 단가·폐기물과 같은 통로)

⚠ 값이 없거나 0 이하인 칸은 받지 않음 — 0 원으로 채우면 「단가 없음」이 금액 0 으로 숨음.
⚠ 코드 없는 「이름 규격」 줄은 자재총괄 몫(다음 단계) — 조립에는 코드 줄만 얹음.
   이름 줄에 나중에 코드가 붙으면 코드 키로 옮겨 붙일 자리 — 아직 안 만듦(브레인 판정).
"""

from __future__ import annotations

import json
import os
import re
from decimal import Decimal, InvalidOperation
from functools import lru_cache
from typing import Any

from B09_Estimation.B09_Estimation_PriceBook import (
    PRICE_SLOT_COUNT,
    PriceBook,
    PriceKind,
    PriceTitle,
)

MATERIAL_PRICES_KEY = "material_prices"
#: 조립에 얹는 키 — 자원 목록 자재 코드(명세 §2 ③ 모양).
_RE_MATERIAL_CODE = re.compile(r"^AR-M-[0-9a-f]{8}$")
#: 출처를 안 적었을 때 쪽수 칸 글.
MANUAL_SOURCE = "수동 입력"
_TEXT_FIELDS = ("source", "name", "spec", "unit")


class MaterialPriceError(ValueError):
    """받을 수 없는 단가 — 저장하지 않고 까닭을 돌려줌."""


def _price(value: Any) -> Decimal | None:
    try:
        price = Decimal(str(value).replace(",", "").strip())
    except (InvalidOperation, ValueError):
        return None
    return price if price.is_finite() and price > 0 else None


def is_code_key(key: str) -> bool:
    """조립에 얹는 코드 키인가 — 아니면 「이름 규격」 키."""
    return bool(_RE_MATERIAL_CODE.match(key))


def normalize(raw: Any) -> dict[str, dict[str, str]]:
    """저장본을 받을 수 있는 줄만 — 값이 없거나 0 이하면 버림."""
    prices: dict[str, dict[str, str]] = {}
    if not isinstance(raw, dict):
        return prices
    for key, entry in raw.items():
        price = _price(entry.get("price_krw")) if isinstance(entry, dict) else None
        if not str(key).strip() or price is None:
            continue
        prices[str(key)] = {
            "price_krw": str(price),
            "entered_at": str(entry.get("entered_at") or ""),
            **{field: str(entry.get(field) or "") for field in _TEXT_FIELDS},
        }
    return prices


def build_key(raw: Any) -> tuple[tuple[str, str, str], ...]:
    """조립 캐시 키 — 코드 키만 (키, 단가, 출처). 같은 값이면 같은 벌."""
    return tuple(
        sorted(
            (key, entry["price_krw"], entry["source"])
            for key, entry in normalize(raw).items()
            if is_code_key(key)
        )
    )


def merge(stored: Any, changes: list[dict[str, Any]], today: str) -> dict[str, dict[str, str]]:
    """바꿀 것 여럿을 한 번에 — `price_krw` 가 비면 그 줄을 지움(단가 없음으로 돌아감)."""
    prices = normalize(stored)
    for change in changes:
        key = str(change.get("key") or "").strip()
        if not key:
            raise MaterialPriceError("자재 키가 비었습니다")
        raw_price = change.get("price_krw")
        if raw_price is None or str(raw_price).strip() == "":
            prices.pop(key, None)
            continue
        price = _price(raw_price)
        if price is None:
            raise MaterialPriceError(f"단가는 0 보다 큰 수여야 합니다: {raw_price}")
        old = prices.get(key) or {}
        source = str(change.get("source") or "").strip()
        same = old.get("price_krw") == str(price) and old.get("source") == source
        prices[key] = {
            "price_krw": str(price),
            "entered_at": old["entered_at"] if same and old.get("entered_at") else today,
            "source": source,
            **{
                field: str(change.get(field) or old.get(field) or "")
                for field in ("name", "spec", "unit")
            },
        }
    return prices


@lru_cache(maxsize=1)
def material_catalog_rows() -> dict[str, dict[str, str]]:
    """자원 목록(`AR-M-`) 코드 → 이름·규격·단위 — 제목을 세우고 화면에 보일 글."""
    from B09_Estimation.B09_Estimation_ResourceAxis_Join import EXT_CATALOG_FILE

    root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    path = os.path.join(root, "resources", "data_resource_catalog", EXT_CATALOG_FILE)
    if not os.path.isfile(path):
        return {}
    with open(path, encoding="utf-8") as handle:
        entries = json.load(handle).get("entries") or []
    return {
        str(row["code"]): {
            "name": str(row.get("name") or ""),
            "spec": str(row.get("spec") or ""),
            "unit": str(row.get("unit") or ""),
        }
        for row in entries
        if is_code_key(str(row.get("code") or ""))
    }


def add_material_titles(
    book: PriceBook, prices: tuple[tuple[str, str, str], ...]
) -> dict[str, str]:
    """코드 키 수동 단가로 자재 제목(M)을 세움 — 돌려주는 값 = 코드 → 출처(미확정 셈에 씀).

    자원 목록에 없는 코드는 안 세움 — 화면 목록이 「단가표에서 사라진 코드」로 드러냄(`listing`).
    """
    catalog = material_catalog_rows()
    manual: dict[str, str] = {}
    for code, price, source in prices:
        row = catalog.get(code)
        if row is None or code in book.titles:
            continue
        slots: list[Decimal | None] = [None] * PRICE_SLOT_COUNT
        pages: list[str | None] = [None] * PRICE_SLOT_COUNT
        slots[-1] = Decimal(price)
        pages[-1] = source or MANUAL_SOURCE
        book.add_title(
            PriceTitle(
                code=code,
                kind=PriceKind.MATERIAL,
                name=row["name"],
                spec=row["spec"],
                unit=row["unit"],
                slots=slots,
                slot_pages=pages,
            )
        )
        manual[code] = source or MANUAL_SOURCE
    return manual


def manual_count(book: PriceBook, code: str | None, manual: dict[str, str]) -> int:
    """그 단가가 밟는 **수동 단가 자재 수**(같은 자재는 한 번) — 내역 줄 「미확정 N건」."""
    if not code or not manual:
        return 0
    found: set[str] = set()
    seen: set[str] = set()
    stack = [code]
    while stack:
        current = stack.pop()
        if current in seen:
            continue
        seen.add(current)
        if current in manual:
            found.add(current)
        stack.extend(detail.ref_code for detail in book.details.get(current, []))
    return len(found)


def listing(
    build: Any, stored: Any, materials: list[dict[str, Any]] | None = None
) -> list[dict[str, Any]]:
    """화면 줄 — 자원 축이 쓰는 자재(코드) + 자재총괄 자재(「이름 규격」) + 저장만 남은 줄.

    `materials` — B08 인계 `materials`(자재총괄). 사급 줄만 단가 칸이 섬(관급은 도급 금액 밖).
    ⚠ 사라진 줄도 조용히 안 버림 — 「사라진 자재」로 보이고 사용자가 지움.
    """
    prices = normalize(stored)
    catalog = material_catalog_rows()
    uses: dict[str, list[str]] = getattr(build, "material_uses", {}) or {}
    rows: list[dict[str, Any]] = []

    def add(key: str, shown: dict[str, Any], **extra: Any) -> None:
        entry = prices.get(key) or {}
        rows.append(
            {
                "key": key,
                **{
                    field: shown.get(field) or entry.get(field, "")
                    for field in ("name", "spec", "unit")
                },
                "price_krw": entry.get("price_krw"),
                "source": entry.get("source", ""),
                "entered_at": entry.get("entered_at", ""),
                "missing": False,
                **extra,
            }
        )

    for code in sorted(uses, key=lambda c: (catalog.get(c, {}).get("name", ""), c)):
        add(code, catalog.get(code, {}), work_items=sorted(uses[code]), origin="unit_price")
    for material in materials or []:
        key = f"{material.get('material_name', '')} {material.get('spec') or ''}".strip()
        add(
            key,
            {**material, "name": material.get("material_name")},
            work_items=list(material.get("source_structure") or []),
            origin="material_sheet",
            supply_type=material.get("supply_type") or "",
            quantity=str(material.get("total_amount")),
        )
    listed = {row["key"] for row in rows}
    for key, entry in sorted(prices.items()):
        if key not in listed:
            rows.append({"key": key, **entry, "work_items": [], "missing": True})
    return rows