eomsangdonandClaude Opus 5.5
3f71305ddb
- main(sub7 a774b4e97 Model 덧) 받아 합침 · Compute 재료 조건 찾기는 sub7 의 틀 채우기 + 빈 단가 알림
- 남은 8: 표에 여럿 맞음 5(옛 find 의 값 칸 있는 줄만 거름 — lookup) · 빈 값이 든 식 3(옛 = 그 줄 비움 · 흐름 = 빈 칸 0)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL
337 lines
16 KiB
Python
337 lines
16 KiB
Python
"""M01 로직 「흐름」 칸(PLAN 36-4 · 정함 1 · 2 · 4 · 10 · 11 · 12).
|
|
|
|
정함 = `docs/raw/M01_쉬운단가/흐름도_적용_정함.md`.
|
|
|
|
· 흐름 = DataFlow Model `CfLogic` 꼴 그대로 · 로직 줄 칸 「흐름」 · 있으면 흐름이 정본
|
|
· 쓰는 때 = `/save` 줄에 실어 온 것뿐(자동 변환 결과는 안 씀) — 꼴 검사(상자 · 선 · 돌고 도는 선) →
|
|
서버 셈(`M02_DF_Compute.run`)으로 다시 셈 → 받는 칸에 값만 되돌려 넣음 · 빈 받는 자리는 경고
|
|
· 새 줄(키 없음) + 흐름 = 자체 파일(`로직_자체.json`)에 더함 → 키 = 대장 `GX` 다음 · 구분 자체
|
|
· 계산 = 흐름이 있으면 흐름 셈 · 없으면 옛 셈(`Store.calc`) · 입력 이름 = 「설계 때 정할 것」
|
|
· 다른 로직 입력 = 마스터 키도 — 흐름 로직은 흐름 셈 · 옛 로직은 옛 셈(노무비 · 재료비 · 경비 · 계)
|
|
· 흐름 받기 = 줄의 흐름 그대로 · 없으면 sub7 변환(`M02_DF_Convert.convert`) · 아직 없으면 빈 흐름
|
|
권한 = `/api/m01` 전체가 시스템 관리자(`main.py`) — 공용 로직 흐름도 관리자가 고침.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
from contextvars import ContextVar
|
|
from decimal import Decimal, InvalidOperation
|
|
from typing import Any
|
|
|
|
from M01_MasterData import M01_MasterData_Store as store
|
|
from M01_MasterData import M01_MasterData_Store_Make as make
|
|
from M02_MasterTemplete.DataFlow import M02_DF_Compute as dc
|
|
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv
|
|
|
|
FLOW = "흐름"
|
|
KINDS = ("input", "function", "output")
|
|
CONVERT = "M02_MasterTemplete.DataFlow.M02_DF_Convert"
|
|
NOT_READY = "변환 준비 중"
|
|
_ACTIVE: ContextVar[tuple] = ContextVar("m01_flow_active", default=()) # 셈 중인 흐름 키
|
|
|
|
|
|
# ── 검사(Model `link` · `ruleWhy` 와 같은 뜻) ──────────────────────────────
|
|
def _outs(box: dict) -> set:
|
|
"""내보내는 알약 — Model `portsOut`."""
|
|
if box.get("kind") == "function":
|
|
return {((box.get("func") or {}).get("output") or {}).get("name")}
|
|
inp = box.get("input") or {}
|
|
if box.get("kind") != "input":
|
|
return set()
|
|
if inp.get("mode") == "logic":
|
|
return {p.get("name") for p in (inp.get("logic") or {}).get("outputs") or []}
|
|
out = {dc._port(e) for e in inp.get("elements") or []} if inp.get("mode") == "group" else set()
|
|
if inp.get("mode") == "group" and inp.get("table"):
|
|
out.add(dc.TABLE_PORT)
|
|
return out | {v.get("name") for v in inp.get("variables") or []}
|
|
|
|
|
|
def _ins(box: dict) -> set:
|
|
"""받는 알약 — Model `portsIn`."""
|
|
if box.get("kind") == "function":
|
|
return {p.get("port") for p in (box.get("func") or {}).get("inputs") or []}
|
|
if box.get("kind") == "output":
|
|
return {dc.OUTPUT_PORT}
|
|
inp = box.get("input") or {}
|
|
return {p.get("name") for p in (inp.get("logic") or {}).get("designs") or []}
|
|
|
|
|
|
def check(flow: Any) -> list[str]:
|
|
"""흐름 꼴 검사 — 틀리면 422 · 돌려받는 것 = 경고(빈 받는 자리)."""
|
|
if not isinstance(flow, dict) or not isinstance(flow.get("boxes"), list):
|
|
raise store.StoreError(422, {"errors": ["흐름 꼴이 아님 — 상자(boxes) 목록이 없음"]})
|
|
if not isinstance(flow.get("links"), list):
|
|
raise store.StoreError(422, {"errors": ["흐름 꼴이 아님 — 선(links) 목록이 없음"]})
|
|
errors, boxes = [], {}
|
|
for b in flow["boxes"]:
|
|
if not isinstance(b, dict) or not b.get("id"):
|
|
errors.append("번호(id) 없는 상자가 있음")
|
|
continue
|
|
name = b.get("name") or b["id"]
|
|
if b["id"] in boxes:
|
|
errors.append(f"상자 번호가 겹침 「{b['id']}」")
|
|
boxes[b["id"]] = b
|
|
if b.get("kind") not in KINDS:
|
|
errors.append(f"「{name}」 모르는 상자 종류 「{b.get('kind')}」")
|
|
elif b["kind"] == "function" and not ((b.get("func") or {}).get("output") or {}).get(
|
|
"name"
|
|
):
|
|
errors.append(f"「{name}」 함수 상자에 결과 알약이 없음")
|
|
taken = set()
|
|
for ln in flow["links"]:
|
|
src, to = (ln or {}).get("from") or {}, (ln or {}).get("to") or {}
|
|
a, b = boxes.get(src.get("box")), boxes.get(to.get("box"))
|
|
if (
|
|
a is None
|
|
or b is None
|
|
or src.get("name") not in _outs(a)
|
|
or to.get("name") not in _ins(b)
|
|
):
|
|
errors.append(f"선 「{(ln or {}).get('id')}」 이 없는 알약을 이음")
|
|
continue
|
|
if (to["box"], to["name"]) in taken:
|
|
errors.append(f"「{b.get('name')} · {to['name']}」 에 선이 둘 이상")
|
|
taken.add((to["box"], to["name"]))
|
|
for p in dc._sources(ln.get("rule")):
|
|
if p.get("name") not in _outs(boxes.get(p.get("box")) or {}):
|
|
errors.append(f"선 「{ln.get('id')}」 규칙의 조건 알약이 없음")
|
|
if not errors:
|
|
errors += [
|
|
f"「{b.get('name')}」 상자가 서로를 기다림(돌고 도는 선)" for b in dc._waves(flow)[1]
|
|
]
|
|
if errors:
|
|
raise store.StoreError(422, {"errors": errors})
|
|
return [
|
|
f"「{b.get('name')} · {port}」 받는 자리에 들어오는 선이 없음"
|
|
for b in boxes.values()
|
|
if b.get("kind") != "input"
|
|
for port in sorted(_ins(b) - {n for bx, n in taken if bx == b["id"]}, key=str)
|
|
]
|
|
|
|
|
|
# ── 다른 로직 · 계산 ───────────────────────────────────────────────────
|
|
def _old(key: str, designs: dict) -> dict:
|
|
"""옛 로직 값 — 옛 셈 · 내보내는 알약 = 노무비 · 재료비 · 경비 · 계(돈 아닌 로직은 결과)."""
|
|
given = {k: v for k, v in designs.items() if v is not None}
|
|
whole, mf = dv.SOURCE.get(), store.cm.mf
|
|
if whole is not None: # 단가 자리(프로젝트 사본)의 옛 로직
|
|
try:
|
|
given = mf.fill_empty(whole.logic(key), {k: _value(v) for k, v in given.items()})
|
|
got = mf.run(whole, key, given)
|
|
except (mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e:
|
|
return {"outputs": {}, "errors": [str(e) or type(e).__name__]}
|
|
if "결과" in got:
|
|
return {"outputs": {"결과": got["결과"]}, "errors": []}
|
|
return {"outputs": {k: got[k] for k in mf.BUNDLE}, "errors": []}
|
|
got = store.calc(key, given)
|
|
if not got.get("ok"):
|
|
return {"outputs": {}, "errors": [got.get("reason") or "옛 셈 오류"]}
|
|
if "result" in got:
|
|
return {"outputs": {"결과": got["result"]}, "errors": []}
|
|
return {"outputs": dict(got["sums"]), "errors": []}
|
|
|
|
|
|
def load(key: str) -> dict:
|
|
"""다른 로직 입력이 부르는 마스터 로직 — 흐름이면 흐름 · 옛 로직이면 옛 셈 함수(`calc`)."""
|
|
whole = dv.SOURCE.get()
|
|
row = whole.logic(key) if whole is not None else store.logic(key)["logic"]
|
|
if isinstance(row.get(FLOW), dict):
|
|
return {**row[FLOW], "key": key}
|
|
return {"key": key, "calc": lambda designs: _old(key, designs)}
|
|
|
|
|
|
def old_designs(key: str) -> list[dict]:
|
|
"""옛 로직의 받는 알약 = 로직 입력 — Model `CfDesign` 꼴."""
|
|
out = []
|
|
for one in store.logic(key)["logic"].get("입력") or []:
|
|
choices = one.get("고르기")
|
|
variable = {"name": one.get("이름"), "type": "choice" if choices else "number",
|
|
"unit": one.get("단위") or "-", "example": None} # fmt: skip
|
|
if choices:
|
|
variable["choices"] = choices
|
|
out.append({"box": "", "port": one.get("이름"), "kind": "변수",
|
|
"unit": one.get("단위") or "-", "variable": variable}) # fmt: skip
|
|
return out
|
|
|
|
|
|
def _value(v: Any, words: set | frozenset = frozenset()) -> Any:
|
|
"""화면이 보낸 값 — 수 글은 수로(「3」 → 3) · 고르기 글 값(「080」 · 「1048」)은 글 그대로."""
|
|
if isinstance(v, str) and v in words:
|
|
return v
|
|
if isinstance(v, str) and v.strip():
|
|
try:
|
|
return Decimal(v.strip())
|
|
except InvalidOperation:
|
|
return v
|
|
return store._dec(v)
|
|
|
|
|
|
def run(flow: dict, key: str, inputs: dict) -> dict:
|
|
"""흐름 셈 — 입력 = 「설계 때 정할 것」 이름 → 값."""
|
|
active = _ACTIVE.get()
|
|
if key and key in active: # 옛 셈을 건너 다시 돌아온 흐름(흐름 → 옛 로직 → 흐름)
|
|
raise store.cm.mf.FormulaError(f"로직이 서로를 부름 — {' → '.join([*active, key])}")
|
|
token = _ACTIVE.set((*active, key))
|
|
try:
|
|
words = { # 고르기에 글로 든 값 — 옛 셈처럼 글 그대로(수로 바꾸면 고르기 밖)
|
|
c
|
|
for d in dc.design_inputs(flow)
|
|
for c in (d.get("variable") or {}).get("choices") or []
|
|
if isinstance(c, str)
|
|
}
|
|
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])
|
|
finally:
|
|
_ACTIVE.reset(token)
|
|
|
|
|
|
def _engine(master, ref: str, row: dict, given: dict) -> dict:
|
|
"""옛 엔진 `mf.run` 갈래(조합 · 프로젝트 단가 · 읽는 식 · 옛 식의 로직(…)) — `run` 과 같은 꼴.
|
|
|
|
단가 · 값 · 표 = 부른 쪽이 넘긴 master(프로젝트 사본이면 그 단가 · `M02_DF_Values.SOURCE`).
|
|
"""
|
|
token = dv.SOURCE.set(master)
|
|
try:
|
|
got = run(row[FLOW], ref, given)
|
|
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["계"], "중간": {}}
|
|
|
|
|
|
store.cm.mf.FLOW_RUN = _engine # 옛 엔진을 부르는 곳이 모두 흐름 로직을 흐름 셈으로(한 곳 갈래)
|
|
|
|
|
|
def auto(key: str) -> dict | None:
|
|
"""견본 입력 — 흐름 로직이면 「설계 때 정할 것」 예시값 · 아니면 None(옛 자동값)."""
|
|
flow = store.logic(key)["logic"].get(FLOW)
|
|
if not isinstance(flow, dict):
|
|
return None
|
|
return {
|
|
"값": {d["port"]: (d.get("variable") or {}).get("example") for d in dc.design_inputs(flow)}
|
|
}
|
|
|
|
|
|
def calc(key: str, inputs: dict, row: dict | None = None) -> dict | None:
|
|
"""계산 갈래 — 줄에 흐름이 없으면 None(옛 셈 그대로) · 있으면 흐름 셈을 `/calc` 꼴로."""
|
|
if row is not None:
|
|
flow, key = row.get(FLOW), str(row.get("키") or key)
|
|
else:
|
|
try:
|
|
flow = store.logic(key)["logic"].get(FLOW)
|
|
except store.StoreError:
|
|
return None
|
|
if not isinstance(flow, dict):
|
|
return None
|
|
try:
|
|
got = run(flow, key, inputs)
|
|
except store.cm.mf.FormulaError as e:
|
|
return {"ok": False, "flow": True, "reason": str(e)}
|
|
names = {b["id"]: b for b in flow.get("boxes") or []}
|
|
outputs = [
|
|
{"box": i, "name": names[i].get("name"), "cost": (names[i].get("output") or {}).get("cost"),
|
|
"unit": (names[i].get("output") or {}).get("unit") or names[i].get("unit"), "value": v}
|
|
for i, v in got["outputs"].items() if i in names
|
|
] # fmt: skip
|
|
cost = got["cost"]
|
|
out = {"ok": not got["errors"], "flow": True, "lines": [], "middle": {},
|
|
"sums": {k: cost[k] for k in (*dc.COSTS, "계")}, "values": cost["값들"],
|
|
"outputs": outputs, "errors": got["errors"], "breaks": got["breaks"]} # fmt: skip
|
|
if got["errors"]:
|
|
out["reason"] = " · ".join(got["errors"])
|
|
return out
|
|
|
|
|
|
# ── 흐름 받기 ──────────────────────────────────────────────────────────
|
|
def _row(file: str, key: str) -> dict:
|
|
if not file:
|
|
return store.logic(key)["logic"]
|
|
data = store.read(file)[0]
|
|
hit = next((r for r in data.get("줄") or [] if str(r.get("키")) == key), None)
|
|
if hit is None:
|
|
raise store.StoreError(404, f"{file} · 없는 키 「{key}」")
|
|
return hit
|
|
|
|
|
|
def get_flow(file: str, key: str) -> dict:
|
|
"""{flow, converted, notes} — 흐름이 있으면 그대로 · 없으면 변환(파일은 안 바뀜)."""
|
|
row = _row(file, key)
|
|
if isinstance(row.get(FLOW), dict):
|
|
return {"flow": {**row[FLOW], "key": key}, "converted": False, "notes": []}
|
|
try:
|
|
convert = importlib.import_module(CONVERT).convert
|
|
except ModuleNotFoundError as e:
|
|
if e.name != CONVERT:
|
|
raise
|
|
empty = {"key": key, "name": row.get("이름") or "", "resultUnit": row.get("결과단위") or "",
|
|
"boxes": [], "links": [], "userElements": []} # fmt: skip
|
|
return {"flow": empty, "converted": True, "notes": [NOT_READY]}
|
|
got = convert(row, {"key": key, "file": file, "master": store.loaded()[1]})
|
|
return {"flow": {**got["flow"], "key": key}, "converted": True, "notes": got.get("notes") or []}
|
|
|
|
|
|
# ── 저장 ──────────────────────────────────────────────────────────────
|
|
def _redo(row: dict, key: str) -> list[str]:
|
|
"""흐름 검사 → 서버 셈으로 다시 셈 → 받는 칸 값만 되돌려 넣음 · 돌려받는 것 = 경고."""
|
|
warns = check(row[FLOW])
|
|
got = run(row[FLOW], key, {})
|
|
filled = got["filled"]
|
|
filled.pop("key", None) # 키는 줄의 「키」 하나 — 흐름 안에 두지 않음
|
|
row[FLOW] = filled
|
|
return warns + got["errors"]
|
|
|
|
|
|
def _bare(row: dict) -> dict:
|
|
"""새 흐름 로직 — 옛 식 칸은 빈 것으로(파일 틀 검사) · 출처 없으면 「자체」."""
|
|
old = {"입력": [], "중간": [], "호표": [], "끝수": None}
|
|
return {**old, **row, "출처": row.get("출처") or "자체"}
|
|
|
|
|
|
def save(batch: list[dict]) -> dict:
|
|
"""`/save` — 흐름 줄은 검사 · 다시 셈 · 새 줄은 자체 GX · 그 밖은 `Store.save` 그대로."""
|
|
warns, fresh = [], False
|
|
for part in batch:
|
|
for ch in part.get("changes") or []:
|
|
row = ch.get("row")
|
|
if not isinstance(row, dict) or FLOW not in row:
|
|
continue
|
|
new = ch.get("op") == "add" or (ch.get("op") == "edit" and not ch.get("key"))
|
|
if new:
|
|
if part.get("file") != make.OWN:
|
|
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)
|
|
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 ""))]
|
|
if part.get("file") == make.OWN and not make._own_path().is_file():
|
|
make._make_file()
|
|
fresh = True
|
|
part["version"] = part.get("version") or store.read(make.OWN)[1]
|
|
before = {p["file"]: _keys(p["file"]) for p in batch}
|
|
try:
|
|
files = store.save(batch)
|
|
except store.StoreError:
|
|
if fresh:
|
|
make._own_path().unlink(missing_ok=True)
|
|
raise
|
|
added = [{"file": f, "key": k} for f, old in before.items() for k in _keys(f) if k not in old]
|
|
return {"files": files, "added": added, "warnings": warns}
|
|
|
|
|
|
def _keys(file: str) -> list[str]:
|
|
try:
|
|
data = store.read(file)[0]
|
|
except store.StoreError:
|
|
return []
|
|
return [str(r.get("키")) for r in data.get(store.items_key(data)) or []]
|