eomsangdonandClaude Opus 5.5
3336902c17
- 흐름으로 저장 = 계산 칸(중간 · 호표 · 결과 · 덧줄)만 지우고 「입력」(이름 · 설명 · 단위 · 고르기 · 범위)은 흐름에서 같은 꼴로 · 조합 · 구조물 · 미리 보기 · 다시 적용은 「입력」 을 그대로 읽음(따로 갈래 없음) - 엔진 갈래: 안 준 입력 = 멈춤 · 빈 값으로 준 입력 · 단가 없는 품목 · 표에 없는 조건 = 그 값만 비움(알림) · 결과 꼴 = 돈 출력 줄 · 값들 · 알림 - 프로젝트에서 고른 자재 = 흐름 재료 자리 이름으로(고름 · 다시 적용 때 살림) - 시험판 길 지움: /list · /get · /save · /delete · /user-elements(서버) · cfList · cfGet · cfSave · cfDelete · cfUserElement*(Api) · 빈 _data 폴더 - 시험: 조합에 담은 로직을 흐름으로 바꿔도 입력 목록 · 로직별 결과 · 비목 합 · 단가 묶음 셈이 같음 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL
418 lines
21 KiB
Python
418 lines
21 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, strict: bool = False) -> dict:
|
|
"""흐름 셈 — 입력 = 「설계 때 정할 것」 이름 → 값 · strict(조합 · 프로젝트) = 안 준 입력은
|
|
예시값
|
|
대신 빈 값 + 알림 「빈 입력」(옛 엔진처럼 그 값을 쓰는 곳만 비움)."""
|
|
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)
|
|
designs = dc.design_inputs(flow) if strict else []
|
|
lack = [d["port"] for d in designs if d["port"] not in given]
|
|
if lack: # 안 준 입력 — 엔진 `_inputs` 와 같이 멈춤(빈 값으로 준 입력은 아래 「빈 입력」)
|
|
raise store.cm.mf.FormulaError(f"「{key}」 입력 「{lack[0]}」 없음")
|
|
blank = [d for d in designs if 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)
|
|
|
|
|
|
def _engine(master, ref: str, row: dict, given: dict) -> dict:
|
|
"""옛 엔진 `mf.run` 갈래(조합 · 프로젝트 단가 · 읽는 식 · 옛 식의 로직(…)) — `run` 과 같은 꼴.
|
|
|
|
단가 · 값 · 표 = 부른 쪽이 넘긴 master(프로젝트 사본이면 그 단가 · `M02_DF_Values.SOURCE`).
|
|
"""
|
|
picks = (
|
|
store.cm.mf.mp.CHOSEN.get() or {}
|
|
) # 고른 자재 — 흐름은 재료 자리 이름으로(호표 차례 아님)
|
|
slots = {d["port"] for d in dc.design_inputs(row[FLOW]) if d.get("kind") == "재료 자리"}
|
|
given = {**given, **{k: v for k, v in picks.items() if k in slots}}
|
|
token = dv.SOURCE.set(master)
|
|
try:
|
|
got = run(row[FLOW], ref, given, strict=True)
|
|
finally:
|
|
dv.SOURCE.reset(token)
|
|
# 빈 값(빈 입력 · 단가 없는 품목 · 표에 없는 조건 → #N/A)을 쓴 칸은 엔진처럼 그 값만 비움 — 알림
|
|
empty = [e for e in got["errors"] if "#N/A" in e or "단가 없는 품목" in e]
|
|
stop = [e for e in got["errors"] if e not in empty]
|
|
if stop:
|
|
raise store.cm.mf.FormulaError(f"「{ref}」 흐름 셈 — {' · '.join(stop)}")
|
|
cost, notes = got["cost"], [*(got.get("notes") or []), *empty]
|
|
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"])
|
|
value = None if value == dc.NA_VALUE else value
|
|
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:
|
|
"""견본 입력 — 흐름 로직이면 「설계 때 정할 것」 예시값 · 아니면 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, draft: dict | None = None) -> dict:
|
|
"""{flow, converted, notes} — 흐름이 있으면 그대로 · 없으면 변환(파일은 안 바뀜)
|
|
· draft = 저장 안 한 고친 줄(옛 편집기) — 늘 그 줄 그대로 변환(쓰기 없음)."""
|
|
row = _row(file, key) if draft is None else store._dec({**draft, "키": draft.get("키") or key})
|
|
if draft is not None:
|
|
row.pop(FLOW, None)
|
|
elif 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:
|
|
"""흐름 로직 줄 — 계산 칸(중간 · 호표 · 결과 · 덧줄)은 지움(흐름이 정본) · 「입력」 은 흐름의
|
|
「설계 때 정할 것」 으로 같은 꼴로 다시 씀(로직 입력 약속 하나 · 설명은 이름이 같으면 살림)
|
|
· 끝수 · 시험입력 · 메타 칸은 둠 · 출처 없으면 「자체」."""
|
|
kept = {k: v for k, v in row.items() if k not in store.cm.mf.CALC_FIELDS}
|
|
said = {s.get("이름"): s.get("설명") for s in row.get("입력") or [] if s.get("설명")}
|
|
inputs = []
|
|
for spec in store.cm.mf.inputs_of(row):
|
|
one = {
|
|
"이름": spec["이름"],
|
|
**({"설명": said[spec["이름"]]} if spec["이름"] in said else {}),
|
|
}
|
|
one |= {k: spec[k] for k in ("단위", "고르기", "범위") if spec.get(k) is not None}
|
|
inputs.append(one)
|
|
return {"끝수": None, **kept, "입력": inputs, "출처": 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
|
|
check(row[FLOW]) # 꼴부터 — 입력 약속을 흐름에서 다시 쓰기 전에
|
|
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)
|
|
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 ""))]
|
|
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 []]
|