"""M02 마스터 템플릿 — 시스템 층 저장소 (`resources/master_template/{종류}/<이름>.json`). M01 `Store` 방식 — 판(파일 sha256 앞 16자) · 판이 다르면 409 · 원자 쓰기. 권한은 등록하는 쪽(`main.py`)이 붙임. """ from __future__ import annotations import hashlib import json import re import threading from datetime import datetime from pathlib import Path from typing import Any from common_util.common_util_json import atomic_write_json from M02_MasterTemplete import M02_Template_Layers as layers FOLDER: Path = Path(__file__).resolve().parent.parent / "resources" / "master_template" KINDS = layers.KINDS # 시험은 FOLDER 를 사본으로 바꿈 # 구조물 도면 = 구조물집계표 열 id 마다 하나(파일 이름 = 열 id) · 새로 만들 때 A1 도각을 깖 STRUCTURE_TABLE = "구조물집계표" STRUCTURE_FRAME = "00_template_A1" _BASIS_HEAD = (("work", "공종"), ("spec", "규격"), ("detail", "산출 내역"), ("unit", "단위")) _NUMBER = re.compile(r"^구-(\d+)$") _BAD_NAME = re.compile(r'[\\/:*?"<>|\x00-\x1f]|\.\.') # ponytail: 저장은 한 번에 하나(프로세스 안 잠금) · 서버를 여럿 띄우면 파일 잠금으로 _LOCK = threading.Lock() class StoreError(Exception): def __init__(self, status: int, detail): super().__init__(detail) self.status, self.detail = status, detail def version_of(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest()[:16] def _path(kind: str, name: str) -> Path: if kind not in KINDS: raise StoreError(404, f"없는 종류 「{kind}」") if not name or name != name.strip() or name.startswith(".") or _BAD_NAME.search(name): raise StoreError(400, f"쓸 수 없는 글자가 있는 이름 「{name}」") return FOLDER / kind / f"{name}.json" def _info(kind: str, path: Path) -> dict[str, Any]: raw = path.read_bytes() stamp = datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds") return {"종류": kind, "이름": path.stem, "판": version_of(raw), "수정일": stamp} def list_all() -> list[dict[str, Any]]: rows = [ _info(kind, p) for kind in KINDS for p in sorted((FOLDER / kind).glob("*.json")) if not p.name.startswith(".") ] return rows def read(kind: str, name: str) -> dict[str, Any]: path = _path(kind, name) if not path.is_file(): raise StoreError(404, f"없는 양식 「{name}」") raw = path.read_bytes() return {"종류": kind, "이름": name, "판": version_of(raw), "문서": json.loads(raw)} def _check_skeleton(kind: str, doc: Any) -> None: """빈 문서·뼈대 없는 문서는 거절 — 층 저장과 같은 규칙(`layers.check_skeleton`).""" try: layers.check_skeleton(kind, doc) except ValueError as e: raise StoreError(400, str(e)) from e def write(kind: str, name: str, version: str, doc: Any) -> dict[str, Any]: """`version` 이 빈 글이면 새로 만듦(이미 있으면 409) · 아니면 그 판일 때만 덮어씀.""" path = _path(kind, name) _check_skeleton(kind, doc) with _LOCK: have = version_of(path.read_bytes()) if path.is_file() else "" if have != (version or ""): raise StoreError(409, {"stale": [name], "판": have}) atomic_write_json(path, doc) return _info(kind, path) def delete(kind: str, name: str, version: str | None = None) -> None: path = _path(kind, name) with _LOCK: if not path.is_file(): raise StoreError(404, f"없는 양식 「{name}」") if version is not None and version_of(path.read_bytes()) != version: raise StoreError(409, {"stale": [name], "판": version_of(path.read_bytes())}) path.unlink() # ── 구조물 도면 ─────────────────────────────────────── def _structures() -> list[tuple[str, Any]]: folder = FOLDER / "structure" return [(p.stem, json.loads(p.read_bytes())) for p in sorted(folder.glob("[!.]*.json"))] def structure_numbers() -> dict[str, str | None]: """`{열 id: 도번}` — 집계표 도면 줄이 한 번에 받음.""" return { name: (doc.get("도번") if isinstance(doc, dict) else None) for name, doc in _structures() } def _number(doc: Any) -> int: found = _NUMBER.match(str(doc.get("도번") or "")) if isinstance(doc, dict) else None return int(found.group(1)) if found else 0 def create_structure(column: str) -> dict[str, Any]: """열 id 하나의 구조물 도면 새로 — 도번 = 있는 것 중 가장 큰 번호 + 1 · 이미 있으면 409.""" path = _path("structure", column) table = read("table", STRUCTURE_TABLE)["문서"] if column not in {col.get("id") for col in table.get("열", []) if isinstance(col, dict)}: raise StoreError(404, f"{STRUCTURE_TABLE}에 없는 열 「{column}」") frame = read("drawing", STRUCTURE_FRAME)["문서"] basis = [{"id": key, "머리": [label], "단위": None, "꼴": "글"} for key, label in _BASIS_HEAD] basis.append({"id": "qty", "머리": ["수량(m당)"], "단위": None, "꼴": "수"}) with _LOCK: if path.is_file(): raise StoreError(409, f"이미 있는 구조물 도면 「{column}」") number = max((_number(doc) for _name, doc in _structures()), default=0) + 1 doc = { "양식": "구조물도면", "종류": "structure", "판": 1, "열": column, "도번": f"구-{number:02d}", "도면": frame, "산출근거": {"양식": "산출근거", "종류": "표", "판": 1, "열": basis, "줄": []}, } atomic_write_json(path, doc) return read("structure", column)