wip(m02-cf): 서버 · 시험 첫 꼴(workbook · fills) — 다음 커밋에서 Model CfLogic 꼴로 바꿈

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL
This commit is contained in:
2026-10-01 16:04:32 +09:00
co-authored by Claude Opus 5.5
parent 73fa2e612a
commit 4f02421acc
4 changed files with 830 additions and 22 deletions
+62 -22
View File
@@ -5,6 +5,7 @@
* 서버 = `M02_CF_Router.py`(`/api/m02/cf/*`) · 정함 `docs/raw/M01_쉬운단가/원가다이어그램_정함.md`
* ========================================================================== */
import { API_BASE_URL } from "@config/config_frontend";
import type { Workbook } from "../../A00_Common/spreadsheet/spreadsheet_types";
/** 입력 그룹 — M01 저장 그룹 + 사용자 요소 */
@@ -17,11 +18,12 @@ export type CfCost = "노무비" | "재료비" | "경비" | "계";
/** 입력 값 물음 한 줄 — 값은 열 때 · 저장 때 마스터(사용자 요소는 `_data`)에서 새로
* · 인력 · 재료 = 그 줄 단가(기존 계산과 같은 셈 · 준용 · 낮은 값) — 재료 key 는 키 또는 검색어 묶음 JSON 글,
* unit(그 줄 단위)을 주면 기존 로직 재료 줄과 같은 단가
* · 유가전력 · 환율 · 요율 · 사용자 = 그 줄 값
* · 유가전력 · 환율 · 사용자 = 그 줄 값 · 요율 = 값 묶음의 field(기본 rate_percent · %)
* · 기계 = key(EQ…) + field(값 칸 이름 — 가격 · 손료계수 …) 또는 element(기계 시간당 경비 로직 부름 글) + inputs → 한 시간 값
* · 소요량 · 계수 = 표 블록(머리 줄 포함 · 조건 칸 → 값 칸 · 범위는 「이름·아래」 「이름·위」) · row 를 주면 그 줄 하나 */
export type CfQuery =
| { id: string; group: "인력" | "유가전력" | "환율" | "요율" | "사용자"; key: string }
| { id: string; group: "인력" | "유가전력" | "환율" | "사용자"; key: string }
| { id: string; group: "요율"; key: string; field?: string }
| { id: string; group: "재료"; key: string; unit?: string }
| { id: string; group: "기계"; key: string; field?: string }
| {
@@ -90,47 +92,85 @@ export interface CfUserElement {
비고?: string;
}
const notReady = (): never => {
throw new Error("준비 중");
};
/** 서버가 준 몸과 상태 번호 — 0 = 서버에 못 닿음 */
export class CfError extends Error {
constructor(
public status: number,
public detail: unknown,
) {
super(typeof detail === "string" ? detail : JSON.stringify(detail));
}
}
async function send<T>(path: string, body?: unknown): Promise<T> {
let response: Response;
try {
response = await fetch(`${API_BASE_URL}/m02/cf${path}`, {
method: body === undefined ? "GET" : "POST",
credentials: "include",
headers: { "Content-Type": "application/json" },
body: body === undefined ? undefined : JSON.stringify(body),
});
} catch {
throw new CfError(0, "");
}
const data = (await response.json().catch(() => ({}))) as { detail?: unknown } & T;
if (!response.ok) throw new CfError(response.status, data.detail ?? response.statusText);
return data;
}
/** 내 다이어그램(저장 차례) + 견본 */
export async function cfList(): Promise<CfListItem[]> {
return notReady();
return (await send<{ logics: CfListItem[] }>("/list")).logics;
}
export async function cfGet(_key: string): Promise<CfLogic> {
return notReady();
export async function cfGet(key: string): Promise<CfLogic> {
return (await send<{ logic: CfLogic }>(`/get?${new URLSearchParams({ key }).toString()}`)).logic;
}
export async function cfSave(_logic: CfLogic): Promise<CfSaved> {
return notReady();
/** 저장 — 서버가 입력 값을 마스터에서 다시 넣고 같은 엔진으로 다시 셈(브라우저 계산값은 버림) ·
* 견본을 저장하면 새 번호 */
export async function cfSave(logic: CfLogic): Promise<CfSaved> {
return send<CfSaved>("/save", { logic });
}
export async function cfDelete(_key: string): Promise<void> {
return notReady();
export async function cfDelete(key: string): Promise<void> {
await send<{ key: string }>("/delete", { key });
}
export async function cfValues(_items: CfQuery[]): Promise<CfValue[]> {
return notReady();
/** 입력 값 한 번에 — 열 때 · 예시값을 바꿀 때 화면이 받아 시트에 넣음 */
export async function cfValues(items: CfQuery[]): Promise<CfValue[]> {
return (await send<{ values: CfValue[] }>("/values", { items })).values;
}
/** 사용자 요소 목록 */
export async function cfUserElements(): Promise<CfUserElement[]> {
return notReady();
return (await send<{ elements: CfUserElement[] }>("/user-elements")).elements;
}
/** 사용자 요소 만들기 — 키는 서버가 줌 */
export async function cfUserElementAdd(
_element: Omit<CfUserElement, "key">,
element: Omit<CfUserElement, "key">,
): Promise<CfUserElement> {
return notReady();
return (await send<{ element: CfUserElement }>("/user-elements", { op: "add", element })).element;
}
export async function cfUserElementDelete(_key: string): Promise<void> {
return notReady();
export async function cfUserElementDelete(key: string): Promise<void> {
await send<{ key: string }>("/user-elements", { op: "delete", key });
}
/** 실패 → 쉬운 말 */
export function cfPlain(_e: unknown): string {
return notReady();
const PLAIN: Record<number, string> = {
0: "서버와 이어지지 않음 — 잠시 뒤 다시 해 주세요",
400: "넣은 값의 모양이 맞지 않음",
403: "견본은 바꾸거나 지울 수 없음 — 저장하면 새 다이어그램으로 남음",
404: "찾을 수 없음 — 이미 지워졌을 수 있음 · 목록을 다시 열어 주세요",
503: "엑셀 계산을 서버에서 못 함 — 잠시 뒤 다시 해 주세요",
};
/** 실패 → 쉬운 말(키 · 날 글이 섞인 서버 글은 뭉뚱그림) */
export function cfPlain(e: unknown): string {
if (!(e instanceof CfError)) return e instanceof Error ? e.message : PLAIN[0];
const text = typeof e.detail === "string" ? e.detail : "";
if (e.status === 400 && text && !/[A-Z]{2}\d{6}|[{}]/.test(text)) return text;
return PLAIN[e.status] ?? "저장하지 못함 — 잠시 뒤 다시 해 주세요";
}
@@ -0,0 +1,488 @@
"""단가산출 다이어그램 시험 길(PLAN 35-7) — 목록 · 열기 · 저장 · 지우기 · 입력 값 · 사용자 요소.
GET /api/m02/cf/list → {logics: [{key, name, resultUnit, saved, sample}]}
GET /api/m02/cf/get?key=CF-0001 → {logic}
POST /api/m02/cf/save {logic} → {key, logic, values}
POST /api/m02/cf/delete {key}
POST /api/m02/cf/values {items: CfQuery[]} → {values: CfValue[]}
GET /api/m02/cf/user-elements → {elements}
POST /api/m02/cf/user-elements {op: add | delete, element? | key?}
정함 `docs/raw/M01_쉬운단가/원가다이어그램_정함.md` — 저장 때 입력 값(인력 노임 · 재료 · 유가전력 ·
기계 · 환율 · 요율 · 소요량 / 계수 표 · 사용자 요소)을 마스터(사용자 요소는 `_data`)에서 다시 넣고
함수 상자 시트를 우리 스프레드시트 엔진(Node · `common_util_spreadsheet`)으로 다시 셈 · 브라우저
계산값은 버림. 단가는 기존 `/calc`(`M01_MasterData_Store.calc`)와 같은 셈. 자리 = 이 폴더
`_data/`(git 밖) · 견본 = `_sample/`(git 안 · 읽기만) · 키 `CF-0001` · 사용자 요소 `UE-0001` ·
마스터 파일은 안 씀. 권한 = 시스템 관리자(`main.py` 등록 한 줄 · 폴더째 지우면 그 줄도 지움).
옛 시험판(SheetLogic)의 서버를 옮겨 고쳐 씀 — 그 폴더를 import 하지 않음.
"""
from __future__ import annotations
import json
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Any
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from common_util import common_util_spreadsheet as spreadsheet
from M01_MasterData import M01_MasterData_Store as store
from M01_MasterData.M01_MasterData_Router import _num
cm = store.cm
router = APIRouter(prefix="/api/m02/cf", tags=["M02 CostFlow (시험)"])
HERE = Path(__file__).resolve().parent
DATA = HERE / "_data" # 시험은 임시 폴더로 바꿈
SAMPLE = HERE / "_sample"
USERS = "_user_elements.json" # 사용자 요소 — `_data` 안
_KEY = re.compile(r"^CF-(\d{4,})$")
_UKEY = re.compile(r"^UE-(\d{4,})$")
_SAMPLE_KEY = re.compile(r"^[\w가-힣·() -]+$")
_CELL = re.compile(r"^([A-Z]+)(\d+)$")
RANGE_LOW, RANGE_HIGH = "·아래", "·위"
FIELDS = (*cm.mf.COST_ITEMS, "계")
GROUPS = ("인력", "재료", "유가전력", "기계", "소요량", "계수", "환율", "요율")
class LogicBody(BaseModel):
logic: dict[str, Any]
class KeyBody(BaseModel):
key: str
class ValuesBody(BaseModel):
items: list[dict[str, Any]]
class UserBody(BaseModel):
op: str
element: dict[str, Any] | None = None
key: str | None = None
# ── 자리 ──────────────────────────────────────────────────────────────
def _write(path: Path, data: Any) -> None:
DATA.mkdir(parents=True, exist_ok=True)
tmp = path.with_name(f".{path.name}.tmp")
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8", newline="\n")
os.replace(tmp, path)
def _mine(key: str) -> Path:
if not _KEY.match(key or ""):
raise store.StoreError(400, f"다이어그램 번호 모양이 아님 「{key}」")
return DATA / f"{key}.json"
def _sample(key: str) -> Path | None:
if not key or _KEY.match(key) or not _SAMPLE_KEY.match(key):
return None
path = SAMPLE / f"{key}.json"
return path if path.is_file() else None
def _read(key: str) -> tuple[dict, bool]:
"""(다이어그램, 견본인가) — 내 것은 `_data` · 견본은 `_sample`."""
path = _sample(key)
if path is not None:
return json.loads(path.read_text(encoding="utf-8")), True
path = _mine(key)
if not path.is_file():
raise store.StoreError(404, f"없는 다이어그램 「{key}」")
return json.loads(path.read_text(encoding="utf-8")), False
def _next(folder_glob: list[Path], pattern: re.Pattern, head: str) -> str:
used = [int(m.group(1)) for p in folder_glob if (m := pattern.match(p.stem))]
return f"{head}-{max(used, default=0) + 1:04d}"
def listing() -> dict:
out = []
for path, sample in [
*((p, False) for p in sorted(DATA.glob("CF-*.json"))),
*((p, True) for p in sorted(SAMPLE.glob("*.json"))),
]:
logic = json.loads(path.read_text(encoding="utf-8"))
out.append(
{
"key": path.stem,
"name": logic.get("name") or path.stem,
"resultUnit": logic.get("resultUnit") or "",
"saved": logic.get("saved") or "",
"sample": sample,
}
)
return {"logics": out}
def get(key: str) -> dict:
logic, sample = _read(key)
return {"logic": {**logic, "key": key}, "sample": sample}
def delete(key: str) -> dict:
if _sample(key) is not None:
raise store.StoreError(403, "견본은 지울 수 없음")
_read(key)
_mine(key).unlink()
return {"key": key}
# ── 사용자 요소 ────────────────────────────────────────────────────────
def _users() -> list[dict]:
path = DATA / USERS
return json.loads(path.read_text(encoding="utf-8")) if path.is_file() else []
def user_elements() -> dict:
return {"elements": _users()}
def user_edit(op: str, element: dict | None, key: str | None) -> dict:
rows = _users()
if op == "delete":
if not any(r["key"] == key for r in rows):
raise store.StoreError(404, f"없는 사용자 요소 「{key}」")
_write(DATA / USERS, [r for r in rows if r["key"] != key])
return {"key": key}
if op != "add" or not isinstance(element, dict):
raise store.StoreError(400, "op 는 add(element) 또는 delete(key)")
name = str(element.get("이름") or "").strip()
if not name:
raise store.StoreError(400, "이름을 넣어 주세요")
group = element.get("group") or "사용자"
if group not in (*GROUPS, "사용자"):
raise store.StoreError(400, f"모르는 그룹 「{group}」")
value = element.get("값")
if value is not None and not isinstance(value, (int, float)):
raise store.StoreError(400, "값은 수 또는 빈 값")
used = [int(m.group(1)) for r in rows if (m := _UKEY.match(r["key"]))]
one = {
"key": f"UE-{max(used, default=0) + 1:04d}",
"group": group,
"이름": name,
"규격": str(element.get("규격") or ""),
"단위": str(element.get("단위") or ""),
"값": value,
"비고": str(element.get("비고") or ""),
}
_write(DATA / USERS, [*rows, one])
return {"element": one}
# ── 입력 값 ────────────────────────────────────────────────────────────
def _row(line: dict, names) -> dict:
"""단가를 `/calc` 와 같은 셈으로 — 호표 한 줄(수량 1)짜리 저장 안 하는 견본 로직."""
return {
"키": "",
"원문번호": "",
"이름": "입력 값",
"결과단위": "원/식",
"출처": "자체",
"입력": [{"이름": n} for n in names],
"중간": [],
"호표": [line],
"덧줄": [],
"끝수": None,
}
def _money(got: dict, field: str | None) -> dict:
if not got["ok"]:
return got
if "result" in got:
return {"ok": True, "value": got["result"]}
pick = field or "계"
if pick not in FIELDS:
return {"ok": False, "reason": f"모르는 칸 「{pick}」"}
return {"ok": True, "value": got["sums"][pick]}
def _pick(key: Any) -> dict | None:
"""재료 검색어 묶음 `{구분, 상세구분, 검색어, 규격?}` — 묶음 · 그 JSON 글 · 아니면 None."""
if isinstance(key, dict):
return key
if isinstance(key, str) and key.lstrip().startswith("{"):
try:
got = json.loads(key)
except ValueError:
return None
return got if isinstance(got, dict) else None
return None
def _material(cond: dict, unit: str) -> dict:
"""검색어 묶음 재료 단가 — 기존 로직 재료 줄과 같은 길(검색어 · 자동값 차례 · 단위 환산)."""
line = {"종류": "재료", "요소": cond, "이름": cond.get("검색어") or "재료", "수량": "1"}
got = store.calc("", {}, _row({**line, "단위": unit, "비목": "재료비"}, ()), None)
if not got["ok"]:
return got
one = got["lines"][0]
if one["단가"] is None:
word = cond.get("검색어") or ""
empty = "값 없음" in str(one.get("까닭") or "")
why = (
"단가가 자료에 비어 있음 — 단가를 채우거나 다른 품목을 고름"
if empty
else "맞는 재료 없음"
)
return {"ok": False, "reason": f"「{word}」 {why}"}
picked = one.get("품목") or {}
return {"ok": True, "value": one["단가"], "name": picked.get("이름"), "unit": unit or None}
def _brief(key: str, want: str) -> tuple[dict | None, str | None]:
"""마스터 줄 하나 + 그룹 확인 — (줄, 까닭)."""
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))
ok = group == want or (want == "유가전력" and key.startswith("MO"))
if not ok:
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]
unit = unit or row.get("단위") or ""
line = {"종류": kind, "요소": key, "이름": row.get("이름") or key, "수량": "1"}
got = store.calc("", {}, _row({**line, "단위": unit, "비목": cost}, ()), 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('이름')}」 에 칸 「{pick}」 없음"}
value = value[pick]
unit = (
unit.get(pick) if isinstance(unit, dict) else ("%" if pick == "rate_percent" else unit)
)
if (
value is not None
and not isinstance(value, (int, float, str))
and not hasattr(value, "quantize")
):
value = str(value)
return {"ok": value is not None, "value": value, "name": row.get("이름"), "unit": unit} | (
{} if value is not None else {"reason": f"「{row.get('이름')}」 값이 자료에 비어 있음"}
)
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(key: str) -> dict:
one = next((r for r in _users() if r["key"] == key), None)
if one is None:
return {"ok": False, "reason": f"없는 사용자 요소 「{key}」"}
return {
"ok": one["값"] is not None,
"value": one["값"],
"name": one["이름"],
"unit": one["단위"],
} | ({} if one["값"] is not None else {"reason": f"「{one['이름']}」 값이 비어 있음"})
def one_value(item: dict) -> dict:
"""물음 한 줄 `CfQuery` → 답 `CfValue`."""
group, key = item.get("group"), item.get("key")
try:
if group in ("소요량", "계수"):
got = _table(str(item.get("table") or ""), item.get("row"))
elif group == "재료" and _pick(key) is not None:
got = _material(_pick(key), str(item.get("unit") or ""))
elif group in ("인력", "재료"):
got = _price(group, str(key or ""), str(item.get("unit") or ""))
elif group == "기계" and item.get("element"):
inputs = item.get("inputs") or {}
line = {"종류": "로직", "요소": item["element"], "이름": "기계", "단위": "시간"}
got = _money(
store.calc("", inputs, _row({**line, "수량": "1"}, inputs), None), item.get("field")
)
elif group in ("기계", "유가전력", "환율", "요율"):
got = _plain_value(group, str(key or ""), item.get("field"))
elif group == "사용자":
got = _user(str(key or ""))
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]) -> dict:
# ponytail: 하나씩 셈(단가 하나 0.3초쯤) — 입력 값이 수십이면 마스터 사본 한 벌로 묶어 셈
return {"values": [one_value(i) 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 _put(book: dict, sheet_id: str, cell: str, value: Any) -> None:
"""입력 값 칸 하나 — 값 칸으로 덮어씀(수식이 있었어도) · 빈 값은 값만 뺌(서식은 둠)."""
sheet = next((s for s in book.get("시트") or [] if s.get("id") == sheet_id), None)
if sheet is None:
raise store.StoreError(400, f"없는 시트 「{sheet_id}」")
cells = sheet.setdefault("칸", {})
keep = {"서식": cells[cell]["서식"]} if "서식" in (cells.get(cell) or {}) else {}
if value is None and not keep:
cells.pop(cell, None)
else:
cells[cell] = {**({"값": value} if value is not None else {}), **keep}
def _fill(book: dict, sheet: str, cell: str, got: dict) -> None:
m = _CELL.match(str(cell or "").upper())
if not m:
raise store.StoreError(400, f"칸 주소 모양이 아님 「{cell}」")
col, row = _col(m.group(1)), int(m.group(2))
if "rows" in got: # 표 — cell 이 왼위
for dy, line in enumerate(got["rows"]):
for dx, one in enumerate(line):
_put(book, sheet, f"{_letters(col + dx)}{row + dy}", one)
else:
_put(book, sheet, f"{m.group(1)}{row}", got.get("value") if got.get("ok") else None)
def save(logic: dict) -> dict:
"""저장 — 입력 값 다시 넣기(fills) → Node 로 다시 셈(브라우저 계산값 버림) → `_data`."""
if not isinstance(logic.get("workbook"), dict):
raise store.StoreError(400, "통합문서(workbook)가 없음")
if not str(logic.get("name") or "").strip():
raise store.StoreError(400, "이름을 넣어 주세요")
key = logic.get("key") or None
if key and _sample(key) is not None:
key = None # 견본은 그대로 두고 새 번호로
elif key:
_read(key) # 있는 것만 고침 — 없으면 404
fills = logic.get("fills") or []
answers = [one_value(f.get("item") or {}) for f in fills]
book = _num(logic["workbook"])
for f, got in zip(fills, answers):
_fill(book, f.get("sheet") or "", f.get("cell") or "", _num(got))
try:
book = spreadsheet.with_server_values(book) # 브라우저 계산값은 버리고 서버 값
except spreadsheet.RecalcError as e:
raise store.StoreError(503, str(e)) from e
key = key or _next(list(DATA.glob("CF-*.json")), _KEY, "CF")
stamp = datetime.now().isoformat(timespec="seconds")
done = {**logic, "key": key, "workbook": book, "saved": stamp}
_write(DATA / f"{key}.json", done)
return {"key": key, "logic": done, "values": answers}
# ── 길 ────────────────────────────────────────────────────────────────
def _call(fn, *args):
try:
return _num(fn(*args))
except store.StoreError as e:
raise HTTPException(status_code=e.status, detail=e.detail) from e
@router.get("/list")
def get_list() -> dict:
return _call(listing)
@router.get("/get")
def get_one(key: str) -> dict:
return _call(get, key)
@router.post("/save")
def post_save(body: LogicBody) -> dict:
return _call(save, body.logic)
@router.post("/delete")
def post_delete(body: KeyBody) -> dict:
return _call(delete, body.key)
@router.post("/values")
def post_values(body: ValuesBody) -> dict:
return _call(values, body.items)
@router.get("/user-elements")
def get_user_elements() -> dict:
return _call(user_elements)
@router.post("/user-elements")
def post_user_elements(body: UserBody) -> dict:
return _call(user_edit, body.op, body.element, body.key)
+5
View File
@@ -648,6 +648,11 @@ app.include_router(
__import__("M01_MasterData.SheetLogic.M01_SL_Router", fromlist=["router"]).router,
dependencies=system_admin_only,
)
# 단가산출 다이어그램 시험 길 — 폴더째 지우면 이 줄도 지움
app.include_router(
__import__("M02_MasterTemplete.CostFlow.M02_CF_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)
+275
View File
@@ -0,0 +1,275 @@
"""단가산출 다이어그램 시험 길(`M02_MasterTemplete/CostFlow/M02_CF_Router.py`) 시험 — PLAN 35-7.
· 임시 `_data` 로 저장 → 열기 → 목록 → 지우기 · 견본(`_sample`)은 읽기만 · 저장하면 새 번호
· 저장 때 입력 값을 다시 넣고 서버가 같은 엔진(Node)으로 다시 셈 — 브라우저 계산값은 버림
· 입력 값 — 인력 노임 · 재료(키 · 검색어 묶음) · 유가전력 · 환율 · 요율 · 기계 값 칸 · 표 블록 · 표 한 줄 · 사용자 요소
· 사용자 요소 만들기 · 지우기 · 권한 없음 403 · 마스터 사본 파일은 안 바뀜
견본 표 · 다이어그램은 시험 안에 둠(임시 폴더 사본에 견본 표를 더함 — 진짜 값에 안 기댐).
"""
from __future__ import annotations
import json
import shutil
from decimal import Decimal
from pathlib import Path
import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
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.CostFlow import M02_CF_Router as cf
REAL = store.FOLDER
OWN = "로직_자체.json"
SAMPLE = "QF999995"
TABLE = {
"키": SAMPLE,
"원문번호": "",
"구분": "",
"상세구분": "",
"이름": "견본 높이 표",
"기준": "1㎡",
"출처": "자체",
"조건": {"돌": "고르기", "높이": "범위"},
"범위규칙": "초과·이하",
"값칸": {"석공": "인", "증가율": "%"},
"줄": [
{"돌": "깬돌", "높이": [None, 3], "석공": 0.5, "증가율": [80, 100]},
{"돌": "견치돌", "높이": 4, "석공": 0.7},
],
"주": [],
}
NO_NODE = shutil.which("node") is None
def _first(name: str, pred) -> dict:
return next(
r for r in json.loads((store.FOLDER / name).read_text(encoding="utf-8"))["줄"] if pred(r)
)
def _labor() -> tuple[str, Decimal]:
row = _first(
"인력.json",
lambda r: (
r.get("상태") == "공표" and not r.get("준용") and isinstance(r.get("값"), (int, float))
),
)
return row["키"], Decimal(str(row["값"]))
@pytest.fixture
def folder(tmp_path: Path, monkeypatch) -> Path:
master = tmp_path / "master"
master.mkdir()
for path in REAL.glob("*.json"):
if path.name == OWN:
continue
if not path.name.startswith("_") or path.name == store.mk.BOOK.name:
shutil.copy(path, master / path.name)
host = sorted(master.glob("소요량_*.json"))[0]
data = json.loads(host.read_text(encoding="utf-8"))
data["표"].append(TABLE)
host.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
sample = tmp_path / "_sample"
sample.mkdir()
(sample / "견본 돌쌓기.json").write_text(
json.dumps(
{"name": "견본 돌쌓기", "resultUnit": "원/㎡", "workbook": _book()}, ensure_ascii=False
),
encoding="utf-8",
)
monkeypatch.setattr(store, "FOLDER", master)
monkeypatch.setattr(cf, "DATA", tmp_path / "_data")
monkeypatch.setattr(cf, "SAMPLE", sample)
return master
@pytest.fixture
def client(folder: Path) -> TestClient:
app = FastAPI()
app.include_router(m01_router.router)
app.include_router(cf.router)
return TestClient(app)
def _post(client: TestClient, url: str, body: dict, want: int = 200) -> dict:
res = client.post(url, json=body)
assert res.status_code == want, res.text
return res.json()
def _book() -> dict:
return {
"종류": "통합문서",
"판": 1,
"서식": [{}],
"시트": [
{"id": "f1", "이름": "노무 함수", "칸": {"A1": {"값": 1}, "B1": {"식": "A1*2"}}},
{"id": "f2", "이름": "표 함수", "칸": {}},
],
"계산값": {"f1": {"B1": {"수": "999"}}},
}
def _logic(labor: str, key: str | None = None) -> dict:
"""노임 입력 값 → 함수 f1 A1 · ×2 = B1 · 표 블록 → 함수 f2 E1 부터 · 다른 칸(상자 · 선)은 그대로."""
return {
"key": key,
"name": "견본 다이어그램",
"resultUnit": "원/㎡",
"workbook": _book(),
"fills": [
{"item": {"id": "a", "group": "인력", "key": labor}, "sheet": "f1", "cell": "A1"},
{"item": {"id": "b", "group": "소요량", "table": SAMPLE}, "sheet": "f2", "cell": "E1"},
],
"boxes": [{"id": "x", "kind": "function", "pos": [10, 20]}],
}
@pytest.mark.skipif(NO_NODE, reason="node 가 없음")
def test_저장_열기_목록_지우기_견본(client: TestClient, folder: Path) -> None:
labor, price = _labor()
before = {p.name: p.read_bytes() for p in folder.glob("*.json")}
saved = _post(client, "/api/m02/cf/save", {"logic": _logic(labor)})
assert saved["key"] == "CF-0001"
book = saved["logic"]["workbook"]
assert book["시트"][0]["칸"]["A1"]["값"] == price # 입력 값을 마스터에서 다시 넣음
assert Decimal(book["계산값"]["f1"]["B1"]["수"]) == 2 * price # 서버 다시 셈(999 는 버림)
t = book["시트"][1]["칸"]
assert [t[c]["값"] for c in ("E1", "F1", "G1", "H1", "I1", "J1")] == [
"돌", "높이·아래", "높이·위", "석공", "증가율·아래", "증가율·위",
] # fmt: skip
assert "F2" not in t and t["G2"]["값"] == 3 and t["F3"]["값"] == t["G3"]["값"] == 4
assert saved["logic"]["boxes"] == _logic(labor)["boxes"]
assert [v["id"] for v in saved["values"]] == ["a", "b"]
listed = client.get("/api/m02/cf/list").json()["logics"]
assert [(x["key"], x["sample"]) for x in listed] == [("CF-0001", False), ("견본 돌쌓기", True)]
sample = client.get("/api/m02/cf/get", params={"key": "견본 돌쌓기"}).json()
assert sample["sample"] is True and sample["logic"]["name"] == "견본 돌쌓기"
copied = _post(client, "/api/m02/cf/save", {"logic": sample["logic"]}) # 견본 저장 = 새 번호
assert copied["key"] == "CF-0002"
_post(client, "/api/m02/cf/delete", {"key": "견본 돌쌓기"}, want=403)
renamed = {**saved["logic"], "name": "고친 이름"}
assert _post(client, "/api/m02/cf/save", {"logic": renamed})["key"] == "CF-0001"
got = client.get("/api/m02/cf/get", params={"key": "CF-0001"}).json()["logic"]
assert got["name"] == "고친 이름" and got["workbook"]["계산값"] == book["계산값"]
_post(client, "/api/m02/cf/delete", {"key": "CF-0002"})
assert client.get("/api/m02/cf/get", params={"key": "CF-0002"}).status_code == 404
_post(client, "/api/m02/cf/save", {"logic": _logic(labor, "CF-0009")}, want=404)
_post(client, "/api/m02/cf/save", {"logic": {**_logic(labor), "name": " "}}, want=400)
assert client.get("/api/m02/cf/get", params={"key": "../x"}).status_code in (400, 404)
assert {p.name: p.read_bytes() for p in folder.glob("*.json")} == before # 마스터 사본 그대로
assert sorted(p.name for p in cf.DATA.glob("*.json")) == ["CF-0001.json"]
def test_입력_값_그룹마다(client: TestClient) -> None:
labor, price = _labor()
mat = _first(
"재료_자재품목.json",
lambda r: isinstance(r.get("물가자료"), (int, float)) and r.get("구분"),
)
cond = {
"구분": mat["구분"],
"상세구분": mat["상세구분"],
"검색어": f"{mat['이름']} {mat['규격'] or ''}".strip(),
}
line = {
"종류": "재료",
"요소": cond,
"이름": "재료",
"단위": mat["단위"],
"수량": "1",
"비목": "재료비",
}
row = {"키": "", "원문번호": "", "이름": "견본", "결과단위": "원/㎡", "출처": "자체",
"입력": [], "중간": [], "호표": [line], "덧줄": [], "끝수": None} # fmt: skip
want = _post(client, "/api/m01/calc", {"key": "", "inputs": {}, "row": row})["lines"][0]["단가"]
oil = _first("재료_유가전력.json", lambda r: isinstance(r.get("값"), (int, float)))
fx = _first("환율_한국은행환율.json", lambda r: isinstance(r.get("값"), (int, float)))
rate = _first("요율_조달청제비율.json", lambda r: isinstance(r.get("값"), dict))
eq = _first("기계.json", lambda r: isinstance((r.get("값") or {}).get("가격"), (int, float)))
user = _post(
client,
"/api/m02/cf/user-elements",
{"op": "add", "element": {"이름": "현장 운반비", "단위": "원/회", "값": 12000}},
)
items = [
{"id": "l", "group": "인력", "key": labor},
{
"id": "m",
"group": "재료",
"key": json.dumps(cond, ensure_ascii=False),
"unit": mat["단위"],
},
{"id": "o", "group": "유가전력", "key": oil["키"]},
{"id": "f", "group": "환율", "key": fx["키"]},
{"id": "r", "group": "요율", "key": rate["키"]},
{"id": "e", "group": "기계", "key": eq["키"], "field": "가격"},
{"id": "e2", "group": "기계", "key": eq["키"]},
{"id": "t", "group": "소요량", "table": SAMPLE},
{"id": "t1", "group": "소요량", "table": SAMPLE, "row": 1},
{"id": "u", "group": "사용자", "key": user["element"]["key"]},
{"id": "w", "group": "인력", "key": oil["키"]},
{"id": "z", "group": "모름", "key": ""},
]
got = {v["id"]: v for v in _post(client, "/api/m02/cf/values", {"items": items})["values"]}
assert Decimal(str(got["l"]["value"])) == price
assert got["m"]["value"] == want # 검색어 묶음 = 기존 /calc 재료 줄 단가
assert got["o"]["value"] == oil["값"] and got["f"]["value"] == fx["값"]
assert got["r"]["value"] == rate["값"]["rate_percent"] and got["r"]["unit"] == "%"
assert (
got["e"]["value"] == eq["값"]["가격"] and got["e2"]["ok"] is False
) # 묶음은 칸을 골라야 함
assert got["t"]["rows"][0] == ["돌", "높이·아래", "높이·위", "석공", "증가율·아래", "증가율·위"]
assert got["t1"]["rows"] == [got["t"]["rows"][0], ["견치돌", 4, 4, 0.7, None, None]]
assert got["u"]["value"] == 12000 and got["u"]["unit"] == "원/회"
for k in ("w", "z", "e2"):
assert got[k]["ok"] is False and got[k]["reason"], got[k]
def test_사용자_요소_만들기_지우기(client: TestClient, folder: Path) -> None:
before = {p.name: p.read_bytes() for p in folder.glob("*.json")}
a = _post(
client,
"/api/m02/cf/user-elements",
{"op": "add", "element": {"이름": "가", "group": "재료", "값": 1.5}},
)
b = _post(client, "/api/m02/cf/user-elements", {"op": "add", "element": {"이름": "나"}})
assert [a["element"]["key"], b["element"]["key"]] == ["UE-0001", "UE-0002"]
assert b["element"]["group"] == "사용자" and b["element"]["값"] is None
_post(client, "/api/m02/cf/user-elements", {"op": "add", "element": {"이름": " "}}, want=400)
_post(
client,
"/api/m02/cf/user-elements",
{"op": "add", "element": {"이름": "다", "값": "x"}},
want=400,
)
_post(client, "/api/m02/cf/user-elements", {"op": "delete", "key": "UE-0001"})
_post(client, "/api/m02/cf/user-elements", {"op": "delete", "key": "UE-0001"}, want=404)
keys = [e["key"] for e in client.get("/api/m02/cf/user-elements").json()["elements"]]
assert keys == ["UE-0002"]
assert {p.name: p.read_bytes() for p in folder.glob("*.json")} == before # 마스터에 안 씀
def test_시스템_관리자가_아니면_403(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)
assert client.get("/api/m02/cf/list").status_code == 403
assert client.post("/api/m02/cf/values", json={"items": []}).status_code == 403
assert client.get("/api/m02/cf/user-elements").status_code == 403
app.dependency_overrides[verify_session] = lambda: {"role": "SYSTEM_ADMIN", "company_id": None}
assert [x["key"] for x in client.get("/api/m02/cf/list").json()["logics"]] == ["견본 돌쌓기"]