feat(M02): 층 인식 집계표 저장소 — 회사 · 개인 층 · 항목 목록 · 항목 복사 (PLAN 21-2)

- Store · Matset · Branch 가 지금 층 폴더(store.at · 기본 시스템 층)를 봄 — 시스템 층 동작 · 옛 길 그대로
- 회사 · 개인 층 첫 진입 때 빈 뼈대(집계표 열 0 · 조합 장부 다음 1 · 자재 조합 빈 표)
- /api/m02/layers/{layer}/… 집계표 길(번호 · 조합 장부 · 구조물 새로 · 복사 · 조합 갈라 내기 · 자재 조합) · 권한 = 시스템 SYSTEM_ADMIN · 회사 관리자 · 개인 본인
- 층 양식 저장(시스템 · 회사 · 개인)은 집계표 규칙(새 열 파일 · 번호 · SS) 거침 · 시스템 층 쓰기는 SYSTEM_ADMIN · 쓰는 양식 지우기 409
- GET /api/m02/items(한 층 · ?all=1 세 층) · POST /api/m02/layers/personal/items/copy · 회사 없음 409 「회사부터 등록」
- 시험 test_m02_layers_store

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EUypcnp5d1gU2aeKh9F2H7
This commit is contained in:
2026-09-29 12:01:08 +09:00
co-authored by Claude Opus 5.5
parent 1758ba0fb2
commit b8195aff2e
8 changed files with 645 additions and 25 deletions
@@ -0,0 +1,130 @@
"""M02 층 집계표 항목 — 회사 · 개인 층 빈 뼈대 · 항목 목록 · 항목 복사(PLAN 21-1 · 21-2).
항목 = 구조물집계표조합(SS) 한 줄 — 그 열의 일위대가(UA) · 자재조합(UM) ·
도번 SD · 면적번 SA · 재료번 SM.
층 폴더(시스템 · 회사 · 개인)는 모양이 같아 저장소 함수를 `store.at(층 폴더)` 안에서 그대로 부름.
권한은 부르는 쪽(`M02_MasterTemplete_Router_Items.py`) 몫 — 여기는 폴더와 파일만.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from common_util.common_util_json import atomic_write_json
from M02_MasterTemplete import M02_MasterTemplete_Matset as matset
from M02_MasterTemplete import M02_MasterTemplete_Store as store
# 21-1 목록 칸 — SS 줄에서 그대로(키는 `key`)
ITEM_FIELDS = (
"열",
"종류",
"분류1",
"분류2",
"분류3",
"단위",
"일위대가",
"자재조합",
"도번",
"면적번",
"재료번",
)
def ensure(folder: Path) -> bool:
"""회사 · 개인 층 첫 진입 — 빈 집계표(열 0 · 층 이름 · 줄 머리는 시스템 것) ·
조합 장부(`다음` 1) · 자재 조합 빈 표 · 종류 폴더.
이미 집계표가 있으면 아무것도 안 함(시스템 층은 늘 있음)."""
table = Path(folder) / "table" / f"{store.STRUCTURE_TABLE}.json"
if table.is_file():
return False
system = json.loads((store.FOLDER / "table" / f"{store.STRUCTURE_TABLE}.json").read_bytes())
doc = {
**system,
"열": [],
"줄": [{**r, "값": {}} for r in system.get("줄", []) if isinstance(r, dict)],
"합계줄": [],
"보기": {},
}
with store.at(folder):
for kind in store.KINDS:
(Path(folder) / kind).mkdir(parents=True, exist_ok=True)
with store._LOCK:
if table.is_file(): # 그 사이 다른 요청이 만듦
return False
atomic_write_json(table, doc)
store.sync_combos()
mat = Path(folder) / matset.FILE
if not mat.is_file():
atomic_write_json(mat, matset._load()[0])
return True
def items(folder: Path) -> list[dict[str, Any]]:
"""그 층 항목 — SS 줄마다 `{key, 열, 종류, 분류1~3, 단위, 일위대가, 자재조합,
도번, 면적번, 재료번}`."""
with store.at(folder):
rows = store.read_combos().get("줄", [])
return [{"key": r.get("키"), **{k: r.get(k) for k in ITEM_FIELDS}} for r in rows]
def _fresh_id(want: str, used: set[str]) -> str:
out, n = want, 2
while out in used:
out, n = f"{want}_{n}", n + 1
return out
def copy_item(source: Path, target: Path, key: str, origin: dict[str, Any]) -> dict[str, Any]:
"""다른 층 항목(SS `key`) → `target` 층 새 열 하나 — 머리 · 단위 · 일위대가 · 바인딩 그대로 ·
도면 · 면적산출 · 재료산출 내용(알약 연결 포함) · 자재 조합 줄(새 UM) 복사 ·
새 번호 · 새 SS 키 ·
열 `본뜬` = `origin`(원래 층 · 주인 · 키). 돌려줌 `{key, 열, 이름}`."""
with store.at(source):
row = next((r for r in store.read_combos().get("줄", []) if r.get("키") == key), None)
if row is None:
raise store.StoreError(404, f"없는 항목 「{key}」")
col_id = row["열"]
col = next(c for c in store._structure_columns() if c["id"] == col_id)
docs = {}
for kind, _, _, _ in store.PARTS:
try:
docs[kind] = store.read(kind, col_id)["문서"]
except store.StoreError:
pass # 없는 파일은 새 층에서 빈 문서로 섬
um = None
if col.get("자재조합"):
try:
um = matset.one(col["자재조합"])
except store.StoreError:
um = None # 가리키던 줄이 없으면 비움
ensure(target)
with store.at(target):
new_um = None
if um:
names = matset.listing()["줄"]
taken = {str(r.get("이름") or "") for r in names}
name, n = str(um.get("이름") or key), 2
base = name
while name in taken:
name, n = f"{base} ({n})", n + 1
got = {k: um.get(k) for k in ("담은자재", "소유", "비고")}
new_um = matset.create({**got, "이름": name, "본뜬키": None})["키"]
table = store.read("table", store.STRUCTURE_TABLE)
doc = table["문서"]
new_id = _fresh_id(col_id, {c.get("id") for c in doc.get("열", []) if isinstance(c, dict)})
made = {**json.loads(json.dumps(col)), "id": new_id, "자재조합": new_um, "본뜬": origin}
doc["열"] = [*doc.get("열", []), made]
store.write("table", store.STRUCTURE_TABLE, table["판"], doc) # 빈 세 파일 · 번호 · SS
for kind, got in docs.items():
number = store._KEY[kind]
have = store.read(kind, new_id)["문서"]
body = {k: v for k, v in got.items() if k != "계산값"}
body = {**body, "열": new_id, number: have.get(number)}
if kind in store.SHEETS:
body = store._server_values(body)
with store._LOCK:
atomic_write_json(store._path(kind, new_id), body)
ss = next(r for r in store.read_combos()["줄"] if r.get("열") == new_id)
return {"key": ss["키"], "열": new_id, "이름": " ".join(h for h in store._heads(made) if h)}
@@ -22,7 +22,7 @@ _HEAD = "UM"
def _load() -> tuple[dict[str, Any], str]:
"""(표, 판) — 파일이 없으면 빈 표 · 판 빈 글."""
path = store.FOLDER / FILE
path = store.root() / FILE
if not path.is_file():
return {"그룹": mmc.GROUP, "판": "2026", "비고": [], "다음": 1, "줄": []}, ""
raw = path.read_bytes()
@@ -95,7 +95,7 @@ def _save(doc: dict, version: str) -> str:
"""틀 검사 → 판이 같을 때만 씀 → 새 판."""
if errors := mmc.check_form(FILE, doc):
raise store.StoreError(422, {"errors": errors})
path = store.FOLDER / FILE
path = store.root() / FILE
with store._LOCK:
have = store.version_of(path.read_bytes()) if path.is_file() else ""
if have != (version or ""):
@@ -0,0 +1,263 @@
"""M02 층 집계표 API — `/api/m02/layers/{layer}/…` · 항목 목록 · 항목 복사(PLAN 21-1 · 21-2).
⚠ 등록은 `main.py`(로그인만) · 층 권한은 여기서(`Router_Layers._layer_dir` 와 같은 규칙):
system 읽기 = 로그인 누구나(항목 고르기) · 쓰기 = SYSTEM_ADMIN
company 읽기 = 같은 회사 · 쓰기 = 회사 관리자(ADMIN · 마스터 · SYSTEM_ADMIN)
personal 읽기 · 쓰기 = 본인(SYSTEM_ADMIN 은 누구나) · 회사 없음 = 쓰기 409 「회사부터 등록」
옛 길(`/api/m02/structures/…` 등 · `M02_MasterTemplete_Router.py`)은 시스템 층 별칭으로 그대로.
회사 · 개인 층은 처음 부를 때 빈 뼈대를 만듦(`Items.ensure`).
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Literal
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from common_util.common_util_auth import verify_session
from M02_MasterTemplete import M02_MasterTemplete_Branch as branch
from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
from M02_MasterTemplete import M02_MasterTemplete_Matset as matset
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_MasterTemplete_Router import (
BranchBody,
MaterialBranchBody,
MatsetBody,
UseComboBody,
)
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Items"])
Layer = Literal["system", "company", "personal"]
Session = dict[str, Any]
NO_COMPANY = "회사부터 등록 — 회사 없이는 회사 · 개인 구조물 집계표를 만들 수 없음"
class CopyItemBody(BaseModel):
from_layer: Literal["system", "company"]
from_owner: str | None = None
key: str
def _admin(session: Session) -> bool:
return session.get("role") == "SYSTEM_ADMIN"
def _company_admin(session: Session) -> bool:
return session.get("role") in ("ADMIN", "SYSTEM_ADMIN") or bool(session.get("is_master"))
def _owner(session: Session, layer: str, owner: str | None) -> str | None:
"""층 주인 글 — 회사 `{회사}` · 개인 `{회사}/{사용자}` · 시스템 None ·
남의 것은 SYSTEM_ADMIN 만."""
if layer == "system":
return None
company, user = session.get("company_id"), session.get("user_id")
mine = str(company) if layer == "company" else f"{company}/{user}"
if owner in (None, "", mine):
return None if company is None else mine
if not _admin(session):
raise HTTPException(status_code=403, detail="다른 회사 · 다른 사람의 구조물 집계표입니다.")
return owner
def _folder(layer: str, owner: str | None) -> Path:
if layer == "system":
return store.FOLDER
try:
if layer == "company":
return layers.company_dir(owner)
company, user = str(owner).split("/", 1)
return layers.personal_dir(company, user)
except ValueError as error:
raise HTTPException(status_code=400, detail=f"층 주인 모양 — {error}") from error
def _root(session: Session, layer: str, owner: str | None = None, write: bool = False) -> Path:
"""층 폴더(권한까지) — 회사 · 개인은 없으면 빈 뼈대를 만듦. 회사 없는 사람: 쓰기 409."""
if layer == "system":
if write and not _admin(session):
raise HTTPException(status_code=403, detail="마스터 템플릿은 시스템 관리자만 고침.")
return store.FOLDER
who = _owner(session, layer, owner)
if who is None:
raise HTTPException(status_code=409, detail=NO_COMPANY)
if write and layer == "company" and not _company_admin(session):
raise HTTPException(
status_code=403, detail="회사용은 회사 관리자만 고침 — 개인용으로 복사해서 씀."
)
folder = _folder(layer, who)
items_store.ensure(folder)
return folder
def _call(folder: Path, fn, *args):
with store.at(folder):
try:
return fn(*args)
except store.StoreError as e:
raise HTTPException(status_code=e.status, detail=e.detail) from e
# ── 항목 목록 · 복사 ───────────────────────────────────
def _listing(session: Session, layer: str, owner: str | None) -> dict[str, Any]:
who = None if layer == "system" else _owner(session, layer, owner)
if layer != "system" and who is None:
return {"layer": layer, "owner": None, "items": []} # 회사 없음 = 빈 그룹
folder = _root(session, layer, who)
return {"layer": layer, "owner": who, "items": items_store.items(folder)}
@router.get("/items")
def get_items(
layer: Layer = "system",
owner: str | None = None,
all: bool = False, # noqa: A002 — 21-1 약속 이름 `?all=1`
session: Session = Depends(verify_session),
):
"""21-1 — 한 층이면 `[{key, 열, 종류, 분류1~3, 단위, 일위대가, 자재조합, 도번, 면적번,
재료번}]` ·
`?all=1` 이면 세 층 `[{layer, owner, items}]`(회사 · 개인은 세션 것 · 회사 없으면 빈 목록)."""
if all:
return [_listing(session, one, None) for one in ("system", "company", "personal")]
return _listing(session, layer, owner)["items"]
@router.post("/layers/personal/items/copy")
def copy_item(body: CopyItemBody, session: Session = Depends(verify_session)) -> dict:
"""회사 · 마스터 항목 → 내 개인 층 새 항목(열 · 세 파일 · 자재 조합 줄 · 새 SS 키 · `본뜬`)."""
target = _root(session, "personal", None, write=True)
source = _root(session, body.from_layer, body.from_owner)
owner = (
None if body.from_layer == "system" else _owner(session, body.from_layer, body.from_owner)
)
origin = {"layer": body.from_layer, "owner": owner, "key": body.key}
try:
return items_store.copy_item(source, target, body.key, origin)
except store.StoreError as e:
raise HTTPException(status_code=e.status, detail=e.detail) from e
# ── 층 길 — 옛 시스템 길과 같은 모양 ─────────────────────
@router.get("/layers/{layer}/structures/numbers")
def get_numbers(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
return _call(_root(session, layer, owner), store.structure_numbers)
@router.get("/layers/{layer}/structure-combos")
def get_combos(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
return _call(_root(session, layer, owner), store.read_combos)
@router.post("/layers/{layer}/structures/{column}")
def post_structure(
layer: Layer, column: str, owner: str | None = None, session: Session = Depends(verify_session)
):
return _call(_root(session, layer, owner, write=True), store.create_structure, column)
@router.post("/layers/{layer}/structures/{column}/combo")
def post_combo(
layer: Layer,
column: str,
body: BranchBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
"""일위대가 조합 갈라 내기 — 새 UA 는 M01 시스템 마스터에 생기므로 SYSTEM_ADMIN 만."""
folder = _root(session, layer, owner, write=True)
if not _admin(session):
raise HTTPException(status_code=403, detail="일위대가 조합은 시스템 관리자만 새로 만듦.")
return _call(
folder, branch.branch_combo, column, body.본뜬키, body.담은로직, body.이름, body.판
)
@router.post("/layers/{layer}/structures/{column}/material-combo")
def post_material_combo_branch(
layer: Layer,
column: str,
body: MaterialBranchBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
folder = _root(session, layer, owner, write=True)
args = (column, body.본뜬키, body.담은자재, body.이름, body.판)
return _call(folder, branch.branch_material_combo, *args)
@router.post("/layers/{layer}/structures/{column}/use-combo")
def post_use_combo(
layer: Layer,
column: str,
body: UseComboBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
folder = _root(session, layer, owner, write=True)
return _call(folder, branch.use_combo, column, body.종류, body.키, body.판)
@router.post("/layers/{layer}/{kind}/{column}/copy")
def post_copy(
layer: Layer,
kind: Literal["structures", "area", "material"],
column: str,
source: str = Query(alias="from"),
owner: str | None = None,
session: Session = Depends(verify_session),
):
folder = _root(session, layer, owner, write=True)
part = "structure" if kind == "structures" else kind
return _call(folder, store.copy_from, part, column, source)
@router.get("/layers/{layer}/material-combos")
def get_matsets(layer: Layer, owner: str | None = None, session: Session = Depends(verify_session)):
return _call(_root(session, layer, owner), matset.listing)
@router.get("/layers/{layer}/material-combos/{key}")
def get_matset(
layer: Layer, key: str, owner: str | None = None, session: Session = Depends(verify_session)
):
return _call(_root(session, layer, owner), matset.one, key)
@router.post("/layers/{layer}/material-combos")
def post_matset(
layer: Layer,
body: MatsetBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
return _call(_root(session, layer, owner, write=True), matset.create, body.줄, body.판)
@router.put("/layers/{layer}/material-combos/{key}")
def put_matset(
layer: Layer,
key: str,
body: MatsetBody,
owner: str | None = None,
session: Session = Depends(verify_session),
):
return _call(_root(session, layer, owner, write=True), matset.edit, key, body.줄, body.판)
@router.delete("/layers/{layer}/material-combos/{key}")
def delete_matset(
layer: Layer,
key: str,
판: str = "",
owner: str | None = None,
session: Session = Depends(verify_session),
):
return _call(_root(session, layer, owner, write=True), matset.delete, key, 판)
@@ -1,7 +1,7 @@
"""M02 양식 층 API — 계약 `tmp/M02_분석/6_계약.md` 「층 (sub4)」.
⚠ 등록은 `main.py`(로그인만) · 층마다 권한은 여기서:
system 읽기만(고치기는 `/api/m02/templates` 시스템 관리자 길)
system 읽기 · 쓰기는 시스템 관리자(옛 `/api/m02/templates` 길과 같은 저장소)
company 같은 회사 읽기 · 쓰기는 회사 관리자(ADMIN · 마스터 · 시스템 관리자)
personal 본인 읽기·쓰기 · 같은 회사 사람 것은 읽기만(가져오기)
project 같은 회사 프로젝트 · 작업본 쓰기 · `_initial/` 은 안 씀
@@ -26,6 +26,8 @@ from common_util import common_util_spreadsheet as spreadsheet
from common_util.common_util_auth import verify_session
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool, run_with_connection
from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Table_Fill as fill
from M02_MasterTemplete import M02_Template_Layers as layers
@@ -152,8 +154,8 @@ async def _project(session: dict[str, Any], project_id: str | None) -> dict[str,
def _company_of(session: dict[str, Any], project: dict[str, Any] | None) -> int:
company_id = project["company_id"] if project else session.get("company_id")
if company_id is None:
raise HTTPException(status_code=403, detail="회사 연결이 필요합니다.")
if company_id is None: # 회사 없는 사용자는 없음(21-0) — 회사 · 개인 층은 회사부터
raise HTTPException(status_code=409, detail="회사부터 등록 — 회사 연결이 필요합니다.")
return int(company_id)
@@ -174,8 +176,8 @@ async def _layer_dir(
) -> Path:
"""층 폴더 — 읽기·쓰기 권한까지 여기서 가름."""
if layer == "system":
if write:
raise HTTPException(status_code=403, detail="시스템 양식은 마스터 템플릿 화면에서만.")
if write and not _is_system_admin(session):
raise HTTPException(status_code=403, detail="시스템 양식은 시스템 관리자만.")
return layers.system_dir()
if layer == "project":
return layers.project_dir((await _project(session, project_id))["root"])
@@ -184,16 +186,22 @@ async def _layer_dir(
if layer == "company":
if write and not _is_company_admin(session):
raise HTTPException(status_code=403, detail="회사 공식 양식은 회사 관리자만.")
return layers.company_dir(company_id)
return _ensured(layers.company_dir(company_id))
if layer == "personal":
owner = int(user_id) if user_id is not None else int(session["user_id"])
if write and owner != session.get("user_id"):
raise HTTPException(status_code=403, detail="개인 양식은 본인만 고칩니다.")
await _same_company_user(session, owner, company_id)
return layers.personal_dir(company_id, owner)
return _ensured(layers.personal_dir(company_id, owner))
raise HTTPException(status_code=404, detail="없는 층입니다.")
def _ensured(folder: Path) -> Path:
"""회사 · 개인 층 첫 진입 — 빈 구조물집계표 뼈대(PLAN 21-2 · `Items.ensure`)."""
items_store.ensure(folder)
return folder
def _bad(error: ValueError) -> HTTPException:
return HTTPException(status_code=400, detail=str(error))
@@ -261,6 +269,13 @@ async def save_layer_template(
if kind == "table" and fill.is_fillable(document):
# 채운 표가 와도 설계값은 안 받아 적음 — 양식 + 손 값만(5장 · 브라우저 값을 믿지 않음)
document = fill.strip_design(document)
if layer != "project":
# 시스템 · 회사 · 개인 층 = 마스터 저장소 규칙 그대로(집계표 저장 때 새 열 파일 ·
# 번호 · SS 장부 ·
# 계산값 서버 풀이 · 번호 칸 보존 — `M02_MasterTemplete_Store.write`)
return await asyncio.to_thread(
_store_save, folder, kind, name, body.판 or "", document, layer
)
if kind in ("area", "material"):
# 면적산출 · 재료산출 계산값은 서버가 같은 TS 엔진을 Node 로 돌려 새로 적음
# (브라우저 값 버림 · 5장 ②)
@@ -287,6 +302,15 @@ async def save_layer_template(
return {"종류": kind, "이름": name, "판": version, "층": layer}
def _store_save(folder: Path, kind: str, name: str, version: str, doc: Any, layer: str) -> dict:
with store.at(folder):
try:
info = store.write(kind, name, version, doc)
except store.StoreError as error:
raise HTTPException(status_code=error.status, detail=error.detail) from error
return {"종류": kind, "이름": name, "판": info["판"], "층": layer}
@router.delete("/layers/{layer}/templates/{kind}/{name}")
async def delete_layer_template(
layer: Layer,
@@ -296,10 +320,15 @@ async def delete_layer_template(
project_id: str | None = Query(None),
session: dict[str, Any] = Depends(verify_session),
) -> dict[str, Any]:
"""개인 = 본인만 · 회사 = 회사 관리자만 · 시스템 · 프로젝트 층은 거절 · 판이 다르면 409."""
if layer in ("system", "project"):
raise HTTPException(status_code=403, detail="개인 · 회사 층 양식만 지울 수 있습니다.")
"""개인 = 본인만 · 회사 = 회사 관리자만 · 시스템 = 시스템 관리자 · 프로젝트 층은 거절 ·
판이 다르면 409 · 프로그램이 이름으로 부르는 양식(`store.IN_USE` — 구조물집계표 등)은 409."""
if layer == "project":
raise HTTPException(
status_code=403, detail="개인 · 회사 · 시스템 층 양식만 지울 수 있습니다."
)
folder = await _layer_dir(session, layer, project_id=project_id, write=True)
if (kind, name) in store.IN_USE:
raise HTTPException(status_code=409, detail=f"쓰는 양식이라 못 지움 「{name}」")
try:
layers.check_kind(kind)
layers.check_name(name)
+34 -11
View File
@@ -1,4 +1,4 @@
"""M02 마스터 템플릿 — 시스템 층 저장소 (`resources/master_template/{종류}/<이름>.json`).
"""M02 마스터 템플릿 — 층 저장소 · 기본 시스템 층(`resources/master_template/{종류}/<이름>.json`).
M01 `Store` 방식 — 판(파일 sha256 앞 16자) · 판이 다르면 409 · 원자 쓰기.
권한은 등록하는 쪽(`main.py`)이 붙임.
@@ -10,6 +10,9 @@ import hashlib
import json
import re
import threading
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from datetime import datetime
from pathlib import Path
from typing import Any
@@ -20,6 +23,26 @@ from M02_MasterTemplete import M02_Template_Layers as layers
FOLDER: Path = Path(__file__).resolve().parent.parent / "resources" / "master_template"
KINDS = layers.KINDS # 시험은 FOLDER 를 사본으로 바꿈
# 지금 다루는 층 폴더 — 층 라우터가 `at(층 폴더)` 로 바꿈 · 안 바꾸면 시스템 층(`FOLDER`).
# 회사 · 개인 층도 시스템과 같은 모양(`table/구조물집계표.json` · 조합 장부 ·
# structure/area/material · 자재조합.json — PLAN 21-1)이라 아래 함수는 그대로 씀
_ROOT: ContextVar[Path | None] = ContextVar("m02_layer_root", default=None)
def root() -> Path:
return _ROOT.get() or FOLDER
@contextmanager
def at(folder: str | Path) -> Iterator[Path]:
"""이 안에서 부르는 저장소 함수는 `folder` 층을 봄(요청 하나 · 스레드 하나 안에서만)."""
token = _ROOT.set(Path(folder))
try:
yield Path(folder)
finally:
_ROOT.reset(token)
# 구조물집계표 구조물 열(꼴=수) 마다 세 파일(파일 이름 = 열 id) —
# 구조물 도면 `structure/` (도각 없는 전용 화면) · 면적산출 `area/` · 재료산출 `material/`
# (둘 다 스프레드시트 통합문서)
@@ -75,7 +98,7 @@ def _path(kind: str, name: str) -> Path:
or _BAD_NAME.search(name)
):
raise StoreError(400, f"쓸 수 없는 글자가 있는 이름 「{name}」")
return FOLDER / kind / f"{name}.json"
return root() / kind / f"{name}.json"
def _info(kind: str, path: Path) -> dict[str, Any]:
@@ -91,7 +114,7 @@ def list_all() -> list[dict[str, Any]]:
rows = [
_info(kind, p)
for kind in KINDS
for p in sorted((FOLDER / kind).glob("*.json"))
for p in sorted((root() / kind).glob("*.json"))
if not p.name.startswith(".")
]
return rows
@@ -162,7 +185,7 @@ def delete(kind: str, name: str, version: str | None = None) -> None:
def _docs(kind: str) -> list[tuple[str, Any]]:
folder = FOLDER / kind
folder = root() / kind
return [(p.stem, json.loads(p.read_bytes())) for p in sorted(folder.glob("[!.]*.json"))]
@@ -260,7 +283,7 @@ def _new_code(column: str) -> str:
def _retired() -> list[str]:
"""지운 열의 8자리 — 조합 장부 `지운번호`(한 번 준 번호는 다시 안 줌)."""
path = FOLDER / COMBO_FILE
path = root() / COMBO_FILE
return json.loads(path.read_bytes()).get("지운번호", []) if path.is_file() else []
@@ -318,7 +341,7 @@ def _take_origins(doc: Any) -> dict[str, str]:
source = col.pop("본뜬열", None)
if not source or source not in by_id or not col.get("id"):
continue
if (FOLDER / "structure" / f"{col['id']}.json").is_file():
if (root() / "structure" / f"{col['id']}.json").is_file():
continue
out[col["id"]] = source
for field in ("일위대가", "자재조합"):
@@ -360,14 +383,14 @@ def prune_removed_structures() -> list[str]:
codes: set[str] = set()
with _LOCK:
for kind, key, _, _ in PARTS:
for path in sorted((FOLDER / kind).glob("[!.]*.json")):
for path in sorted((root() / kind).glob("[!.]*.json")):
if path.stem not in keep:
code = _code(json.loads(path.read_bytes()), key)
codes.update([code] if code else [])
path.unlink()
gone.append(f"{kind}/{path.stem}")
if codes: # 지운 번호는 장부에 남겨 새 열이 다시 안 받게
book = FOLDER / COMBO_FILE
book = root() / COMBO_FILE
doc = json.loads(book.read_bytes()) if book.is_file() else {}
doc["지운번호"] = sorted(codes | set(doc.get("지운번호", [])))
atomic_write_json(book, doc)
@@ -377,7 +400,7 @@ def prune_removed_structures() -> list[str]:
def _move_basis() -> list[str]:
"""12장 산출근거 `basis/` → 면적산출 `area/` (내용 · 알약 연결 그대로 · 표번 칸 → 면적번 ·
시트 이름 산출근거 → 면적산출) · 옮긴 뒤 빈 `basis/` 폴더를 지움."""
old = FOLDER / "basis"
old = root() / "basis"
if not old.is_dir():
return []
moved: list[str] = []
@@ -478,7 +501,7 @@ def _combo_row(key: str, col: dict[str, Any], numbers: dict[str, str]) -> dict[s
def sync_combos() -> dict[str, Any]:
"""조합 표를 집계표에 맞춤 — 새 열 = 새 SS 키 · 지운 열 = 줄 지움 · 고친 열 = 칸 갱신 ·
줄 차례 = 집계표 열 차례. 바뀐 것이 없으면 파일을 안 씀."""
path = FOLDER / COMBO_FILE
path = root() / COMBO_FILE
with _LOCK:
old = json.loads(path.read_bytes()) if path.is_file() else {}
keys = {row["열"]: row["키"] for row in old.get("줄", [])}
@@ -502,5 +525,5 @@ def sync_combos() -> dict[str, Any]:
def read_combos() -> dict[str, Any]:
"""조합 표 — 파일이 아직 없으면 지금 열 전부로 처음 한 번 만듦."""
path = FOLDER / COMBO_FILE
path = root() / COMBO_FILE
return json.loads(path.read_bytes()) if path.is_file() else sync_combos()
+2
View File
@@ -63,6 +63,7 @@ from B07_DesignDetail.B07_DesignDetail_Router_Frame import router as b07_frame_r
from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router
from M02_MasterTemplete.M02_MasterTemplete_Router import router as m02_master_template_router
from M02_MasterTemplete.M02_MasterTemplete_Router_Drawing import router as m02_drawing_router
from M02_MasterTemplete.M02_MasterTemplete_Router_Items import router as m02_items_router
from M02_MasterTemplete.M02_MasterTemplete_Router_Layers import router as m02_layers_router
from common_util.common_util_audit import note_api_call, record_call_burst
from common_util.common_util_auth import (
@@ -638,6 +639,7 @@ app.include_router(m01_master_data_router, dependencies=system_admin_only)
# 마스터 템플릿 — 시스템 층은 시스템 관리자만 · 층(회사·개인·프로젝트)·도면 길은 로그인만(권한은 길 안에서).
app.include_router(m02_master_template_router, dependencies=system_admin_only)
app.include_router(m02_layers_router, dependencies=protected)
app.include_router(m02_items_router, dependencies=protected)
app.include_router(m02_drawing_router, dependencies=protected)
# 개발 전용 잠금 해제 — 다른 라우터와 **같은 보호**를 받는다(로그인·회사·프로젝트 접근).
# 그 위에 서버가 환경까지 한 번 더 본다.
+172
View File
@@ -0,0 +1,172 @@
"""M02 층 집계표(PLAN 21-2) — 층별 읽기 · 쓰기 · 권한 · 첫 진입 뼈대 · 항목 목록 · 항목 복사 ·
시스템 회귀.
임시 storage(회사 · 개인 층) + 견본 시스템 집계표(`write_sample_table`)로만 봄 —
실제 양식은 안 읽음(시스템 층 두 자리 `store.FOLDER` · `layers.SYSTEM_ROOT` 모두 사본).
자재 코드는 실제 M01 마스터에 있는 MT 키를 씀(자재 조합 저장이 마스터 대조를 함).
"""
from __future__ import annotations
import json
from pathlib import Path
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 M01_MasterData import M01_MasterData_Store as m01
from M02_MasterTemplete import M02_MasterTemplete_Router_Items as items_router
from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layers_router
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Template_Layers as layers
from resources.tester.test_m02_structure import write_sample_table
TABLE = "table/구조물집계표"
def _material() -> dict[str, Any]:
for data in m01.loaded()[0].values():
for row in data.get("줄") or []:
if str(row.get("키", "")).startswith("MT"):
return {"코드": row["키"], "이름": row.get("이름"), "규격": None, "단위": None}
raise AssertionError("마스터에 자재 키가 없음")
@pytest.fixture()
def world(tmp_path: Path, monkeypatch: pytest.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)
store.renumber()
storage = tmp_path / "storage"
storage.mkdir()
monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
session: dict[str, Any] = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False}
app = FastAPI()
app.include_router(layers_router.router)
app.include_router(items_router.router)
app.dependency_overrides[verify_session] = lambda: session
return {"client": TestClient(app), "session": session, "storage": storage, "system": system}
def _as(world: dict, role: str = "USER", user: int = 42, company: int | None = 7) -> TestClient:
world["session"].update(role=role, user_id=user, company_id=company)
return world["client"]
def _add_column(client: TestClient, layer: str, col: dict[str, Any]) -> None:
url = f"/api/m02/layers/{layer}/templates/{TABLE}"
got = client.get(url).json()
got["문서"]["열"].append(col)
put = client.put(url, json={"판": got["판"], "문서": got["문서"]})
assert put.status_code == 200, put.text
COL = {"id": "cw", "머리": ["돌기슭막이", "찰쌓기", "회사형", None], "단위": "m", "꼴": "수"}
def test_시스템_층은_그대로_항목_목록은_SS_줄(world):
client = _as(world)
rows = client.get("/api/m02/items").json()
ss = store.read_combos()["줄"]
assert [r["key"] for r in rows] == [r["키"] for r in ss] and rows[0]["key"] == "SS000001"
assert set(rows[0]) == {"key", *items_router.items_store.ITEM_FIELDS}
assert client.get("/api/m02/layers/system/structure-combos").json() == store.read_combos()
# 시스템 층 쓰기는 SYSTEM_ADMIN 만
assert client.post("/api/m02/layers/system/structures/rv").status_code == 403
url = f"/api/m02/layers/system/templates/{TABLE}"
got = client.get(url).json()
assert client.put(url, json={"판": got["판"], "문서": got["문서"]}).status_code == 403
assert client.delete(f"{url}?판={got['판']}").status_code == 403
admin = _as(world, "SYSTEM_ADMIN")
assert admin.put(url, json={"판": got["판"], "문서": got["문서"]}).status_code == 200
assert admin.delete(f"{url}?판={got['판']}").status_code == 409 # 쓰는 양식
def test_회사_층_첫_진입_뼈대와_쓰기_권한(world):
client = _as(world)
assert client.get("/api/m02/items?layer=company").json() == []
root = world["storage"] / "7" / "templates"
table = json.loads((root / "table/구조물집계표.json").read_text("utf-8"))
assert (
table["열"] == []
and table["층"]
== json.loads((world["system"] / "table/구조물집계표.json").read_text("utf-8"))["층"]
)
assert json.loads((root / "구조물집계표조합.json").read_text("utf-8"))["다음"] == 1
assert json.loads((root / "자재조합.json").read_text("utf-8"))["다음"] == 1
url = f"/api/m02/layers/company/templates/{TABLE}"
got = client.get(url).json()
got["문서"]["열"].append(COL)
assert client.put(url, json={"판": got["판"], "문서": got["문서"]}).status_code == 403
_add_column(_as(world, "ADMIN"), "company", COL)
items = _as(world).get("/api/m02/items?layer=company").json() # 같은 회사 일반 사용자
assert [(r["key"], r["열"]) for r in items] == [("SS000001", "cw")]
assert items[0]["도번"].startswith("SD") and (root / "structure/cw.json").is_file()
assert client.get("/api/m02/items?layer=company&owner=9").status_code == 403 # 남의 회사
# 일위대가(UA) 갈라 내기는 M01 시스템 마스터에 쓰므로 회사 관리자도 못 함
body = {"본뜬키": "UA000001", "담은로직": [], "이름": "x"}
admin = _as(world, "ADMIN")
assert admin.post("/api/m02/layers/company/structures/cw/combo", json=body).status_code == 403
assert admin.delete(f"{url}?판={client.get(url).json()['판']}").status_code == 409 # 쓰는 양식
def test_개인_층은_본인만_회사_없으면_409(world):
client = _as(world)
_add_column(client, "personal", {**COL, "id": "p1"})
assert [r["열"] for r in client.get("/api/m02/items?layer=personal").json()] == ["p1"]
assert (world["storage"] / "7" / "42" / "templates" / "area" / "p1.json").is_file()
other = client.get("/api/m02/items?layer=personal&owner=7/43")
assert other.status_code == 403
lonely = _as(world, company=None)
groups = lonely.get("/api/m02/items?all=1").json()
assert [(g["layer"], g["items"]) for g in groups][1:] == [("company", []), ("personal", [])]
assert len(groups[0]["items"]) == len(store.read_combos()["줄"]) # 마스터는 보임
made = lonely.post(
"/api/m02/layers/personal/material-combos", json={"줄": {"이름": "a", "담은자재": []}}
)
assert made.status_code == 409 and "회사부터" in made.json()["detail"]
def test_회사_항목을_개인으로_복사(world):
admin = _as(world, "ADMIN")
_add_column(admin, "company", COL)
base = "/api/m02/layers/company"
drawing = admin.get(f"{base}/templates/structure/cw").json()
doc = {**drawing["문서"], "도면": {**drawing["문서"]["도면"], "entities": [{"type": "line"}]}}
admin.put(f"{base}/templates/structure/cw", json={"판": drawing["판"], "문서": doc})
판 = admin.get(f"{base}/material-combos").json()["판"]
row = {"이름": "회사 자재", "담은자재": [_material()]}
um = admin.post(f"{base}/material-combos", json={"판": 판, "줄": row}).json()["키"]
admin.post(f"{base}/structures/cw/use-combo", json={"종류": "자재조합", "키": um})
ss = admin.get("/api/m02/items?layer=company").json()[0]["key"]
user = _as(world)
body = {"from_layer": "company", "from_owner": "7", "key": ss}
got = user.post("/api/m02/layers/personal/items/copy", json=body)
assert got.status_code == 200, got.text
made = got.json()
assert made["열"] == "cw" and made["key"] == "SS000001"
mine = "/api/m02/layers/personal"
copy = user.get(f"{mine}/templates/structure/cw").json()["문서"]
assert copy["도면"]["entities"] == [{"type": "line"}] and copy["열"] == "cw"
col = next(c for c in user.get(f"{mine}/templates/{TABLE}").json()["문서"]["열"])
assert col["본뜬"] == {"layer": "company", "owner": "7", "key": ss}
mat = user.get(f"{mine}/material-combos/{col['자재조합']}").json()
assert mat["이름"] == "회사 자재" and mat["담은자재"][0]["코드"] == _material()["코드"]
# 또 복사하면 새 열 id · 새 SS 키 · 자재 조합 이름 (2)
again = user.post("/api/m02/layers/personal/items/copy", json=body).json()
assert (again["열"], again["key"]) == ("cw_2", "SS000002")
names = [r["이름"] for r in user.get(f"{mine}/material-combos").json()["줄"]]
assert names == ["회사 자재", "회사 자재 (2)"]
# 마스터 항목도 개인으로 · 없는 키 404
master = {"from_layer": "system", "key": "SS000001"}
assert user.post("/api/m02/layers/personal/items/copy", json=master).status_code == 200
missing = {**body, "key": "SS999999"}
assert user.post("/api/m02/layers/personal/items/copy", json=missing).status_code == 404
+3 -2
View File
@@ -143,12 +143,13 @@ def test_개인_양식은_본인만_쓰고_같은_회사는_읽어_가져옴(wor
session.update(user_id=43)
theirs = mine + "&user_id=42"
assert client.get(theirs).status_code == 200
# 첫 진입 때 구조물집계표 빈 뼈대가 서므로(PLAN 21-2) 새 이름으로 씀
put = client.put(
f"/api/m02/layers/personal/templates/table/구조물집계표?project_id={P1}&user_id=42",
f"/api/m02/layers/personal/templates/table/내표?project_id={P1}&user_id=42",
json={"판": None, "문서": {"열": []}},
)
assert put.status_code == 200 # user_id 는 쓰기에서 무시 — 본인(43) 자리에 새로 씀
assert (world["storage"] / "7/43/templates/table/구조물집계표.json").is_file()
assert (world["storage"] / "7/43/templates/table/내표.json").is_file()
got = client.post(
f"/api/m02/projects/{P2}/templates/apply", json={"from": "personal", "user_id": 42}
)