eomsangdonandClaude Opus 5.5
f609dd2767
- /cf/values · 저장 다시 넣기가 기계 ref 「EQ키#칸」 과 {ref: 키, field: 칸} 둘 다 받음
- 칸 「시간당 경비」 = 기종(분류번호)을 받는 기계경비 로직 가운데 짝 차례 첫째(옛 EasyCalc pickMachine 규칙을 옮김) · 빈 입력은 첫 고르기(지역 = 전국평균) · inputs 로 바꿈
- 시험 — 키#칸 = {ref, field} = 기계 표 값 · 시간당 경비 = 그 로직 /calc 계 · 저장 때 같은 값
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL
665 lines
28 KiB
Python
665 lines
28 KiB
Python
"""단가산출 다이어그램 시험 길(PLAN 35-7) — 목록 · 열기 · 저장 · 지우기 · 입력 값 · 사용자 요소.
|
|
|
|
GET /api/m02/cf/list → {logics: [{key, name, resultUnit, saved, sample}]}
|
|
GET /api/m02/cf/get?key=CF-0001 → {logic, sample}
|
|
POST /api/m02/cf/save {logic} → {key, logic, values: {ports, outputs, cost, errors}}
|
|
POST /api/m02/cf/delete {key}
|
|
POST /api/m02/cf/values {items, userElements?} → {values: CfAnswer[]}
|
|
GET /api/m02/cf/user-elements → {elements}
|
|
POST /api/m02/cf/user-elements {op: add | delete, element? | id?}
|
|
|
|
모양 = sub7 Model `M02_CF_Model.ts`
|
|
(CfLogic = key · name · resultUnit · boxes · links · userElements).
|
|
알약 이름(브레인 정함): 입력 상자가 내보내는 알약 = 요소 name · 표 「표」 · 변수 name /
|
|
함수 상자 받는 알약 = inputs[].port · 내보내는 알약 = output.name / 출력 상자 받는 알약 = 「값」.
|
|
저장 = 입력 값(노임 · 재료 · 유가전력 · 기계 · 환율 · 요율 · 소요량 / 계수 표 · 사용자 요소)을
|
|
마스터(사용자 요소는 다이어그램 안 · `_data`)에서 다시 받아 → 함수 상자 시트들을 한 통합문서로
|
|
묶어(받는 칸에 값 · 함수 → 함수는 시트 사이 참조) 우리 스프레드시트 엔진(Node ·
|
|
`common_util_spreadsheet`)으로 셈 → 받는 칸에 값만 되돌려 넣어 `_data` 에 씀 · 브라우저 값은
|
|
안 받아 적음. 단가는 기존 `/calc` 와 같은 셈.
|
|
자리 = 이 폴더 `_data/`(git 밖) · 견본 `_sample/`(git 안 · 읽기만 · 저장하면 새 번호) ·
|
|
키 `CF-0001` · 사용자 요소 `UE-0001` · 마스터 파일은 안 씀. 권한 = 시스템 관리자(`main.py` 한 줄).
|
|
옛 시험판(SheetLogic) 서버를 옮겨 고쳐 씀 — 그 폴더를 import 하지 않음.
|
|
정함 = `docs/raw/M01_쉬운단가/원가다이어그램_정함.md`.
|
|
"""
|
|
|
|
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
|
|
|
|
cm = store.cm
|
|
router = APIRouter(prefix="/api/m02/cf", tags=["M02 CostFlow (시험)"])
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
DATA = HERE / "_data" # 시험은 임시 폴더로 바꿈
|
|
SAMPLE = HERE / "_sample"
|
|
USERS = "_user_elements.json" # 사용자 요소 공용 목록 — `_data` 안
|
|
_KEY = re.compile(r"^CF-(\d{4,})$")
|
|
_UKEY = re.compile(r"^UE-(\d{4,})$")
|
|
_SAMPLE_KEY = re.compile(r"^[\w가-힣·() -]+$")
|
|
_CELL = re.compile(r"^([A-Z]+)(\d+)$")
|
|
RANGE_LOW, RANGE_HIGH = "·아래", "·위"
|
|
FIELDS = (*cm.mf.COST_ITEMS, "계")
|
|
GROUPS = ("인력", "재료", "유가전력", "기계", "소요량", "계수", "환율", "요율")
|
|
TABLE_PORT, OUTPUT_PORT = "표", "값"
|
|
COSTS = ("노무비", "재료비", "경비")
|
|
|
|
|
|
class LogicBody(BaseModel):
|
|
logic: dict[str, Any]
|
|
|
|
|
|
class KeyBody(BaseModel):
|
|
key: str
|
|
|
|
|
|
class ValuesBody(BaseModel):
|
|
items: list[dict[str, Any]]
|
|
userElements: list[dict[str, Any]] = []
|
|
|
|
|
|
class UserBody(BaseModel):
|
|
op: str
|
|
element: dict[str, Any] | None = None
|
|
id: str | None = None
|
|
|
|
|
|
# ── 자리 ──────────────────────────────────────────────────────────────
|
|
def _write(path: Path, data: Any) -> None:
|
|
DATA.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_name(f".{path.name}.tmp")
|
|
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8", newline="\n")
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def _mine(key: str) -> Path:
|
|
if not _KEY.match(key or ""):
|
|
raise store.StoreError(400, f"다이어그램 번호 모양이 아님 「{key}」")
|
|
return DATA / f"{key}.json"
|
|
|
|
|
|
def _sample(key: str) -> Path | None:
|
|
if not key or _KEY.match(key) or not _SAMPLE_KEY.match(key):
|
|
return None
|
|
path = SAMPLE / f"{key}.json"
|
|
return path if path.is_file() else None
|
|
|
|
|
|
def _read(key: str) -> tuple[dict, bool]:
|
|
"""(다이어그램, 견본인가) — 내 것은 `_data` · 견본은 `_sample`."""
|
|
path = _sample(key)
|
|
if path is not None:
|
|
return json.loads(path.read_text(encoding="utf-8")), True
|
|
path = _mine(key)
|
|
if not path.is_file():
|
|
raise store.StoreError(404, f"없는 다이어그램 「{key}」")
|
|
return json.loads(path.read_text(encoding="utf-8")), False
|
|
|
|
|
|
def listing() -> dict:
|
|
out = []
|
|
for path, sample in [
|
|
*((p, False) for p in sorted(DATA.glob("CF-*.json"))),
|
|
*((p, True) for p in sorted(SAMPLE.glob("*.json"))),
|
|
]:
|
|
logic = json.loads(path.read_text(encoding="utf-8"))
|
|
out.append(
|
|
{
|
|
"key": path.stem,
|
|
"name": logic.get("name") or path.stem,
|
|
"resultUnit": logic.get("resultUnit") or "",
|
|
"saved": logic.get("saved") or "",
|
|
"sample": sample,
|
|
}
|
|
)
|
|
return {"logics": out}
|
|
|
|
|
|
def get(key: str) -> dict:
|
|
logic, sample = _read(key)
|
|
return {"logic": {**logic, "key": key}, "sample": sample}
|
|
|
|
|
|
def delete(key: str) -> dict:
|
|
if _sample(key) is not None:
|
|
raise store.StoreError(403, "견본은 지울 수 없음")
|
|
_read(key)
|
|
_mine(key).unlink()
|
|
return {"key": key}
|
|
|
|
|
|
# ── 사용자 요소(Model CfUserElement — id · group · name · spec · unit · value) ─────
|
|
def _users() -> list[dict]:
|
|
path = DATA / USERS
|
|
return json.loads(path.read_text(encoding="utf-8")) if path.is_file() else []
|
|
|
|
|
|
def user_elements() -> dict:
|
|
return {"elements": _users()}
|
|
|
|
|
|
def user_edit(op: str, element: dict | None, uid: str | None) -> dict:
|
|
rows = _users()
|
|
if op == "delete":
|
|
if not any(r["id"] == uid for r in rows):
|
|
raise store.StoreError(404, f"없는 사용자 요소 「{uid}」")
|
|
_write(DATA / USERS, [r for r in rows if r["id"] != uid])
|
|
return {"id": uid}
|
|
if op != "add" or not isinstance(element, dict):
|
|
raise store.StoreError(400, "op 는 add(element) 또는 delete(id)")
|
|
name = str(element.get("name") or "").strip()
|
|
if not name:
|
|
raise store.StoreError(400, "이름을 넣어 주세요")
|
|
group = element.get("group") or "재료"
|
|
if group not in GROUPS:
|
|
raise store.StoreError(400, f"모르는 그룹 「{group}」")
|
|
value = element.get("value")
|
|
if value is not None and (isinstance(value, bool) or not isinstance(value, (int, float))):
|
|
raise store.StoreError(400, "값은 수 또는 빈 값")
|
|
used = [int(m.group(1)) for r in rows if (m := _UKEY.match(r["id"]))]
|
|
one = {
|
|
"id": f"UE-{max(used, default=0) + 1:04d}",
|
|
"group": group,
|
|
"name": name,
|
|
"spec": str(element.get("spec") or ""),
|
|
"unit": str(element.get("unit") or ""),
|
|
"value": value,
|
|
}
|
|
_write(DATA / USERS, [*rows, one])
|
|
return {"element": one}
|
|
|
|
|
|
# ── 입력 값 ────────────────────────────────────────────────────────────
|
|
def _row(line: dict, names) -> dict:
|
|
"""단가를 `/calc` 와 같은 셈으로 — 호표 한 줄(수량 1)짜리 저장 안 하는 견본 로직."""
|
|
return {
|
|
"키": "",
|
|
"원문번호": "",
|
|
"이름": "입력 값",
|
|
"결과단위": "원/식",
|
|
"출처": "자체",
|
|
"입력": [{"이름": n} for n in names],
|
|
"중간": [],
|
|
"호표": [line],
|
|
"덧줄": [],
|
|
"끝수": None,
|
|
}
|
|
|
|
|
|
def _money(got: dict, field: str | None) -> dict:
|
|
if not got["ok"]:
|
|
return got
|
|
if "result" in got:
|
|
return {"ok": True, "value": got["result"]}
|
|
pick = field or "계"
|
|
if pick not in FIELDS:
|
|
return {"ok": False, "reason": f"모르는 칸 「{pick}」"}
|
|
return {"ok": True, "value": got["sums"][pick]}
|
|
|
|
|
|
def _pick(key: Any) -> dict | None:
|
|
"""재료 검색어 묶음 `{구분, 상세구분, 검색어, 규격?}` — 묶음 · 그 JSON 글 · 아니면 None."""
|
|
if isinstance(key, dict):
|
|
return key
|
|
if isinstance(key, str) and key.lstrip().startswith("{"):
|
|
try:
|
|
got = json.loads(key)
|
|
except ValueError:
|
|
return None
|
|
return got if isinstance(got, dict) else None
|
|
return None
|
|
|
|
|
|
def _material(cond: dict, unit: str) -> dict:
|
|
"""검색어 묶음 재료 단가 — 기존 로직 재료 줄과 같은 길(검색어 · 자동값 차례 · 단위 환산)."""
|
|
line = {"종류": "재료", "요소": cond, "이름": cond.get("검색어") or "재료", "수량": "1"}
|
|
got = store.calc("", {}, _row({**line, "단위": unit, "비목": "재료비"}, ()), None)
|
|
if not got["ok"]:
|
|
return got
|
|
one = got["lines"][0]
|
|
if one["단가"] is None:
|
|
word = cond.get("검색어") or ""
|
|
empty = "값 없음" in str(one.get("까닭") or "")
|
|
why = (
|
|
"단가가 자료에 비어 있음 — 단가를 채우거나 다른 품목을 고름"
|
|
if empty
|
|
else "맞는 재료 없음"
|
|
)
|
|
return {"ok": False, "reason": f"「{word}」 {why}"}
|
|
picked = one.get("품목") or {}
|
|
return {"ok": True, "value": one["단가"], "name": picked.get("이름"), "unit": unit or None}
|
|
|
|
|
|
def _brief(key: str, want: str) -> tuple[dict | None, str | None]:
|
|
"""마스터 줄 하나 + 그룹 확인 — (줄, 까닭) · 유가전력은 재료 그룹 안 MO 키."""
|
|
try:
|
|
row = store.loaded()[1].get(key)
|
|
except cm.mf.FormulaError as e:
|
|
return None, str(e)
|
|
group = store.mk.GROUP_OF.get(store.mk.id_of(key))
|
|
if not (group == want or (want == "유가전력" and key.startswith("MO"))):
|
|
return None, f"「{key}」 는 {want} 아님"
|
|
return row, None
|
|
|
|
|
|
def _price(group: str, key: str, unit: str = "") -> dict:
|
|
"""인력 노임 · 재료 단가 — 준용 · 낮은 값 · 출처는 엔진 그대로."""
|
|
row, why = _brief(key, group)
|
|
if row is None:
|
|
return {"ok": False, "reason": why}
|
|
kind, cost = {"인력": ("인력", "노무비"), "재료": ("재료", "재료비")}[group]
|
|
line = {"종류": kind, "요소": key, "이름": row.get("이름") or key, "수량": "1"}
|
|
line = {**line, "단위": unit or row.get("단위") or "", "비목": cost}
|
|
got = store.calc("", {}, _row(line, ()), None)
|
|
if not got["ok"]:
|
|
return got
|
|
one = got["lines"][0]
|
|
out = {"ok": one["단가"] is not None, "value": one["단가"], "name": row.get("이름")}
|
|
return {**out, "unit": row.get("단위"), **({"reason": one["까닭"]} if one.get("까닭") else {})}
|
|
|
|
|
|
def _plain_value(group: str, key: str, field: str | None) -> dict:
|
|
"""유가전력 · 환율 · 요율 · 기계 값 칸 — 그 줄 값(묶음이면 field · 요율 기본 rate_percent)."""
|
|
row, why = _brief(key, group)
|
|
if row is None:
|
|
return {"ok": False, "reason": why}
|
|
value, unit = row.get("값"), row.get("단위")
|
|
if isinstance(value, dict):
|
|
pick = field or ("rate_percent" if group == "요율" else "")
|
|
if pick not in value:
|
|
return {
|
|
"ok": False,
|
|
"reason": f"「{row.get('이름')}」 에서 쓸 칸을 고름 — {', '.join(value)}",
|
|
}
|
|
value = value[pick]
|
|
unit = (
|
|
unit.get(pick) if isinstance(unit, dict) else ("%" if pick == "rate_percent" else unit)
|
|
)
|
|
if value is None:
|
|
return {"ok": False, "reason": f"「{row.get('이름')}」 값이 자료에 비어 있음"}
|
|
return {"ok": True, "value": value, "name": row.get("이름"), "unit": unit}
|
|
|
|
|
|
def _table(key: str, at: Any = None) -> dict:
|
|
"""소요량 · 계수 표 → 머리 줄 + 줄(조건 칸 → 값 칸 · 범위가 있는 칸은 「·아래」 「·위」 두 칸 ·
|
|
그 칸에 수 하나면 아래 · 위 같은 수) · at = 줄 차례(0부터)면 머리 줄 + 그 줄 하나."""
|
|
files = store.loaded()[0]
|
|
table = next(
|
|
(
|
|
t
|
|
for d in files.values()
|
|
if d.get("그룹") in cm.mf.TABLE_GROUPS
|
|
for t in d.get("표") or []
|
|
if str(t.get("키")) == key
|
|
),
|
|
None,
|
|
)
|
|
if table is None:
|
|
return {"ok": False, "reason": f"없는 표 「{key}」"}
|
|
conds, lines = table.get("조건") or {}, table.get("줄") or []
|
|
names = [*conds, *(table.get("값칸") or {})]
|
|
split = {
|
|
n for n in names if conds.get(n) == "범위" or any(isinstance(r.get(n), list) for r in lines)
|
|
}
|
|
head = [h for n in names for h in ((n + RANGE_LOW, n + RANGE_HIGH) if n in split else (n,))]
|
|
if at is not None:
|
|
if not isinstance(at, int) or not 0 <= at < len(lines):
|
|
return {"ok": False, "reason": f"「{table.get('이름')}」 에 {at} 번째 줄 없음"}
|
|
lines = [lines[at]]
|
|
rows = []
|
|
for r in lines:
|
|
line = []
|
|
for n in names:
|
|
v = r.get(n)
|
|
if n in split:
|
|
line += list(v) if isinstance(v, list) else [v, v]
|
|
else:
|
|
line.append(None if v == "" else v if not isinstance(v, dict) else str(v))
|
|
rows.append(line)
|
|
return {"ok": True, "rows": [head, *rows], "name": table.get("이름"), "unit": table.get("기준")}
|
|
|
|
|
|
def _user(uid: str, local: list[dict]) -> dict:
|
|
one = next((r for r in [*local, *_users()] if r.get("id") == uid), None)
|
|
if one is None:
|
|
return {"ok": False, "reason": f"없는 사용자 요소 「{uid}」"}
|
|
if one.get("value") is None:
|
|
return {"ok": False, "reason": f"「{one.get('name')}」 값이 비어 있음"}
|
|
return {"ok": True, "value": one["value"], "name": one.get("name"), "unit": one.get("unit")}
|
|
|
|
|
|
MACHINE_SEP = "#" # 입력 판(sub3) 기계 요소 ref 「EQ키#칸」
|
|
HOURLY = "시간당 경비" # 기계 칸 이름 — 기계 시간당 경비 로직으로 셈(기종 → 로직 짝)
|
|
_MACHINE_PLAIN = {"기계", "지역", "보정작업", "조종원직종"} # 이 밖의 입력 = 부착 장비 전용
|
|
|
|
|
|
def _machine_family() -> list[dict]:
|
|
"""기계 시간당 경비 로직 — 「건설기계」 장 파일 안 · 입력 「기계」 고르기가 있는 돈 로직."""
|
|
files = store.loaded()[0]
|
|
return [
|
|
row
|
|
for name, data in files.items()
|
|
if data.get("그룹") == "로직" and "건설기계" in name
|
|
for row in data.get("줄") or []
|
|
if str(row.get("결과단위") or "").startswith("원")
|
|
and any(i.get("이름") == "기계" and i.get("고르기") for i in row.get("입력") or [])
|
|
]
|
|
|
|
|
|
def _machine_rank(row: dict) -> tuple:
|
|
"""짝 차례(옛 EasyCalc `pickMachine` 와 같음) — 부착 전용 · 손료만 · 보정 · 기종 수 · 키."""
|
|
names = [i.get("이름") for i in row.get("입력") or []]
|
|
codes = next(i.get("고르기") for i in row["입력"] if i.get("이름") == "기계")
|
|
attach = any(n not in _MACHINE_PLAIN for n in names)
|
|
loss = "지역" not in names
|
|
return (attach, loss, loss and "보정작업" in names, len(codes), str(row.get("키")))
|
|
|
|
|
|
def _machine_hourly(key: str, inputs: dict) -> dict:
|
|
"""기계 한 시간 값 — 기종을 받는 기계경비 로직 가운데 짝 차례 첫째 · 빈 입력은 첫 고르기."""
|
|
row, why = _brief(key, "기계")
|
|
if row is None:
|
|
return {"ok": False, "reason": why}
|
|
code = str(row.get("원문번호") or "")
|
|
fit = [r for r in _machine_family() if code in [str(c) for c in _codes(r)]]
|
|
if not fit:
|
|
return {"ok": False, "reason": f"「{row.get('이름')}」 은 시간당 경비를 셀 자료가 없음"}
|
|
logic = min(fit, key=_machine_rank)
|
|
given = {}
|
|
for spec in logic.get("입력") or []:
|
|
name = spec.get("이름")
|
|
if name != "기계":
|
|
given[name] = inputs.get(name, (spec.get("고르기") or [None])[0])
|
|
args = "".join(f", {n}='{v}'" for n, v in given.items())
|
|
ref = f"로직({logic['키']}, 기계='{code}'{args})"
|
|
line = {"종류": "로직", "요소": ref, "이름": "기계", "단위": "시간", "수량": "1"}
|
|
got = _money(store.calc("", {}, _row(line, ()), None), "계")
|
|
return {**got, "name": row.get("이름"), "unit": "원/시간"} if got["ok"] else got
|
|
|
|
|
|
def _codes(row: dict) -> list:
|
|
return next(
|
|
(i.get("고르기") or [] for i in row.get("입력") or [] if i.get("이름") == "기계"), []
|
|
)
|
|
|
|
|
|
def one_value(item: dict, local: list[dict] | None = None) -> dict:
|
|
"""물음 한 줄 `CfQuery {id, group, ref? · table? · unit? · field? · inputs?}` → `CfAnswer`.
|
|
|
|
기계 ref 「EQ키#칸」(입력 판 요소 ref)도 받음 · 칸 「시간당 경비」 = 기계경비 로직으로 셈.
|
|
"""
|
|
group, ref = item.get("group"), item.get("ref")
|
|
table, field = item.get("table") or {}, item.get("field")
|
|
if group == "기계" and isinstance(ref, str) and MACHINE_SEP in ref:
|
|
ref, _, field = ref.partition(MACHINE_SEP)
|
|
try:
|
|
if isinstance(ref, str) and _UKEY.match(ref):
|
|
got = _user(ref, local or [])
|
|
elif group in ("소요량", "계수"):
|
|
got = _table(str(table.get("key") or ref or ""), table.get("row"))
|
|
elif group == "재료" and _pick(ref) is not None:
|
|
got = _material(_pick(ref), str(item.get("unit") or ""))
|
|
elif group in ("인력", "재료"):
|
|
got = _price(group, str(ref or ""), str(item.get("unit") or ""))
|
|
elif group == "기계" and str(ref or "").startswith("로직("):
|
|
inputs = item.get("inputs") or {}
|
|
line = {"종류": "로직", "요소": ref, "이름": "기계", "단위": "시간", "수량": "1"}
|
|
got = _money(store.calc("", inputs, _row(line, inputs), None), field)
|
|
elif group == "기계" and field == HOURLY:
|
|
got = _machine_hourly(str(ref or ""), item.get("inputs") or {})
|
|
elif group in ("기계", "유가전력", "환율", "요율"):
|
|
got = _plain_value(group, str(ref or ""), field)
|
|
else:
|
|
got = {"ok": False, "reason": f"모르는 그룹 「{group}」"}
|
|
except store.StoreError as e:
|
|
got = {"ok": False, "reason": str(e.detail)}
|
|
return {"id": item.get("id") or "", **got}
|
|
|
|
|
|
def values(items: list[dict], local: list[dict]) -> dict:
|
|
# ponytail: 하나씩 셈(단가 하나 0.3초쯤) — 입력 값이 수십이면 마스터 사본 한 벌로 묶어 셈
|
|
return {"values": [one_value(i, local) for i in items]}
|
|
|
|
|
|
# ── 저장 · 계산 ────────────────────────────────────────────────────────
|
|
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 _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 = 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 save(logic: dict) -> dict:
|
|
"""저장 — 입력 값 다시 받기 → 함수 상자 시트를 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):
|
|
raise store.StoreError(400, "상자(boxes) · 선(links) 이 없음")
|
|
key = logic.get("key") or None
|
|
if key and _sample(key) is not None:
|
|
key = None # 견본은 그대로 두고 새 번호로
|
|
elif key:
|
|
_read(key) # 있는 것만 고침 — 없으면 404
|
|
done, result = compute(logic, logic.get("userElements") or [])
|
|
if not key:
|
|
used = [int(m.group(1)) for p in DATA.glob("CF-*.json") if (m := _KEY.match(p.stem))]
|
|
key = f"CF-{max(used, default=0) + 1:04d}"
|
|
done = {**done, "key": key, "saved": datetime.now().isoformat(timespec="seconds")}
|
|
_write(DATA / f"{key}.json", _num(done))
|
|
return {"key": key, "logic": done, "values": result}
|
|
|
|
|
|
# ── 길 ────────────────────────────────────────────────────────────────
|
|
def _call(fn, *args):
|
|
try:
|
|
return _num(fn(*args))
|
|
except store.StoreError as e:
|
|
raise HTTPException(status_code=e.status, detail=e.detail) from e
|
|
|
|
|
|
@router.get("/list")
|
|
def get_list() -> dict:
|
|
return _call(listing)
|
|
|
|
|
|
@router.get("/get")
|
|
def get_one(key: str) -> dict:
|
|
return _call(get, key)
|
|
|
|
|
|
@router.post("/save")
|
|
def post_save(body: LogicBody) -> dict:
|
|
return _call(save, body.logic)
|
|
|
|
|
|
@router.post("/delete")
|
|
def post_delete(body: KeyBody) -> dict:
|
|
return _call(delete, body.key)
|
|
|
|
|
|
@router.post("/values")
|
|
def post_values(body: ValuesBody) -> dict:
|
|
return _call(values, body.items, body.userElements)
|
|
|
|
|
|
@router.get("/user-elements")
|
|
def get_user_elements() -> dict:
|
|
return _call(user_elements)
|
|
|
|
|
|
@router.post("/user-elements")
|
|
def post_user_elements(body: UserBody) -> dict:
|
|
return _call(user_edit, body.op, body.element, body.id)
|