diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Router.py b/M02_MasterTemplete/DataFlow/M02_DF_Router.py index 70f4d841a..c965d5db1 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Router.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Router.py @@ -40,6 +40,7 @@ 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_Values as dv cm = store.cm router = APIRouter(prefix="/api/m02/df", tags=["M02 DataFlow (시험)"]) @@ -183,258 +184,6 @@ 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, - } - - -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 @@ -509,7 +258,7 @@ def _input_ports(box: dict, local: list[dict], errors: list[str]) -> dict[str, A if inp.get("table"): asks.append((TABLE_PORT, {"table": inp["table"]})) for name, ask in asks: - got = one_value({"group": group, **ask}, local) + got = dv.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')}") @@ -612,7 +361,7 @@ def save(logic: dict) -> dict: key = None # 견본은 그대로 두고 새 번호로 elif key: _read(key) # 있는 것만 고침 — 없으면 404 - done, result = compute(logic, logic.get("userElements") or []) + done, result = compute(logic, [*(logic.get("userElements") or []), *_users()]) 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}" @@ -651,7 +400,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 +411,45 @@ 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 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..9e8c5cfd5 --- /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]} + + +RULES = ("규격", "이름", "같음") # 규격에 값 포함 · 이름에 값 포함 · 규격 또는 이름이 값과 같음 + + +def slot_price(slot: dict, value: Any = None, pick: str | None = None) -> dict: + """재료 자리 단가 — 고정 = 품목 · 선택 = 고른 품목(없으면 기본) · 조건 = 범위 + 규칙으로 찾기. + + slot = {mode: fixed | choice | cond, ref?, default?, scope?: {구분?, 상세구분?, q?}, rule?} + 조건 찾기: 하나면 그 단가 · 0 이면 「맞는 재료 없음」 · 여럿이면 몇 개인지 + 후보 다섯. + """ + mode = slot.get("mode") + 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 != "cond": + return {"ok": False, "reason": f"모르는 재료 방식 「{mode}」"} + rule, scope = slot.get("rule") or "규격", slot.get("scope") or {} + if rule not in RULES: + return {"ok": False, "reason": f"모르는 찾기 규칙 「{rule}」"} + if value in (None, ""): + return {"ok": False, "reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요"} + want = _squash(value) + rows = _found( + str(scope.get("구분") or ""), str(scope.get("상세구분") or ""), scope.get("q") or "" + ) + fits = { + "규격": lambda r: want in _squash(r.get("규격")), + "이름": lambda r: want in _squash(r.get("이름")), + "같음": lambda r: want in (_squash(r.get("규격")), _squash(r.get("이름"))), + }[rule] + hits = [r for r in rows if fits(r)] + if not hits: + where = ( + " · ".join(str(x) for x in (scope.get("구분"), scope.get("상세구분")) if x) or "전체" + ) + return {"ok": False, "reason": f"맞는 재료 없음 — {where} 에서 {rule} 「{value}」"} + 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/main.py b/main.py index 1d641f20f..0f3965eed 100644 --- a/main.py +++ b/main.py @@ -653,6 +653,11 @@ app.include_router( __import__("M02_MasterTemplete.CostFlow.M02_CF_Router", fromlist=["router"]).router, dependencies=system_admin_only, ) +# 단가산출 흐름도 시험 길 — 폴더째 지우면 이 줄도 지움 +app.include_router( + __import__("M02_MasterTemplete.DataFlow.M02_DF_Router", fromlist=["router"]).router, + dependencies=system_admin_only, +) # 마스터 템플릿 — 시스템 층은 시스템 관리자만 · 층(회사·개인·프로젝트)·도면 길은 로그인만(권한은 길 안에서). app.include_router(m02_master_template_router, dependencies=system_admin_only) app.include_router(m02_layers_router, dependencies=protected) diff --git a/resources/tester/test_m02_df_router.py b/resources/tester/test_m02_df_router.py index 4b13147c9..59e14a1a3 100644 --- a/resources/tester/test_m02_df_router.py +++ b/resources/tester/test_m02_df_router.py @@ -25,6 +25,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 +312,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["입력"]