feat(M02): 개인 자재 파일 · 목록 · 추가 길 · 품번 PM 전체 일련(148-2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015sDZSjFvoiDjdGpgj3muc3

M02_MasterTemplete/M02_Labor_Router.py

@@ -2,7 +2,8 @@
 
 문서 읽기 · 쓰기는 양식 길(`GET/PUT /api/m02/layers/{층}/templates/labor/{열}`) 그대로 ·
 여기는 저장하지 않는 셈만 + 서고(단가승격 · 사전조합 · 50-3) 새로 · 복사(키는 서버가 줌 · 고침 ·
-지우기는 양식 길 그대로). 층 권한 = `Router_Items._root`(`M02_Layer_Access` 한 벌).
+지우기는 양식 길 그대로) + 개인 자재 목록 · 추가(148-2 · 개인 층만 · 품번은 서버).
+층 권한 = `Router_Items._root`(`M02_Layer_Access` 한 벌).
 ⚠ 등록은 `main.py`(로그인만).
 """
 
@@ -18,6 +19,7 @@ from M02_MasterTemplete import M02_Labor_Doc as labor
 from M02_MasterTemplete import M02_Labor_Ref as refs
 from M02_MasterTemplete import M02_Layer_Access as access
 from M02_MasterTemplete import M02_MasterTemplete_Store as store
+from M02_MasterTemplete import M02_Material_Own as own
 from M02_MasterTemplete.M02_MasterTemplete_Router_Items import Layer, Session, _call, _root
 
 router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Labor"])
@@ -101,3 +103,30 @@ def new_library(
     if doc is None:
         raise HTTPException(status_code=404, detail=f"없는 {refs.TITLE[kind]} 「{ref['키']}」")
     return _call(folder, _copy, kind, doc, (got.layer, got.owner), ref)
+
+
+@router.get("/layers/{layer}/materials")
+def own_materials(
+    layer: Layer,
+    owner: str | None = None,
+    session: Session = Depends(verify_session),
+) -> dict:
+    """개인 자재 목록(148-2) — 개인 층만 · 회사 · 마스터 = 빈 목록."""
+    if layer != "personal":
+        return {"줄": []}
+    return {"줄": own.rows(_root(session, layer, owner))}
+
+
+@router.post("/layers/{layer}/materials")
+def new_material(
+    layer: Layer,
+    body: dict[str, Any],
+    owner: str | None = None,
+    session: Session = Depends(verify_session),
+) -> dict:
+    """신규 자재(148-2) — `{이름, 규격, 단위, 값}` → 새 줄(품번 `PM` + 6자리 서버 일련) ·
+    개인 층만."""
+    if layer != "personal":
+        raise HTTPException(status_code=400, detail="신규 자재는 개인 층에서만")
+    folder = _root(session, layer, owner, write=True)
+    return _call(folder, own.add, folder, body)

M02_MasterTemplete/M02_Material_Own.py

@@ -0,0 +1,87 @@
+"""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

resources/tester/test_m02_material_own.py

@@ -0,0 +1,99 @@
+"""M02 개인 자재(PLAN 148) — 개인 층 파일 `개인자재.json` · 서버 길(목록 · 추가) ·
+품번 `PM` + 6자리.
+
+- 품번 = 모든 사람 통틀어 한 줄 번호(두 사람 · 두 회사 안 겹침) · 화면이 보낸 키는 안 받음
+- 품명 · 단위 빈 = 400 · 단가 = 0 이상 수 또는 빈 값 · 같은 품명 + 규격 + 단위 = 409
+- 개인 층만(회사 = 빈 목록 · 추가 400) · 남의 개인 층 = 403
+층은 임시 폴더(마스터 · storage).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from common_util.common_util_auth import verify_session
+from config import config_system
+from M02_MasterTemplete import M02_Labor_Router as labor_router
+from M02_MasterTemplete import M02_MasterTemplete_Store as store
+from M02_MasterTemplete import M02_Material_Own as own
+from M02_MasterTemplete import M02_Template_Layers as layers
+from resources.tester.test_m02_structure import write_sample_table
+
+URL = "/api/m02/layers/{}/materials"
+BOLT = {"이름": "앵커볼트", "규격": "M16", "단위": "EA", "값": 1200}
+
+
+@pytest.fixture()
+def world(tmp_path, monkeypatch) -> dict[str, Any]:
+    system = tmp_path / "system"
+    system.mkdir()
+    write_sample_table(system)
+    monkeypatch.setattr(store, "FOLDER", system)
+    monkeypatch.setattr(layers, "SYSTEM_ROOT", system)
+    (tmp_path / "storage").mkdir()
+    monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(tmp_path / "storage"))
+    session: dict[str, Any] = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False}
+    app = FastAPI()
+    app.include_router(labor_router.router)
+    app.dependency_overrides[verify_session] = lambda: session
+    return {"client": TestClient(app), "session": session}
+
+
+def _as(world, user: int, company: int = 7) -> TestClient:
+    world["session"].update(user_id=user, company_id=company)
+    return world["client"]
+
+
+def test_추가하면_PM_6자리_키_목록에_보임_화면_키_무시(world):
+    c = _as(world, 42)
+    got = c.post(URL.format("personal"), json={**BOLT, "키": "MT000008"})
+    assert got.status_code == 200, got.text
+    row = got.json()
+    assert row == {"키": "PM000001", "이름": "앵커볼트", "규격": "M16", "단위": "EA", "값": 1200}
+    assert c.get(URL.format("personal")).json()["줄"] == [row]
+    path = layers.personal_dir(7, 42) / own.FILE
+    assert path.is_file()
+    assert own.read(path.parent)["그룹"] == "재료"
+
+
+def test_두_사람_두_회사_번호_안_겹침(world):
+    a = _as(world, 42).post(URL.format("personal"), json=BOLT).json()["키"]
+    b = _as(world, 43).post(URL.format("personal"), json=BOLT).json()["키"]
+    c = _as(world, 50, company=8).post(URL.format("personal"), json=BOLT).json()["키"]
+    d = _as(world, 42).post(URL.format("personal"), json={**BOLT, "규격": "M20"}).json()["키"]
+    assert [a, b, c, d] == ["PM000001", "PM000002", "PM000003", "PM000004"]
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        {**BOLT, "이름": " "},
+        {**BOLT, "단위": ""},
+        {**BOLT, "값": -1},
+        {**BOLT, "값": "abc"},
+        {**BOLT, "값": True},
+    ],
+)
+def test_빈_품명_단위_틀린_단가_400(world, body):
+    assert _as(world, 42).post(URL.format("personal"), json=body).status_code == 400
+
+
+def test_단가_빈_값_받음_같은_자재_409(world):
+    c = _as(world, 42)
+    row = c.post(URL.format("personal"), json={**BOLT, "값": ""}).json()
+    assert row["값"] is None
+    assert c.post(URL.format("personal"), json={**BOLT, "값": 5}).status_code == 409
+
+
+def test_회사_층은_빈_목록_추가_400_남의_개인_403(world):
+    c = _as(world, 42)
+    c.post(URL.format("personal"), json=BOLT)
+    assert c.get(URL.format("company")).json() == {"줄": []}
+    assert c.post(URL.format("company"), json=BOLT).status_code == 400
+    other = _as(world, 43)
+    assert other.get(URL.format("personal"), params={"owner": "7/42"}).status_code == 403
+    assert other.post(URL.format("personal") + "?owner=7/42", json=BOLT).status_code == 403