eomsangdonandClaude Opus 5.5
6c0210ebc5
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BU9gbNE3aWFUvE61pS2WGt
397 lines
17 KiB
Python
397 lines
17 KiB
Python
"""M02 일위대가 조합 층 저장소(PLAN 27-2a) — 마스터 · 회사 · 개인 세 층 한 길.
|
||
|
||
층 자리 · 키
|
||
마스터 `resources/master_data/일위대가조합.json` · 키 UA(`_키대장.json`)
|
||
쓰기는 M01 조합 저장소(`M01_MasterData_Store_Combo` 의 검사 · 대장 그대로)
|
||
회사 `storage/{회사}/templates/일위대가조합.json` · 키 `UC{회사}-######`
|
||
개인 `storage/{회사}/{사용자}/templates/일위대가조합.json` · 키 `UP{사용자}-######`
|
||
회사 · 개인 파일 = 마스터와 같은 머리 · 열 한 벌 + 머리 `다음`(키 장부 · 지운 키는 다시 안 씀).
|
||
지금 층 = `store.at(층 폴더)`(`store.root()`) — 키 머리도 그 자리에서 셈 · 권한은 라우터 몫.
|
||
나무 = 폴더 [구분] · [구분, 상세구분] 깊이 2 · 머리 `폴더` = 폴더 전부(빈 폴더 포함).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from common_util.common_util_json import atomic_write_json
|
||
from M01_MasterData import M01_MasterData_Store as m01
|
||
from M01_MasterData import M01_MasterData_Store_Combo as combo_store
|
||
from M02_MasterTemplete import M02_MasterTemplete_Store as store
|
||
from M02_MasterTemplete import M02_MasterTemplete_Tree as tree
|
||
from M02_MasterTemplete import M02_Template_Layers as layers
|
||
|
||
mcb = combo_store.mcb
|
||
FILE = combo_store.FILE
|
||
OWNERS = combo_store.OWNERS
|
||
MASTER = mcb.TID # UA
|
||
_HEAD = re.compile(r"^U([CP])(\d+)-")
|
||
DEPTH = 2 # 구분 · 상세구분
|
||
|
||
|
||
def _m01(fn, *args):
|
||
"""M01 막힘은 M02 막힘으로 — 상태 · 까닭 그대로."""
|
||
try:
|
||
return fn(*args)
|
||
except m01.StoreError as e:
|
||
raise store.StoreError(e.status, e.detail) from e
|
||
|
||
|
||
# ── 자리 ──────────────────────────────────────────────────────────────
|
||
def heads() -> tuple[str, ...]:
|
||
"""지금 층이 쓸 수 있는 키 머리 — 마지막이 제 층 것(마스터 UA · 회사 UC · 개인 UP)."""
|
||
root = store.root().resolve()
|
||
if root == Path(store.FOLDER).resolve():
|
||
return (MASTER,)
|
||
try:
|
||
parts = root.relative_to(layers.storage_root()).parts
|
||
except ValueError:
|
||
parts = ()
|
||
if len(parts) == 2:
|
||
return (MASTER, f"UC{parts[0]}-")
|
||
if len(parts) == 3:
|
||
return (MASTER, f"UC{parts[0]}-", f"UP{parts[1]}-")
|
||
raise store.StoreError(400, "일위대가 조합 층 자리를 모름")
|
||
|
||
|
||
def _file(head: str) -> Path | None:
|
||
"""키 머리의 조합 파일 — 개인은 사용자 id 로 찾음(회사 폴더를 모름)."""
|
||
if head.startswith(MASTER):
|
||
return Path(m01.FOLDER) / FILE
|
||
got = _HEAD.match(head)
|
||
if not got:
|
||
return None
|
||
if got[1] == "C":
|
||
return layers.company_dir(got[2]) / FILE
|
||
pattern = f"*/{got[2]}/{layers.TEMPLATES_DIRNAME}/{FILE}"
|
||
return next(layers.storage_root().glob(pattern), None)
|
||
|
||
|
||
def _rows_at(path: Path | None) -> list[dict]:
|
||
if path is None or not path.is_file():
|
||
return []
|
||
return json.loads(path.read_bytes()).get("줄") or []
|
||
|
||
|
||
def find(key: str) -> dict[str, Any] | None:
|
||
"""어느 층 조합이든 키로 한 줄 — 없으면 None."""
|
||
key = str(key or "")
|
||
return next((r for r in _rows_at(_file(key)) if str(r.get("키")) == key), None)
|
||
|
||
|
||
def where(key: str) -> tuple[str, str | None] | None:
|
||
"""키의 층 · 주인 — ("system", None) · ("company", "7") · ("personal", "7/42") · 모르면 None."""
|
||
key = str(key or "")
|
||
got, path = _HEAD.match(key), _file(key)
|
||
if key.startswith(MASTER) or path is None or not got:
|
||
return ("system", None) if key.startswith(MASTER) else None
|
||
if got[1] == "C":
|
||
return ("company", got[2])
|
||
return ("personal", f"{path.parent.parent.parent.name}/{got[2]}") # {회사}/{사용자}/templates
|
||
|
||
|
||
def names() -> dict[str, str]:
|
||
"""지금 층이 쓸 수 있는 조합 키 → 이름(마스터 · 회사 · 개인)."""
|
||
out: dict[str, str] = {}
|
||
for head in heads():
|
||
for r in _rows_at(_file(head)):
|
||
out[str(r.get("키"))] = str(r.get("이름") or "")
|
||
return out
|
||
|
||
|
||
def _load() -> tuple[dict[str, Any], str]:
|
||
"""(지금 층 조합 표, 판) — 회사 · 개인 파일이 없으면 빈 표 · 판 빈 글."""
|
||
if heads()[-1] == MASTER:
|
||
return _m01(combo_store._read)
|
||
path = store.root() / FILE
|
||
if not path.is_file():
|
||
head = {"그룹": mcb.GROUP, "판": str(date.today().year), "폴더": [], "다음": 1}
|
||
return {**head, "줄": []}, ""
|
||
raw = path.read_bytes()
|
||
return json.loads(raw), store.version_of(raw)
|
||
|
||
|
||
def rows() -> list[dict[str, Any]]:
|
||
return _load()[0].get("줄") or []
|
||
|
||
|
||
def _find(doc: dict, key: str) -> dict[str, Any]:
|
||
row = next((r for r in doc.get("줄") or [] if str(r.get("키")) == key), None)
|
||
if row is None:
|
||
raise store.StoreError(404, f"없는 조합 「{key}」")
|
||
return row
|
||
|
||
|
||
# ── 나무 폴더 ─────────────────────────────────────────────────────────
|
||
def path_of(row: dict) -> list[str]:
|
||
"""조합의 폴더 — [구분] · [구분, 상세구분] · 구분이 없으면 뿌리."""
|
||
a, b = (str(row.get(k) or "").strip() for k in ("구분", "상세구분"))
|
||
return [a, b] if a and b else [a] if a else []
|
||
|
||
|
||
def _set_path(row: dict, path: list[str]) -> None:
|
||
row["구분"] = path[0] if path else None
|
||
row["상세구분"] = path[1] if len(path) > 1 else None
|
||
|
||
|
||
def folders_of(doc: dict) -> list[list[str]]:
|
||
"""폴더 전부 — 적어 둔 것 + 조합 자리 · 부모까지."""
|
||
out: list[list[str]] = []
|
||
written = [p for p in doc.get("폴더") or [] if isinstance(p, list)]
|
||
for path in [*written, *(path_of(r) for r in doc.get("줄") or [])]:
|
||
path = [str(x) for x in path][:DEPTH]
|
||
for at in range(1, len(path) + 1):
|
||
if path[:at] not in out:
|
||
out.append(path[:at])
|
||
return out
|
||
|
||
|
||
# ── 읽기 ──────────────────────────────────────────────────────────────
|
||
def _brief(whole, row: dict) -> dict[str, Any]:
|
||
problems = mcb.check_row(whole, row)
|
||
return {
|
||
**{k: row.get(k) for k in mcb.COLS if k != "담은로직"},
|
||
"count": len(mcb.items(row)),
|
||
"blocked": bool(problems),
|
||
"reasons": problems,
|
||
}
|
||
|
||
|
||
def listing() -> dict[str, Any]:
|
||
"""지금 층 — `{판, 폴더, 줄}` · 줄 = 담은 로직 수 · 막힘 까닭(담은 로직 목록은 빼고)."""
|
||
doc, version = _load()
|
||
whole = m01.loaded()[1]
|
||
줄 = [_brief(whole, r) for r in doc.get("줄") or []]
|
||
return {"판": version, "폴더": folders_of(doc), "줄": 줄}
|
||
|
||
|
||
def one(key: str) -> dict[str, Any]:
|
||
"""조합 하나 — 담은 로직에 이름 · 단위를 붙여 줌."""
|
||
doc, version = _load()
|
||
row = _find(doc, key)
|
||
whole = m01.loaded()[1]
|
||
problems = mcb.check_row(whole, row)
|
||
brief = combo_store._brief(whole, row)
|
||
return {
|
||
"판": version,
|
||
"combo": row,
|
||
"줄": brief,
|
||
"blocked": bool(problems),
|
||
"reasons": problems,
|
||
}
|
||
|
||
|
||
def pills(key: str) -> dict[str, Any]:
|
||
return combo_store.combo_pills(_find(_load()[0], key))
|
||
|
||
|
||
def preview(combo: dict | str, inputs: dict | None = None) -> dict[str, Any]:
|
||
"""저장 안 함 — `combo` 는 조합 줄 통째 또는 지금 층 키."""
|
||
row = _find(_load()[0], combo) if isinstance(combo, str) else combo
|
||
return _m01(combo_store.combo_preview, row, inputs)
|
||
|
||
|
||
# ── 쓰기(회사 · 개인) ──────────────────────────────────────────────────
|
||
def _clean(row: Any, key: str, was: dict | None = None) -> dict[str, Any]:
|
||
"""서버가 잡는 칸 — 키 · 열 한 벌 · 소유 · 본뜬키는 있는 다른 조합(어느 층이든)."""
|
||
if not isinstance(row, dict):
|
||
raise store.StoreError(400, "조합 줄이 묶음 아님")
|
||
name = row.get("이름")
|
||
if not isinstance(name, str) or not name.strip():
|
||
raise store.StoreError(400, "조합 줄에 「이름」 이 없거나 글자 아님")
|
||
owner = row.get("소유") or (was or {}).get("소유") or OWNERS[0]
|
||
if owner not in OWNERS:
|
||
raise store.StoreError(400, f"모르는 소유 「{owner}」 — {' · '.join(OWNERS)}")
|
||
origin = row.get("본뜬키", (was or {}).get("본뜬키"))
|
||
if origin is not None and (origin == key or find(origin) is None):
|
||
raise store.StoreError(400, f"본뜬키 「{origin}」 — 비었거나 있는 다른 조합 키여야 함")
|
||
one = {k: row.get(k) for k in mcb.COLS}
|
||
one.update({"키": key, "이름": name.strip(), "소유": owner, "본뜬키": origin})
|
||
one["출처"] = one["출처"] or "자체"
|
||
return one
|
||
|
||
|
||
def _check(doc: dict, one: dict) -> None:
|
||
"""이름 길이 · 겹침 · 담은 로직(없는 키 · 두 번 · 빈 조합) — 막히면 422."""
|
||
name = one["이름"]
|
||
errors = [f"이름이 너무 김 — {mcb.MAX_이름}자까지"] if len(name) > mcb.MAX_이름 else []
|
||
if any(r.get("키") != one["키"] and str(r.get("이름", "")).strip() == name for r in doc["줄"]):
|
||
errors.append(f"「{name}」 이름의 조합이 이미 있음 — 다른 이름으로")
|
||
errors += mcb.check_row(m01.loaded()[1], one)
|
||
if errors:
|
||
raise store.StoreError(422, {"errors": errors})
|
||
|
||
|
||
def _save(doc: dict, version: str) -> str:
|
||
path = store.root() / FILE
|
||
with store._LOCK:
|
||
have = store.version_of(path.read_bytes()) if path.is_file() else ""
|
||
if have != (version or ""):
|
||
raise store.StoreError(409, {"stale": [FILE], "판": have})
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
atomic_write_json(path, doc)
|
||
return store.version_of(path.read_bytes())
|
||
|
||
|
||
def _users(key: str) -> list[str]:
|
||
"""이 층 구조물 조합에서 이 조합을 쓰는 항목 이름."""
|
||
if heads()[-1] == MASTER:
|
||
return combo_store.combo_users(key)
|
||
return [
|
||
tree.name_of(tree.path_of(c)) or c["id"]
|
||
for c in store._structure_columns()
|
||
if c.get("일위대가") == key
|
||
]
|
||
|
||
|
||
def _refuse_used(keys: list[str]) -> None:
|
||
used = [u for key in keys for u in _users(key)]
|
||
if used:
|
||
msg = f"구조물 {len(used)} 곳이 쓰는 조합이라 못 지움 — {' · '.join(used)}"
|
||
raise store.StoreError(409, {"errors": [msg], "users": used})
|
||
|
||
|
||
def create(row: Any, version: str | None = None) -> dict[str, Any]:
|
||
"""새 조합 — 키 = 층 머리 + 「다음」 번호. `version` None = 지금 판(갈라 내기)."""
|
||
if heads()[-1] == MASTER:
|
||
owner = row.get("소유") if isinstance(row, dict) else None
|
||
done = _m01(combo_store.combo_new, row, owner or OWNERS[0])
|
||
return {"키": done["key"], "판": done["version"], "combo": done["combo"]}
|
||
doc, have = _load()
|
||
number = int(doc.get("다음") or 1)
|
||
fixed = _clean(row, f"{heads()[-1]}{number:06d}")
|
||
_check(doc, fixed)
|
||
doc = {**doc, "다음": number + 1, "줄": [*doc["줄"], fixed]}
|
||
return {
|
||
"키": fixed["키"],
|
||
"판": _save(doc, have if version is None else version),
|
||
"combo": fixed,
|
||
}
|
||
|
||
|
||
def edit(key: str, row: Any, version: str) -> dict[str, Any]:
|
||
if heads()[-1] == MASTER:
|
||
owner = row.get("소유") if isinstance(row, dict) else ""
|
||
done = _m01(combo_store.combo_edit, key, version, row, owner or "")
|
||
return {"키": key, "판": done["version"], "combo": done["combo"]}
|
||
doc, _ = _load()
|
||
fixed = _clean(row, key, _find(doc, key))
|
||
_check(doc, fixed)
|
||
doc = {**doc, "줄": [fixed if r.get("키") == key else r for r in doc["줄"]]}
|
||
return {"키": key, "판": _save(doc, version), "combo": fixed}
|
||
|
||
|
||
def delete(key: str, version: str) -> dict[str, Any]:
|
||
"""지우기 — 이 층 구조물 조합이 쓰면 409 + 항목 이름들(지우지 않음)."""
|
||
if heads()[-1] == MASTER:
|
||
return {"판": _m01(combo_store.combo_delete, key, version)["version"]}
|
||
doc, _ = _load()
|
||
_find(doc, key)
|
||
_refuse_used([key])
|
||
return {"판": _save({**doc, "줄": [r for r in doc["줄"] if r.get("키") != key]}, version)}
|
||
|
||
|
||
# ── 나무 고침 ─────────────────────────────────────────────────────────
|
||
def _moved(path: list[str], old: list[str], new: list[str]) -> list[str]:
|
||
return new + path[len(old) :] if path[: len(old)] == old else path
|
||
|
||
|
||
def _folder(folders: list[list[str]], path: list[str] | None, what: str) -> list[str]:
|
||
path = [str(x) for x in path or []]
|
||
if not path or path not in folders:
|
||
raise store.StoreError(404, f"{what} 폴더 「{' › '.join(path)}」 없음")
|
||
return path
|
||
|
||
|
||
def _fresh(name: Any) -> str:
|
||
text = str(name or "").strip()
|
||
if not text or "›" in text or len(text) > mcb.MAX_이름:
|
||
raise store.StoreError(422, "폴더 이름 — 빈 이름 · 「›」 · 너무 긴 이름은 못 씀")
|
||
return text
|
||
|
||
|
||
def _apply(doc: dict, body: dict) -> tuple[dict, list[str]]:
|
||
"""나무 고침 한 번 → (새 표, 지운 조합 키). 조합 더하기 · 고치기는 `create` · `edit` 몫."""
|
||
op, key = body.get("op"), body.get("키")
|
||
folders, rows = folders_of(doc), [dict(r) for r in doc.get("줄") or []]
|
||
gone: list[str] = []
|
||
target = [str(x) for x in body.get("새경로") or []]
|
||
if op in ("move", "add-folder") and target and target not in folders:
|
||
raise store.StoreError(404, f"없는 폴더 「{' › '.join(target)}」")
|
||
if op == "add-folder":
|
||
made = [*[str(x) for x in body.get("경로") or []], _fresh(body.get("이름"))]
|
||
if len(made) > DEPTH or (made[:-1] and made[:-1] not in folders):
|
||
raise store.StoreError(422, f"폴더는 {DEPTH} 단까지 · 있는 폴더 아래에만")
|
||
if made in folders:
|
||
raise store.StoreError(409, f"이미 있는 폴더 「{' › '.join(made)}」")
|
||
folders.append(made)
|
||
elif op == "move" and key:
|
||
_set_path(_find({"줄": rows}, str(key)), target)
|
||
elif op == "delete" and key:
|
||
_find({"줄": rows}, str(key))
|
||
gone = [str(key)]
|
||
elif op in ("rename", "move", "delete"):
|
||
old = _folder(folders, body.get("경로"), "고칠")
|
||
if op == "delete":
|
||
gone = [str(r["키"]) for r in rows if path_of(r)[: len(old)] == old]
|
||
folders = [f for f in folders if f[: len(old)] != old]
|
||
else:
|
||
new = [*old[:-1], _fresh(body.get("이름"))] if op == "rename" else [*target, old[-1]]
|
||
deepest = max(len(f) for f in folders if f[: len(old)] == old)
|
||
if new[: len(old)] == old or deepest - len(old) + len(new) > DEPTH:
|
||
raise store.StoreError(422, f"폴더는 {DEPTH} 단까지 · 제 안으로 못 옮김")
|
||
if new in folders:
|
||
raise store.StoreError(409, f"이미 있는 폴더 「{' › '.join(new)}」")
|
||
folders = [_moved(f, old, new) for f in folders]
|
||
for r in rows:
|
||
if path_of(r)[: len(old)] == old:
|
||
_set_path(r, _moved(path_of(r), old, new))
|
||
else:
|
||
raise store.StoreError(
|
||
422, f"모르는 나무 고침 「{op}」 — add-folder · rename · move · delete"
|
||
)
|
||
rows = [r for r in rows if str(r.get("키")) not in gone]
|
||
return {**doc, "폴더": folders, "줄": rows}, gone
|
||
|
||
|
||
def edit_tree(body: dict) -> dict[str, Any]:
|
||
"""폴더 더하기 · 이름 바꾸기 · 옮기기(폴더 · 조합) · 지우기(폴더 = 안 조합까지 · 조합)
|
||
→ 새 목록 · 쓰이는 조합이 들면 409 · 판 다르면 409."""
|
||
doc, have = _load()
|
||
if have != (body.get("판") or ""):
|
||
raise store.StoreError(409, {"stale": [FILE], "판": have})
|
||
new, gone = _apply(doc, body)
|
||
_refuse_used(gone)
|
||
if heads()[-1] != MASTER:
|
||
_save(new, have)
|
||
return {**listing(), "지운조합": gone}
|
||
was = {str(r["키"]): r for r in doc["줄"]}
|
||
changes = [{"op": "delete", "key": k} for k in gone] + [
|
||
{"op": "edit", "key": str(r["키"]), "row": r} for r in new["줄"] if was[str(r["키"])] != r
|
||
]
|
||
version = _m01(combo_store._write, changes, have)["version"] if changes else have
|
||
_write_master_folders(new["폴더"], version)
|
||
return {**listing(), "지운조합": gone}
|
||
|
||
|
||
def _write_master_folders(folders: list[list[str]], version: str) -> None:
|
||
"""마스터 조합 파일 머리 `폴더` — 줄은 안 건드림(줄 고침은 M01 저장 길)."""
|
||
path = Path(m01.FOLDER) / FILE
|
||
with m01._LOCK:
|
||
doc, have = m01.read(FILE)
|
||
if have != version:
|
||
raise store.StoreError(409, {"stale": [FILE], "판": have})
|
||
if doc.get("폴더") == folders:
|
||
return
|
||
head = {k: v for k, v in doc.items() if k != "줄"}
|
||
text = m01.dump({**head, "폴더": folders, "줄": doc.get("줄") or []})
|
||
temporary = path.with_name(f".{FILE}.tmp")
|
||
temporary.write_text(text, encoding="utf-8", newline="\n")
|
||
os.replace(temporary, path)
|
||
m01.cache.stale([FILE])
|