eomsangdonandClaude Opus 5.5
28da700d47
- master_formula.inputs_of · FLOW_INPUTS · OLD_FIELDS · mf.run 이 흐름 줄이면 스스로 서버 갈래를 깖(스크립트 · 시험도) - check_master 틀 검사 · 시험입력 · 로직(…) 입력 검사 · master_auto 자동값 · 프로젝트 빈 입력 · 연결 · 다시 적용 값이 흐름 입력으로 - 프로젝트 단가 묶음(collect)이 흐름의 요소 · 표 · 다른 로직 · 기계 시간당 경비 로직을 모음 - 옛 엔진 갈래 결과 = 옛 꼴(돈 출력 줄 · 값들 · 알림) · 조합 · 프로젝트에서 안 준 입력은 빈 입력 알림(예시값 아님) - /text 흐름 로직 = 출력 상자 줄 · GET /logic 은 보기용으로 설계 때 정할 것을 입력 꼴로 얹음(파일은 그대로) - 찾을 값이 빈 조건 찾기는 그 값만 비움(옛 셈처럼) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL
297 lines
13 KiB
Python
297 lines
13 KiB
Python
"""집계 항목 복사본의 단가 묶음 `단가.json` · 복사본만으로 계산 (PLAN 25-1b · 25-2b · 25-2e).
|
|
|
|
단가 묶음 = 그 항목 일위대가 조합이 담은 로직을 계산할 때 닿을 수 있는 마스터 줄 전부(닫힌 묶음):
|
|
- 담은 로직 · 로직이 부르는 하위 로직(`로직(GC…)`) 끝까지
|
|
- 식이 가리키는 요소(인력 · 기계 · 재료 · 유가전력 · 요율 · 환율 …) · 준용 대상
|
|
- 찾기 표(`찾기(QC…)` · `찾기(QF…)` 소요량 · 계수)
|
|
- 재료 고르기 조건의 후보 자재품목 줄 · 품셈재료 연결 · 지역 · 계약종별 후보
|
|
원문번호 틀의 `{입력}` 은 그 입력의 고르기 값으로 모두 끼워 모음(고르기가 없으면 그 테이블 전부 ·
|
|
자재품목만은 너무 커서 뺌). 줄은 마스터 파일 차례 그대로 담음 — 고르기의 「첫 줄」 이 같게.
|
|
|
|
`단가.json` = `{파일 이름: {그룹 · 머리 칸 …, 줄|표: [줄]}}` — `master_formula.Master` 가
|
|
그대로 읽는 꼴.
|
|
계산(`calc`)은 이 묶음과 `입력값.json` · `고른자재.json` 만 씀 — 마스터 현재값을 읽지 않음.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from M01_MasterData import M01_MasterData_Store as m01
|
|
|
|
mf = m01.cm.mf
|
|
mk = m01.mk
|
|
mcb = m01.mcb
|
|
mm = mf.mp.mm # 품셈재료 · 자재품목 후보(`master_material`)
|
|
|
|
#: 틀을 못 풀었을 때 통째로 담지 않는 테이블 — 자재품목(3만 줄)
|
|
_TOO_BIG = ("MT",)
|
|
|
|
|
|
def master() -> tuple[dict[str, dict], Any]:
|
|
"""(마스터 파일 묶음, 색인) — 읽기 전용(시험은 `m01.FOLDER` 를 사본으로 바꿈)."""
|
|
return m01.loaded()
|
|
|
|
|
|
def _texts(row: dict) -> list[str]:
|
|
out = [step["식"] for step in row.get("중간", [])]
|
|
if "결과" in row:
|
|
out.append(row["결과"]["식"])
|
|
for item in row.get("호표", []):
|
|
out.append(item["수량"])
|
|
if not isinstance(item.get("요소"), dict):
|
|
out.append(item["요소"])
|
|
out += [extra["식"] for extra in row.get("덧줄", [])]
|
|
return out
|
|
|
|
|
|
class _Collector:
|
|
def __init__(self, whole: Any):
|
|
self.whole = whole
|
|
self.keys: set[str] = set()
|
|
self.logics: set[str] = set()
|
|
self.missing: list[str] = []
|
|
|
|
def refs(self, row: dict, template: str) -> list[str]:
|
|
"""원문번호 틀 → 닿을 수 있는 요소 참조 — 고르기로 끼우고, 못 풀면 그 테이블 전부."""
|
|
if "{" not in template:
|
|
return [template]
|
|
expanded = mf._expand(row, template, self.whole)
|
|
if expanded is not None:
|
|
return expanded
|
|
tid = mk.id_of(template)
|
|
if not tid or tid in _TOO_BIG:
|
|
self.missing.append(f"{row.get('키')} 「{template}」 틀을 못 풂")
|
|
return []
|
|
return list(self.whole.index.get(tid, {}))
|
|
|
|
def add(self, ref: str) -> None:
|
|
try:
|
|
row = self.whole.get(ref)
|
|
except mf.FormulaError:
|
|
return # 없는 요소 — 계산이 같은 까닭으로 멈춤(마스터와 같게)
|
|
key = str(row.get("키"))
|
|
if key in self.keys:
|
|
return
|
|
self.keys.add(key)
|
|
if row.get("준용") not in (None, ""):
|
|
self.add(str(row["준용"]))
|
|
link = row.get("연결") if key[:2] == mf.LINKED else None
|
|
if isinstance(link, str):
|
|
self.add(link)
|
|
elif isinstance(link, dict):
|
|
self.widely(link, row.get("단위"))
|
|
|
|
def widely(self, cond: dict, unit: Any) -> None:
|
|
"""지역 · 계약종별 「입력」 조건 — 입력값과 상관없이 닿을 수 있는 후보 전부."""
|
|
want = mm.plain(cond.get("이름"))
|
|
if cond.get("계약종별") == "입력":
|
|
for key, one in self.whole.index.get(mm.PRICED, {}).items():
|
|
if one.get("구분") == "전력" and mm.plain(one.get("이름")).startswith(want):
|
|
self.keys.add(key)
|
|
return
|
|
spec, unit_p = mm.plain_spec(cond.get("규격")), mm.plain_unit(unit)
|
|
for root, _, s, u, key in mm._market_index(self.whole):
|
|
if root.endswith(want) and u == unit_p and (not spec or s == spec):
|
|
self.keys.add(key)
|
|
|
|
def pick(self, row: dict, cond: dict, unit: Any) -> None:
|
|
"""재료 고르기 조건 — 대표 줄 · 후보 줄(`{입력}` 은 고르기 값으로 끼움)."""
|
|
if cond.get("지역") == "입력" or cond.get("계약종별") == "입력":
|
|
self.widely(cond, unit)
|
|
return
|
|
holes = sorted(
|
|
{h for v in cond.values() if isinstance(v, str) for h in mf._HOLE.findall(v)}
|
|
)
|
|
variants = [cond]
|
|
if holes:
|
|
env_sets = mf._expand(row, "|".join(f"{{{h}}}" for h in holes), self.whole)
|
|
if env_sets is None:
|
|
variants = [mf.static(cond)]
|
|
else:
|
|
variants = []
|
|
for filled in env_sets:
|
|
env = dict(zip(holes, filled.split("|")))
|
|
variants.append(
|
|
{k: mf.fill(v, env) if isinstance(v, str) else v for k, v in cond.items()}
|
|
)
|
|
for one in variants:
|
|
if one.get("대표"):
|
|
self.add(str(one["대표"]))
|
|
if not one.get("구분") and not one.get("검색어"):
|
|
continue
|
|
for key in mf.candidates(self.whole, one, None, None):
|
|
self.add(key)
|
|
|
|
def flow(self, row: dict) -> None:
|
|
"""흐름 입력 상자 — 요소(고정 · 기본 · 검색어 묶음 · 조건 찾기 후보) · 표 · 다른 로직 ·
|
|
기계 시간당 경비 로직."""
|
|
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv # 흐름 값 찾기와 같은 뜻
|
|
|
|
for box in row["흐름"].get("boxes") or []:
|
|
inp = (box.get("input") or {}) if box.get("kind") == "input" else {}
|
|
if inp.get("mode") == "logic":
|
|
self.logic(str((inp.get("logic") or {}).get("key") or ""))
|
|
if inp.get("table"):
|
|
self.add(str(inp["table"].get("key") or ""))
|
|
for el in inp.get("elements") or []:
|
|
ref, unit = str(el.get("ref") or ""), el.get("unit")
|
|
for one in (ref, str(el.get("default") or "")):
|
|
cond = dv._pick(one)
|
|
if cond is not None:
|
|
self.pick(row, cond, unit)
|
|
elif one and not one.startswith("UE-"):
|
|
self.element(one)
|
|
if el.get("mode") != "rule":
|
|
continue
|
|
if inp.get("group") in dv.GROUP_RULE: # 인력 · 기계 · 유가전력 찾기 = 그 그룹 전부
|
|
self.keys.update(str(r.get("키")) for r in dv._group_rows(inp["group"]))
|
|
elif dv._pick(ref) is None: # 재료 범위 찾기 — 범위(구분 · 상세구분) 후보
|
|
scope = el.get("scope") or {}
|
|
self.pick(
|
|
row, {k: scope[k] for k in ("구분", "상세구분") if scope.get(k)}, unit
|
|
)
|
|
|
|
def element(self, ref: str) -> None:
|
|
"""요소 키 · 「EQ키#칸」(시간당 경비면 그 기계경비 로직까지) · 「ID:원문번호{…}」 틀."""
|
|
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv
|
|
|
|
base, _, field = ref.partition(dv.MACHINE_SEP)
|
|
if "{" in base: # 「EQ:{기계}」 같은 틀 — 그 테이블 전부
|
|
self.keys.update(self.whole.index.get(mk.id_of(base) or "", {}))
|
|
return
|
|
self.add(base)
|
|
if field == dv.HOURLY:
|
|
code = str((self.whole.index.get(base[:2], {}).get(base) or {}).get("원문번호") or "")
|
|
for fit in dv._machine_family():
|
|
if code in [str(c) for c in dv._codes(fit)]:
|
|
self.logic(str(fit.get("키")))
|
|
|
|
def logic(self, key: str) -> None:
|
|
if key in self.logics:
|
|
return
|
|
try:
|
|
row = self.whole.logic(key)
|
|
except mf.FormulaError as error:
|
|
self.missing.append(f"{key} · {error}")
|
|
return
|
|
self.logics.add(key)
|
|
self.keys.add(key)
|
|
if isinstance(row.get("흐름"), dict): # 흐름 로직 — 옛 칸 대신 흐름이 쓰는 것을 모음
|
|
self.flow(row)
|
|
return
|
|
for text in _texts(row):
|
|
try:
|
|
node = mf.parse(text)
|
|
except mf.FormulaError:
|
|
continue # 못 읽는 식 — 계산도 같은 까닭으로 멈춤
|
|
for part in mf.walk(node):
|
|
kind = part[0]
|
|
if kind == "ref" or kind == "find":
|
|
for ref in self.refs(row, part[1]):
|
|
self.add(ref)
|
|
elif kind == "logic":
|
|
for ref in self.refs(row, part[1]):
|
|
self.logic(ref)
|
|
for item in row.get("호표", []):
|
|
if isinstance(item.get("요소"), dict):
|
|
self.pick(row, item["요소"], item.get("단위"))
|
|
|
|
|
|
def collect(logic_keys: list[str]) -> dict[str, Any]:
|
|
"""담은 로직의 닫힌 단가 묶음 — `{파일 이름: 파일(줄은 모은 것만 · 차례 그대로)}`."""
|
|
files, whole = master()
|
|
collector = _Collector(whole)
|
|
for key in logic_keys:
|
|
collector.logic(key)
|
|
out: dict[str, Any] = {}
|
|
for name, data in files.items():
|
|
slot = m01.items_key(data)
|
|
rows = [row for row in data.get(slot) or [] if str(row.get("키")) in collector.keys]
|
|
if rows:
|
|
out[name] = {**{k: v for k, v in data.items() if k != slot}, slot: rows}
|
|
if collector.missing:
|
|
out["_빠진"] = sorted(set(collector.missing))
|
|
return out
|
|
|
|
|
|
def to_json(value: Any) -> Any:
|
|
"""Decimal 을 JSON 수로 — 정수는 int · 아니면 원래 자릿수 그대로(float 로 흔들지 않음)."""
|
|
return json.loads(json.dumps(value, ensure_ascii=False, default=_number))
|
|
|
|
|
|
def _number(value: Any) -> Any:
|
|
if isinstance(value, Decimal):
|
|
text = str(value) # 「14480.0」 은 소수 그대로 — 끝수 · 표기가 마스터와 같게
|
|
return int(value) if "." not in text and "E" not in text.upper() else float(value)
|
|
raise TypeError(type(value).__name__)
|
|
|
|
|
|
def load_prices(prices: dict[str, Any]) -> Any:
|
|
"""`단가.json` → 계산 색인(수는 Decimal · 마스터 적재와 같게)."""
|
|
raw = json.dumps({k: v for k, v in prices.items() if not k.startswith("_")}, ensure_ascii=False)
|
|
files = json.loads(raw, parse_float=Decimal, parse_int=Decimal)
|
|
return mf.Master(files)
|
|
|
|
|
|
# ── 계산(25-2e) ─────────────────────────────────────────────────────
|
|
|
|
|
|
def missing_inputs(snapshot: Any, labor: dict[str, Any], values: dict[str, Any]) -> list[dict]:
|
|
"""담은 로직마다 값이 없는(null · 안 적음) 입력 — `{로직, 입력}`."""
|
|
out = []
|
|
for key in mcb.logic_keys(labor.get("줄") or {}):
|
|
try:
|
|
row = snapshot.logic(key)
|
|
except mf.FormulaError:
|
|
continue
|
|
mine = values.get(key) or {}
|
|
for spec in mf.inputs_of(row):
|
|
got = mine.get(spec["이름"])
|
|
if not isinstance(got, dict) or got.get("값") in (None, ""):
|
|
out.append({"로직": key, "입력": spec["이름"]})
|
|
return out
|
|
|
|
|
|
def calc(
|
|
labor: dict[str, Any], prices: dict[str, Any], values: dict[str, Any], picks: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""복사본만으로 계산 — 입력이 빠진 로직은 안 돌리고 `멈춤` · 세 비목 · 계는 돈 로직 합."""
|
|
snapshot = load_prices(prices)
|
|
gaps = missing_inputs(snapshot, labor, values)
|
|
blocked = {gap["로직"] for gap in gaps}
|
|
sums = dict.fromkeys(mf.COST_ITEMS, Decimal(0))
|
|
lines, stuck = [], []
|
|
for at, key in enumerate(mcb.logic_keys(labor.get("줄") or {}), start=1):
|
|
head: dict[str, Any] = {"차례": at, "로직": key}
|
|
if key in blocked:
|
|
names = [gap["입력"] for gap in gaps if gap["로직"] == key]
|
|
lines.append({**head, "ok": False, "까닭": f"입력 없음 — {', '.join(names)}"})
|
|
stuck.append(key)
|
|
continue
|
|
given = {name: one.get("값") for name, one in (values.get(key) or {}).items()}
|
|
try:
|
|
with mf.mp.choosing((picks or {}).get(key)):
|
|
got = mf.run(snapshot, key, given)
|
|
except (mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as error:
|
|
lines.append({**head, "ok": False, "까닭": str(error) or type(error).__name__})
|
|
stuck.append(key)
|
|
continue
|
|
if "결과" in got:
|
|
lines.append({**head, "ok": True, "갈래": "결과", "결과": got["결과"]})
|
|
continue
|
|
for name in mf.COST_ITEMS:
|
|
sums[name] += got[name]
|
|
lines.append(
|
|
{
|
|
**head,
|
|
"ok": True,
|
|
"갈래": "돈",
|
|
**{k: got[k] for k in mf.COST_ITEMS},
|
|
"계": got["계"],
|
|
}
|
|
)
|
|
return to_json({"줄": lines, **sums, "계": sum(sums.values()), "빠진입력": gaps, "멈춤": stuck})
|