From 4a62f05caf8842630ad4d7da9fbb02346dce9c3d Mon Sep 17 00:00:00 2001 From: umsangdon Date: Thu, 1 Oct 2026 19:50:41 +0900 Subject: [PATCH] =?UTF-8?q?wip(m02-df):=20=EC=84=9C=EB=B2=84=20=EA=B3=84?= =?UTF-8?q?=EC=82=B0=20M02=5FDF=5FCompute.py(=EC=9E=AC=EB=A3=8C=20?= =?UTF-8?q?=EC=85=8B=20=C2=B7=20=EB=8B=A4=EB=A5=B8=20=EB=A1=9C=EC=A7=81=20?= =?UTF-8?q?=C2=B7=20=EC=84=A0=20=EA=B7=9C=EC=B9=99=20=C2=B7=20=EB=81=9D?= =?UTF-8?q?=EC=88=98)=20=C2=B7=20=EC=84=A4=EA=B3=84=20=EB=95=8C=20?= =?UTF-8?q?=EC=A0=95=ED=95=A0=20=EA=B2=83=20=C2=B7=20=EB=8B=A4=EB=A5=B8=20?= =?UTF-8?q?=EB=A1=9C=EC=A7=81=20=EA=B0=92=20=EA=B8=B8=20=C2=B7=20=EC=8B=9C?= =?UTF-8?q?=ED=97=98=20=EB=8D=94=ED=95=A8(=EA=B7=9C=EC=B9=99=20=ED=95=98?= =?UTF-8?q?=EB=82=98=20=EB=A7=9E=EC=B6=94=EB=8A=94=20=EC=A4=91)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL --- M02_MasterTemplete/DataFlow/M02_DF_Compute.py | 518 ++++++++++++++++++ M02_MasterTemplete/DataFlow/M02_DF_Router.py | 210 ++----- M02_MasterTemplete/DataFlow/M02_DF_Values.py | 44 +- resources/tester/test_m02_df_router.py | 224 ++++++++ 4 files changed, 805 insertions(+), 191 deletions(-) create mode 100644 M02_MasterTemplete/DataFlow/M02_DF_Compute.py diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py new file mode 100644 index 000000000..fc88da736 --- /dev/null +++ b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py @@ -0,0 +1,518 @@ +"""단가산출 흐름도 서버 계산(PLAN 35-9 · 정함 4 · 5 · 6 · 7 · 12) — 저장 다시 셈 · 다른 로직 값. + +모양 = sub7 Model `M02_DF_Model.ts` +(CfLogic · CfLink.rule · CfElementRef.mode · CfInput.logic · CfRound). +화면 계산(`M02_DF_Eval.evalLogic`)과 같은 차례 · 같은 엔진 — 서버는 함수 상자 시트들을 한 통합문서로 +묶어(시트 id = 상자 id) 우리 스프레드시트 엔진(Node · `common_util_spreadsheet`)으로 한 번에 셈. +· 입력 상자 알약 = 요소(재료는 고정 · 선택 · 조건 찾기) · 표 · 변수 · 다른 로직 출력 — + 단가는 마스터에서 새로(`M02_DF_Values`) · 설계 값 values = {상자 id: {알약: 값}} 이 이김 + (재료 선택 자리는 고른 품목 키 · 수) +· 선 → 받는 칸: 값 · 표 블록(왼위) · 앞 함수는 시트 사이 참조 · 선 규칙은 받는 칸의 엑셀 수식 + (round = ROUND · ROUNDUP · ROUNDDOWN · scale = × ÷ +% −% · gate = IF(조건, 값, "") · + lookup = 보조 시트에 표 블록 → COUNTIFS 로 하나인지 · SUMIFS 로 꺼냄 · + 범위 칸 「·아래」 이상 · 「·위」 이하) +· 함수 출력 · 출력 상자 끝수 = 보조 시트 ROUND 계열 · 원가 = 출력 상자 비용 종류별 합 +· 다른 로직 = 저장 로직 · 견본을 받은 설계 값(없으면 그 로직 예시값)으로 셈 · 서로 부르면 막음 +마스터 파일은 읽기만. +""" + +from __future__ import annotations + +import copy +import re +from decimal import Decimal +from typing import Any, Callable + +from common_util import common_util_spreadsheet as spreadsheet +from M01_MasterData import M01_MasterData_Store as store +from M01_MasterData.M01_MasterData_Router import _num +from M02_MasterTemplete.DataFlow import M02_DF_Values as dv + +_CELL = re.compile(r"^([A-Z]+)(\d+)$") +TABLE_PORT, OUTPUT_PORT = "표", "값" +COSTS = ("노무비", "재료비", "경비") +GATE = {"=": "=", "≠": "<>", "≥": ">=", "≤": "<=", ">": ">", "<": "<"} +ROUND = {"반올림": "ROUND", "올림": "ROUNDUP", "버림": "ROUNDDOWN"} +HELP = "_df_" # 보조 시트 이름 머리(표 블록 · 끝수) +FAR = Decimal("1E+300") # 범위 칸 열린 끝 — 엑셀 비교가 늘 참이 되게 +Loader = Callable[[str], dict] + + +def _col(letters: str) -> int: + n = 0 + for ch in letters: + n = n * 26 + ord(ch) - 64 + return n + + +def _letters(n: int) -> str: + out = "" + while n: + n, rest = divmod(n - 1, 26) + out = chr(65 + rest) + out + return out + + +def _at(cell: str) -> tuple[int, int, str]: + m = _CELL.match(str(cell or "").strip().upper()) + if not m: + raise store.StoreError(400, f"칸 주소 모양이 아님 「{cell}」") + return _col(m.group(1)), int(m.group(2)), f"{m.group(1)}{m.group(2)}" + + +def _style(cells: dict, cell: str) -> dict: + return {"서식": cells[cell]["서식"]} if "서식" in (cells.get(cell) or {}) else {} + + +def _put(cells: dict, cell: str, value: Any) -> None: + """칸 하나 — 값 · 수식 `{"식"}` · 빈 값은 값만 뺌(서식은 둠).""" + keep = _style(cells, cell) + if isinstance(value, dict) and "식" in value: + cells[cell] = {"식": value["식"], **keep} + elif value is None or value == "": + cells.pop(cell, None) if not keep else cells.__setitem__(cell, keep) + else: + cells[cell] = {"값": value, **keep} + + +def _place(cells: dict, cell: str, value: Any) -> None: + """받는 칸에 값 · 수식 · 표 블록(cell 이 왼위).""" + col, row, at = _at(cell) + if isinstance(value, list): + for dy, line in enumerate(value): + for dx, one in enumerate(line): + _put(cells, f"{_letters(col + dx)}{row + dy}", one) + else: + _put(cells, at, value) + + +def _plain(v: Any) -> Any: + """엔진 계산값 한 칸 → 수 · 글 · 참거짓 · 빈 값(오류는 None).""" + if not isinstance(v, dict): + return v + if "수" in v: + return Decimal(str(v["수"])) + return v.get("글", v.get("참")) + + +def _literal(v: Any) -> str: + """수식 안 값 글 — 수는 그대로 · 글은 따옴표 · 빈 값은 빈 글.""" + if v is None: + return '""' + if isinstance(v, bool): + return "TRUE" if v else "FALSE" + if isinstance(v, (int, float, Decimal)): + return str(v) + return '"' + str(v).replace('"', '""') + '"' + + +# ── 입력 상자 ────────────────────────────────────────────────────────── +def _element_value( + group: str, el: dict, given: Any, ports: dict, errors: list, box: dict, ctx: dict +) -> Any: + """요소 알약 값 — 재료는 방식 셋 · 그 밖은 그룹 값 · 설계 값이 이김(선택 자리 = 키 · 수).""" + mode = el.get("mode") or "fixed" + if group == "재료" and mode == "choice": + if isinstance(given, (int, float, Decimal)) and not isinstance(given, bool): + return given + got = dv.slot_price({"mode": "choice", "default": el.get("default")}, pick=given or None) + elif group == "재료" and mode == "rule": + rule = el.get("rule") or {} + src = rule.get("from") or {} + slot = { + "mode": "rule", + "scope": el.get("scope"), + "field": rule.get("field"), + "op": rule.get("op"), + } + got = dv.slot_price(slot, ports.get(src.get("box"), {}).get(src.get("name"))) + if not got["ok"] and el.get("default"): + errors.append( + f"「{box.get('name')} · {el.get('name')}」 {got['reason']} — 기본 품목으로 셈" + ) + got = dv.slot_price({"mode": "fixed", "ref": el["default"]}) + elif given is not None: + return given + else: + got = dv.one_value( + {"group": group, "ref": el.get("ref"), "unit": el.get("unit")}, ctx["local"] + ) + if not got["ok"]: + errors.append(f"「{box.get('name')} · {el.get('name')}」 {got.get('reason')}") + return None + return got.get("value") + + +def _input_ports(lg: dict, box: dict, values: dict, ports: dict, errors: list, ctx: dict) -> dict: + inp, given, out = box.get("input") or {}, values.get(box["id"]) or {}, {} + mode = inp.get("mode") + if mode == "logic": + return _logic_ports(lg, box, given, ports, errors, ctx) + if mode == "group": + group = inp.get("group") + for el in inp.get("elements") or []: + out[el.get("name")] = _element_value( + group, el, given.get(el.get("name")), ports, errors, box, ctx + ) + if inp.get("table"): + got = dv.one_value({"group": group, "table": inp["table"]}, ctx["local"]) + out[TABLE_PORT] = got.get("rows") if got["ok"] else None + if not got["ok"]: + errors.append(f"「{box.get('name')} · 표」 {got.get('reason')}") + for v in inp.get("variables") or []: + out[v.get("name")] = given.get(v.get("name"), v.get("example")) + return out + + +def _logic_ports(lg: dict, box: dict, given: dict, ports: dict, errors: list, ctx: dict) -> dict: + """다른 로직 — 받는 알약(설계 값) = 선 · 검증 칸 값 · 없으면 그 로직 예시값 → 그 로직 출력.""" + ref = (box.get("input") or {}).get("logic") or {} + key = str(ref.get("key") or "") + if key in ctx["stack"]: + errors.append( + f"「{box.get('name')}」 로직이 서로를 부름 — {' → '.join([*ctx['stack'], key])}" + ) + return {} + try: + other = ctx["load"](key) + except store.StoreError as e: + errors.append(f"「{box.get('name')}」 {e.detail}") + return {} + designs = {**given} + for link in lg.get("links") or []: + if link.get("to", {}).get("box") == box["id"]: + src = link.get("from") or {} + designs[link["to"].get("name")] = ports.get(src.get("box"), {}).get(src.get("name")) + stack = [*ctx["stack"], key] + result = run(other, _design_values(other, designs), ctx["load"], ctx["local"], stack) + errors += [f"「{box.get('name')}」 {x}" for x in result["errors"]] + names = {b["id"]: b.get("name") for b in other.get("boxes") or []} + return {names[i]: v for i, v in result["outputs"].items() if i in names} + + +def _design_values(lg: dict, designs: dict) -> dict: + """다른 로직의 설계 값 {알약 이름: 값} → 그 로직의 values {상자 id: {알약: 값}}.""" + out: dict[str, dict] = {} + for d in design_inputs(lg): + if d["port"] in designs and designs[d["port"]] is not None: + out.setdefault(d["box"], {})[d["port"]] = designs[d["port"]] + return out + + +# ── 설계 때 정할 것 ──────────────────────────────────────────────────── +def design_inputs(lg: dict) -> list[dict]: + """변수 · 재료 자리(선택) · 다른 로직의 받는 알약(선이 없을 때) — Model `CfDesign` 꼴.""" + links = lg.get("links") or [] + fed = { + (ln.get("to") or {}).get("box", "") + "\0" + (ln.get("to") or {}).get("name", "") + for ln in links + } + out = [] + for box in lg.get("boxes") or []: + inp = box.get("input") or {} + if box.get("kind") != "input": + for v in (box.get("func") or {}).get("variables") or []: + out.append( + { + "box": box["id"], + "port": v.get("name"), + "kind": "변수", + "unit": v.get("unit") or "-", + "variable": v, + } + ) + continue + for v in inp.get("variables") or []: + out.append( + { + "box": box["id"], + "port": v.get("name"), + "kind": "변수", + "unit": v.get("unit") or "-", + "variable": v, + } + ) + for el in inp.get("elements") or [] if inp.get("group") == "재료" else []: + if el.get("mode") == "choice": + out.append( + { + "box": box["id"], + "port": el.get("name"), + "kind": "재료 자리", + "unit": el.get("unit") or "-", + "element": el, + } + ) + for pill in ( + ((inp.get("logic") or {}).get("designs") or []) if inp.get("mode") == "logic" else [] + ): + if box["id"] + "\0" + pill.get("name", "") not in fed: + out.append( + { + "box": box["id"], + "port": pill.get("name"), + "kind": "다른 로직", + "unit": pill.get("unit") or "-", + } + ) + return out + + +# ── 선 규칙 → 받는 칸 수식 ───────────────────────────────────────────── +class _Book: + """한 통합문서 — 함수 시트(id = 상자 id) + 보조 시트(표 블록 · 끝수).""" + + def __init__(self) -> None: + self.sheets: list[dict] = [] + self.helpers: dict[str, dict] = {} + + def helper(self, name: str) -> dict: + if name not in self.helpers: + self.helpers[name] = {"id": name, "이름": name, "칸": {}} + self.sheets.append(self.helpers[name]) + return self.helpers[name] + + +def _ref(sheet: str, cell: str) -> str: + return f"'{sheet}'!{cell}" + + +def _lookup(book: _Book, link: dict, block: Any, rule: dict, expr: Callable) -> Any: + """표에서 찾기 — 보조 시트에 표 블록(열린 범위 끝 ±1E300) · 하나면 그 칸 · 아니면 까닭 글.""" + if not isinstance(block, list) or not block: + return {"식": '"표 없음"'} + head = [str(h) for h in block[0]] + sheet = book.helper(f"{HELP}{link['id']}") + for y, line in enumerate(block, start=1): + for x, v in enumerate(line, start=1): + name = head[x - 1] + if ( + y > 1 + and v is None + and (name.endswith(dv.RANGE_LOW) or name.endswith(dv.RANGE_HIGH)) + ): + v = -FAR if name.endswith(dv.RANGE_LOW) else FAR + _put(sheet["칸"], f"{_letters(x)}{y}", v) + last = len(block) + + def rng(name: str) -> str: + c = _letters(head.index(name) + 1) + return f"'{sheet['id']}'!{c}2:{c}{last}" + + crit = [] + for cond in rule.get("conds") or []: + col, value = str(cond.get("col")), expr(cond.get("from") or {}) + if col in head: + crit.append(f"{rng(col)},{value}") + elif col + dv.RANGE_LOW in head and col + dv.RANGE_HIGH in head: + crit.append( + f'{rng(col + dv.RANGE_LOW)},"<="&{value},{rng(col + dv.RANGE_HIGH)},">="&{value}' + ) + else: + return {"식": f'"표에 「{col}」 칸 없음"'} + want = str(rule.get("col")) + if want not in head or not crit: + return {"식": f'"표에 「{want}」 칸 없음"'} + joined = ",".join(crit) + return { + "식": f"IF(COUNTIFS({joined})=1,SUMIFS({rng(want)},{joined})," + f'IF(COUNTIFS({joined})=0,"표에 없는 조건","표에 여럿 맞음"))' + } + + +def _apply(rule: dict | None, src: str, round_: Callable) -> str: + """값 수식 src 에 규칙(round · scale · gate) — lookup 은 따로.""" + kind = (rule or {}).get("kind") + if kind == "round": + return round_(src, rule) + if kind == "scale": + v = _literal(rule.get("value")) + return { + "×": f"({src})*{v}", + "÷": f"({src})/{v}", + "+%": f"({src})*(1+{v}/100)", + "-%": f"({src})*(1-{v}/100)", + }.get(rule.get("op"), src) + return src + + +def _round(src: str, r: dict) -> str: + return f"{ROUND.get(r.get('method'), 'ROUND')}({src},{int(r.get('places') or 0)})" + + +# ── 계산 ─────────────────────────────────────────────────────────────── +def _order(lg: dict) -> tuple[list[dict], list[dict]]: + """함수 상자 차례(Eval `order` 와 같음) — 앞 함수가 먼저 · 돌고 도는 상자는 따로.""" + funcs = [b for b in lg.get("boxes") or [] if b.get("kind") == "function" and b.get("func")] + links, done, left = lg.get("links") or [], [], {f["id"] for f in funcs} + moved = True + while moved and left: + moved = False + for f in funcs: + if f["id"] in left and not any( + ln["to"].get("box") == f["id"] and ln["from"].get("box") in left for ln in links + ): + done.append(f) + left.discard(f["id"]) + moved = True + return done, [f for f in funcs if f["id"] in left] + + +def _gate(rule: dict, src: str, expr: Callable) -> str: + cond = ( + expr(rule.get("from") or {}) + GATE.get(rule.get("op"), "=") + _literal(rule.get("value")) + ) + return f'IF({cond},{src},"")' + + +def _build(lg: dict, funcs: list[dict], ports: dict, values: dict) -> tuple[_Book, list, dict]: + """(통합문서, 받는 칸 목록 [(상자, 칸, 표인가, 넣은 것)], 출력 자리 {상자: (시트, 칸)}).""" + boxes = {b["id"]: b for b in lg.get("boxes") or []} + links = lg.get("links") or [] + book, fills, outs = _Book(), [], {} + + def expr(port: dict) -> str: + b = boxes.get(port.get("box")) + if b is None: + return '""' + if b.get("kind") == "function": + return _ref(*outs[b["id"]]) if b["id"] in outs else '""' + v = ports.get(b["id"], {}).get(port.get("name")) + return '""' if isinstance(v, list) else _literal(v) + + def incoming(link: dict) -> Any: + """선 하나가 받는 칸에 넣을 것 — 값 · 표 블록 · 수식.""" + src, rule = link["from"], link.get("rule") or {} + up = boxes.get(src.get("box")) or {} + func = up.get("kind") == "function" + raw = None if func else ports.get(up.get("id"), {}).get(src.get("name")) + if rule.get("kind") == "lookup": + return _lookup(book, link, raw, rule, expr) + if isinstance(raw, list): + return raw # 표 블록 — 규칙 없이 그대로 펼침 + if not rule and not func: + return raw + if rule.get("kind") == "gate": + return {"식": _gate(rule, expr(src), expr)} + return {"식": _apply(rule, expr(src), _round)} + + def to(box: str, port: str) -> dict | None: + return next((ln for ln in links if ln.get("to") == {"box": box, "name": port}), None) + + for f in funcs: + func = f["func"] + sheet = {**copy.deepcopy(func.get("sheet") or {}), "id": f["id"], "이름": f["id"]} + cells, given = sheet.setdefault("칸", {}), values.get(f["id"]) or {} + for v in func.get("variables") or []: + if v.get("cell"): + _place(cells, v["cell"], given.get(v.get("name"), v.get("example"))) + for p in func.get("inputs") or []: + if not str(p.get("cell") or "").strip(): + continue + link = to(f["id"], p.get("port")) + value = incoming(link) if link else given.get(p.get("port")) + _place(cells, p["cell"], value) + fills.append((f["id"], _at(p["cell"])[2], isinstance(value, list), value)) + book.sheets.append(sheet) + out = func.get("output") or {} + cell = _at(out.get("cell") or "A1")[2] + if out.get("round"): # 함수 출력 끝수 — 보조 시트 칸 · 뒤 상자는 이 칸을 봄 + helper = book.helper(f"{HELP}r") + at = f"A{len(helper['칸']) + 1}" + _put(helper["칸"], at, {"식": _round(_ref(f["id"], cell), out["round"])}) + outs[f["id"]] = (helper["id"], at) + else: + outs[f["id"]] = (f["id"], cell) + for b in boxes.values(): # 출력 상자 — 받은 값 · 규칙 · 끝수를 보조 시트 칸에 + link = to(b["id"], OUTPUT_PORT) if b.get("kind") == "output" else None + if link is None: + continue + value = incoming(link) + src = value["식"] if isinstance(value, dict) else _literal(value) + if (b.get("output") or {}).get("round"): + src = _round(src, b["output"]["round"]) + helper = book.helper(f"{HELP}o") + at = f"A{len(helper['칸']) + 1}" + _put(helper["칸"], at, {"식": src}) + outs["out:" + b["id"]] = (helper["id"], at) + return book, fills, outs + + +def _needs_func(boxes: dict) -> bool: + """재료 조건 찾기가 함수 출력을 쓰나 — 그러면 한 번 더 셈.""" + for b in boxes.values(): + for el in (b.get("input") or {}).get("elements") or []: + src = (el.get("rule") or {}).get("from") or {} + if el.get("mode") == "rule" and boxes.get(src.get("box"), {}).get("kind") == "function": + return True + return False + + +def run(lg: dict, values: dict, load: Loader, local: list[dict], stack: list | None = None) -> dict: + """셈 — {ports, outputs, cost, errors, filled}(filled = 받는 칸에 값만 되돌려 넣음).""" + ctx = {"load": load, "local": local, "stack": stack or [lg.get("key") or ""]} + boxes = {b["id"]: b for b in lg.get("boxes") or []} + funcs, loop = _order(lg) + ports: dict[str, dict] = {} + for _ in range(2 if _needs_func(boxes) else 1): + errs = [f"「{f.get('name')}」 함수 상자가 서로를 기다림(돌고 도는 선)" for f in loop] + for b in boxes.values(): + if b.get("kind") == "input": + ports[b["id"]] = _input_ports(lg, b, values, ports, errs, ctx) + book, fills, outs = _build(lg, funcs, ports, values) + doc = _num({"종류": "통합문서", "판": 1, "서식": [{}] * 64, "시트": book.sheets}) + try: + calc = spreadsheet.recalc_workbooks([doc])[0] if book.sheets else {} + except spreadsheet.RecalcError as e: + raise store.StoreError(503, str(e)) from e + sheets = {s["id"]: s for s in book.sheets} + + def read(sheet: str, cell: str) -> Any: + hit = (calc.get(sheet) or {}).get(cell) + if hit is not None: + return _plain(hit) + return ((sheets.get(sheet) or {"칸": {}})["칸"].get(cell) or {}).get("값") + + for f in funcs: + ports[f["id"]] = {f["func"]["output"].get("name"): read(*outs[f["id"]])} + errors = errs + for sheet, cells in calc.items(): + for cell, v in cells.items(): + if "오류" in v: + name = boxes.get(sheet, {}).get("name") or "끝수 · 규칙" + errors.append(f"「{name}」 {cell} 칸 계산 오류 {v['오류']}") + elif v.get("글") in ("표에 없는 조건", "표에 여럿 맞음", "표 없음"): + errors.append(f"표에서 찾기 — {v['글']}") + outputs, cost = {}, {**dict.fromkeys(COSTS, Decimal(0)), "값들": {}} + for b in boxes.values(): + if b.get("kind") != "output": + continue + at = outs.get("out:" + b["id"]) + value = read(*at) if at else None + value = None if value == "" else value + outputs[b["id"]] = value + kind = (b.get("output") or {}).get("cost") or "값" + if ( + kind in COSTS + and isinstance(value, (int, float, Decimal)) + and not isinstance(value, bool) + ): + cost[kind] += Decimal(str(value)) + elif kind == "값": + cost["값들"][b.get("name")] = value + elif value is not None: + errors.append(f"「{b.get('name')}」 출력 값이 수가 아님") + cost["계"] = sum(cost[k] for k in COSTS) + filled = copy.deepcopy(lg) # 받는 칸에 값만 되돌려 넣음 — 수식 · 참조는 셈한 값으로 + for b in filled.get("boxes") or []: + if b.get("kind") != "function" or not b.get("func"): + continue + cells = b["func"].setdefault("sheet", {}).setdefault("칸", {}) + given = values.get(b["id"]) or {} + for v in b["func"].get("variables") or []: + if v.get("cell"): + _place(cells, v["cell"], given.get(v.get("name"), v.get("example"))) + for box_id, cell, block, value in fills: + if box_id == b["id"]: + _place(cells, cell, value if block else read(box_id, cell)) + return {"ports": ports, "outputs": outputs, "cost": cost, "errors": errors, "filled": filled} diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Router.py b/M02_MasterTemplete/DataFlow/M02_DF_Router.py index c965d5db1..c7375090c 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Router.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Router.py @@ -25,21 +25,19 @@ from __future__ import annotations -import copy import json import os import re from datetime import datetime -from decimal import Decimal from pathlib import Path from typing import Any from fastapi import APIRouter, HTTPException from pydantic import BaseModel -from common_util import common_util_spreadsheet as spreadsheet from M01_MasterData import M01_MasterData_Store as store from M01_MasterData.M01_MasterData_Router import _num +from M02_MasterTemplete.DataFlow import M02_DF_Compute as dc from M02_MasterTemplete.DataFlow import M02_DF_Values as dv cm = store.cm @@ -184,174 +182,19 @@ def user_edit(op: str, element: dict | None, uid: str | None) -> dict: return {"element": one} -# ── 저장 · 계산 ──────────────────────────────────────────────────────── -def _col(letters: str) -> int: - n = 0 - for ch in letters: - n = n * 26 + ord(ch) - 64 - return n +# ── 저장 · 계산(`M02_DF_Compute`) ───────────────────────────────────── +def _load(key: str) -> dict: + """다른 로직 입력이 부르는 로직 — 내 것(`_data`) · 견본(`_sample`).""" + logic, _ = _read(key) + return {**logic, "key": key} -def _letters(n: int) -> str: - out = "" - while n: - n, rest = divmod(n - 1, 26) - out = chr(65 + rest) + out - return out - - -def _style(cells: dict, cell: str) -> dict: - return {"서식": cells[cell]["서식"]} if "서식" in (cells.get(cell) or {}) else {} - - -def _put(cells: dict, cell: str, value: Any) -> None: - """받는 칸 하나 — 값 칸으로 덮어씀(수식이 있었어도) · 빈 값은 값만 뺌(서식은 둠).""" - keep = _style(cells, cell) - if value is None and not keep: - cells.pop(cell, None) - else: - cells[cell] = {**({"값": value} if value is not None else {}), **keep} - - -def _place(cells: dict, cell: str, value: Any) -> None: - """값 하나 · 표 블록(cell 이 왼위) · 수식 `{"식": …}` 을 받는 칸에.""" - m = _CELL.match(str(cell or "").upper()) - if not m: - raise store.StoreError(400, f"칸 주소 모양이 아님 「{cell}」") - col, row, at = _col(m.group(1)), int(m.group(2)), f"{m.group(1)}{m.group(2)}" - if isinstance(value, dict) and "식" in value: - cells[at] = {**value, **_style(cells, at)} - elif isinstance(value, list): - for dy, line in enumerate(value): - for dx, one in enumerate(line): - _put(cells, f"{_letters(col + dx)}{row + dy}", one) - else: - _put(cells, at, value) - - -def _plain(v: Any) -> Any: - """엔진 계산값 한 칸 → 수 · 글 · 참거짓 · 빈 값(오류는 None).""" - if not isinstance(v, dict): - return v - if "수" in v: - return Decimal(str(v["수"])) - if "글" in v: - return v["글"] - if "참" in v: - return v["참"] - return None - - -def _source(links: list[dict], box: str, port: str) -> dict | None: - return next((ln["from"] for ln in links if ln.get("to") == {"box": box, "name": port}), None) - - -def _input_ports(box: dict, local: list[dict], errors: list[str]) -> dict[str, Any]: - """입력 상자 알약 값 — 요소 name · 표 「표」 · 변수 name(예시값).""" - inp, out = box.get("input") or {}, {} - if inp.get("mode") == "group": - group = inp.get("group") - asks = [ - (el.get("name"), {"ref": el.get("ref"), "unit": el.get("unit")}) - for el in inp.get("elements") or [] - ] - if inp.get("table"): - asks.append((TABLE_PORT, {"table": inp["table"]})) - for name, ask in asks: - got = dv.one_value({"group": group, **ask}, local) - out[name] = (got.get("rows") or got.get("value")) if got["ok"] else None - if not got["ok"]: - errors.append(f"「{box.get('name')} · {name}」 {got.get('reason')}") - for v in inp.get("variables") or []: - out[v.get("name")] = v.get("example") - return out - - -def compute(logic: dict, local: list[dict]) -> tuple[dict, dict]: - """함수 상자 시트를 한 통합문서로 셈 → (값을 되돌려 넣은 다이어그램, 계산 결과).""" - boxes = {b.get("id"): b for b in logic.get("boxes") or []} - links = logic.get("links") or [] - errors: list[str] = [] - ports = { - i: _input_ports(b, local, errors) for i, b in boxes.items() if b.get("kind") == "input" - } - funcs = [b for b in boxes.values() if b.get("kind") == "function" and b.get("func")] - sheets, fills = [], [] - for b in funcs: # 한 통합문서 — 시트 id · 이름 = 상자 id - f = b["func"] - sheet = {**copy.deepcopy(f.get("sheet") or {}), "id": b["id"], "이름": b["id"]} - cells = sheet.setdefault("칸", {}) - for v in f.get("variables") or []: - if v.get("cell"): - _place(cells, v["cell"], v.get("example")) - for p in f.get("inputs") or []: - src = _source(links, b["id"], p.get("port")) - if src is None or src.get("box") not in boxes: - continue - up = boxes[src["box"]] - if up.get("kind") == "function": - at = ((up.get("func") or {}).get("output") or {}).get("cell") or "" - _place(cells, p["cell"], {"식": f"'{up['id']}'!{at}"}) - fills.append((b["id"], p["cell"], ("ref", up["id"], at))) - else: - value = ports.get(up["id"], {}).get(src.get("name")) - _place(cells, p["cell"], value) - fills.append((b["id"], p["cell"], ("value", value, None))) - sheets.append(sheet) - top = [c.get("서식", 0) for s in sheets for c in s["칸"].values() if isinstance(c, dict)] - book = _num( - {"종류": "통합문서", "판": 1, "서식": [{}] * (int(max(top, default=0)) + 1), "시트": sheets} - ) - try: - calc = spreadsheet.recalc_workbooks([book])[0] if sheets else {} - except spreadsheet.RecalcError as e: - raise store.StoreError(503, str(e)) from e - - def cell_value(box_id: str, cell: str) -> Any: - hit = (calc.get(box_id) or {}).get(cell) - if hit is None: - sheet = next(s for s in book["시트"] if s["id"] == box_id) - return (sheet["칸"].get(cell) or {}).get("값") - if isinstance(hit, dict) and "오류" in hit: - errors.append(f"「{boxes[box_id].get('name')}」 {cell} 칸 계산 오류 {hit.get('오류')}") - return _plain(hit) - - for b in funcs: - out = b["func"].get("output") or {} - ports[b["id"]] = {out.get("name"): cell_value(b["id"], out.get("cell") or "")} - outputs, cost = {}, {**dict.fromkeys(COSTS, Decimal(0)), "값들": {}} - for b in boxes.values(): # 출력 상자 · 원가 - if b.get("kind") != "output": - continue - src = _source(links, b["id"], OUTPUT_PORT) - value = ports.get(src["box"], {}).get(src.get("name")) if src else None - outputs[b["id"]] = value - kind = (b.get("output") or {}).get("cost") - if ( - kind in COSTS - and isinstance(value, (int, float, Decimal)) - and not isinstance(value, bool) - ): - cost[kind] += Decimal(str(value)) - elif kind == "값": - cost["값들"][b.get("name")] = value - cost["계"] = sum(cost[k] for k in COSTS) - done = copy.deepcopy(logic) # 받는 칸에 값만 되돌려 넣음 — 시트 사이 참조는 셈한 값으로 - for b in done.get("boxes") or []: - if b.get("kind") != "function" or not b.get("func"): - continue - cells = b["func"].setdefault("sheet", {}).setdefault("칸", {}) - for v in b["func"].get("variables") or []: - if v.get("cell"): - _place(cells, v["cell"], v.get("example")) - for box_id, cell, (how, a, at) in fills: - if box_id == b["id"]: - _place(cells, cell, cell_value(a, at) if how == "ref" else a) - return done, {"ports": ports, "outputs": outputs, "cost": cost, "errors": errors} +def _local(logic: dict) -> list[dict]: + return [*(logic.get("userElements") or []), *_users()] def save(logic: dict) -> dict: - """저장 — 입력 값 다시 받기 → 함수 상자 시트를 Node 로 셈 → 값만 되돌려 넣어 `_data`.""" + """저장 — 입력 값 다시 받기 → 함수 상자 시트를 Node 로 셈 → 받는 칸에 값만 되돌려 `_data`.""" if not str(logic.get("name") or "").strip(): raise store.StoreError(400, "이름을 넣어 주세요") if not isinstance(logic.get("boxes"), list) or not isinstance(logic.get("links"), list): @@ -361,13 +204,27 @@ def save(logic: dict) -> dict: key = None # 견본은 그대로 두고 새 번호로 elif key: _read(key) # 있는 것만 고침 — 없으면 404 - done, result = compute(logic, [*(logic.get("userElements") or []), *_users()]) + got = dc.run(logic, {}, _load, _local(logic), [key or ""]) if not key: used = [int(m.group(1)) for p in DATA.glob("DF-*.json") if (m := _KEY.match(p.stem))] key = f"DF-{max(used, default=0) + 1:04d}" - done = {**done, "key": key, "saved": datetime.now().isoformat(timespec="seconds")} + done = {**got.pop("filled"), "key": key, "saved": datetime.now().isoformat(timespec="seconds")} _write(DATA / f"{key}.json", _num(done)) - return {"key": key, "logic": done, "values": result} + return {"key": key, "logic": done, "values": got} + + +def logic_value(key: str, designs: dict) -> dict: + """다른 로직 값 — 저장 로직 · 견본을 받은 설계 값(없으면 예시값)으로 셈 · 출력 이름 → 값.""" + other = _load(key) + got = dc.run(other, dc._design_values(other, designs), _load, _local(other), [key]) + names = {b["id"]: b.get("name") for b in other.get("boxes") or []} + outputs = {names[i]: v for i, v in got["outputs"].items() if i in names} + return {"key": key, "outputs": outputs, "cost": got["cost"], "errors": got["errors"]} + + +def design_inputs(key: str) -> dict: + """설계 때 정할 것 — 변수 · 재료 자리 · 다른 로직 받는 알약(선이 없을 때) · CfDesign 꼴.""" + return {"key": key, "designs": dc.design_inputs(_load(key))} # ── 길 ──────────────────────────────────────────────────────────────── @@ -417,6 +274,21 @@ def post_user_elements(body: UserBody) -> dict: STANDARD = HERE / "_standard" +class LogicValueBody(BaseModel): + key: str + designs: dict[str, Any] = {} + + +@router.post("/logic-value") +def post_logic_value(body: LogicValueBody) -> dict: + return _call(logic_value, body.key, body.designs) + + +@router.get("/design-inputs") +def get_design_inputs(key: str) -> dict: + return _call(design_inputs, key) + + class StandardsBody(BaseModel): unit: dict[str, Any] money: dict[str, Any] diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Values.py b/M02_MasterTemplete/DataFlow/M02_DF_Values.py index 9e8c5cfd5..cb4cfad15 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Values.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Values.py @@ -335,16 +335,17 @@ def materials(q: str, sub: str, detail: str, page: int, size: int) -> dict: return {"total": len(hits), "page": page, "size": size, "items": [_brief_row(r) for r in part]} -RULES = ("규격", "이름", "같음") # 규격에 값 포함 · 이름에 값 포함 · 규격 또는 이름이 값과 같음 +FIELDS_RULE, OPS_RULE = ("규격", "이름"), ("포함", "같음") # Model CfMaterialRule def slot_price(slot: dict, value: Any = None, pick: str | None = None) -> dict: - """재료 자리 단가 — 고정 = 품목 · 선택 = 고른 품목(없으면 기본) · 조건 = 범위 + 규칙으로 찾기. + """재료 자리 단가 — 고정 = 품목 · 선택 = 고른 품목(없으면 기본) · 조건 찾기 = 범위 + 규칙. - slot = {mode: fixed | choice | cond, ref?, default?, scope?: {구분?, 상세구분?, q?}, rule?} - 조건 찾기: 하나면 그 단가 · 0 이면 「맞는 재료 없음」 · 여럿이면 몇 개인지 + 후보 다섯. + slot = Model CfElementRef 꼴 {mode: fixed | choice | rule, ref?, default?, + scope?: {구분?, 상세구분?, 찾는말?}, field?: 규격 | 이름, op?: 포함 | 같음} + 조건 찾기(value = 이어진 알약 값): 하나면 그 단가 · 0 이면 맞는 재료 없음 · 여럿이면 후보 다섯. """ - mode = slot.get("mode") + mode = slot.get("mode") or "fixed" if mode == "fixed": return _price("재료", str(slot.get("ref") or "")) if mode == "choice": @@ -352,28 +353,27 @@ def slot_price(slot: dict, value: Any = None, pick: str | None = None) -> dict: if not ref: return {"ok": False, "reason": "고른 품목도 기본 품목도 없음 — 품목을 골라 주세요"} return _price("재료", str(ref)) - if mode != "cond": + if mode != "rule": return {"ok": False, "reason": f"모르는 재료 방식 「{mode}」"} - rule, scope = slot.get("rule") or "규격", slot.get("scope") or {} - if rule not in RULES: - return {"ok": False, "reason": f"모르는 찾기 규칙 「{rule}」"} + field, op, scope = ( + slot.get("field") or "규격", + slot.get("op") or "포함", + slot.get("scope") or {}, + ) + if field not in FIELDS_RULE or op not in OPS_RULE: + return {"ok": False, "reason": f"모르는 찾기 규칙 「{field} {op}」"} if value in (None, ""): return {"ok": False, "reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요"} want = _squash(value) - rows = _found( - str(scope.get("구분") or ""), str(scope.get("상세구분") or ""), scope.get("q") or "" - ) - fits = { - "규격": lambda r: want in _squash(r.get("규격")), - "이름": lambda r: want in _squash(r.get("이름")), - "같음": lambda r: want in (_squash(r.get("규격")), _squash(r.get("이름"))), - }[rule] - hits = [r for r in rows if fits(r)] + sub, detail = str(scope.get("구분") or ""), str(scope.get("상세구분") or "") + rows = _found(sub, detail, scope.get("찾는말") or "") + if op == "같음": + hits = [r for r in rows if _squash(r.get(field)) == want] + else: + hits = [r for r in rows if want in _squash(r.get(field))] if not hits: - where = ( - " · ".join(str(x) for x in (scope.get("구분"), scope.get("상세구분")) if x) or "전체" - ) - return {"ok": False, "reason": f"맞는 재료 없음 — {where} 에서 {rule} 「{value}」"} + where = " · ".join(x for x in (sub, detail) if x) or "전체" + return {"ok": False, "reason": f"맞는 재료 없음 — {where} 에서 {field}에 「{value}」 {op}"} if len(hits) > 1: names = ", ".join(f"{r.get('이름')} {r.get('규격') or ''}".strip() for r in hits[:5]) return { diff --git a/resources/tester/test_m02_df_router.py b/resources/tester/test_m02_df_router.py index 59e14a1a3..9f4277ee3 100644 --- a/resources/tester/test_m02_df_router.py +++ b/resources/tester/test_m02_df_router.py @@ -352,3 +352,227 @@ def test_기계_키_칸_꼴과_시간당_경비(client: TestClient) -> None: saved = _post(client, "/api/m02/df/save", {"logic": body}) assert _d(saved["values"]["cost"]["경비"]) == 2 * _d(want) _post(client, "/api/m02/df/delete", {"key": saved["key"]}) + + +# ── 35-9 넓힘 — 재료 셋 · 자재 찾기 · 다른 로직 · 설계 때 정할 것 · 선 규칙 · 끝수 · 표준 표 ───── +PIPES = [ # 견본 자재(사본 자재품목에만 더함) + {"이름": "견본관", "규격": "D300", "물가자료": 30000}, + {"이름": "견본관", "규격": "D450", "물가자료": 45000}, + {"이름": "견본관", "규격": "D450 B종", "물가자료": 47000}, +] + + +def _add_pipes(folder: Path) -> list[str]: + path = folder / "재료_자재품목.json" + data = json.loads(path.read_text(encoding="utf-8")) + cols = list(data["줄"][0]) + keys = [] + for i, p in enumerate(PIPES): + row = dict.fromkeys(cols) + row.update( + {"키": f"MT99999{i}", "구분": "시험구분", "상세구분": "시험상세", "단위": "m", **p} + ) + data["줄"].append(row) + keys.append(row["키"]) + path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + return keys + + +def _slot(**slot) -> dict: + return {"id": "s", "group": "재료", "slot": slot} + + +def test_재료_세_방식(client: TestClient, folder: Path) -> None: + keys = _add_pipes(folder) + scope = {"구분": "시험구분", "상세구분": "시험상세", "찾는말": "견본관"} + ask = [ + {**_slot(mode="fixed", ref=keys[0]), "id": "fixed"}, + {**_slot(mode="choice", default=keys[0]), "id": "choice"}, + {**_slot(mode="choice", default=keys[0]), "id": "picked", "pick": keys[1]}, + {**_slot(mode="choice"), "id": "none"}, + {**_slot(mode="rule", scope=scope, field="규격", op="포함"), "id": "one", "value": "D300"}, + {**_slot(mode="rule", scope=scope, field="규격", op="포함"), "id": "many", "value": "D450"}, + {**_slot(mode="rule", scope=scope, field="규격", op="같음"), "id": "same", "value": "D450"}, + { + **_slot(mode="rule", scope=scope, field="이름", op="포함"), + "id": "zero", + "value": "없는관", + }, + ] + got = {v["id"]: v for v in _post(client, "/api/m02/df/values", {"items": ask})["values"]} + assert got["fixed"]["value"] == got["choice"]["value"] == 30000 + assert got["picked"]["value"] == 45000 and got["none"]["ok"] is False + assert got["one"]["value"] == 30000 and got["same"]["value"] == 45000 + assert got["many"]["ok"] is False and "2개" in got["many"]["reason"] + assert [c["spec"] for c in got["many"]["candidates"]] == ["D450", "D450 B종"] + assert got["zero"]["ok"] is False and got["zero"]["reason"].startswith("맞는 재료 없음") + + +def test_자재_찾기_띄어쓰기_AND_쪽(client: TestClient, folder: Path) -> None: + keys = _add_pipes(folder) + + def find(**q) -> dict: + res = client.get("/api/m02/df/materials", params=q) + assert res.status_code == 200, res.text + return res.json() + + assert find(q="견본관 D450")["total"] == 2 + assert [x["ref"] for x in find(q="견본관 b종")["items"]] == [keys[2]] + assert find(q=f"{keys[1]}")["total"] == 1 # 키로도 + page = find(q="견본관", sub="시험구분", detail="시험상세", page=2, size=2) + assert page["total"] == 3 and page["page"] == 2 and len(page["items"]) == 1 + one = page["items"][0] + assert one == {"ref": keys[2], "name": "견본관", "spec": "D450 B종", "unit": "m", "price": 47000, + "구분": "시험구분", "상세구분": "시험상세"} # fmt: skip + + +def _area(x: int = 100) -> dict: + """면적(예시 2) × x = 결과(값) — 다른 로직이 부를 견본.""" + return { + "key": None, + "name": "면적 곱", + "resultUnit": "-", + "boxes": [ + _box("v", "input", "면적", input={"mode": "variable", "elements": [], "variables": [ + {"name": "면적", "type": "number", "unit": "㎡", "example": 2}]}), + _box("f", "function", "곱", func=_func({"id": "s", "이름": "곱", "칸": {"B1": {"식": f"A1*{x}"}}}, + [{"port": "면적", "cell": "A1"}], {"cell": "B1", "name": "곱", "unit": "-"})), + _box("o", "output", "결과", output={"cost": "값", "unit": "-"}), + ], + "links": [_link("a", "v", "면적", "f", "면적"), _link("b", "f", "곱", "o", "값")], + "userElements": [], + } # fmt: skip + + +@pytest.mark.skipif(NO_NODE, reason="node 가 없음") +def test_다른_로직_값_설계_값_순환(client: TestClient) -> None: + a = _post(client, "/api/m02/df/save", {"logic": _area()})["key"] + assert _post(client, "/api/m02/df/logic-value", {"key": a})["outputs"] == {"결과": 200} + assert _post(client, "/api/m02/df/logic-value", {"key": a, "designs": {"면적": 5}})[ + "outputs" + ] == {"결과": 500} + + ref = {"key": a, "name": "면적 곱", "outputs": [{"name": "결과", "unit": "-"}], + "designs": [{"name": "면적", "unit": "㎡"}]} # fmt: skip + b = { + "key": None, "name": "다른 로직 잇기", "resultUnit": "원", + "boxes": [ + _box("L", "input", "면적 곱", input={"mode": "logic", "elements": [], "variables": [], "logic": ref}), + _box("o", "output", "경비", output={"cost": "경비", "unit": "원"}), + ], + "links": [_link("a", "L", "결과", "o", "값")], + "userElements": [], + } # fmt: skip + saved = _post(client, "/api/m02/df/save", {"logic": b}) + assert ( + saved["values"]["ports"]["L"] == {"결과": 200} + and _d(saved["values"]["cost"]["경비"]) == 200 + ) + designs = client.get("/api/m02/df/design-inputs", params={"key": saved["key"]}).json()[ + "designs" + ] + assert [(d["box"], d["port"], d["kind"]) for d in designs] == [("L", "면적", "다른 로직")] + + # 순환 — 로직이 자기를 부름 + loop = {**b, "key": saved["key"]} + loop["boxes"][0]["input"]["logic"] = {**ref, "key": saved["key"]} + got = _post(client, "/api/m02/df/save", {"logic": loop})["values"] + assert any("서로를 부름" in e for e in got["errors"]) + + +@pytest.mark.skipif(NO_NODE, reason="node 가 없음") +def test_선_규칙_끝수_재료_조건_설계_때_정할_것(client: TestClient, folder: Path) -> None: + keys = _add_pipes(folder) + scope = {"구분": "시험구분", "상세구분": "시험상세", "찾는말": "견본관"} + vars_ = [{"name": "돌", "type": "choice", "unit": "-", "example": "견치돌"}, + {"name": "높이", "type": "number", "unit": "m", "example": 4}, + {"name": "관경", "type": "text", "unit": "-", "example": "D300"}] # fmt: skip + pipe = {"ref": "", "name": "관", "spec": "", "unit": "m", "mode": "rule", "scope": scope, + "rule": {"field": "규격", "op": "포함", "from": {"box": "v", "name": "관경"}}} # fmt: skip + pick = { + "ref": "", + "name": "골라 쓰는 관", + "spec": "", + "unit": "m", + "mode": "choice", + "default": keys[1], + } + sheet = {"id": "s", "이름": "셈", "칸": {"B1": {"식": "A1+A2+A3+A4"}}} + lg = { + "key": None, "name": "규칙 견본", "resultUnit": "원", + "boxes": [ + _box("v", "input", "값들", input={"mode": "variable", "elements": [], "variables": vars_}), + _box("t", "input", "표", input={"mode": "group", "group": "소요량", "elements": [], + "variables": [], "table": {"key": SAMPLE}}), + _box("m", "input", "재료", input={"mode": "group", "group": "재료", "elements": [pipe, pick], + "variables": []}), + _box("f", "function", "셈", func={**_func(sheet, [{"port": "품", "cell": "A1"}, + {"port": "관", "cell": "A2"}, {"port": "통과", "cell": "A3"}, {"port": "막힘", "cell": "A4"}], + {"cell": "B1", "name": "합", "unit": "원", "round": {"places": 0, "method": "버림"}})}), + _box("o", "output", "재료비", output={"cost": "재료비", "unit": "원", "round": {"places": -1, "method": "버림"}}), + _box("o2", "output", "골라 쓴 관", output={"cost": "값", "unit": "원"}), + ], + "links": [ + {**_link("l1", "t", "표", "f", "품"), "rule": {"kind": "lookup", "col": "석공", "conds": [ + {"col": "돌", "from": {"box": "v", "name": "돌"}}, {"col": "높이", "from": {"box": "v", "name": "높이"}}]}}, + {**_link("l2", "m", "관", "f", "관"), "rule": {"kind": "scale", "op": "+%", "value": 10}}, + {**_link("l3", "v", "높이", "f", "통과"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"}, "op": "≥", "value": 3}}, + {**_link("l4", "v", "높이", "f", "막힘"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"}, "op": "<", "value": 3}}, + _link("l5", "f", "합", "o", "값"), + {**_link("l6", "m", "골라 쓰는 관", "o2", "값"), "rule": {"kind": "round", "places": -3, "method": "반올림"}}, + ], + "userElements": [], + } # fmt: skip + got = _post(client, "/api/m02/df/save", {"logic": lg}) + v = got["values"] + assert v["ports"]["m"] == {"관": 30000, "골라 쓰는 관": 45000} # 조건 찾기 D300 · 선택 = 기본 + total = Decimal("0.7") + 30000 * Decimal("1.1") + 4 # 표에서 찾기 + 10% + 통과(4) + 막힘("") + assert _d(v["ports"]["f"]["합"]) == int(total) # 함수 출력 끝수 0 자리 버림 + assert _d(v["cost"]["재료비"]) == int(total) // 10 * 10 # 출력 끝수 −1 자리 버림 + assert _d(v["cost"]["값들"]["골라 쓴 관"]) == 45000 and v["errors"] == [] + cells = {b["id"]: b for b in got["logic"]["boxes"]}["f"]["func"]["sheet"]["칸"] + assert _d(cells["A1"]["값"]) == Decimal("0.7") and "A4" not in cells # 받는 칸에 값만 + + designs = client.get("/api/m02/df/design-inputs", params={"key": got["key"]}).json()["designs"] + assert [(d["port"], d["kind"]) for d in designs] == [ + ("돌", "변수"), ("높이", "변수"), ("관경", "변수"), ("골라 쓰는 관", "재료 자리")] # fmt: skip + + lg["boxes"][0]["input"]["variables"][2]["example"] = "D999" # 맞는 재료 없음 · 표에 없는 조건 + lg["boxes"][0]["input"]["variables"][0]["example"] = "없는돌" + bad = _post(client, "/api/m02/df/save", {"logic": lg})["values"]["errors"] + assert any("맞는 재료 없음" in e for e in bad) and any("표에 없는 조건" in e for e in bad) + + +def test_표준_표_읽기_쓰기_처음값(client: TestClient, folder: Path) -> None: + first = client.get("/api/m02/df/standards").json() + assert first["saved"] is False and first["unit"]["줄"] and first["money"]["줄"] + edited = json.loads(json.dumps(first)) + edited["money"]["줄"][4]["지위"] = 1 + res = client.put( + "/api/m02/df/standards", json={"unit": edited["unit"], "money": edited["money"]} + ) + assert res.status_code == 200 and res.json()["saved"] is True + assert client.get("/api/m02/df/standards").json()["money"]["줄"][4]["지위"] == 1 + assert client.put("/api/m02/df/standards", json={"unit": {}, "money": {}}).status_code == 400 + back = client.delete("/api/m02/df/standards").json() + assert ( + back["saved"] is False and back["money"]["줄"][4]["지위"] == first["money"]["줄"][4]["지위"] + ) + assert (cf.STANDARD / "금액단위표준.json").is_file() # 처음 값 파일은 그대로 + + +def test_새_길도_시스템_관리자만(folder: Path) -> None: + app = FastAPI() + app.include_router( + cf.router, dependencies=[Depends(verify_session), Depends(require_system_admin)] + ) + app.dependency_overrides[verify_session] = lambda: {"role": "USER", "company_id": 1} + client = TestClient(app) + for method, url in ( + ("get", "/api/m02/df/materials"), + ("get", "/api/m02/df/standards"), + ("delete", "/api/m02/df/standards"), + ("get", "/api/m02/df/design-inputs?key=x"), + ): + assert getattr(client, method)(url).status_code == 403 # fmt: skip + assert client.post("/api/m02/df/logic-value", json={"key": "x"}).status_code == 403