Merge remote-tracking branch 'origin/sub_laptop_4' into main_laptop_1
This commit is contained in:
@@ -2,9 +2,14 @@
|
||||
* M01_LB_Api.ts
|
||||
* 로직 만들기 판 Api — 고친 표로 예시값 계산(0.3초 · 최신만) · 표를 쓰는 로직(사용처) · 표 읽기
|
||||
* 서버 = `M01_LB_Calc.py`(`/api/m01/lb/*`) · 모양은 `docs/raw/M01_쉬운단가/로직판_A_정함.md` 「맞출 모양」
|
||||
* 시험: `resources/tester/test_m01_lb_calc.py`
|
||||
* ========================================================================== */
|
||||
|
||||
import type { CalcAnswer, LogicRow } from "../M01_MasterData_UI_Logic_Api";
|
||||
import { API_BASE_URL } from "@config/config_frontend";
|
||||
import { fetchTable, fetchTables } from "../M01_MasterData_Api_Fetch";
|
||||
import { ApiError, type CalcAnswer, type LogicRow } from "../M01_MasterData_UI_Logic_Api";
|
||||
import { safeReason } from "../M01_MasterData_UI_Logic_Text";
|
||||
import { plainError } from "../EasyCalc/M01_EasyCalc_Api";
|
||||
|
||||
/** 표 키 → 표 사본(고친 값 포함 · 표 통째) — 서버가 메모리에서만 바꿔 끼움 */
|
||||
export type LbTables = Record<string, unknown>;
|
||||
@@ -23,24 +28,76 @@ export interface LbTable {
|
||||
table: Record<string, unknown> & { 줄: Record<string, unknown>[] };
|
||||
}
|
||||
|
||||
const notReady = (): never => {
|
||||
throw new Error("준비 중");
|
||||
};
|
||||
const CALC_WAIT_MS = 300;
|
||||
const TABLE_GROUPS = ["소요량", "계수"]; // 표가 드는 그룹(계수 표가 소요량 파일에 든 것도 있음)
|
||||
|
||||
async function send<T>(path: string, body?: unknown): Promise<T> {
|
||||
let response: Response;
|
||||
try {
|
||||
response = await fetch(`${API_BASE_URL}/m01/lb${path}`, {
|
||||
method: body === undefined ? "GET" : "POST",
|
||||
credentials: "include",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: body === undefined ? undefined : JSON.stringify(body),
|
||||
});
|
||||
} catch {
|
||||
throw new ApiError(0, "");
|
||||
}
|
||||
const data = (await response.json().catch(() => ({}))) as { detail?: unknown } & T;
|
||||
if (!response.ok) throw new ApiError(response.status, data.detail ?? response.statusText);
|
||||
return data;
|
||||
}
|
||||
|
||||
/** 로직이 받는 입력값만 — 모르는 이름을 보내면 서버가 멈춤 */
|
||||
function inputsFor(row: LogicRow, inputs: Record<string, unknown>): Record<string, unknown> {
|
||||
return Object.fromEntries(
|
||||
(row.입력 ?? []).filter((i) => i.이름 in inputs).map((i) => [i.이름, inputs[i.이름]]),
|
||||
);
|
||||
}
|
||||
|
||||
/** 계산 멈춤 까닭 — 표 키(`QF000421`)가 섞인 글은 뭉뚱그림 */
|
||||
function plainStop(reason: string): string {
|
||||
const plain = safeReason(reason);
|
||||
return /[A-Z]{2}\d{6}/.test(plain) ? safeReason("{}") : plain;
|
||||
}
|
||||
|
||||
let calcSeq = 0;
|
||||
let calcTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
|
||||
/** 0.3초 뒤 고친 표로 서버 계산 · 최신 응답만 `onDone` · 멈춤 · 실패는 `plain` 쉬운 말 */
|
||||
export function lbCalcLater(
|
||||
_row: LogicRow,
|
||||
_inputs: Record<string, unknown>,
|
||||
_tables: LbTables,
|
||||
_onDone: (ans: CalcAnswer | null, plain: string | null) => void,
|
||||
row: LogicRow,
|
||||
inputs: Record<string, unknown>,
|
||||
tables: LbTables,
|
||||
onDone: (ans: CalcAnswer | null, plain: string | null) => void,
|
||||
): void {
|
||||
notReady();
|
||||
const mine = ++calcSeq;
|
||||
clearTimeout(calcTimer);
|
||||
calcTimer = setTimeout(() => {
|
||||
send<CalcAnswer>("/calc", { row, inputs: inputsFor(row, inputs), tables })
|
||||
.then((ans) => {
|
||||
if (mine === calcSeq) onDone(ans, ans.ok ? null : plainStop(ans.reason));
|
||||
})
|
||||
.catch((e: unknown) => {
|
||||
if (mine === calcSeq) onDone(null, plainError(e));
|
||||
});
|
||||
}, CALC_WAIT_MS);
|
||||
}
|
||||
|
||||
export async function tableUsers(_key: string): Promise<LbTableUser[]> {
|
||||
return notReady();
|
||||
/** 사용처 — 그 표를 쓰는 로직(표 「용도」 + 로직 글 대조) */
|
||||
export async function tableUsers(key: string): Promise<LbTableUser[]> {
|
||||
const got = await send<{ users: LbTableUser[] }>(
|
||||
`/table-users?${new URLSearchParams({ key }).toString()}`,
|
||||
);
|
||||
return got.users;
|
||||
}
|
||||
|
||||
export async function fetchLbTable(_key: string): Promise<LbTable> {
|
||||
return notReady();
|
||||
/** 표 키 하나로 표 통째 — 든 파일을 찾아(소요량 · 계수) Api_Fetch 의 표 읽기로 */
|
||||
export async function fetchLbTable(key: string): Promise<LbTable> {
|
||||
for (const group of TABLE_GROUPS) {
|
||||
const page = await fetchTables(group, "", "", key, 1, 5);
|
||||
const head = page.tables.find((t) => t.키 === key);
|
||||
if (head) return { file: head.file, ...(await fetchTable(head.file, key)) };
|
||||
}
|
||||
throw new ApiError(404, `없는 표 「${key}」`);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
"""로직 만들기 판 시험 길(PLAN 35-2) — 고친 표로 시험 계산 · 표를 쓰는 로직.
|
||||
|
||||
POST /api/m01/lb/calc {row, inputs, tables: {표 키: 표 통째}} → `/calc` 와 같은 답
|
||||
GET /api/m01/lb/table-users ?key=표 키 → {key, users: [{key, 이름, via}]}
|
||||
|
||||
계산은 `/calc`(`M01_MasterData_Store.calc`)와 같은 길 — 부를 때마다 새로 읽은 마스터 사본
|
||||
(`cm.load`)에 로직 줄 · 표를 바꿔 끼워 셈. 전역 캐시 · 파일은 안 건드림(동시 요청에 안전) ·
|
||||
표 고침은 저장하지 않음(정함 5). 권한 = `/api/m01` 과 같음(등록 자리 `main.py` 한 줄 ·
|
||||
폴더째 지우면 그 줄도 지움).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
from M01_MasterData import M01_MasterData_Store as store
|
||||
from M01_MasterData.M01_MasterData_Router import _num
|
||||
|
||||
cm = store.cm
|
||||
router = APIRouter(prefix="/api/m01/lb", tags=["M01 LogicBuilder (시험)"])
|
||||
|
||||
|
||||
class LbCalcBody(BaseModel):
|
||||
row: dict[str, Any]
|
||||
inputs: dict[str, Any] = {}
|
||||
tables: dict[str, dict[str, Any]] = {}
|
||||
|
||||
|
||||
def _put_tables(files: dict[str, dict], tables: dict[str, dict]) -> None:
|
||||
"""표 키 → 고친 표 — 그 표가 든 파일 사본 자리에 바꿔 끼움 · 키는 못 바꿈 · 없는 표는 404."""
|
||||
for key, table in tables.items():
|
||||
for data in files.values():
|
||||
if data.get("그룹") not in cm.mf.TABLE_GROUPS:
|
||||
continue
|
||||
items = data.get("표") or []
|
||||
at = next((i for i, t in enumerate(items) if str(t.get("키")) == key), None)
|
||||
if at is not None:
|
||||
items[at] = {**store._dec(table), "키": key}
|
||||
break
|
||||
else:
|
||||
raise store.StoreError(404, f"없는 표 「{key}」")
|
||||
|
||||
|
||||
def calc(row: dict, inputs: dict, tables: dict[str, dict]) -> dict:
|
||||
"""`Store.calc` 와 같은 셈 + 표 바꿔 끼우기 — 답 모양도 같음."""
|
||||
given = {k: store._dec(v) for k, v in inputs.items()}
|
||||
files = cm.load(folder=store.FOLDER) # 부를 때마다 새 사본 — 캐시 아님
|
||||
_put_tables(files, tables)
|
||||
key = store._swap(files, str(row.get("키") or ""), row, None)
|
||||
try:
|
||||
given = cm.mf.fill_empty(cm.mf.Master(files).logic(key), given)
|
||||
result = cm.calc(files, key, given)
|
||||
except (cm.mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e:
|
||||
return {"ok": False, "reason": str(e) or type(e).__name__}
|
||||
if "결과" in result:
|
||||
return {"ok": True, "result": result["결과"], "middle": result["중간"]}
|
||||
return {
|
||||
"ok": True,
|
||||
"lines": result["줄"],
|
||||
"sums": {k: result[k] for k in cm.mf.BUNDLE},
|
||||
"middle": result["중간"],
|
||||
}
|
||||
|
||||
|
||||
def table_users(key: str) -> dict:
|
||||
"""표를 쓰는 로직 — 표 「용도.로직키」 + 로직 글에 표 키가 든 것(읽기 전용 캐시에서)."""
|
||||
files, _ = store.loaded()
|
||||
table = next(
|
||||
(
|
||||
t
|
||||
for data in files.values()
|
||||
if data.get("그룹") in cm.mf.TABLE_GROUPS
|
||||
for t in data.get("표") or []
|
||||
if str(t.get("키")) == key
|
||||
),
|
||||
None,
|
||||
)
|
||||
if table is None:
|
||||
raise store.StoreError(404, f"없는 표 「{key}」")
|
||||
listed = {str(k) for k in (table.get("용도") or {}).get("로직키") or []}
|
||||
named = re.compile(rf"(?<![A-Z0-9]){re.escape(key)}(?![0-9])")
|
||||
users = []
|
||||
for data in files.values():
|
||||
if data.get("그룹") != "로직":
|
||||
continue
|
||||
for row in data.get("줄") or []:
|
||||
in_use = str(row.get("키")) in listed
|
||||
in_text = bool(named.search(json.dumps(row, ensure_ascii=False, default=str)))
|
||||
if in_use or in_text:
|
||||
via = "둘" if in_use and in_text else "용도" if in_use else "글"
|
||||
users.append({"key": row.get("키"), "이름": row.get("이름"), "via": via})
|
||||
return {"key": key, "users": users}
|
||||
|
||||
|
||||
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.post("/calc")
|
||||
def post_calc(body: LbCalcBody) -> dict:
|
||||
return _call(calc, body.row, body.inputs, body.tables)
|
||||
|
||||
|
||||
@router.get("/table-users")
|
||||
def get_table_users(key: str) -> dict:
|
||||
return _call(table_users, key)
|
||||
@@ -638,6 +638,11 @@ app.include_router(b07_frame_router, dependencies=protected_with_company)
|
||||
# 마스터 데이터 — 회사·프로젝트가 아니라 시스템 관리자만 본다.
|
||||
system_admin_only = [Depends(verify_session), Depends(require_system_admin)]
|
||||
app.include_router(m01_master_data_router, dependencies=system_admin_only)
|
||||
# 로직 만들기 시험 길 — 폴더째 지우면 이 줄도 지움
|
||||
app.include_router(
|
||||
__import__("M01_MasterData.LogicBuilder.M01_LB_Calc", 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)
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
"""로직 만들기 판 시험 길(`M01_MasterData/LogicBuilder/M01_LB_Calc.py`) 시험 — PLAN 35-2.
|
||||
|
||||
· 고친 표 사본으로 셈하면 값이 바뀌고 · 전역 캐시 · 파일은 그대로
|
||||
· 표 없이 부르면 `/calc` 와 같은 값 · 없는 표는 404
|
||||
· 사용처 — 표 「용도.로직키」 + 로직 글 대조
|
||||
· 권한 — 시스템 관리자 아니면 403
|
||||
견본 표 · 로직 줄은 시험 안에 둠(임시 폴더 사본에 견본 표를 더함 — 진짜 마스터 값에 안 기댐).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
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 router_module
|
||||
from M01_MasterData import M01_MasterData_Store as store
|
||||
from M01_MasterData.LogicBuilder import M01_LB_Calc as lb
|
||||
|
||||
REAL = store.FOLDER
|
||||
OWN = "로직_자체.json"
|
||||
SAMPLE = "QF999999" # 견본 표 — 사본 폴더에만 있음
|
||||
TABLE = {
|
||||
"키": SAMPLE,
|
||||
"원문번호": "",
|
||||
"구분": "",
|
||||
"상세구분": "",
|
||||
"이름": "견본 쌓기 표",
|
||||
"기준": "1㎡",
|
||||
"출처": "자체",
|
||||
"용도": {"공종": "견본", "대상": ["인력"], "로직키": []},
|
||||
"조건": {"돌": "고르기"},
|
||||
"값칸": {"석공": "인"},
|
||||
"줄": [{"돌": "깬돌", "석공": 0.5}, {"돌": "견치돌", "석공": 0.7}],
|
||||
"주": [],
|
||||
}
|
||||
|
||||
|
||||
def _labor() -> tuple[str, Decimal]:
|
||||
rows = json.loads((store.FOLDER / "인력.json").read_text(encoding="utf-8"))["줄"]
|
||||
row = next(r for r in rows if r.get("상태") == "공표" and isinstance(r.get("값"), (int, float)))
|
||||
return row["키"], Decimal(str(row["값"]))
|
||||
|
||||
|
||||
def _row(labor: str) -> dict:
|
||||
"""견본 로직 — 입력 「돌」 로 견본 표에서 석공 품을 찾아 노임에 곱함."""
|
||||
return {
|
||||
"키": "",
|
||||
"원문번호": "",
|
||||
"이름": "견본 쌓기",
|
||||
"결과단위": "원/㎡",
|
||||
"출처": "자체",
|
||||
"입력": [{"이름": "돌", "고르기": ["깬돌", "견치돌"]}],
|
||||
"중간": [],
|
||||
"호표": [
|
||||
{
|
||||
"종류": "인력",
|
||||
"요소": labor,
|
||||
"이름": "석공",
|
||||
"단위": "인",
|
||||
"수량": f"찾기({SAMPLE}, 돌=돌).석공",
|
||||
"비목": "노무비",
|
||||
}
|
||||
],
|
||||
"덧줄": [],
|
||||
"끝수": None,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def folder(tmp_path: Path, monkeypatch) -> Path:
|
||||
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, tmp_path / path.name)
|
||||
host = sorted(tmp_path.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")
|
||||
monkeypatch.setattr(store, "FOLDER", tmp_path)
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(folder: Path) -> TestClient:
|
||||
app = FastAPI()
|
||||
app.include_router(router_module.router)
|
||||
app.include_router(lb.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 test_고친_표로_값이_바뀌고_캐시와_파일은_그대로(client: TestClient, folder: Path) -> None:
|
||||
labor, price = _labor()
|
||||
row = _row(labor)
|
||||
before = {p.name: p.read_bytes() for p in folder.glob("*.json")}
|
||||
|
||||
plain = _post(client, "/api/m01/lb/calc", {"row": row, "inputs": {"돌": "깬돌"}})
|
||||
assert plain["ok"] is True
|
||||
assert Decimal(str(plain["sums"]["계"])) == Decimal("0.5") * price
|
||||
|
||||
edited = copy.deepcopy(TABLE)
|
||||
edited["줄"][0]["석공"] = 0.8 # 깬돌 줄 품 0.5 → 0.8
|
||||
body = {"row": row, "inputs": {"돌": "깬돌"}, "tables": {SAMPLE: edited}}
|
||||
changed = _post(client, "/api/m01/lb/calc", body)
|
||||
assert Decimal(str(changed["sums"]["계"])) == Decimal("0.8") * price
|
||||
assert changed["lines"][0]["수량"] == 0.8
|
||||
|
||||
# 전역 캐시 · 파일 · 같은 요청 뒤 표 없는 셈은 그대로
|
||||
files = store.loaded()[0].values()
|
||||
cached = next(
|
||||
t for d in files if d.get("그룹") == "소요량" for t in d["표"] if t["키"] == SAMPLE
|
||||
)
|
||||
assert cached["줄"][0]["석공"] == Decimal("0.5")
|
||||
assert {p.name: p.read_bytes() for p in folder.glob("*.json")} == before
|
||||
again = _post(client, "/api/m01/lb/calc", {"row": row, "inputs": {"돌": "깬돌"}})
|
||||
assert again == plain
|
||||
|
||||
# 키는 못 바꿈 — 사본 표의 「키」 가 달라도 그 자리 키로 셈
|
||||
_post(client, "/api/m01/lb/calc", {**body, "tables": {SAMPLE: {**edited, "키": "QF000001"}}})
|
||||
|
||||
|
||||
def test_표_없이_부르면_calc_와_같고_없는_표는_404(client: TestClient) -> None:
|
||||
labor, _ = _labor()
|
||||
row = _row(labor)
|
||||
for stone in ("깬돌", "견치돌"):
|
||||
mine = _post(client, "/api/m01/lb/calc", {"row": row, "inputs": {"돌": stone}})
|
||||
theirs = _post(client, "/api/m01/calc", {"key": "", "inputs": {"돌": stone}, "row": row})
|
||||
assert mine == theirs
|
||||
|
||||
stop = _post(client, "/api/m01/lb/calc", {"row": row, "inputs": {"돌": "호박돌"}})
|
||||
assert stop["ok"] is False and stop == _post(
|
||||
client, "/api/m01/calc", {"key": "", "inputs": {"돌": "호박돌"}, "row": row}
|
||||
)
|
||||
_post(client, "/api/m01/lb/calc", {"row": row, "tables": {"QF999998": TABLE}}, want=404)
|
||||
|
||||
|
||||
def test_사용처_용도와_로직_글(client: TestClient, folder: Path) -> None:
|
||||
labor, _ = _labor()
|
||||
made = _post(client, "/api/m01/logic/new", {"logic": _row(labor)})
|
||||
first = next(
|
||||
r["키"] for p in sorted(folder.glob("로직_*.json")) if p.name != OWN
|
||||
for r in json.loads(p.read_text(encoding="utf-8"))["줄"]
|
||||
) # fmt: skip
|
||||
host = sorted(folder.glob("소요량_*.json"))[0]
|
||||
data = json.loads(host.read_text(encoding="utf-8"))
|
||||
next(t for t in data["표"] if t["키"] == SAMPLE)["용도"]["로직키"] = [first, made["key"]]
|
||||
host.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
store.cache.stale([host.name])
|
||||
|
||||
res = client.get("/api/m01/lb/table-users", params={"key": SAMPLE})
|
||||
assert res.status_code == 200, res.text
|
||||
users = {u["key"]: u["via"] for u in res.json()["users"]}
|
||||
assert users == {first: "용도", made["key"]: "둘"}
|
||||
assert client.get("/api/m01/lb/table-users", params={"key": "QF999998"}).status_code == 404
|
||||
|
||||
|
||||
def test_시스템_관리자가_아니면_403(folder: Path) -> None:
|
||||
app = FastAPI()
|
||||
app.include_router(
|
||||
lb.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.post("/api/m01/lb/calc", json={"row": {}}).status_code == 403
|
||||
assert client.get("/api/m01/lb/table-users", params={"key": SAMPLE}).status_code == 403
|
||||
app.dependency_overrides[verify_session] = lambda: {"role": "SYSTEM_ADMIN", "company_id": None}
|
||||
assert client.get("/api/m01/lb/table-users", params={"key": SAMPLE}).status_code == 200
|
||||
Reference in New Issue
Block a user