diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Api.ts b/M02_MasterTemplete/DataFlow/M02_DF_Api.ts index 83706b0fb..35596a0c9 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Api.ts +++ b/M02_MasterTemplete/DataFlow/M02_DF_Api.ts @@ -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; + slot?: Partial & { 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[]; [k: string]: unknown }; + money: { 이름: string; 줄: Record[]; [k: string]: unknown }; + saved: boolean; +} + +/** 서버 계산 결과 — 저장 답(화면 Eval 과 같은 차례 · 같은 규칙 · arrived = 상자마다 받은 값 · breaks = 끊긴 곳) */ export interface CfComputed { + arrived?: Record>; + breaks?: { kind: string; box: string; port?: string; link?: string; text: string }[]; /** 상자 id → 내보내는 알약 이름 → 값(표는 칸 덩어리) */ ports: Record>; /** 출력 상자 id → 값 */ @@ -82,11 +109,11 @@ export class CfError extends Error { } } -async function send(path: string, body?: unknown): Promise { +async function send(path: string, body?: unknown, method?: string): Promise { 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 = {}, +): Promise<{ outputs: Record; cost: CfComputed["cost"]; errors: string[] }> { + return send("/logic-value", { key, designs }); +} + +/** 설계 때 정할 것 — 변수 · 재료 자리 · 다른 로직 받는 알약(프로젝트 일위대가가 쓸 꼴) */ +export async function cfDesignInputs(key: string): Promise { + const qs = new URLSearchParams({ key }).toString(); + return (await send<{ designs: CfDesign[] }>(`/design-inputs?${qs}`)).designs; +} + +/** 단위 · 금액 표준 표 읽기 · 고치기(시스템 관리자) · 처음 값으로 */ +export const cfStandards = (): Promise => send("/standards"); +export const cfStandardsSave = (body: Omit): Promise => + send("/standards", body, "PUT"); +export const cfStandardsReset = (): Promise => send("/standards", undefined, "DELETE"); diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py new file mode 100644 index 000000000..c50191906 --- /dev/null +++ b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py @@ -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 diff --git a/M02_MasterTemplete/DataFlow/M02_DF_InputPanel.css b/M02_MasterTemplete/DataFlow/M02_DF_InputPanel.css index 3ac3c5845..18a1b25f5 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_InputPanel.css +++ b/M02_MasterTemplete/DataFlow/M02_DF_InputPanel.css @@ -143,3 +143,9 @@ max-height: 24vh; overflow: auto; } + +.m02-dfip__unitfield { + display: flex; + gap: var(--spacing-8); + align-items: center; +} diff --git a/M02_MasterTemplete/DataFlow/M02_DF_InputPanel.ts b/M02_MasterTemplete/DataFlow/M02_DF_InputPanel.ts index e268be002..3fb3a69c5 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_InputPanel.ts +++ b/M02_MasterTemplete/DataFlow/M02_DF_InputPanel.ts @@ -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; } diff --git a/M02_MasterTemplete/DataFlow/M02_DF_InputPanel_Vars.ts b/M02_MasterTemplete/DataFlow/M02_DF_InputPanel_Vars.ts index 3c0d9cd0f..555ad4a9d 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_InputPanel_Vars.ts +++ b/M02_MasterTemplete/DataFlow/M02_DF_InputPanel_Vars.ts @@ -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] }), diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Router.py b/M02_MasterTemplete/DataFlow/M02_DF_Router.py index 70f4d841a..c7375090c 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Router.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Router.py @@ -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) diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Values.py b/M02_MasterTemplete/DataFlow/M02_DF_Values.py new file mode 100644 index 000000000..cb4cfad15 --- /dev/null +++ b/M02_MasterTemplete/DataFlow/M02_DF_Values.py @@ -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}」 표 모양이 아님 — 줄 목록이 있어야 함") diff --git a/resources/tester/test_m02_df_router.py b/resources/tester/test_m02_df_router.py index 4b13147c9..833a40096 100644 --- a/resources/tester/test_m02_df_router.py +++ b/resources/tester/test_m02_df_router.py @@ -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