Aislo/M02_MasterTemplete/DataFlow/M02_DF_Router.py

"""단가산출 다이어그램 시험 길(PLAN 35-7) — 목록 · 열기 · 저장 · 지우기 · 입력 값 · 사용자 요소.

    GET  /api/m02/df/list                     → {logics: [{key, name, resultUnit, saved, sample}]}
    GET  /api/m02/df/get?key=DF-0001          → {logic, sample}
    POST /api/m02/df/save     {logic}         → {key, logic, values: {ports, outputs, cost, errors}}
    POST /api/m02/df/delete   {key}
    POST /api/m02/df/values   {items, userElements?} → {values: CfAnswer[]}
    GET  /api/m02/df/user-elements            → {elements}
    POST /api/m02/df/user-elements {op: add | delete, element? | id?}

모양 = sub7 Model `M02_DF_Model.ts`
(CfLogic = key · name · resultUnit · boxes · links · userElements).
알약 이름(브레인 정함): 입력 상자가 내보내는 알약 = 요소 name · 표 「표」 · 변수 name /
함수 상자 받는 알약 = inputs[].port · 내보내는 알약 = output.name / 출력 상자 받는 알약 = 「값」.
저장 = 입력 값(노임 · 재료 · 유가전력 · 기계 · 환율 · 요율 · 소요량 / 계수 표 · 사용자 요소)을
마스터(사용자 요소는 다이어그램 안 · `_data`)에서 다시 받아 → 함수 상자 시트들을 한 통합문서로
묶어(받는 칸에 값 · 함수 → 함수는 시트 사이 참조) 우리 스프레드시트 엔진(Node ·
`common_util_spreadsheet`)으로 셈 → 받는 칸에 값만 되돌려 넣어 `_data` 에 씀 · 브라우저 값은
안 받아 적음. 단가는 기존 `/calc` 와 같은 셈.
자리 = 이 폴더 `_data/`(git 밖) · 견본 `_sample/`(git 안 · 읽기만 · 저장하면 새 번호) ·
키 `DF-0001` · 사용자 요소 `UE-0001` · 마스터 파일은 안 씀. 권한 = 시스템 관리자(`main.py` 한 줄).
옛 시험판(SheetLogic) 서버를 옮겨 고쳐 씀 — 그 폴더를 import 하지 않음.
정함 = `docs/raw/M01_쉬운단가/원가다이어그램_정함.md`.
"""

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 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 (시험)"])

HERE = Path(__file__).resolve().parent
DATA = HERE / "_data"  # 시험은 임시 폴더로 바꿈
SAMPLE = HERE / "_sample"
USERS = "_user_elements.json"  # 사용자 요소 공용 목록 — `_data` 안
_KEY = re.compile(r"^DF-(\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 = ("인력", "재료", "유가전력", "기계", "소요량", "계수", "환율", "요율")
TABLE_PORT, OUTPUT_PORT = "표", "값"
COSTS = ("노무비", "재료비", "경비")


class LogicBody(BaseModel):
    logic: dict[str, Any]


class KeyBody(BaseModel):
    key: str


class ValuesBody(BaseModel):
    items: list[dict[str, Any]]
    userElements: list[dict[str, Any]] = []


class UserBody(BaseModel):
    op: str
    element: dict[str, Any] | None = None
    id: 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 listing() -> dict:
    out = []
    for path, sample in [
        *((p, False) for p in sorted(DATA.glob("DF-*.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}


# ── 사용자 요소(Model CfUserElement — id · group · name · spec · unit · value) ─────
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, uid: str | None) -> dict:
    rows = _users()
    if op == "delete":
        if not any(r["id"] == uid for r in rows):
            raise store.StoreError(404, f"없는 사용자 요소 「{uid}」")
        _write(DATA / USERS, [r for r in rows if r["id"] != uid])
        return {"id": uid}
    if op != "add" or not isinstance(element, dict):
        raise store.StoreError(400, "op 는 add(element) 또는 delete(id)")
    name = str(element.get("name") 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("value")
    if value is not None and (isinstance(value, bool) or not isinstance(value, (int, float))):
        raise store.StoreError(400, "값은 수 또는 빈 값")
    used = [int(m.group(1)) for r in rows if (m := _UKEY.match(r["id"]))]
    one = {
        "id": f"UE-{max(used, default=0) + 1:04d}",
        "group": group,
        "name": name,
        "spec": str(element.get("spec") or ""),
        "unit": str(element.get("unit") or ""),
        "value": value,
    }
    _write(DATA / USERS, [*rows, one])
    return {"element": one}


# ── 저장 · 계산(`M02_DF_Compute`) ─────────────────────────────────────
def _load(key: str) -> dict:
    """다른 로직 입력이 부르는 로직 — 내 것(`_data`) · 견본(`_sample`)."""
    logic, _ = _read(key)
    return {**logic, "key": key}


def _local(logic: dict) -> list[dict]:
    return [*(logic.get("userElements") or []), *_users()]


def save(logic: dict) -> dict:
    """저장 — 입력 값 다시 받기 → 함수 상자 시트를 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):
        raise store.StoreError(400, "상자(boxes) · 선(links) 이 없음")
    key = logic.get("key") or None
    if key and _sample(key) is not None:
        key = None  # 견본은 그대로 두고 새 번호로
    elif key:
        _read(key)  # 있는 것만 고침 — 없으면 404
    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 = {**got.pop("filled"), "key": key, "saved": datetime.now().isoformat(timespec="seconds")}
    _write(DATA / f"{key}.json", _num(done))
    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))}


# ── 길 ────────────────────────────────────────────────────────────────
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(dv.values, body.items, [*body.userElements, *_users()])


@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.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)