From 28da700d4714992ae3eb84166580a537a2ce69c0 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Fri, 2 Oct 2026 08:55:36 +0900 Subject: [PATCH] =?UTF-8?q?feat(M01):=20=ED=9D=90=EB=A6=84=20=EB=A1=9C?= =?UTF-8?q?=EC=A7=81=20=EC=A4=84=20=3D=20=EC=98=9B=20=EC=8B=9D=20=EC=B9=B8?= =?UTF-8?q?(=EC=9E=85=EB=A0=A5=20=C2=B7=20=EC=A4=91=EA=B0=84=20=C2=B7=20?= =?UTF-8?q?=ED=98=B8=ED=91=9C=20=C2=B7=20=EA=B2=B0=EA=B3=BC=20=C2=B7=20?= =?UTF-8?q?=EB=8D=A7=EC=A4=84)=20=EC=A7=80=EC=9B=80=20=C2=B7=20=EC=9D=BD?= =?UTF-8?q?=EB=8A=94=20=EA=B3=B3=EC=9D=80=20=ED=9D=90=EB=A6=84=EC=9D=98=20?= =?UTF-8?q?=EC=84=A4=EA=B3=84=20=EB=95=8C=20=EC=A0=95=ED=95=A0=20=EA=B2=83?= =?UTF-8?q?=20=C2=B7=20=EC=B6=9C=EB=A0=A5=EC=97=90=EC=84=9C(=EC=82=AC?= =?UTF-8?q?=EC=9A=A9=EC=9E=90=20=EA=B2=B0=EC=A0=95)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL --- M01_MasterData/M01_MasterData_Router.py | 11 ++- M01_MasterData/M01_MasterData_Store_Flow.py | 80 ++++++++++++++++--- M02_MasterTemplete/DataFlow/M02_DF_Compute.py | 2 +- M02_MasterTemplete/DataFlow/M02_DF_Values.py | 18 ++++- M02_MasterTemplete/M02_Item_Copy.py | 2 +- M02_MasterTemplete/M02_Item_Prices.py | 49 +++++++++++- .../M02_MasterTemplete_Inputs.py | 8 +- resources/master_data/scripts/check_master.py | 7 +- resources/master_data/scripts/master_auto.py | 4 +- .../master_data/scripts/master_formula.py | 32 +++++++- .../tester/test_m01_logic_flow_engine.py | 2 +- .../tester/test_m01_logic_flow_project.py | 65 +++++++++++++++ resources/tester/test_m01_logic_flow_save.py | 28 +++++++ 13 files changed, 275 insertions(+), 33 deletions(-) create mode 100644 resources/tester/test_m01_logic_flow_project.py diff --git a/M01_MasterData/M01_MasterData_Router.py b/M01_MasterData/M01_MasterData_Router.py index 76f24e24e..c9dff1c54 100644 --- a/M01_MasterData/M01_MasterData_Router.py +++ b/M01_MasterData/M01_MasterData_Router.py @@ -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) diff --git a/M01_MasterData/M01_MasterData_Store_Flow.py b/M01_MasterData/M01_MasterData_Store_Flow.py index 9b5089f05..9aadb06bd 100644 --- a/M01_MasterData/M01_MasterData_Store_Flow.py +++ b/M01_MasterData/M01_MasterData_Store_Flow.py @@ -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 ""))] diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py index cef97b616..86659f59f 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py @@ -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: diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Values.py b/M02_MasterTemplete/DataFlow/M02_DF_Values.py index 4befa7693..060bad891 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Values.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Values.py @@ -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() diff --git a/M02_MasterTemplete/M02_Item_Copy.py b/M02_MasterTemplete/M02_Item_Copy.py index 7708e664f..faf70a09d 100644 --- a/M02_MasterTemplete/M02_Item_Copy.py +++ b/M02_MasterTemplete/M02_Item_Copy.py @@ -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] = {} diff --git a/M02_MasterTemplete/M02_Item_Prices.py b/M02_MasterTemplete/M02_Item_Prices.py index 9db657711..baa316471 100644 --- a/M02_MasterTemplete/M02_Item_Prices.py +++ b/M02_MasterTemplete/M02_Item_Prices.py @@ -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["이름"]}) diff --git a/M02_MasterTemplete/M02_MasterTemplete_Inputs.py b/M02_MasterTemplete/M02_MasterTemplete_Inputs.py index 91bc32d87..5c48d3232 100644 --- a/M02_MasterTemplete/M02_MasterTemplete_Inputs.py +++ b/M02_MasterTemplete/M02_MasterTemplete_Inputs.py @@ -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( { diff --git a/resources/master_data/scripts/check_master.py b/resources/master_data/scripts/check_master.py index 28e14b02e..86ee20f45 100644 --- a/resources/master_data/scripts/check_master.py +++ b/resources/master_data/scripts/check_master.py @@ -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: diff --git a/resources/master_data/scripts/master_auto.py b/resources/master_data/scripts/master_auto.py index 329116042..23e8e55bf 100644 --- a/resources/master_data/scripts/master_auto.py +++ b/resources/master_data/scripts/master_auto.py @@ -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("입력", []): diff --git a/resources/master_data/scripts/master_formula.py b/resources/master_data/scripts/master_formula.py index 4bb9afb14..66ba92833 100644 --- a/resources/master_data/scripts/master_formula.py +++ b/resources/master_data/scripts/master_formula.py @@ -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}」 없음") diff --git a/resources/tester/test_m01_logic_flow_engine.py b/resources/tester/test_m01_logic_flow_engine.py index 58ccea2e3..c825a15eb 100644 --- a/resources/tester/test_m01_logic_flow_engine.py +++ b/resources/tester/test_m01_logic_flow_engine.py @@ -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)", diff --git a/resources/tester/test_m01_logic_flow_project.py b/resources/tester/test_m01_logic_flow_project.py new file mode 100644 index 000000000..2c0d8264f --- /dev/null +++ b/resources/tester/test_m01_logic_flow_project.py @@ -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 []) diff --git a/resources/tester/test_m01_logic_flow_save.py b/resources/tester/test_m01_logic_flow_save.py index dd1f29e5e..7a0efd9fd 100644 --- a/resources/tester/test_m01_logic_flow_save.py +++ b/resources/tester/test_m01_logic_flow_save.py @@ -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