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 copy
import json
import os
import re
from datetime import datetime
from decimal import Decimal
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
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}


# ── 저장 · 계산 ────────────────────────────────────────────────────────
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 _style(cells: dict, cell: str) -> dict:
    return {"서식": cells[cell]["서식"]} if "서식" in (cells.get(cell) or {}) else {}


def _put(cells: dict, cell: str, value: Any) -> None:
    """받는 칸 하나 — 값 칸으로 덮어씀(수식이 있었어도) · 빈 값은 값만 뺌(서식은 둠)."""
    keep = _style(cells, cell)
    if value is None and not keep:
        cells.pop(cell, None)
    else:
        cells[cell] = {**({"값": value} if value is not None else {}), **keep}


def _place(cells: dict, cell: str, value: Any) -> None:
    """값 하나 · 표 블록(cell 이 왼위) · 수식 `{"식": …}` 을 받는 칸에."""
    m = _CELL.match(str(cell or "").upper())
    if not m:
        raise store.StoreError(400, f"칸 주소 모양이 아님 「{cell}」")
    col, row, at = _col(m.group(1)), int(m.group(2)), f"{m.group(1)}{m.group(2)}"
    if isinstance(value, dict) and "식" in value:
        cells[at] = {**value, **_style(cells, at)}
    elif isinstance(value, list):
        for dy, line in enumerate(value):
            for dx, one in enumerate(line):
                _put(cells, f"{_letters(col + dx)}{row + dy}", one)
    else:
        _put(cells, at, value)


def _plain(v: Any) -> Any:
    """엔진 계산값 한 칸 → 수 · 글 · 참거짓 · 빈 값(오류는 None)."""
    if not isinstance(v, dict):
        return v
    if "수" in v:
        return Decimal(str(v["수"]))
    if "글" in v:
        return v["글"]
    if "참" in v:
        return v["참"]
    return None


def _source(links: list[dict], box: str, port: str) -> dict | None:
    return next((ln["from"] for ln in links if ln.get("to") == {"box": box, "name": port}), None)


def _input_ports(box: dict, local: list[dict], errors: list[str]) -> dict[str, Any]:
    """입력 상자 알약 값 — 요소 name · 표 「표」 · 변수 name(예시값)."""
    inp, out = box.get("input") or {}, {}
    if inp.get("mode") == "group":
        group = inp.get("group")
        asks = [
            (el.get("name"), {"ref": el.get("ref"), "unit": el.get("unit")})
            for el in inp.get("elements") or []
        ]
        if inp.get("table"):
            asks.append((TABLE_PORT, {"table": inp["table"]}))
        for name, ask in asks:
            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')}")
    for v in inp.get("variables") or []:
        out[v.get("name")] = v.get("example")
    return out


def compute(logic: dict, local: list[dict]) -> tuple[dict, dict]:
    """함수 상자 시트를 한 통합문서로 셈 → (값을 되돌려 넣은 다이어그램, 계산 결과)."""
    boxes = {b.get("id"): b for b in logic.get("boxes") or []}
    links = logic.get("links") or []
    errors: list[str] = []
    ports = {
        i: _input_ports(b, local, errors) for i, b in boxes.items() if b.get("kind") == "input"
    }
    funcs = [b for b in boxes.values() if b.get("kind") == "function" and b.get("func")]
    sheets, fills = [], []
    for b in funcs:  # 한 통합문서 — 시트 id · 이름 = 상자 id
        f = b["func"]
        sheet = {**copy.deepcopy(f.get("sheet") or {}), "id": b["id"], "이름": b["id"]}
        cells = sheet.setdefault("칸", {})
        for v in f.get("variables") or []:
            if v.get("cell"):
                _place(cells, v["cell"], v.get("example"))
        for p in f.get("inputs") or []:
            src = _source(links, b["id"], p.get("port"))
            if src is None or src.get("box") not in boxes:
                continue
            up = boxes[src["box"]]
            if up.get("kind") == "function":
                at = ((up.get("func") or {}).get("output") or {}).get("cell") or ""
                _place(cells, p["cell"], {"식": f"'{up['id']}'!{at}"})
                fills.append((b["id"], p["cell"], ("ref", up["id"], at)))
            else:
                value = ports.get(up["id"], {}).get(src.get("name"))
                _place(cells, p["cell"], value)
                fills.append((b["id"], p["cell"], ("value", value, None)))
        sheets.append(sheet)
    top = [c.get("서식", 0) for s in sheets for c in s["칸"].values() if isinstance(c, dict)]
    book = _num(
        {"종류": "통합문서", "판": 1, "서식": [{}] * (int(max(top, default=0)) + 1), "시트": sheets}
    )
    try:
        calc = spreadsheet.recalc_workbooks([book])[0] if sheets else {}
    except spreadsheet.RecalcError as e:
        raise store.StoreError(503, str(e)) from e

    def cell_value(box_id: str, cell: str) -> Any:
        hit = (calc.get(box_id) or {}).get(cell)
        if hit is None:
            sheet = next(s for s in book["시트"] if s["id"] == box_id)
            return (sheet["칸"].get(cell) or {}).get("값")
        if isinstance(hit, dict) and "오류" in hit:
            errors.append(f"「{boxes[box_id].get('name')}」 {cell} 칸 계산 오류 {hit.get('오류')}")
        return _plain(hit)

    for b in funcs:
        out = b["func"].get("output") or {}
        ports[b["id"]] = {out.get("name"): cell_value(b["id"], out.get("cell") or "")}
    outputs, cost = {}, {**dict.fromkeys(COSTS, Decimal(0)), "값들": {}}
    for b in boxes.values():  # 출력 상자 · 원가
        if b.get("kind") != "output":
            continue
        src = _source(links, b["id"], OUTPUT_PORT)
        value = ports.get(src["box"], {}).get(src.get("name")) if src else None
        outputs[b["id"]] = value
        kind = (b.get("output") or {}).get("cost")
        if (
            kind in COSTS
            and isinstance(value, (int, float, Decimal))
            and not isinstance(value, bool)
        ):
            cost[kind] += Decimal(str(value))
        elif kind == "값":
            cost["값들"][b.get("name")] = value
    cost["계"] = sum(cost[k] for k in COSTS)
    done = copy.deepcopy(logic)  # 받는 칸에 값만 되돌려 넣음 — 시트 사이 참조는 셈한 값으로
    for b in done.get("boxes") or []:
        if b.get("kind") != "function" or not b.get("func"):
            continue
        cells = b["func"].setdefault("sheet", {}).setdefault("칸", {})
        for v in b["func"].get("variables") or []:
            if v.get("cell"):
                _place(cells, v["cell"], v.get("example"))
        for box_id, cell, (how, a, at) in fills:
            if box_id == b["id"]:
                _place(cells, cell, cell_value(a, at) if how == "ref" else a)
    return done, {"ports": ports, "outputs": outputs, "cost": cost, "errors": errors}


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
    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}"
    done = {**done, "key": key, "saved": datetime.now().isoformat(timespec="seconds")}
    _write(DATA / f"{key}.json", _num(done))
    return {"key": key, "logic": done, "values": result}


# ── 길 ────────────────────────────────────────────────────────────────
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 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)