Aislo/M02_MasterTemplete/M02_Material_Own.py

"""M02 개인 자재(PLAN 148-2) — 개인 층 뿌리 한 파일 `개인자재.json`(종류 폴더 밖 · 양식 목록 ·
프로젝트 복사 · 층 복사에 안 섞임 · `구조물집계표조합.json` 꼴).

꼴 = M01 자재 줄 꼴 `{그룹: 재료, 원문: 개인, 줄: [{키, 이름, 규격, 단위, 값}]}` —
사본 `단가.json` 에 그대로 넣으면 `mf.Master` 가 읽음(MT 아닌 키 = 줄 「값」 ·
`master_material.element`). 품번 = `PM` + 6자리 — 모든 사람 통틀어 한 줄 번호(모든 개인 층 파일
가장 큰 번호 + 1 · `store._LOCK` 안 · 화면이 보낸 키는 안 받음). 신규 자재는 개인 층에서만
(회사 · 마스터 층 = 마스터 자재만).
"""

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

from common_util.common_util_json import atomic_write_json
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Template_Layers as layers

FILE = "개인자재.json"
PREFIX = "PM"
KEY = re.compile(rf"^{PREFIX}(\d{{6}})$")
COLS = ("이름", "규격", "단위")


def _blank() -> dict[str, Any]:
    return {"그룹": "재료", "원문": "개인", "줄": []}


def read(folder: str | Path) -> dict[str, Any]:
    """층 폴더의 개인 자재 묶음 — 없거나 못 읽으면 빈 묶음."""
    path = Path(folder) / FILE
    try:
        got = json.loads(path.read_bytes()) if path.is_file() else None
    except (OSError, ValueError):
        got = None
    return got if isinstance(got, dict) and isinstance(got.get("줄"), list) else _blank()


def rows(folder: str | Path) -> list[dict[str, Any]]:
    return read(folder)["줄"]


def _last() -> int:
    """모든 개인 층 파일의 가장 큰 PM 번호 — 사람 사이 번호 안 겹침."""
    found = [
        int(m.group(1))
        for path in layers.storage_root().glob(f"*/*/{layers.TEMPLATES_DIRNAME}/{FILE}")
        for row in read(path.parent)["줄"]
        if (m := KEY.match(str(row.get("키") or "")))
    ]
    return max(found, default=0)


def _text(v: Any) -> str:
    return str(v).strip() if v is not None else ""


def _value(v: Any) -> float | int | None:
    if v is None or (isinstance(v, str) and not v.strip()):
        return None
    try:
        got = float("nan") if isinstance(v, bool) else float(v)
    except (TypeError, ValueError):
        got = float("nan")
    if not 0 <= got < float("inf"):  # NaN 도 여기서 걸림
        raise store.StoreError(400, "단가는 0 이상 수 또는 빈 값")
    return int(got) if got.is_integer() else got


def add(folder: str | Path, body: dict[str, Any]) -> dict[str, Any]:
    """새 개인 자재 한 줄 → 그 줄(키는 서버) · 품명 · 단위 빈 = 400 ·
    같은 품명 + 규격 + 단위 = 409."""
    row = {k: _text(body.get(k)) for k in COLS}
    if not row["이름"] or not row["단위"]:
        raise store.StoreError(400, "품명 · 단위는 꼭 적음")
    row["값"] = _value(body.get("값"))
    with store._LOCK:
        doc = read(folder)
        if any(all(_text(x.get(k)) == row[k] for k in COLS) for x in doc["줄"]):
            raise store.StoreError(409, f"이미 있는 자재 「{row['이름']} {row['규격']}」")
        got = {"키": f"{PREFIX}{_last() + 1:06d}", **row}
        doc["줄"].append(got)
        atomic_write_json(Path(folder) / FILE, doc)
    return got


# ── 148-3 — 저장 검사 · 셈 · 사본 · 층 사이 ─────────────────────────────
BUNDLE = "재료_개인자재.json"  # 사본 `단가.json` 안 이름


def is_own(code: str) -> bool:
    return bool(KEY.match(code))


def _personal(folder: Path) -> bool:
    from M02_MasterTemplete import M02_Labor_Ref as refs

    return refs.where(folder)[0] == "personal"


def find(code: str, folder: str | Path | None = None) -> dict[str, Any] | None:
    """개인 자재 한 줄 — `folder`(없으면 지금 층)가 개인 층일 때만."""
    folder = Path(folder) if folder is not None else store.root()
    if not is_own(code) or not _personal(folder):
        return None
    return next((r for r in rows(folder) if r.get("키") == code), None)


def codes(doc: Any) -> list[str]:
    """일위대가 문서 사용자 자재의 개인 자재 코드."""
    mats = doc.get("자재") if isinstance(doc, dict) else None
    return [c for m in mats or [] if isinstance(m, dict) and is_own(c := str(m.get("코드") or ""))]


def bundle(wanted: list[str], folder: str | Path) -> dict[str, Any] | None:
    """사본 `단가.json` 에 넣을 개인 자재 묶음(원본 층 목록에서 쓰는 줄만) — 없으면 None."""
    got = [r for r in rows(folder) if r.get("키") in set(wanted)]
    return {**_blank(), "줄": got} if got else None


def adopt(found: list[dict[str, Any]], folder: str | Path | None = None) -> None:
    """남의 줄을 같은 키로 내 목록에 — 이미 있는 키는 그대로 · 개인 층 아니면 400."""
    folder = Path(folder) if folder is not None else store.root()
    if not found:
        return
    if not _personal(folder):
        raise store.StoreError(400, "개인 자재는 개인 층에서만 — " + found[0].get("키", ""))
    with store._LOCK:
        doc = read(folder)
        have = {r.get("키") for r in doc["줄"]}
        new = [{k: r.get(k) for k in ("키", *COLS, "값")} for r in found if r.get("키") not in have]
        if new:
            doc["줄"] += new
            atomic_write_json(folder / FILE, doc)


def carry(doc: Any, source: str | Path) -> None:
    """층 사이 복사(`Store._server`) — 원본 개인 층 자재를 받는 층(지금 층) 목록에 같은 키로."""
    wanted = codes(doc)
    if wanted and Path(source) != store.root():
        adopt([r for r in rows(source) if r.get("키") in set(wanted)])