Files
Aislo/M02_MasterTemplete/DataFlow/M02_DF_Router.py
T
eomsangdonandClaude Opus 5.5 3336902c17 feat(M01): 흐름 로직 입력 약속 하나 — 저장 때 「입력」 은 흐름의 설계 때 정할 것으로 다시 씀 · 지우는 것은 계산 칸만 · 시험판 길 지움
- 흐름으로 저장 = 계산 칸(중간 · 호표 · 결과 · 덧줄)만 지우고 「입력」(이름 · 설명 · 단위 · 고르기 · 범위)은 흐름에서 같은 꼴로 · 조합 · 구조물 · 미리 보기 · 다시 적용은 「입력」 을 그대로 읽음(따로 갈래 없음)
- 엔진 갈래: 안 준 입력 = 멈춤 · 빈 값으로 준 입력 · 단가 없는 품목 · 표에 없는 조건 = 그 값만 비움(알림) · 결과 꼴 = 돈 출력 줄 · 값들 · 알림
- 프로젝트에서 고른 자재 = 흐름 재료 자리 이름으로(고름 · 다시 적용 때 살림)
- 시험판 길 지움: /list · /get · /save · /delete · /user-elements(서버) · cfList · cfGet · cfSave · cfDelete · cfUserElement*(Api) · 빈 _data 폴더
- 시험: 조합에 담은 로직을 흐름으로 바꿔도 입력 목록 · 로직별 결과 · 비목 합 · 단가 묶음 셈이 같음

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL
2026-10-02 09:57:30 +09:00

137 lines
5.4 KiB
Python

"""단가산출 흐름도 서버 길 — 입력 값 · 다른 로직 값 · 설계 때 정할 것 · 자재 찾기 · 표준 표.
POST /api/m02/df/values {items, userElements?} → {values: CfAnswer[]}
POST /api/m02/df/logic-value {key, designs} → {key, outputs, cost, errors}
GET /api/m02/df/design-inputs?key= → {key, designs: CfDesign[]}
GET /api/m02/df/materials?q=&sub=&detail=&page=&size= → {total, page, size, items}
GET · PUT · DELETE /api/m02/df/standards → {unit, money, saved}
모양 = sub7 Model `M02_DF_Model.ts` CfLogic.
흐름 저장 = 로직 줄 「흐름」 칸(M01 `/save` · `M01_MasterData_Store_Flow`) — 이 길은 쓰지 않음.
다른 로직 = 마스터 로직 키(흐름이면 흐름 셈 · 아니면 지금 계산 · `M01_MasterData_Store_Flow.load`).
사용자 요소 = 흐름 안 `userElements` 만. 권한 = 시스템 관리자(`main.py` 한 줄).
"""
from __future__ import annotations
import json
import os
from pathlib import Path
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 import M01_MasterData_Store_Flow as sflow
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
router = APIRouter(prefix="/api/m02/df", tags=["M02 DataFlow"])
HERE = Path(__file__).resolve().parent
STANDARD = HERE / "_standard" # 단위 · 금액 표준 표 — 고치면 이 파일에 바로(git 이 따라감)
class ValuesBody(BaseModel):
items: list[dict[str, Any]]
userElements: list[dict[str, Any]] = []
class LogicValueBody(BaseModel):
key: str
designs: dict[str, Any] = {}
class StandardsBody(BaseModel):
unit: dict[str, Any]
money: dict[str, Any]
# ── 다른 로직(`M02_DF_Compute`) ────────────────────────────────────────
def _load(key: str) -> dict:
"""다른 로직 입력이 부르는 마스터 로직 — 흐름이면 흐름 · 아니면 지금 계산(정함 11)."""
return sflow.load(key)
def logic_value(key: str, designs: dict) -> dict:
"""다른 로직 값 — 받은 설계 값(없으면 예시값)으로 셈 · 출력 이름 → 값."""
other = _load(key)
if callable(other.get("calc")): # 흐름 없는 로직 — 노무비 · 재료비 · 경비 · 계
got = other["calc"](designs)
return {"key": key, "outputs": got["outputs"], "cost": {}, "errors": got["errors"]}
local = other.get("userElements") or []
got = dc.run(other, dc._design_values(other, designs), _load, local, [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 꼴."""
other = _load(key)
if callable(other.get("calc")):
return {"key": key, "designs": sflow.old_designs(key)}
return {"key": key, "designs": dc.design_inputs(other)}
# ── 표준 표 ────────────────────────────────────────────────────────────
def _standards_put(body: dict) -> dict:
"""고친 표준 표 = 본 프로그램 값 — git 이 따라가는 `_standard/` 파일에 바로(시스템 관리자)."""
dv.standards_check(body)
for part, name in dv.STANDARDS.items():
tmp = STANDARD / f".{name}.tmp"
text = json.dumps(body[part], ensure_ascii=False, indent=2) + "\n"
tmp.write_text(text, encoding="utf-8", newline="\n")
os.replace(tmp, STANDARD / name)
return dv.standards(STANDARD)
def _standards_reset() -> dict:
"""고친 값을 따로 두지 않음(바로 저장) — 되돌릴 처음 값 없음 · 지금 값 그대로."""
return dv.standards(STANDARD)
# ── 길 ────────────────────────────────────────────────────────────────
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("/values")
def post_values(body: ValuesBody) -> dict:
return _call(dv.values, body.items, body.userElements)
@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)
@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)
@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)