- resources/master_template/구조물집계표조합.json (그룹 · 판 · 비고 · 다음 · 줄) · 지금 열 51개로 처음 만듦(SS000001~SS000051) - 줄 칸: 키 · 열 · 종류 · 분류1~3 · 단위 · 일위대가 · 도번 · 면적번 · 재료번 · 바인딩 · 펼침 · 꼴=수 열만 - 새 열 = 새 SS 키 · 지운 열 = 줄 지움(키 다시 안 씀) · 고친 열 = 칸 갱신 · 새 프로젝트 복사에도 실림 - GET /api/m02/structure-combos · Api_Fetch fetchStructureCombos() - 시험: 견본 표로 키 유지 · 새 열 · 지운 열 · 글 열 제외 · 키대장 머리글과 안 겹침 · 프로젝트 복사 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EUypcnp5d1gU2aeKh9F2H7
429 lines
18 KiB
Python
429 lines
18 KiB
Python
"""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 import common_util_spreadsheet as spreadsheet
|
|
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) —
|
|
# 구조물 도면 `structure/` (도각 없는 전용 화면) · 면적산출 `area/` · 재료산출 `material/`
|
|
# (둘 다 스프레드시트 통합문서)
|
|
# 번호 = 도번 SD · 면적번 SA · 재료번 SM + 8자리 [종류][분류1][분류2][분류3]
|
|
# (제목행 = 열 `머리` 네 층 · 없는 층 00) — 한 열의 세 번호는 같은 8자리 ·
|
|
# 한 번 준 번호는 파일에 남아 안 바뀜(PLAN 13장)
|
|
STRUCTURE_TABLE = "구조물집계표"
|
|
# (종류, 문서 번호 칸, 머리 글자, 시트 이름) — 시트 이름이 None 이면 구조물 도면
|
|
PARTS = (
|
|
("structure", "도번", "SD", None),
|
|
("area", "면적번", "SA", "면적산출"),
|
|
("material", "재료번", "SM", "재료산출"),
|
|
)
|
|
_KEY = {kind: key for kind, key, _, _ in PARTS}
|
|
SHEETS = ("area", "material") # 통합문서 종류 — [저장] 때 서버가 계산값을 새로 적음
|
|
_LEVELS = 4
|
|
_CODE = re.compile(r"^S[DAM](\d{8})$")
|
|
_OLD_CODE = re.compile(r"^S[DCA](\d{6})$") # PLAN 12장 6자리 — 끝에 00 을 붙여 읽음
|
|
_BAD_NAME = re.compile(r'[\\/:*?"<>|\x00-\x1f]|\.\.')
|
|
# 도각 없는 빈 CAD 문서의 기본 도면층 — 없으면 B07 CAD 가 못 엶("Cannot read properties
|
|
# of undefined (reading 'length')") · CAD 쪽도 없으면 새로 채우지만(2026-09-27 sub1 고침)
|
|
# 뿌리도 처음부터 채움
|
|
_BLANK_LAYER = {"id": "기본", "name": "기본", "isVisible": True, "isLocked": False}
|
|
# 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) · 아니면 그 판일 때만 덮어씀.
|
|
구조물집계표를 저장하면 — 새 구조물 열의 빈 구조물 도면 · 면적산출 · 재료산출을 번호와 같이
|
|
만들고 지운 열(또는 구조물이 아니게 된 열)의 세 파일을 지움(잠금 밖에서 · 안에서 부르면 죽음).
|
|
통합문서(area · material)는 서버가 같은 TS 엔진을 Node 로 돌려 `계산값` 을 새로 적음
|
|
(브라우저 값 버림 · 5장 ②) · 알약 `연결` 은 문서와 같이 그대로 적힘.
|
|
번호 칸(도번 · 면적번 · 재료번)은 서버 것 — 있던 파일 번호를 그대로 둠(화면 값을 안 받음)."""
|
|
path = _path(kind, name)
|
|
_check_skeleton(kind, doc)
|
|
if kind in SHEETS:
|
|
doc = _server_values(doc)
|
|
with _LOCK:
|
|
have = version_of(path.read_bytes()) if path.is_file() else ""
|
|
if have != (version or ""):
|
|
raise StoreError(409, {"stale": [name], "판": have})
|
|
if have and kind in _KEY:
|
|
number = json.loads(path.read_bytes()).get(_KEY[kind])
|
|
doc = {**doc, _KEY[kind]: number} if number else doc
|
|
atomic_write_json(path, doc)
|
|
info = _info(kind, path)
|
|
if kind == "table" and name == STRUCTURE_TABLE:
|
|
backfill_missing_structures() # 지우기 앞 — 같은 저장의 새 열이 지운 번호를 안 받게
|
|
prune_removed_structures()
|
|
sync_combos() # 번호가 다 선 뒤 — 조합 표 도번 · 면적번 · 재료번 칸
|
|
return info
|
|
|
|
|
|
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 _docs(kind: str) -> list[tuple[str, Any]]:
|
|
folder = FOLDER / kind
|
|
return [(p.stem, json.loads(p.read_bytes())) for p in sorted(folder.glob("[!.]*.json"))]
|
|
|
|
|
|
def _code(doc: Any, key: str) -> str | None:
|
|
"""문서의 번호 8자리 — 12장 6자리는 끝에 00 · 옛 `구-01` · `표-01` · 없음은 None."""
|
|
text = str(doc.get(key) or "") if isinstance(doc, dict) else ""
|
|
if found := _CODE.match(text):
|
|
return found.group(1)
|
|
found = _OLD_CODE.match(text)
|
|
return found.group(1) + "00" if found else None
|
|
|
|
|
|
def _codes() -> dict[str, str]:
|
|
"""`{열 id: 8자리}` — 구조물 도면 도번 먼저 · 없으면 면적번 · 재료번."""
|
|
codes: dict[str, str] = {}
|
|
for kind, key, _, _ in reversed(PARTS):
|
|
codes.update({name: c for name, doc in _docs(kind) if (c := _code(doc, key))})
|
|
return codes
|
|
|
|
|
|
def structure_numbers() -> dict[str, dict[str, str]]:
|
|
"""`{열 id: {도번, 면적번, 재료번}}` — 집계표 줄 · 좌측 줄이 한 번에 받음(없는 쪽은 빈 글)."""
|
|
out: dict[str, dict[str, str]] = {}
|
|
for kind, key, _, _ in PARTS:
|
|
for name, doc in _docs(kind):
|
|
row = out.setdefault(name, {k: "" for _, k, _, _ in PARTS})
|
|
row[key] = str(doc.get(key) or "")
|
|
return out
|
|
|
|
|
|
def _heads(col: dict[str, Any]) -> tuple[str | None, ...]:
|
|
"""제목행 네 층(종류 · 분류1 · 분류2 · 분류3) — 빈 층은 None."""
|
|
parts = [*(col.get("머리") or []), *[None] * _LEVELS][:_LEVELS]
|
|
return tuple((str(p).strip() if p is not None else "") or None for p in parts)
|
|
|
|
|
|
def _structure_columns() -> list[dict[str, Any]]:
|
|
"""번호 · 파일을 받는 열 — 꼴=수(NO · 측점 · 비고 같은 글 열은 구조물 아님)."""
|
|
table = read("table", STRUCTURE_TABLE)["문서"]
|
|
return [
|
|
col
|
|
for col in table.get("열", [])
|
|
if isinstance(col, dict) and col.get("id") and col.get("꼴") == "수"
|
|
]
|
|
|
|
|
|
def _column_ids() -> list[str]:
|
|
return [col["id"] for col in _structure_columns()]
|
|
|
|
|
|
def _pick_code(col: dict[str, Any], taken: dict[str, str], heads: dict[str, tuple]) -> str:
|
|
"""새 열 번호 — 층마다 같은 무리(그 층까지 제목행이 같은 열)가 이미 번호를 가졌으면 그 두 자리를
|
|
이어 받고 · 없으면 그 층 가장 큰 번호 + 1 · 빈 층(종류 밖)은 00 · 맨 아래 층(분류3)은 늘 다음
|
|
번호. 같은 8자리가 이미 있으면(같은 제목행 · 빈 분류3 겹침) 맨 아래 층을 그 무리 다음 번호로.
|
|
`taken` = 이미 번호 있는 열 `{열 id: 8자리}` · `heads` = 열 id → 제목행."""
|
|
parts = _heads(col)
|
|
codes = list(taken.values())
|
|
mates = [(heads[cid], code) for cid, code in taken.items() if cid in heads]
|
|
|
|
def next_of(prefix: str) -> str:
|
|
at = slice(len(prefix), len(prefix) + 2)
|
|
found = [int(c[at]) for c in codes if c.startswith(prefix)]
|
|
return f"{max(found, default=0) + 1:02d}"
|
|
|
|
code = ""
|
|
for depth, head in enumerate(parts):
|
|
if depth and head is None:
|
|
code += "00"
|
|
continue
|
|
at = slice(2 * depth, 2 * depth + 2)
|
|
group = parts[: depth + 1]
|
|
same = (c[at] for h, c in mates if h[: depth + 1] == group and c.startswith(code))
|
|
leaf = depth == _LEVELS - 1
|
|
code += (None if leaf else next(same, None)) or next_of(code)
|
|
if code in codes:
|
|
code = code[:-2] + next_of(code[:-2])
|
|
return code
|
|
|
|
|
|
def _new_code(column: str) -> str:
|
|
"""이 열이 쓸 8자리 — 짝 파일에 이미 있으면 그것 · 없으면 새로(잠금 안에서 부름)."""
|
|
taken = _codes()
|
|
if column in taken:
|
|
return taken[column]
|
|
cols = _structure_columns()
|
|
col = next((c for c in cols if c["id"] == column), {"id": column})
|
|
return _pick_code(col, taken, {c["id"]: _heads(c) for c in cols})
|
|
|
|
|
|
def _blank(kind: str, column: str) -> dict[str, Any]:
|
|
"""빈 문서(번호는 부르는 쪽이 붙임) — 구조물 도면 = 도각 없음 · 기본 도면층 하나 /
|
|
면적산출 · 재료산출 = 진짜 빈 시트 하나(식 없어 `계산값` 없음 · 첫 [저장] 때 서버가 적음)."""
|
|
if kind == "structure":
|
|
return {
|
|
"양식": "구조물도면",
|
|
"종류": "structure",
|
|
"판": 1,
|
|
"열": column,
|
|
"도면": {"format": 6, "entities": [], "layers": [dict(_BLANK_LAYER)]},
|
|
}
|
|
sheet = next(name for k, _, _, name in PARTS if k == kind)
|
|
return {
|
|
"종류": "통합문서",
|
|
"판": 1,
|
|
"열": column,
|
|
"서식": [{}],
|
|
"시트": [{"id": "s1", "이름": sheet, "칸": {}}],
|
|
"활성": "s1",
|
|
}
|
|
|
|
|
|
def _ensure(kind: str, column: str) -> bool:
|
|
"""그 열의 파일이 없으면 빈 문서를 번호(세 파일이 같은 8자리)와 같이 만듦 · 있으면 그대로."""
|
|
path = _path(kind, column)
|
|
head = next(h for k, _, h, _ in PARTS if k == kind)
|
|
with _LOCK:
|
|
if path.is_file():
|
|
return False
|
|
atomic_write_json(path, {**_blank(kind, column), _KEY[kind]: head + _new_code(column)})
|
|
return True
|
|
|
|
|
|
def create_structure(column: str) -> dict[str, Any]:
|
|
"""구조물 열 하나의 구조물 도면 새로 — 도번 SD + 8자리 · 이미 있으면 409.
|
|
면적산출 · 재료산출 파일이 없으면 빈 통합문서도 같이 만듦(같은 8자리)."""
|
|
if column not in _column_ids():
|
|
raise StoreError(404, f"{STRUCTURE_TABLE}에 없는 구조물 열 「{column}」")
|
|
if not _ensure("structure", column):
|
|
raise StoreError(409, f"이미 있는 구조물 도면 「{column}」")
|
|
for kind in SHEETS:
|
|
_ensure(kind, column)
|
|
return read("structure", column)
|
|
|
|
|
|
def copy_from(kind: str, column: str, source: str) -> dict[str, Any]:
|
|
"""다른 열(`source`) 내용을 받는 열(`column`)로 가져옴 — 번호 · 열은 받는 쪽 그대로(두 열은
|
|
따로 움직임) · 통합문서는 알약 `연결` 도 같이 오고 계산값은 서버가 다시 풂."""
|
|
if kind not in _KEY:
|
|
raise StoreError(404, f"없는 종류 「{kind}」")
|
|
target = read(kind, column)["문서"]
|
|
doc = {**read(kind, source)["문서"], **{k: target.get(k) for k in ("열", _KEY[kind])}}
|
|
if kind in SHEETS:
|
|
doc = _server_values({k: v for k, v in doc.items() if k != "계산값"})
|
|
with _LOCK:
|
|
atomic_write_json(_path(kind, column), doc)
|
|
return _info(kind, _path(kind, column))
|
|
|
|
|
|
def prune_removed_structures() -> list[str]:
|
|
"""집계표에서 지운 열(또는 구조물이 아니게 된 열)의 세 파일을 같이 지움 —
|
|
번호는 비워 둠(다른 열 번호를 안 당김)."""
|
|
keep = set(_column_ids())
|
|
gone: list[str] = []
|
|
with _LOCK:
|
|
for kind in _KEY:
|
|
for path in sorted((FOLDER / kind).glob("[!.]*.json")):
|
|
if path.stem not in keep:
|
|
path.unlink()
|
|
gone.append(f"{kind}/{path.stem}")
|
|
return gone
|
|
|
|
|
|
def _move_basis() -> list[str]:
|
|
"""12장 산출근거 `basis/` → 면적산출 `area/` (내용 · 알약 연결 그대로 · 표번 칸 → 면적번 ·
|
|
시트 이름 산출근거 → 면적산출) · 옮긴 뒤 빈 `basis/` 폴더를 지움."""
|
|
old = FOLDER / "basis"
|
|
if not old.is_dir():
|
|
return []
|
|
moved: list[str] = []
|
|
with _LOCK:
|
|
for path in sorted(old.glob("[!.]*.json")):
|
|
doc = json.loads(path.read_bytes())
|
|
doc["면적번"] = doc.pop("표번", None) # 번호는 renumber 가 SA 8자리로
|
|
for sheet in doc.get("시트") or []:
|
|
if isinstance(sheet, dict) and sheet.get("이름") == "산출근거":
|
|
sheet["이름"] = "면적산출"
|
|
target = _path("area", path.stem)
|
|
if not target.is_file():
|
|
atomic_write_json(target, doc)
|
|
path.unlink()
|
|
moved.append(path.stem)
|
|
if not any(old.iterdir()):
|
|
old.rmdir()
|
|
return moved
|
|
|
|
|
|
def renumber() -> dict[str, list[str]]:
|
|
"""한 번 — 옛 산출근거를 면적산출로 옮김 · 옛 번호(6자리 · 구-NN · 표-NN · 없음)를 8자리로
|
|
(6자리는 끝에 00 · 없는 열 차례 = 종류 · 분류1 · 분류2 · 분류3 차례) ·
|
|
구조물 아닌 열 파일 지움 · 빠진 파일(재료산출 등) 만듦. 내용은 그대로 · 두 번 돌려도 같음."""
|
|
moved = _move_basis()
|
|
gone = prune_removed_structures()
|
|
changed: list[str] = []
|
|
made: list[str] = []
|
|
with _LOCK:
|
|
cols = _structure_columns()
|
|
heads = {c["id"]: _heads(c) for c in cols}
|
|
taken = _codes()
|
|
for col in cols:
|
|
if col["id"] not in taken:
|
|
taken[col["id"]] = _pick_code(col, taken, heads)
|
|
docs = {kind: dict(_docs(kind)) for kind in _KEY}
|
|
for col in cols:
|
|
cid = col["id"]
|
|
for kind, key, head, _ in PARTS:
|
|
number = f"{head}{taken[cid]}"
|
|
doc = docs[kind].get(cid)
|
|
if doc is None:
|
|
atomic_write_json(_path(kind, cid), {**_blank(kind, cid), key: number})
|
|
made.append(f"{kind}/{cid}")
|
|
elif doc.get(key) != number:
|
|
doc[key] = number
|
|
atomic_write_json(_path(kind, cid), doc)
|
|
changed.append(f"{kind}/{cid}")
|
|
return {"옮김": moved, "지움": gone, "바꿈": changed, "새로": made}
|
|
|
|
|
|
def _server_values(doc: Any) -> dict[str, Any]:
|
|
try:
|
|
return spreadsheet.with_server_values(doc)
|
|
except spreadsheet.RecalcError as e:
|
|
raise StoreError(503, str(e)) from e
|
|
|
|
|
|
def backfill_missing_structures() -> list[str]:
|
|
"""구조물 열마다 세 파일이 있게 — 없는 것만 열 차례로 새로 만듦(그 무리 다음 번호).
|
|
그 사이 남이 만들었으면 그대로 둠."""
|
|
made: list[str] = []
|
|
for cid in _column_ids():
|
|
made += [f"{kind}/{cid}" for kind in _KEY if _ensure(kind, cid)]
|
|
return made
|
|
|
|
|
|
# ── 구조물집계표조합 ─────────────────────
|
|
# 구조물 열마다 줄 하나(제어용 장부 · 관리자는 집계표 화면으로 고침 · 서버가 집계표로 맞춤) ·
|
|
# 키 = SS + 6자리 · 한 번 준 키는 그 열에 남음 · 지운 열의 키는 다시 안 씀(`다음` 이 기억)
|
|
COMBO_FILE = layers.COMBO_FILE
|
|
_COMBO_HEAD = "SS"
|
|
_COMBO_NOTE = (
|
|
"구조물집계표 구조물 열(꼴=수)마다 한 줄 · 서버가 집계표 [저장] 때 맞춤(손으로 고치지 않음) · "
|
|
"키 SS 는 한 번 주면 안 바뀌고 지운 열의 키는 다시 안 씀"
|
|
)
|
|
|
|
|
|
def _combo_row(key: str, col: dict[str, Any], numbers: dict[str, str]) -> dict[str, Any]:
|
|
kind, level1, level2, level3 = _heads(col)
|
|
return {
|
|
"키": key,
|
|
"열": col["id"],
|
|
"종류": kind,
|
|
"분류1": level1,
|
|
"분류2": level2,
|
|
"분류3": level3,
|
|
"단위": col.get("단위"),
|
|
"일위대가": col.get("일위대가"),
|
|
**{field: numbers.get(field, "") for _, field, _, _ in PARTS},
|
|
"바인딩": col.get("바인딩"),
|
|
"펼침": col.get("펼침"),
|
|
}
|
|
|
|
|
|
def sync_combos() -> dict[str, Any]:
|
|
"""조합 표를 집계표에 맞춤 — 새 열 = 새 SS 키 · 지운 열 = 줄 지움 · 고친 열 = 칸 갱신 ·
|
|
줄 차례 = 집계표 열 차례. 바뀐 것이 없으면 파일을 안 씀."""
|
|
path = FOLDER / COMBO_FILE
|
|
with _LOCK:
|
|
old = json.loads(path.read_bytes()) if path.is_file() else {}
|
|
keys = {row["열"]: row["키"] for row in old.get("줄", [])}
|
|
taken = [int(k[len(_COMBO_HEAD) :]) for k in keys.values()]
|
|
after = max(old.get("다음", 1), max(taken, default=0) + 1)
|
|
numbers = structure_numbers()
|
|
rows = []
|
|
for col in _structure_columns():
|
|
if col["id"] not in keys:
|
|
keys[col["id"]] = f"{_COMBO_HEAD}{after:06d}"
|
|
after += 1
|
|
rows.append(_combo_row(keys[col["id"]], col, numbers.get(col["id"], {})))
|
|
doc = {"그룹": "구조물집계표조합", "판": 1, "비고": _COMBO_NOTE, "다음": after, "줄": rows}
|
|
if doc != old:
|
|
atomic_write_json(path, doc)
|
|
return doc
|
|
|
|
|
|
def read_combos() -> dict[str, Any]:
|
|
"""조합 표 — 파일이 아직 없으면 지금 열 전부로 처음 한 번 만듦."""
|
|
path = FOLDER / COMBO_FILE
|
|
return json.loads(path.read_bytes()) if path.is_file() else sync_combos()
|