Aislo/M02_MasterTemplete/CostFlow/M02_CF_Router.py

"""단가산출 다이어그램 시험 길(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)