eomsangdon/Aislo
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branches 'origin/sub_laptop_3' and 'origin/sub_laptop_4' into main_laptop_1

Browse Source
This commit is contained in:
eomsangdon
2026-10-01 19:59:53 +09:00
parent ba650fe63f 8c200dfc77 868dc8e37f
commit 3e5dd641f1
8 changed files with 1395 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files
M02_MasterTemplete/DataFlow/M02_DF_Api.ts
+70 -9
View File
@@ -1,13 +1,13 @@
/* =============================================================================
* M02_DF_Api.ts
* 단가산출 다이어그램 Api — 목록(내 것 + 견본) · 열기 · 저장(서버가 입력 값을 마스터에서 다시 받아
* 함수 상자 시트를 같은 엔진으로 셈하고 받는 칸에 값만 되돌려 넣음) · 지우기 · 입력 값 한 번에 받기 ·
* 사용자 요소. 모양 = sub7 Model(CfLogic · CfGroup · CfUserElement) · 아래는 Api 몫만.
* 서버 = `M02_DF_Router.py`(`/api/m02/df/*`) · 정함 `docs/raw/M01_쉬운단가/원가다이어그램_정함.md`
* 단가산출 흐름도 Api — 목록(내 것 + 견본) · 열기 · 저장(서버가 입력 값을 마스터에서 다시 받아 화면과
* 같은 차례 · 같은 규칙으로 셈하고 받는 칸에 값만 되돌려 넣음) · 지우기 · 입력 값 한 번에 받기 · 사용자 요소 ·
* 자재 찾기 · 다른 로직 값 · 설계 때 정할 것 · 단위 · 금액 표준 표. 모양 = sub7 Model · 아래는 Api 몫만.
* 서버 = `M02_DF_Router.py`(`/api/m02/df/*`) · 정함 `docs/raw/M01_쉬운단가/흐름도_정함.md`
* ========================================================================== */
import { API_BASE_URL } from "@config/config_frontend";
import type { CfGroup, CfLogic, CfUserElement } from "./M02_DF_Model";
import type { CfDesign, CfElementRef, CfGroup, CfLogic, CfUserElement } from "./M02_DF_Model";
/** 돈 로직 결과에서 꺼낼 칸(기계 시간당 경비) */
export type CfField = "노무비" | "재료비" | "경비" | "계";
@@ -17,7 +17,9 @@ export type CfField = "노무비" | "재료비" | "경비" | "계";
* 기존 로직 재료 줄과 같은 단가)
* · 유가전력 · 환율 = 그 줄 값 · 요율 = 값 묶음의 field(기본 rate_percent · %)
* · 기계 ref = EQ 키 + field(값 칸 — 가격 · 손료계수 …) 또는 `로직(GC…, 기계='…', 지역=지역)` + inputs → 한 시간 값
* · 소요량 · 계수 table = {key, row?} → 표 블록(머리 줄 포함 · 조건 칸 → 값 칸 · 범위는 「·아래」 「·위」) */
* · 소요량 · 계수 table = {key, row?} → 표 블록(머리 줄 포함 · 조건 칸 → 값 칸 · 범위는 「·아래」 「·위」)
* · 재료 자리 slot = 요소 그대로(mode fixed · choice · rule · scope · default) + field · op(조건 찾기 규칙) ·
* value = 이어진 알약 값(조건 찾기) · pick = 고른 품목 키(선택) → 단가 · 여럿이면 candidates */
export interface CfQuery {
id: string;
group: CfGroup;
@@ -26,6 +28,9 @@ export interface CfQuery {
unit?: string;
field?: string;
inputs?: Record<string, unknown>;
slot?: Partial<CfElementRef> & { field?: "규격" | "이름"; op?: "포함" | "같음" };
value?: unknown;
pick?: string;
}
/** 입력 값 답 — 표는 `rows` · 그 밖은 `value` · 못 구하면 ok false + 쉬운 말 */
@@ -37,10 +42,32 @@ export interface CfAnswer {
name?: string;
unit?: string;
reason?: string;
/** 재료 조건 찾기가 여럿 맞을 때 후보 다섯 */
candidates?: CfMaterial[];
}
/** 서버 계산 결과 — 저장 답 */
/** 자재 찾기 한 줄 */
export interface CfMaterial {
ref: string;
name: string;
spec: string;
unit: string;
price: number | null;
구분: string;
상세구분: string;
}
/** 단위 · 금액 표준 표 — 고친 것(`_data`)이 있으면 saved */
export interface CfStandards {
unit: { 이름: string; 줄: Record<string, unknown>[]; [k: string]: unknown };
money: { 이름: string; 줄: Record<string, unknown>[]; [k: string]: unknown };
saved: boolean;
}
/** 서버 계산 결과 — 저장 답(화면 Eval 과 같은 차례 · 같은 규칙 · arrived = 상자마다 받은 값 · breaks = 끊긴 곳) */
export interface CfComputed {
arrived?: Record<string, Record<string, unknown>>;
breaks?: { kind: string; box: string; port?: string; link?: string; text: string }[];
/** 상자 id → 내보내는 알약 이름 → 값(표는 칸 덩어리) */
ports: Record<string, Record<string, unknown>>;
/** 출력 상자 id → 값 */
@@ -82,11 +109,11 @@ export class CfError extends Error {
}
}
async function send<T>(path: string, body?: unknown): Promise<T> {
async function send<T>(path: string, body?: unknown, method?: string): Promise<T> {
let response: Response;
try {
response = await fetch(`${API_BASE_URL}/m02/df${path}`, {
method: body === undefined ? "GET" : "POST",
method: method ?? (body === undefined ? "GET" : "POST"),
credentials: "include",
headers: { "Content-Type": "application/json" },
body: body === undefined ? undefined : JSON.stringify(body),
@@ -156,3 +183,37 @@ export function cfPlain(e: unknown): string {
if (e.status === 400 && text && !/[A-Z]{2}\d{6}|[{}]/.test(text)) return text;
return PLAIN[e.status] ?? "저장하지 못함 — 잠시 뒤 다시 해 주세요";
}
/** 자재 찾기 — 띄어쓰기로 나눈 말을 모두 포함(이름 · 규격 · 키 · 원문번호) · 쪽 넘기기 */
export async function cfMaterials(q: {
q?: string;
sub?: string;
detail?: string;
page?: number;
size?: number;
}): Promise<{ total: number; page: number; size: number; items: CfMaterial[] }> {
const qs = new URLSearchParams(
Object.entries(q).flatMap(([k, v]) => (v === undefined || v === "" ? [] : [[k, String(v)]])),
);
return send(`/materials?${qs.toString()}`);
}
/** 다른 로직 값 — 저장 로직 · 견본을 받은 설계 값(없으면 예시값)으로 서버가 셈 · 출력 이름 → 값 */
export async function cfLogicValue(
key: string,
designs: Record<string, unknown> = {},
): Promise<{ outputs: Record<string, unknown>; cost: CfComputed["cost"]; errors: string[] }> {
return send("/logic-value", { key, designs });
}
/** 설계 때 정할 것 — 변수 · 재료 자리 · 다른 로직 받는 알약(프로젝트 일위대가가 쓸 꼴) */
export async function cfDesignInputs(key: string): Promise<CfDesign[]> {
const qs = new URLSearchParams({ key }).toString();
return (await send<{ designs: CfDesign[] }>(`/design-inputs?${qs}`)).designs;
}
/** 단위 · 금액 표준 표 읽기 · 고치기(시스템 관리자) · 처음 값으로 */
export const cfStandards = (): Promise<CfStandards> => send("/standards");
export const cfStandardsSave = (body: Omit<CfStandards, "saved">): Promise<CfStandards> =>
send("/standards", body, "PUT");
export const cfStandardsReset = (): Promise<CfStandards> => send("/standards", undefined, "DELETE");
M02_MasterTemplete/DataFlow/M02_DF_Compute.py
+554
View File
@@ -0,0 +1,554 @@
"""단가산출 흐름도 서버 계산(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` 과 같은 차례 · 같은 뜻으로 셈(짝 ① — 화면은 TS · 서버는
이 파일 · 함수 시트는 같은 엔진(Node · `common_util_spreadsheet`)):
· 상자 차례 = 선 · 선 규칙 조건 알약 · 재료 조건 찾기 알약에 기댐(Model `depends`) — 서로 안 기대는
함수 상자들은 한 번에 Node 로 셈(시트마다 통합문서 하나)
· 입력 상자 알약 = 요소(재료는 고정 · 선택 · 조건 찾기 — 알약 이름 = `elementPort`) · 표 · 변수 ·
다른 로직 출력 — 단가는 마스터에서 새로(`M02_DF_Values`) · 설계 값 values = {상자: {알약: 값}}
· 선 규칙 = `M02_DF_Rule.applyRule` 과 같은 뜻(값 기준) — 막히면(통과 안 됨 · 표에 없는 조건 ·
여럿 ·
오류) 받는 함수는 셈하지 않고 빈 값 · 끝수 = 엔진 ROUND · ROUNDUP · ROUNDDOWN 과 같음
· 다른 로직 = 저장 로직 · 견본을 받은 설계 값(선 · 검증 칸 · 없으면 그 로직 예시값)으로 셈 ·
서로 부르면 막음
마스터 파일은 읽기만.
"""
from __future__ import annotations
import copy
import re
from decimal import ROUND_DOWN, ROUND_HALF_UP, ROUND_UP, 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 _port(el: dict) -> str:
"""요소 알약 이름 — Model `elementPort`(고정 아니면 자리 이름 slot · 없으면 name)."""
slot = str(el.get("slot") or "").strip()
return slot if el.get("mode") not in (None, "fixed") and slot else el.get("name")
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[_port(el)] = _element_value(
group, el, given.get(_port(el)), 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": _port(el),
"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
# ── 선 규칙(`M02_DF_Rule.applyRule` 와 같은 뜻 · 값 기준) ────────────────────
def _num_of(v: Any) -> Decimal | None:
if isinstance(v, bool) or v is None:
return None
if isinstance(v, (int, float, Decimal)):
return Decimal(str(v))
return None
def _same(a: Any, b: Any) -> bool:
"""엑셀 「=」 — 수끼리는 수로 · 글끼리는 대소문자 무시 · 빈 칸은 0 · 빈 글과 같음."""
na, nb = _num_of(a), _num_of(b)
if a is None or b is None:
other = b if a is None else a
return other is None or other == "" or _num_of(other) == 0
if na is not None and nb is not None:
return na == nb
if isinstance(a, str) and isinstance(b, str):
return a.lower() == b.lower()
return False
def _compare(a: Any, op: str, b: Any) -> bool:
if op in ("=", "≠"):
return _same(a, b) == (op == "=")
na, nb = _num_of(a if a is not None else 0), _num_of(b)
if na is None or nb is None:
sa, sb = str(a or "").lower(), str(b or "").lower()
na, nb = (sa, sb) if na is None and nb is None else (None, None)
if na is None:
return False
return {"≥": na >= nb, "≤": na <= nb, ">": na > nb, "<": na < nb}.get(op, False)
def _rounded(value: Any, r: dict) -> Any:
"""끝수 — 엔진 ROUND(반올림 · 0 에서 먼 쪽) · ROUNDUP · ROUNDDOWN · 자리는 음수도."""
n = _num_of(value)
if n is None:
return value
mode = {"반올림": ROUND_HALF_UP, "올림": ROUND_UP, "버림": ROUND_DOWN}.get(
r.get("method"), ROUND_HALF_UP
)
places = int(r.get("places") or 0)
return (
n.quantize(Decimal(1).scaleb(-places), rounding=mode)
if places >= 0
else ((n.scaleb(places).quantize(Decimal(1), rounding=mode)).scaleb(-places))
)
def _lookup(rule: dict, block: Any, conds: dict) -> tuple[Any, str | None]:
"""표에서 찾기 — 같음 · 범위(「이름·아래」 < 값 ≤ 「이름·위」 · 빈 끝 열림) · 맞은 줄 하나만."""
if not isinstance(block, list) or len(block) < 2:
return None, "오류"
head = [str(h if h is not None else "") for h in block[0]]
out = rule.get("col")
if out not in head:
return None, "오류"
tests = []
for c in rule.get("conds") or []:
col, want = str(c.get("col")), conds.get((c.get("from") or {}).get("name"))
if col in head:
i = head.index(col)
tests.append(lambda row, i=i, want=want: _same(row[i], want))
elif col + dv.RANGE_LOW in head and col + dv.RANGE_HIGH in head:
lo, hi, x = (
head.index(col + dv.RANGE_LOW),
head.index(col + dv.RANGE_HIGH),
_num_of(want),
)
def fits(row, lo=lo, hi=hi, x=x) -> bool:
low, high = _num_of(row[lo]), _num_of(row[hi])
return (
x is not None
and (row[lo] is None or (low is not None and low < x))
and (row[hi] is None or (high is not None and x <= high))
)
tests.append(fits)
else:
return None, "오류"
hits = [row for row in block[1:] if all(t(list(row) + [None] * len(head)) for t in tests)]
if not hits:
return None, "표에 없는 조건"
if len(hits) > 1:
return None, "표에 여럿 맞음"
value = hits[0][head.index(out)]
return (None, "오류") if value is None else (value, None)
def apply_rule(rule: dict, value: Any, conds: dict) -> tuple[Any, str | None]:
"""(값, 까닭) — 까닭 = 통과 안 됨 · 표에 없는 조건 · 표에 여럿 맞음 · 오류 · None."""
kind = rule.get("kind")
if kind == "lookup":
return _lookup(rule, value, conds)
if kind == "gate":
src = (rule.get("from") or {}).get("name")
return (
(value, None)
if _compare(conds.get(src), rule.get("op"), rule.get("value"))
else (None, "통과 안 됨")
)
n = _num_of(value)
if n is None:
return (None, None) if value is None else (None, "오류")
if kind == "round":
return _rounded(n, rule), None
v = Decimal(str(rule.get("value")))
op = rule.get("op")
if op == "÷" and v == 0:
return None, "오류"
return {"×": n * v, "÷": n / v, "+%": n * (1 + v / 100), "-%": n * (1 - v / 100)}.get(
op, n
), None
def _sources(rule: dict | None) -> list[dict]:
if not rule:
return []
if rule.get("kind") == "lookup":
return [c.get("from") or {} for c in rule.get("conds") or []]
return [rule.get("from") or {}] if rule.get("kind") == "gate" else []
# ── 계산(`M02_DF_Eval.evalLogic` 와 같은 차례) ─────────────────────────────
def _depends(lg: dict) -> list[tuple[str, str]]:
"""상자 사이 기댐 — 선 + 선 규칙 조건 알약 + 재료 조건 찾기 알약(Model `depends`)."""
out = []
for ln in lg.get("links") or []:
out.append((ln["from"].get("box"), ln["to"].get("box")))
out += [(p.get("box"), ln["to"].get("box")) for p in _sources(ln.get("rule"))]
for b in lg.get("boxes") or []:
for el in (b.get("input") or {}).get("elements") or []:
if el.get("mode") == "rule" and el.get("rule"):
out.append(((el["rule"].get("from") or {}).get("box"), b["id"]))
return out
def _waves(lg: dict) -> tuple[list[list[dict]], list[dict]]:
"""차례 물결 — 한 물결 안 상자는 서로 안 기댐(함수 시트를 한 번에 셈) · 돌고 도는 것은 따로."""
deps, boxes = _depends(lg), lg.get("boxes") or []
left, waves = {b["id"] for b in boxes}, []
while left:
ready = [
b
for b in boxes
if b["id"] in left and not any(t == b["id"] and f != t and f in left for f, t in deps)
]
if not ready:
break
waves.append(ready)
left -= {b["id"] for b in ready}
return waves, [b for b in boxes if b["id"] in left]
def _sheet_of(box: dict, arrived: dict, given: dict) -> dict:
"""함수 상자 시트 — 받는 칸 · 변수 칸에 값(Eval `runFunc` 와 같음)."""
func = box["func"]
sheet = copy.deepcopy(func.get("sheet") or {})
cells = sheet.setdefault("칸", {})
for p in func.get("inputs") or []:
if str(p.get("cell") or "").strip():
port = p.get("port")
_place(
cells,
p["cell"],
arrived[port] if arrived.get(port) is not None else given.get(port),
)
for v in func.get("variables") or []:
if v.get("cell"):
_place(cells, v["cell"], given.get(v.get("name"), v.get("example")))
return sheet
def run(lg: dict, values: dict, load: Loader, local: list[dict], stack: list | None = None) -> dict:
"""셈 — {ports, outputs, cost, errors, breaks, 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 []}
links = lg.get("links") or []
name = lambda i: boxes.get(i, {}).get("name") or i # noqa: E731
errors, breaks, ports, arrived, outputs = [], [], {}, {}, {}
blocked: set[str] = set()
cost: dict[str, Any] = {**dict.fromkeys(COSTS, Decimal(0)), "값들": {}}
def through(ln: dict) -> Any:
src = ln.get("from") or {}
v = ports.get(src.get("box"), {}).get(src.get("name"))
rule = ln.get("rule")
if not rule:
return v
if v is None and rule.get("kind") in ("scale", "round"):
return None
conds = {
p.get("name"): ports.get(p.get("box"), {}).get(p.get("name")) for p in _sources(rule)
}
got, why = apply_rule(rule, v, conds)
if why:
blocked.add(ln["to"].get("box"))
text = f"「{name(src.get('box'))}」 → 「{name(ln['to'].get('box'))}」 {why}"
kind = "오류" if why in ("표에 여럿 맞음", "오류") else why
to = ln["to"]
breaks.append(
{
"kind": kind,
"box": to.get("box"),
"port": to.get("name"),
"link": ln.get("id"),
"text": text,
}
)
if why != "통과 안 됨":
errors.append(text)
return None if why else got
def arrive(box: dict) -> dict:
got = {
ln["to"].get("name"): through(ln) for ln in links if ln["to"].get("box") == box["id"]
}
arrived[box["id"]] = got
return got
waves, loop = _waves(lg)
errors += [f"「{b.get('name')}」 상자가 서로를 기다림(돌고 도는 선)" for b in loop]
sheets: dict[str, dict] = {}
for wave in waves:
funcs = []
for b in wave:
given = values.get(b["id"]) or {}
if b.get("kind") == "input" and (b.get("input") or {}).get("mode") == "logic":
ports[b["id"]] = _logic_ports(lg, b, {**given, **arrive(b)}, ports, errors, ctx)
elif b.get("kind") == "input":
ports[b["id"]] = _input_ports(lg, b, values, ports, errors, ctx)
elif b.get("kind") == "function" and b.get("func"):
sheet = _sheet_of(b, arrive(b), given)
sheets[b["id"]] = sheet
if b["id"] in blocked:
ports[b["id"]] = {b["func"]["output"].get("name"): None}
else:
funcs.append(b)
elif b.get("kind") == "output":
outputs[b["id"]] = arrive(b).get(OUTPUT_PORT)
if funcs: # 이 물결의 함수 시트를 한 번에(시트마다 통합문서 하나 — Eval 과 같음)
books = [
_num({"종류": "통합문서", "판": 1, "서식": [{}] * 64, "시트": [sheets[f["id"]]]})
for f in funcs
]
try:
calcs = spreadsheet.recalc_workbooks(books)
except spreadsheet.RecalcError as e:
raise store.StoreError(503, str(e)) from e
for f, book, calc in zip(funcs, books, calcs):
sheet = book["시트"][0]
for cell, v in (calc.get(sheet.get("id")) or {}).items():
if "오류" in v:
errors.append(f"「{f.get('name')}」 {cell} 칸 계산 오류 {v['오류']}")
out = f["func"].get("output") or {}
at = _at(out.get("cell") or "A1")[2]
hit = (calc.get(sheet.get("id")) or {}).get(at)
value = (
_plain(hit)
if hit is not None
else ((sheet.get("칸") or {}).get(at) or {}).get("값")
)
if out.get("round") and _num_of(value) is not None:
value = _rounded(value, out["round"])
ports[f["id"]] = {out.get("name"): value}
for b in boxes.values(): # 출력 상자 · 원가(Eval 과 같음)
if b.get("kind") != "output":
continue
value = outputs.get(b["id"])
value = None if isinstance(value, list) else value
if (b.get("output") or {}).get("round") and _num_of(value) is not None:
value = _rounded(value, b["output"]["round"])
outputs[b["id"]] = value
kind = (b.get("output") or {}).get("cost") or "값"
if kind == "값":
cost["값들"][b.get("name")] = value
elif _num_of(value) is not None:
cost[kind] += _num_of(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["id"] in sheets:
b["func"]["sheet"] = sheets[b["id"]]
return {"ports": ports, "arrived": arrived, "outputs": outputs, "cost": cost, "errors": errors,
"breaks": breaks, "filled": filled} # fmt: skip
M02_MasterTemplete/DataFlow/M02_DF_InputPanel.css
+6
View File
@@ -143,3 +143,9 @@
max-height: 24vh;
overflow: auto;
}
.m02-dfip__unitfield {
display: flex;
gap: var(--spacing-8);
align-items: center;
}
M02_MasterTemplete/DataFlow/M02_DF_InputPanel.ts
+15 -10
View File
@@ -38,6 +38,7 @@ import {
type M01Row,
} from "./M02_DF_InputPanel_Build";
import { renderLogicInput } from "./M02_DF_InputPanel_Logic";
import { NO_UNIT, createUnitPick } from "./M02_DF_UnitPick";
import { renderMaterial } from "./M02_DF_InputPanel_Material";
import { renderTablePick } from "./M02_DF_InputPanel_Table";
import { renderVariables } from "./M02_DF_InputPanel_Vars";
@@ -417,7 +418,18 @@ function newForm(
): HTMLElement {
const name = createInputField({ type: "text", label: t("NewName"), compact: true });
const spec = createInputField({ type: "text", label: t("NewSpec"), compact: true });
const unit = createInputField({ type: "text", label: t("NewUnit"), compact: true });
let unitValue = "";
const unit = el("div", { className: "m02-dfip__unitfield" });
const drawUnit = (): void =>
unit.replaceChildren(
el("span", { className: "m02-dfip__muted", text: t("NewUnit") }),
createUnitPick({
value: unitValue || NO_UNIT,
ariaLabel: t("NewUnit"),
onPick: (u) => ((unitValue = u === NO_UNIT ? "" : u), drawUnit()),
}),
);
drawUnit();
const value = createInputField({ type: "text", label: t("NewValue"), compact: true });
const go = createButton({
label: t("Make"),
@@ -432,7 +444,7 @@ function newForm(
group,
name: name.input.value.trim(),
spec: spec.input.value.trim(),
unit: unit.input.value.trim(),
unit: unitValue,
value: n,
}).then(
(made) => {
@@ -455,14 +467,7 @@ function newForm(
});
const d = el("details", {
className: "m02-dfip__new",
children: [
el("summary", { text: t("NewHead") }),
name.root,
spec.root,
unit.root,
value.root,
go,
],
children: [el("summary", { text: t("NewHead") }), name.root, spec.root, unit, value.root, go],
});
return d;
}
M02_MasterTemplete/DataFlow/M02_DF_InputPanel_Vars.ts
+16 -9
View File
@@ -7,6 +7,7 @@
import { currentLanguageIndex } from "@ui/ui_template_locale";
import { createButton, createInputField, createSelectField, el } from "@ui/ui_template_elements";
import type { CfLogic, CfVariable } from "./M02_DF_Model";
import { NO_UNIT, createUnitPick } from "./M02_DF_UnitPick";
import { checkVariable, putVariable, removeVariable } from "./M02_DF_InputPanel_Build";
const TEXT = {
@@ -84,13 +85,19 @@ export function renderVariables(
compact: true,
onChange: () => paint(),
});
const unit = createInputField({
type: "text",
label: t("Unit"),
value: cur?.unit ?? "",
placeholder: "예: ㎡",
compact: true,
});
// 단위 = 고르기만(표준 단위 목록) — 손글 안 받음
let unitValue = cur?.unit ?? "";
const unit = el("div", { className: "m02-dfip__unitfield" });
const drawUnit = (): void =>
unit.replaceChildren(
el("span", { className: "m02-dfip__muted", text: t("Unit") }),
createUnitPick({
value: unitValue || NO_UNIT,
ariaLabel: t("Unit"),
onPick: (u) => ((unitValue = u === NO_UNIT ? "" : u), drawUnit()),
}),
);
drawUnit();
const lo = createInputField({
type: "text",
label: t("Min"),
@@ -129,7 +136,7 @@ export function renderVariables(
const v: CfVariable = {
name: name.input.value.trim(),
type: ty,
unit: unit.input.value.trim(),
unit: unitValue,
example: ex === "" ? null : ty === "number" ? (numOf(ex) ?? ex) : ex,
...(ty === "number" && numOf(lo.input.value) !== null && numOf(hi.input.value) !== null
? { range: [Number(lo.input.value), Number(hi.input.value)] as [number, number] }
@@ -164,7 +171,7 @@ export function renderVariables(
children: [
name.root,
type.root,
unit.root,
unit,
extra,
example.root,
el("div", { className: "m02-dfip__chips", children: cancel ? [ok, cancel] : [ok] }),
M02_MasterTemplete/DataFlow/M02_DF_Router.py
+85 -422
View File
@@ -25,21 +25,20 @@
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
router = APIRouter(prefix="/api/m02/df", tags=["M02 DataFlow (시험)"])
@@ -183,426 +182,19 @@ def user_edit(op: str, element: dict | None, uid: str | None) -> dict:
return {"element": one}
# ── 입력 값 ────────────────────────────────────────────────────────────
def _row(line: dict, names) -> dict:
"""단가를 `/calc` 와 같은 셈으로 — 호표 한 줄(수량 1)짜리 저장 안 하는 견본 로직."""
return {
"키": "",
"원문번호": "",
"이름": "입력 값",
"결과단위": "원/식",
"출처": "자체",
"입력": [{"이름": n} for n in names],
"중간": [],
"호표": [line],
"덧줄": [],
"끝수": None,
}
# ── 저장 · 계산(`M02_DF_Compute`) ─────────────────────────────────────
def _load(key: str) -> dict:
"""다른 로직 입력이 부르는 로직 — 내 것(`_data`) · 견본(`_sample`)."""
logic, _ = _read(key)
return {**logic, "key": key}
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 _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):
@@ -612,13 +204,27 @@ def save(logic: dict) -> dict:
key = None # 견본은 그대로 두고 새 번호로
elif key:
_read(key) # 있는 것만 고침 — 없으면 404
done, result = compute(logic, logic.get("userElements") or [])
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))}
# ── 길 ────────────────────────────────────────────────────────────────
@@ -651,7 +257,7 @@ def post_delete(body: KeyBody) -> dict:
@router.post("/values")
def post_values(body: ValuesBody) -> dict:
return _call(values, body.items, body.userElements)
return _call(dv.values, body.items, [*body.userElements, *_users()])
@router.get("/user-elements")
@@ -662,3 +268,60 @@ def get_user_elements() -> dict:
@router.post("/user-elements")
def post_user_elements(body: UserBody) -> dict:
return _call(user_edit, body.op, body.element, body.id)
# ── 자재 찾기 · 표준 표(정함 5 · 11) ─────────────────────────────────────
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]
@router.get("/materials")
def get_materials(q: str = "", sub: str = "", detail: str = "", page: int = 1, size: int = 30):
return _call(dv.materials, q, sub, detail, page, size)
@router.get("/standards")
def get_standards() -> dict:
return _call(dv.standards, STANDARD, DATA)
def _standards_put(body: dict) -> dict:
dv.standards_check(body)
for part, name in dv.STANDARDS.items():
_write(DATA / name, body[part])
return dv.standards(STANDARD, DATA)
def _standards_reset() -> dict:
for name in dv.STANDARDS.values():
(DATA / name).unlink(missing_ok=True)
return dv.standards(STANDARD, DATA)
@router.put("/standards")
def put_standards(body: StandardsBody) -> dict:
return _call(_standards_put, body.model_dump())
@router.delete("/standards")
def delete_standards() -> dict:
return _call(_standards_reset)
M02_MasterTemplete/DataFlow/M02_DF_Values.py
+406
View File
@@ -0,0 +1,406 @@
"""단가산출 흐름도 입력 값(PLAN 35-9 · 정함 5 · 6 · 11) — 서버 길은 `M02_DF_Router.py`.
· 그룹 요소 값 — 인력 노임 · 재료(키 · 검색어 묶음) · 유가전력 · 환율 · 요율 ·
기계(값 칸 · 시간당 경비) · 소요량 / 계수 표(블록 · 한 줄) · 사용자 요소
— 단가는 기존 `/calc`(`M01_MasterData_Store.calc`)와 같은 셈
· 재료 자리 셋(고정 · 선택 · 조건 찾기) · 자재 찾기(띄어쓰기 AND · 쪽) · 단위 · 금액 표준 표
마스터 파일은 읽기만.
"""
from __future__ import annotations
import json
import re
from typing import Any
from M01_MasterData import M01_MasterData_Store as store
cm = store.cm
_UKEY = re.compile(r"^UE-(\d{4,})$")
RANGE_LOW, RANGE_HIGH = "·아래", "·위"
FIELDS = (*cm.mf.COST_ITEMS, "계")
# ── 입력 값 ────────────────────────────────────────────────────────────
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:
"""사용자 요소 값 — local = 다이어그램 안 것 + 공용 목록(부르는 쪽이 모음)."""
one = next((r for r in local 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(item.get("slot"), dict): # 재료 자리(고정 · 선택 · 조건 찾기)
got = slot_price(item["slot"], item.get("value"), item.get("pick"))
elif 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]}
# ── 자재 찾기 · 재료 자리 ──────────────────────────────────────────────
MATERIALS = "재료_자재품목.json"
_LEDGER: dict[str, Any] = {}
def _numbers() -> dict[str, str]:
"""자재품목 키 → 원문번호(품번) — 줄에 없고 키대장에만 있음 · 파일이 바뀌면 다시 읽음."""
path = store.FOLDER / store.mk.BOOK.name
stamp = (str(path), path.stat().st_mtime if path.is_file() else 0)
if _LEDGER.get("stamp") != stamp:
book = json.loads(path.read_text(encoding="utf-8")) if path.is_file() else {"키": {}}
keys = book.get("키") or {}
found = {k: str(v.get("원문번호") or "") for k, v in keys.items() if k.startswith("MT")}
_LEDGER.update(stamp=stamp, map=found)
return _LEDGER["map"]
def _squash(text: Any) -> str:
return re.sub(r"\s+", "", str(text or "")).lower()
def _found(sub: str, detail: str, q: str) -> list[dict]:
"""범위(구분 · 상세구분) 안 · 띄어쓰기로 나눈 말을 모두(AND) 이름 · 규격 · 키 · 원문번호에서."""
words = [w.lower() for w in str(q or "").split()]
numbers = _numbers()
rows = (store.loaded()[0].get(MATERIALS) or {}).get("줄") or []
out = []
for r in rows:
if (sub and r.get("구분") != sub) or (detail and r.get("상세구분") != detail):
continue
hay = " ".join(
str(x or "")
for x in (r.get("이름"), r.get("규격"), r.get("키"), numbers.get(r.get("키")))
).lower()
if all(w in hay for w in words):
out.append(r)
return out
def _brief_row(r: dict) -> dict:
return {
"ref": r.get("키"),
"name": r.get("이름"),
"spec": r.get("규격") or "",
"unit": r.get("단위") or "",
"price": store._lowest(r),
"구분": r.get("구분"),
"상세구분": r.get("상세구분"),
}
def materials(q: str, sub: str, detail: str, page: int, size: int) -> dict:
"""자재 찾기 — {total, page, size, items: [{ref, name, spec, unit, price, 구분, 상세구분}]}."""
hits = _found(sub, detail, q)
page, size = max(page, 1), min(max(size, 1), 200)
part = hits[(page - 1) * size : page * size]
return {"total": len(hits), "page": page, "size": size, "items": [_brief_row(r) for r in part]}
FIELDS_RULE, OPS_RULE = ("규격", "이름"), ("포함", "같음") # Model CfMaterialRule
def slot_price(slot: dict, value: Any = None, pick: str | None = None) -> dict:
"""재료 자리 단가 — 고정 = 품목 · 선택 = 고른 품목(없으면 기본) · 조건 찾기 = 범위 + 규칙.
slot = Model CfElementRef 꼴 {mode: fixed | choice | rule, ref?, default?,
scope?: {구분?, 상세구분?, 찾는말?}, field?: 규격 | 이름, op?: 포함 | 같음}
조건 찾기(value = 이어진 알약 값): 하나면 그 단가 · 0 이면 맞는 재료 없음 · 여럿이면 후보 다섯.
"""
mode = slot.get("mode") or "fixed"
if mode == "fixed":
return _price("재료", str(slot.get("ref") or ""))
if mode == "choice":
ref = pick or slot.get("default")
if not ref:
return {"ok": False, "reason": "고른 품목도 기본 품목도 없음 — 품목을 골라 주세요"}
return _price("재료", str(ref))
if mode != "rule":
return {"ok": False, "reason": f"모르는 재료 방식 「{mode}」"}
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)
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(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 {
"ok": False,
"reason": f"맞는 재료 {len(hits)}개 — 범위나 규칙을 좁혀 주세요({names} …)",
"candidates": [_brief_row(r) for r in hits[:5]],
}
got = _price("재료", str(hits[0].get("키")))
return {**got, "ref": hits[0].get("키"), "name": hits[0].get("이름")}
# ── 단위 · 금액 표준 표 ────────────────────────────────────────────────
STANDARDS = {"unit": "단위표준.json", "money": "금액단위표준.json"}
def standards(first: Any, mine: Any) -> dict:
"""{unit, money, saved} — 고친 것(`_data`)이 있으면 그것 · 없으면 처음 값(`_standard`)."""
out: dict[str, Any] = {"saved": False}
for part, name in STANDARDS.items():
path = mine / name if (mine / name).is_file() else first / name
out["saved"] = out["saved"] or path.parent == mine
out[part] = json.loads(path.read_text(encoding="utf-8"))
return out
def standards_check(body: dict) -> None:
for part in STANDARDS:
table = body.get(part)
if not isinstance(table, dict) or not isinstance(table.get("줄"), list):
raise store.StoreError(400, f"「{part}」 표 모양이 아님 — 줄 목록이 있어야 함")
resources/tester/test_m02_df_router.py
+243 -5
View File
@@ -1,3 +1,4 @@
# ruff: noqa: E501 — 견본 상자 · 선 덩어리는 한눈에 보이게 한 줄로 둠
"""단가산출 다이어그램 시험 길(`M02_MasterTemplete/DataFlow/M02_DF_Router.py`) 시험 — PLAN 35-7.
· Model CfLogic(boxes · links · userElements) 저장 → 열기 → 목록 → 지우기 · 견본(`_sample`)은 읽기만
@@ -25,6 +26,7 @@ from common_util.common_util_auth import require_system_admin, verify_session
from M01_MasterData import M01_MasterData_Router as m01_router
from M01_MasterData import M01_MasterData_Store as store
from M02_MasterTemplete.DataFlow import M02_DF_Router as cf
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv
REAL = store.FOLDER
OWN = "로직_자체.json"
@@ -311,13 +313,13 @@ def test_시스템_관리자가_아니면_403(folder: Path) -> None:
def test_기계_키_칸_꼴과_시간당_경비(client: TestClient) -> None:
"""입력 판(sub3) 기계 ref 「EQ키#칸」 · 물음 {ref, field} · 「시간당 경비」 = 기계경비 로직."""
logic = min(
(r for r in cf._machine_family() if "지역" in [i["이름"] for i in r["입력"]]),
key=cf._machine_rank,
(r for r in dv._machine_family() if "지역" in [i["이름"] for i in r["입력"]]),
key=dv._machine_rank,
)
code = str(cf._codes(logic)[0])
code = str(dv._codes(logic)[0])
eq = _first("기계.json", lambda r: str(r.get("원문번호")) == code)
family = [r for r in cf._machine_family() if code in [str(c) for c in cf._codes(r)]]
chosen = min(family, key=cf._machine_rank)
family = [r for r in dv._machine_family() if code in [str(c) for c in dv._codes(r)]]
chosen = min(family, key=dv._machine_rank)
given = {
i["이름"]: (code if i["이름"] == "기계" else (i.get("고르기") or [None])[0])
for i in chosen["입력"]
@@ -351,3 +353,239 @@ 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:
"""Eval 과 같은 뜻 — 규칙은 값 기준 · 막힌 선(통과 안 됨 · 표에 없는 조건)은 받는 상자를 멈춤."""
keys = _add_pipes(folder)
scope = {"구분": "시험구분", "상세구분": "시험상세", "찾는말": "견본관"}
vars_ = [{"name": "돌", "type": "choice", "unit": "-", "example": "깬돌"},
{"name": "높이", "type": "number", "unit": "m", "example": 2},
{"name": "관경", "type": "text", "unit": "-", "example": "D300"}] # fmt: skip
pipe = {"ref": "", "name": "관", "slot": "관 자리", "spec": "", "unit": "m", "mode": "rule", "scope": scope,
"rule": {"field": "규격", "op": "포함", "from": {"box": "v", "name": "관경"}}} # fmt: skip
pick = {"ref": "", "name": "관2", "slot": "골라 쓰는 관", "spec": "", "unit": "m", "mode": "choice",
"default": keys[1]} # fmt: skip
sheet = {"id": "s", "이름": "셈", "칸": {"B1": {"식": "A1+A2+A3"}}}
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"}],
{"cell": "B1", "name": "합", "unit": "원", "round": {"places": 0, "method": "버림"}})),
_box("o", "output", "재료비", output={"cost": "재료비", "unit": "원",
"round": {"places": -1, "method": "버림"}}),
_box("o2", "output", "골라 쓴 관", output={"cost": "값", "unit": "원"}),
_box("o3", "output", "낮을 때만", output={"cost": "값", "unit": "m"}),
],
"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": 1}},
_link("l4", "f", "합", "o", "값"),
{**_link("l5", "m", "골라 쓰는 관", "o2", "값"), "rule": {"kind": "round", "places": -3, "method": "반올림"}},
{**_link("l6", "v", "높이", "o3", "값"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"},
"op": "<", "value": 1}},
],
"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.5") + 30000 * Decimal("1.1") + 2 # 깬돌 · 높이 2 → 「~3」 줄 · +10% · 통과
assert _d(v["ports"]["f"]["합"]) == int(total) # 함수 출력 끝수 0 자리 버림
assert _d(v["cost"]["재료비"]) == int(total) // 10 * 10 # 출력 끝수 −1 자리 버림
assert _d(v["cost"]["값들"]["골라 쓴 관"]) == 45000 and v["cost"]["값들"]["낮을 때만"] is None
assert [b["kind"] for b in v["breaks"]] == ["통과 안 됨"] and v["errors"] == []
cells = {b["id"]: b for b in got["logic"]["boxes"]}["f"]["func"]["sheet"]["칸"]
assert _d(cells["A1"]["값"]) == Decimal("0.5") and cells["B1"] == {
"식": "A1+A2+A3"
} # 받는 칸에 값만
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"][1]["example"] = 3 # 「~3」 줄(위 끝 포함)
assert _d(
_post(client, "/api/m02/df/save", {"logic": lg})["values"]["ports"]["f"]["합"]
) == int(Decimal("0.5") + 33000 + 3)
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"]
assert any("맞는 재료 없음" in e for e in bad["errors"]) and any(
"표에 없는 조건" in e for e in bad["errors"]
)
assert bad["ports"]["f"]["합"] is None and _d(bad["cost"]["재료비"]) == 0
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