feat(M01): 흐름 로직 줄 = 옛 식 칸(입력 · 중간 · 호표 · 결과 · 덧줄) 지움 · 읽는 곳은 흐름의 설계 때 정할 것 · 출력에서(사용자 결정)
- 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
This commit is contained in:
Claude Opus 5.5
@@ -187,7 +187,11 @@ def get_logics(
|
||||
|
||||
@router.get("/logic")
|
||||
def get_logic(key: str) -> dict:
|
||||
return _call(store.logic, key)
|
||||
"""흐름 로직은 「입력」 칸 없음 — 보기용 「설계 때 정할 것」 을 입력 꼴로(파일 그대로)."""
|
||||
got = _call(store.logic, key)
|
||||
if isinstance(got["logic"].get("흐름"), dict):
|
||||
got["logic"] = {**got["logic"], "입력": _num(store.cm.mf.inputs_of(got["logic"]))}
|
||||
return got
|
||||
|
||||
|
||||
@router.get("/logic/auto")
|
||||
@@ -233,7 +237,10 @@ def post_calc(body: CalcBody) -> dict:
|
||||
|
||||
@router.post("/text")
|
||||
def post_text(body: CalcBody) -> dict:
|
||||
"""읽는 식 줄 — 몸은 `/calc` 와 같음."""
|
||||
"""읽는 식 줄 — 몸은 `/calc` 와 같음 · 흐름 로직은 출력 상자 줄."""
|
||||
got = _call(flow.text, body.key, body.inputs, body.row)
|
||||
if got is not None:
|
||||
return got
|
||||
with store.cm.mf.mp.choosing(body.고름):
|
||||
return _call(store.text, body.key, body.inputs, body.row, body.file)
|
||||
|
||||
|
||||
@@ -167,8 +167,10 @@ def _value(v: Any, words: set | frozenset = frozenset()) -> Any:
|
||||
return store._dec(v)
|
||||
|
||||
|
||||
def run(flow: dict, key: str, inputs: dict) -> dict:
|
||||
"""흐름 셈 — 입력 = 「설계 때 정할 것」 이름 → 값."""
|
||||
def run(flow: dict, key: str, inputs: dict, strict: bool = False) -> dict:
|
||||
"""흐름 셈 — 입력 = 「설계 때 정할 것」 이름 → 값 · strict(조합 · 프로젝트) = 안 준 입력은
|
||||
예시값
|
||||
대신 빈 값 + 알림 「빈 입력」(옛 엔진처럼 그 값을 쓰는 곳만 비움)."""
|
||||
active = _ACTIVE.get()
|
||||
if key and key in active: # 옛 셈을 건너 다시 돌아온 흐름(흐름 → 옛 로직 → 흐름)
|
||||
raise store.cm.mf.FormulaError(f"로직이 서로를 부름 — {' → '.join([*active, key])}")
|
||||
@@ -183,7 +185,12 @@ def run(flow: dict, key: str, inputs: dict) -> dict:
|
||||
given = {k: _value(v, words) for k, v in (inputs or {}).items()}
|
||||
flow = {**flow, "key": key}
|
||||
values = dc._design_values(flow, given)
|
||||
return dc.run(flow, values, load, flow.get("userElements") or [], [key])
|
||||
blank = [d for d in dc.design_inputs(flow) if strict and given.get(d["port"]) is None]
|
||||
for d in blank:
|
||||
values.setdefault(d["box"], {})[d["port"]] = None
|
||||
got = dc.run(flow, values, load, flow.get("userElements") or [], [key])
|
||||
got["notes"] = [*(f"입력 「{d['port']}」 빈 입력" for d in blank), *got.get("notes", [])]
|
||||
return got
|
||||
finally:
|
||||
_ACTIVE.reset(token)
|
||||
|
||||
@@ -195,20 +202,65 @@ def _engine(master, ref: str, row: dict, given: dict) -> dict:
|
||||
"""
|
||||
token = dv.SOURCE.set(master)
|
||||
try:
|
||||
got = run(row[FLOW], ref, given)
|
||||
got = run(row[FLOW], ref, given, strict=True)
|
||||
finally:
|
||||
dv.SOURCE.reset(token)
|
||||
if got["errors"]:
|
||||
raise store.cm.mf.FormulaError(f"「{ref}」 흐름 셈 — {' · '.join(got['errors'])}")
|
||||
cost = got["cost"]
|
||||
kinds = {(b.get("output") or {}).get("cost") for b in row[FLOW].get("boxes") or []
|
||||
if b.get("kind") == "output"} # fmt: skip
|
||||
if kinds and kinds <= {"값"} and len(cost["값들"]) == 1: # 돈 아닌 로직 = 결과 하나
|
||||
return {"결과": next(iter(cost["값들"].values())), "중간": {}}
|
||||
return {"줄": [], **{k: cost[k] for k in dc.COSTS}, "계": cost["계"], "중간": {}}
|
||||
cost, notes = got["cost"], got.get("notes") or []
|
||||
outs = [b for b in row[FLOW].get("boxes") or [] if b.get("kind") == "output"]
|
||||
extra = {"중간": {}, "값들": cost["값들"], **({"알림": notes} if notes else {})}
|
||||
if outs and all((b.get("output") or {}).get("cost") in (None, "값") for b in outs):
|
||||
first = got["outputs"].get(outs[0]["id"]) # 돈 아닌 로직 = 첫 출력이 결과 · 나머지는 값들
|
||||
return {"결과": first, **extra}
|
||||
lines = [] # 옛 「줄」 꼴 — 돈 출력 상자마다 한 줄(수량 1 · 단가 = 금액)
|
||||
for b in outs:
|
||||
kind, value = (b.get("output") or {}).get("cost"), got["outputs"].get(b["id"])
|
||||
if kind in dc.COSTS:
|
||||
unit = (b.get("output") or {}).get("unit") or ""
|
||||
lines.append({"이름": b.get("name") or "", "단위": unit,
|
||||
"수량": Decimal(1), "단가": value, "금액": value, "비목": {kind: value},
|
||||
**({} if value is not None else {"까닭": "값 없음"})}) # fmt: skip
|
||||
sums = {k: cost[k] for k in dc.COSTS}
|
||||
return {"줄": lines, **sums, "계": cost["계"], **extra}
|
||||
|
||||
|
||||
def _inputs(row: dict) -> list[dict]:
|
||||
"""흐름 로직 입력 = 「설계 때 정할 것」 → 옛 입력 꼴(이름 · 단위 · 고르기 · 범위 · 예시)."""
|
||||
out = []
|
||||
for d in dc.design_inputs(row[FLOW]):
|
||||
v = d.get("variable") or {}
|
||||
one = {"이름": d["port"], "단위": None if d.get("unit") in (None, "-") else d["unit"]}
|
||||
one |= {"고르기": v["choices"]} if v.get("choices") else {}
|
||||
one |= {"범위": v["range"]} if v.get("range") else {}
|
||||
out.append(one | {"예시": v.get("example")})
|
||||
return out
|
||||
|
||||
|
||||
store.cm.mf.FLOW_RUN = _engine # 옛 엔진을 부르는 곳이 모두 흐름 로직을 흐름 셈으로(한 곳 갈래)
|
||||
store.cm.mf.FLOW_INPUTS = (
|
||||
_inputs # 입력을 읽는 곳(시험입력 · 자동값 · 프로젝트 입력 · 연결)도 흐름에서
|
||||
)
|
||||
|
||||
|
||||
def text(key: str, inputs: dict, row: dict | None = None) -> dict | None:
|
||||
"""읽는 식 줄(`/text` 꼴) — 흐름 로직이면 출력 상자를 비목 묶음 줄로 · 아니면 None(옛 글)."""
|
||||
got = calc(key, inputs, row)
|
||||
if got is None or not got.get("flow") or "outputs" not in got:
|
||||
return got
|
||||
groups: dict[str, list] = {}
|
||||
for o in got["outputs"]:
|
||||
value = o["value"]
|
||||
shown = f"{value:,}" if isinstance(value, (int, Decimal)) else str(value)
|
||||
line = {"이름": o["name"], "글": shown, "이름글": o["name"], "금액": value, "조각": []}
|
||||
groups.setdefault(o["cost"] or "값", []).append(line)
|
||||
out = [{"비목": k, "줄": v} for k, v in groups.items()]
|
||||
return {
|
||||
"ok": got["ok"],
|
||||
"flow": True,
|
||||
"groups": out,
|
||||
**({"reason": got["reason"]} if "reason" in got else {}),
|
||||
}
|
||||
|
||||
|
||||
def auto(key: str) -> dict | None:
|
||||
@@ -294,9 +346,10 @@ def _redo(row: dict, key: str) -> list[str]:
|
||||
|
||||
|
||||
def _bare(row: dict) -> dict:
|
||||
"""새 흐름 로직 — 옛 식 칸은 빈 것으로(파일 틀 검사) · 출처 없으면 「자체」."""
|
||||
old = {"입력": [], "중간": [], "호표": [], "끝수": None}
|
||||
return {**old, **row, "출처": row.get("출처") or "자체"}
|
||||
"""흐름 로직 줄 — 옛 식 칸(입력 · 중간 · 호표 · 결과 · 덧줄)은 지움(흐름이 정본)
|
||||
· 끝수 · 시험입력 · 메타 칸은 둠 · 출처 없으면 「자체」."""
|
||||
kept = {k: v for k, v in row.items() if k not in store.cm.mf.OLD_FIELDS}
|
||||
return {"끝수": None, **kept, "출처": row.get("출처") or "자체"}
|
||||
|
||||
|
||||
def save(batch: list[dict]) -> dict:
|
||||
@@ -313,6 +366,7 @@ def save(batch: list[dict]) -> dict:
|
||||
raise store.StoreError(400, f"새 흐름 로직은 「{make.OWN}」 에만 더함")
|
||||
row = make._clean(_bare(row), str(row.get("소유") or make.OWNERS[0]))
|
||||
ch.update(op="add", key=None)
|
||||
row = _bare(row) if not new else row
|
||||
ch["row"] = row
|
||||
name = row.get("이름") or ch.get("key") or "새 로직"
|
||||
warns += [f"「{name}」 {w}" for w in _redo(row, str(ch.get("key") or ""))]
|
||||
|
||||
@@ -146,7 +146,7 @@ def _element_value(
|
||||
slot = {"field": rule.get("field"), "op": rule.get("op"), "scope": el.get("scope"),
|
||||
"ref": el.get("ref"), "unit": el.get("unit")} # fmt: skip
|
||||
got = dv.group_find(group, slot, ports.get(src.get("box"), {}).get(src.get("name")))
|
||||
if not got["ok"] and el.get("default"):
|
||||
if not got["ok"] and not got.get("blank") and el.get("default"):
|
||||
errors.append(f"「{box.get('name')} · {el.get('name')}」 {got['reason']} — 기본으로 셈")
|
||||
got = dv.one_value({"group": group, "ref": el["default"], "unit": el.get("unit")})
|
||||
elif given is not None:
|
||||
|
||||
@@ -431,7 +431,11 @@ def _filled(template: str, value: Any) -> str:
|
||||
def _search(slot: dict, value: Any) -> dict:
|
||||
"""조건 찾기 field 검색어 — 틀(요소 ref 묶음)의 {자리} 를 알약 값으로 채워 옛 찾기 그대로."""
|
||||
if value in (None, ""):
|
||||
return {"ok": False, "reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요"}
|
||||
return {
|
||||
"ok": False,
|
||||
"blank": True,
|
||||
"reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요",
|
||||
}
|
||||
cond = _pick(slot.get("ref")) # 요소 ref = 옛 검색어 묶음(규격 · 지역 칸도 그대로)
|
||||
if cond is None: # 묶음이 없으면 범위(구분 · 상세구분) + 채운 검색어
|
||||
scope = slot.get("scope") or {}
|
||||
@@ -468,7 +472,11 @@ def slot_price(slot: dict, value: Any = None, pick: str | None = None) -> dict:
|
||||
if field == "검색어":
|
||||
return _search(slot, value)
|
||||
if value in (None, ""):
|
||||
return {"ok": False, "reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요"}
|
||||
return {
|
||||
"ok": False,
|
||||
"blank": True,
|
||||
"reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요",
|
||||
}
|
||||
want = _squash(value)
|
||||
sub, detail = str(scope.get("구분") or ""), str(scope.get("상세구분") or "")
|
||||
rows = _found(sub, detail, scope.get("찾는말") or "")
|
||||
@@ -520,7 +528,11 @@ def group_find(group: str, slot: dict, value: Any) -> dict:
|
||||
· field 가 없으면 키(원문번호) 또는 이름이 같은 줄 · 하나면 그 값(`one_value`).
|
||||
"""
|
||||
if value in (None, ""):
|
||||
return {"ok": False, "reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요"}
|
||||
return {
|
||||
"ok": False,
|
||||
"blank": True,
|
||||
"reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요",
|
||||
}
|
||||
field, op, want = slot.get("field"), slot.get("op") or "같음", _squash(value)
|
||||
words = str((slot.get("scope") or {}).get("찾는말") or "").split()
|
||||
|
||||
|
||||
@@ -211,7 +211,7 @@ def write_file(
|
||||
def _kept_values(values: Any, labor: dict[str, Any]) -> dict[str, Any]:
|
||||
"""같은 로직 · 같은 입력 이름만 살림."""
|
||||
names = {
|
||||
str(row.get("키")): {spec["이름"] for spec in row.get("입력", [])}
|
||||
str(row.get("키")): {spec["이름"] for spec in prices.mf.inputs_of(row)}
|
||||
for row in labor.get("로직", [])
|
||||
}
|
||||
out: dict[str, Any] = {}
|
||||
|
||||
@@ -125,6 +125,50 @@ class _Collector:
|
||||
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
|
||||
@@ -135,6 +179,9 @@ class _Collector:
|
||||
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)
|
||||
@@ -201,7 +248,7 @@ def missing_inputs(snapshot: Any, labor: dict[str, Any], values: dict[str, Any])
|
||||
except mf.FormulaError:
|
||||
continue
|
||||
mine = values.get(key) or {}
|
||||
for spec in row.get("입력", []):
|
||||
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["이름"]})
|
||||
|
||||
@@ -84,7 +84,7 @@ def inputs(key: str) -> dict[str, Any]:
|
||||
for logic in logics:
|
||||
auto = m01.mau.values(whole, logic)
|
||||
specs = []
|
||||
for spec in logic.get("입력") or []:
|
||||
for spec in m01.cm.mf.inputs_of(logic):
|
||||
one = {k: spec[k] for k in _SPEC if k in spec}
|
||||
one.setdefault("설명", None)
|
||||
one.setdefault("단위", None)
|
||||
@@ -126,7 +126,9 @@ def write_file(column: str, doc: dict[str, Any]) -> None:
|
||||
def _known(key: str) -> dict[str, set[str]]:
|
||||
"""{로직 키: {입력 이름}} — 그 항목 조합이 지금 담은 것."""
|
||||
_, logics, _ = combo_logics(_item(key).get("일위대가"))
|
||||
return {str(lg.get("키")): {str(s.get("이름")) for s in lg.get("입력") or []} for lg in logics}
|
||||
return {
|
||||
str(lg.get("키")): {str(s.get("이름")) for s in m01.cm.mf.inputs_of(lg)} for lg in logics
|
||||
}
|
||||
|
||||
|
||||
def read_links(key: str) -> dict[str, Any]:
|
||||
@@ -204,7 +206,7 @@ def pills(key: str, place: str) -> dict[str, Any]:
|
||||
if row.get("자리") != place or row.get("사라짐"):
|
||||
continue
|
||||
logic = by_key[str(row["로직"])]
|
||||
spec = next(s for s in logic.get("입력") or [] if s.get("이름") == row["입력"])
|
||||
spec = next(s for s in m01.cm.mf.inputs_of(logic) if s.get("이름") == row["입력"])
|
||||
short = combo_store._TAIL.sub("", str(logic.get("이름") or "")) or row["로직"]
|
||||
linked.append(
|
||||
{
|
||||
|
||||
@@ -295,8 +295,9 @@ def check_form(name: str, data: dict) -> list[str]:
|
||||
f" 이 파일 머리·이름 「{want[0]} · {want[1]}」 과 다름"
|
||||
)
|
||||
cols = COLS.get(book)
|
||||
flow = isinstance(row.get("흐름"), dict) # 흐름 로직 — 옛 식 칸 없음
|
||||
for need in cols or ROW_KEYS[kind]:
|
||||
if need not in row:
|
||||
if need not in row and not (flow and need in mf.OLD_FIELDS):
|
||||
out.append(f"{where} · 칸 「{need}」 없음")
|
||||
if "출처" in row and not _source_ok(row["출처"]):
|
||||
out.append(f"{where} · 출처 모양 「{row['출처']}」")
|
||||
@@ -312,7 +313,7 @@ def check_form(name: str, data: dict) -> list[str]:
|
||||
out.append(f"{where} · 「{bundle}」 묶음 칸이 값 묶음과 다름")
|
||||
if kind == "표":
|
||||
out += _check_table(where, row)
|
||||
if kind == "로직" and "결과" not in row and "호표" not in row:
|
||||
if kind == "로직" and "결과" not in row and "호표" not in row and not flow:
|
||||
out.append(f"{where} · 「호표」 도 「결과」 도 없음")
|
||||
return out
|
||||
|
||||
@@ -497,7 +498,7 @@ def 시험입력(row: dict) -> dict:
|
||||
"""일괄 시험 계산용 입력 한 벌 — 고르기 첫 값 · 범위 위끝 · 줄의 「시험입력」 이 이김.
|
||||
시험 계산 전용 — 정본 계산은 화면이 준 값만 씀."""
|
||||
given = {}
|
||||
for spec in row.get("입력", []):
|
||||
for spec in mf.inputs_of(row):
|
||||
if "고르기" in spec:
|
||||
given[spec["이름"]] = spec["고르기"][0]
|
||||
elif "범위" in spec:
|
||||
|
||||
@@ -78,7 +78,9 @@ def _lowest(master: mf.Master, uses: list):
|
||||
|
||||
|
||||
def values(master: mf.Master, row: dict) -> dict:
|
||||
"""로직 줄 하나의 입력마다 자동값."""
|
||||
"""로직 줄 하나의 입력마다 자동값 · 흐름 로직은 「설계 때 정할 것」 예시값."""
|
||||
if isinstance(row.get("흐름"), dict):
|
||||
return {s["이름"]: s.get("예시") for s in mf.inputs_of(row)}
|
||||
finds, divs = _uses(row)
|
||||
out = {}
|
||||
for spec in row.get("입력", []):
|
||||
|
||||
@@ -7,9 +7,12 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import itertools
|
||||
import re
|
||||
import sys
|
||||
from decimal import ROUND_DOWN, ROUND_HALF_UP, ROUND_UP, Decimal
|
||||
from pathlib import Path
|
||||
|
||||
import master_keys as mk
|
||||
|
||||
@@ -31,6 +34,26 @@ _ARITH = re.compile(r"[\d.\s+\-*/()]*\d[\d.\s+\-*/()]*") # 분수·식 값 칸(
|
||||
# 흐름도 로직(줄에 「흐름」) 셈 — 서버(`M01_MasterData_Store_Flow`)가 꽂음 · 없으면 옛 셈만
|
||||
# 꼴 = (master, 키, 줄, 입력) → `run` 과 같은 꼴(돈 = 줄 · 비목 합 · 계 / 돈 아님 = 결과)
|
||||
FLOW_RUN = None
|
||||
# 흐름 로직 입력 = 「설계 때 정할 것」 → 입력 꼴(이름 · 단위 · 고르기 · 범위 · 예시) — 서버가 꽂음
|
||||
FLOW_INPUTS = None
|
||||
OLD_FIELDS = ("입력", "중간", "호표", "결과", "덧줄") # 흐름 로직 줄에는 없음(흐름 하나가 정본)
|
||||
|
||||
|
||||
def _flow_hooks() -> None:
|
||||
"""흐름 셈 · 입력 갈래를 깖 — 서버가 안 깔았으면(스크립트 · 시험) 여기서 서버 모듈을 불러 깖."""
|
||||
if FLOW_RUN is None or FLOW_INPUTS is None:
|
||||
root = str(Path(__file__).resolve().parents[3])
|
||||
if root not in sys.path:
|
||||
sys.path.insert(0, root)
|
||||
importlib.import_module("M01_MasterData.M01_MasterData_Store_Flow")
|
||||
|
||||
|
||||
def inputs_of(row: dict) -> list[dict]:
|
||||
"""로직 입력 — 흐름 로직이면 흐름의 「설계 때 정할 것」 · 아니면 줄의 「입력」."""
|
||||
if isinstance(row.get("흐름"), dict):
|
||||
_flow_hooks()
|
||||
return FLOW_INPUTS(row)
|
||||
return row.get("입력", [])
|
||||
|
||||
|
||||
class FormulaError(ValueError):
|
||||
@@ -485,7 +508,7 @@ def fill_empty(row: dict, given: dict) -> dict:
|
||||
"""화면이 안 보낸 가격 입력(단위에 「원」)은 null 로 채움 — 빈 입력과 같이 그 줄만 비움."""
|
||||
gap = {
|
||||
s["이름"]: None
|
||||
for s in row.get("입력", [])
|
||||
for s in inputs_of(row)
|
||||
if s["이름"] not in given and "원" in str(s.get("단위") or "")
|
||||
}
|
||||
return {**given, **gap}
|
||||
@@ -543,7 +566,8 @@ def _blank_line(item: dict, why: str) -> dict:
|
||||
def run(master: Master, ref: str, given: dict, depth: int = 0) -> dict:
|
||||
"""로직 한 줄 계산 — 돈 로직은 호표 줄 · 비목 합 · 계, 돈 아닌 로직은 결과 수 하나."""
|
||||
row = master.logic(ref)
|
||||
if FLOW_RUN is not None and isinstance(row.get("흐름"), dict):
|
||||
if isinstance(row.get("흐름"), dict): # 흐름 로직 — 깔림에 기대지 않고 스스로 깖
|
||||
_flow_hooks()
|
||||
return FLOW_RUN(master, ref, row, given)
|
||||
env = _inputs(row, given)
|
||||
for step in row.get("중간", []):
|
||||
@@ -641,7 +665,7 @@ def run(master: Master, ref: str, given: dict, depth: int = 0) -> dict:
|
||||
# ── 검사 ──────────────────────────────────────────────────────────────
|
||||
def _choices(row: dict, name: str, master: Master):
|
||||
"""이름이 가질 수 있는 값 — 고르기 입력, 또는 고르기 입력만 끼운 요소 칸인 중간. 모르면 None."""
|
||||
for spec in row.get("입력", []):
|
||||
for spec in inputs_of(row):
|
||||
if spec["이름"] == name:
|
||||
return spec.get("고르기")
|
||||
for step in row.get("중간", []):
|
||||
@@ -729,7 +753,7 @@ def _check_node(row, node, known: set, master: Master, where: str, out: list, ca
|
||||
except FormulaError as e:
|
||||
out.append(f"{where} · {e}")
|
||||
continue
|
||||
names = {s["이름"] for s in target.get("입력", [])}
|
||||
names = {s["이름"] for s in inputs_of(target)}
|
||||
given = {k for k, _ in part[2]}
|
||||
for extra in sorted(given - names):
|
||||
out.append(f"{where} · 「{master.label(part[1])}」 에 입력 「{extra}」 없음")
|
||||
|
||||
@@ -40,7 +40,7 @@ def test_옛_엔진도_흐름_로직은_흐름_셈(client: TestClient) -> None:
|
||||
_save_flow(client, KEY, _flow())
|
||||
whole = store.cm.mf.Master(store.cm.load(folder=store.FOLDER))
|
||||
got = store.cm.mf.run(whole, KEY, {"길이": Decimal(4), "폭": Decimal(5)})
|
||||
assert got["줄"] == [] and got["노무비"] == 20000 and got["계"] == 20000
|
||||
assert [(x["이름"], x["금액"]) for x in got["줄"]] == [("노무비", 20000)] and got["계"] == 20000
|
||||
line = {
|
||||
"종류": "로직",
|
||||
"요소": f"로직({KEY}, 길이=4, 폭=5)",
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
# ruff: noqa: E501
|
||||
"""흐름 로직(옛 식 칸 지운 줄)이 조합 · 프로젝트 길에서도 옛 셈과 같음(PLAN 36 · 사용자 결정 · 사본 폴더).
|
||||
|
||||
못박는 것 — 옛 로직 하나를 변환해 흐름으로 저장(옛 칸 지움)한 뒤
|
||||
- 옛 엔진 `mf.run`(조합 · 프로젝트 길) 합계 = 바꾸기 전 옛 셈 합계
|
||||
- 프로젝트 단가 묶음(`M02_Item_Prices.collect`)만으로 셈 = 같은 합계(흐름이 쓰는 요소 · 표 · 다른 로직이 담김)
|
||||
- 입력 목록(`mf.inputs_of`) = 옛 입력 이름 · 안 준 입력은 「빈 입력」 알림(예시값으로 안 넘어감)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from M01_MasterData import M01_MasterData_Store as store
|
||||
from M01_MasterData import M01_MasterData_Store_Flow as sflow
|
||||
from M02_MasterTemplete import M02_Item_Prices as prices
|
||||
from M02_MasterTemplete.DataFlow import M02_DF_Convert as cv
|
||||
|
||||
REAL = store.FOLDER
|
||||
KEYS = ("GF000185", "GF000219") # 레미콘 검색어 · 펌프차 기계 · 인력 / 표 찾기 · 인력
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def folder(tmp_path: Path, monkeypatch) -> Path:
|
||||
for path in REAL.glob("*.json"):
|
||||
shutil.copy(path, tmp_path / path.name)
|
||||
monkeypatch.setattr(store, "FOLDER", tmp_path)
|
||||
return tmp_path
|
||||
|
||||
|
||||
def _sums(got: dict) -> dict:
|
||||
return {k: round(Decimal(str(got[k])), 6) for k in ("노무비", "재료비", "경비", "계")}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("key", KEYS)
|
||||
def test_흐름으로_바꾼_로직_조합_프로젝트_길도_옛_셈과_같음(folder: Path, key: str) -> None:
|
||||
mf = store.cm.mf
|
||||
got = store.logic(key)
|
||||
old_row = got["logic"]
|
||||
names = [s["이름"] for s in old_row.get("입력") or []]
|
||||
given = {k: v for k, v in store.cm.시험입력(old_row).items()}
|
||||
before = _sums(mf.run(mf.Master(store.cm.load(folder=folder)), key, dict(given)))
|
||||
flow = cv.convert(old_row, {"key": key, "master": store.loaded()[1]})["flow"]
|
||||
flow.pop("key", None)
|
||||
part = {
|
||||
"file": got["file"],
|
||||
"version": got["version"],
|
||||
"changes": [{"op": "edit", "key": key, "row": {**old_row, "흐름": flow}}],
|
||||
}
|
||||
sflow.save([store._dec(part)])
|
||||
row = store.logic(key)["logic"]
|
||||
assert not set(row) & set(mf.OLD_FIELDS)
|
||||
assert [s["이름"] for s in mf.inputs_of(row)] == names
|
||||
assert _sums(mf.run(mf.Master(store.cm.load(folder=folder)), key, dict(given))) == before
|
||||
bundle = prices.collect([key])
|
||||
assert "_빠진" not in bundle, bundle.get("_빠진")
|
||||
snapshot = prices.load_prices(prices.to_json(bundle))
|
||||
assert _sums(mf.run(snapshot, key, dict(given))) == before # 단가 묶음만으로 같은 합계
|
||||
if key == "GF000185": # 빈 입력 — 재료 찾기 자리(표 찾기 빈 조건은 sub7 lookup 몫)
|
||||
blank = mf.run(snapshot, key, {k: v for k, v in given.items() if k != names[0]})
|
||||
assert any(f"「{names[0]}」 빈 입력" in n for n in blank.get("알림") or [])
|
||||
@@ -168,6 +168,9 @@ def test_신규_흐름_로직은_자체_GX_복사는_흐름도(client: TestClien
|
||||
assert key.startswith("GX")
|
||||
saved = _get(client, "/api/m01/logic", key=key)["logic"]
|
||||
assert saved["구분"] == "자체" and saved["흐름"]["name"] == "시험 흐름 로직"
|
||||
assert not set(store.logic(key)["logic"]) & set(
|
||||
store.cm.mf.OLD_FIELDS
|
||||
) # 파일 줄에 옛 식 칸 없음
|
||||
assert (
|
||||
Decimal(str(_post(client, "/api/m01/calc", {"key": key, "inputs": {}})["sums"]["계"]))
|
||||
== 6000
|
||||
@@ -252,3 +255,28 @@ def test_저장_안_한_고친_줄_변환(client: TestClient, folder: Path) -> N
|
||||
if importlib.util.find_spec(sflow.CONVERT) is not None:
|
||||
assert got["flow"]["name"] == "고친 이름" and got["flow"]["boxes"]
|
||||
assert _hashes(folder) == before
|
||||
|
||||
|
||||
def test_흐름_로직_줄은_옛_식_칸_없음_읽는_곳은_흐름에서(client: TestClient, folder: Path) -> None:
|
||||
"""흐름으로 저장 = 입력 · 중간 · 호표 · 결과 · 덧줄 지움(사용자 결정) · 입력을 읽는 곳은 「설계 때 정할 것」."""
|
||||
mf = store.cm.mf
|
||||
_save_flow(client, KEY, _flow())
|
||||
row = store.logic(KEY)["logic"] # 파일 줄 그대로
|
||||
assert not set(row) & set(mf.OLD_FIELDS) and "끝수" in row and row["흐름"]
|
||||
view = _get(client, "/api/m01/logic", key=KEY)[
|
||||
"logic"
|
||||
] # 화면 보기 = 설계 때 정할 것을 입력 꼴로
|
||||
assert [x["이름"] for x in view["입력"]] == ["길이", "폭"]
|
||||
assert [(s["이름"], s["단위"], s["예시"]) for s in mf.inputs_of(row)] == [
|
||||
("길이", "m", 2),
|
||||
("폭", "m", 3),
|
||||
]
|
||||
assert {"길이", "폭"} <= set(store.cm.시험입력(row)) # 줄의 시험입력 칸은 그대로 둠(이김)
|
||||
assert store.mau.values(store.loaded()[1], row) == {"길이": 2, "폭": 3}
|
||||
assert _get(client, "/api/m01/logic/auto", key=KEY) == {"값": {"길이": 2, "폭": 3}}
|
||||
text = _post(client, "/api/m01/text", {"key": KEY, "inputs": {"길이": 4, "폭": 5}})
|
||||
assert text["flow"] is True and text["groups"][0]["비목"] == "노무비"
|
||||
assert Decimal(str(text["groups"][0]["줄"][0]["금액"])) == 20000
|
||||
whole = mf.Master(store.cm.load(folder=folder)) # 옛 엔진 길(조합 · 프로젝트 단가)
|
||||
assert mf.run(whole, KEY, {"길이": Decimal(4), "폭": Decimal(5)})["계"] == 20000
|
||||
assert _get(client, "/api/m01/logic", key=KEY)["blocked"] is False
|
||||
|
||||
Reference in New Issue
Block a user