218 lines
8.7 KiB
Python
218 lines
8.7 KiB
Python
"""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) -> list[dict[str, Any]]:
|
|
"""화면 줄 — 자원 축이 쓰는 자재(코드) + 저장만 있고 자원 축에서 사라진 줄.
|
|
|
|
⚠ 사라진 줄도 조용히 안 버림 — 「단가표에서 사라진 코드」로 보이고 사용자가 지움.
|
|
"""
|
|
prices = normalize(stored)
|
|
catalog = material_catalog_rows()
|
|
uses: dict[str, list[str]] = getattr(build, "material_uses", {}) or {}
|
|
rows: list[dict[str, Any]] = []
|
|
for code in sorted(uses, key=lambda c: (catalog.get(c, {}).get("name", ""), c)):
|
|
info = catalog.get(code, {})
|
|
entry = prices.get(code) or {}
|
|
rows.append(
|
|
{
|
|
"key": code,
|
|
"name": info.get("name") or entry.get("name", ""),
|
|
"spec": info.get("spec") or entry.get("spec", ""),
|
|
"unit": info.get("unit") or entry.get("unit", ""),
|
|
"work_items": sorted(uses[code]),
|
|
"price_krw": entry.get("price_krw"),
|
|
"source": entry.get("source", ""),
|
|
"entered_at": entry.get("entered_at", ""),
|
|
"missing": False,
|
|
}
|
|
)
|
|
for key, entry in sorted(prices.items()):
|
|
if key in uses or not is_code_key(key):
|
|
continue
|
|
rows.append({"key": key, **entry, "work_items": [], "missing": True})
|
|
return rows
|