feat(M02): 양식 층 넷의 자리 · 읽기 · 쓰기 · 복사 · manifest (PLAN 10-5)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015LuapLYqN1GGFD8Y1PStD5
This commit is contained in:
@@ -0,0 +1,312 @@
|
||||
"""M02 양식 층 — 네 층의 자리 · 읽기 · 쓰기 · 복사 · manifest.
|
||||
|
||||
층 넷 (PLAN 10-5):
|
||||
system `resources/master_template/{table,drawing}/<이름>.json` (git)
|
||||
company `storage/{회사}/templates/{table,drawing}/`
|
||||
personal `storage/{회사}/{사용자}/templates/{table,drawing}/`
|
||||
project `storage/{회사}/{사용자}/{프로젝트}/templates/{table,drawing}/`
|
||||
+ 초기 사본 `templates/_initial/`
|
||||
|
||||
- 층 폴더마다 `manifest.json` — 그 폴더에 든 양식의 출처 `{"table/이름": {층, 이름, 판, 적용일}}`.
|
||||
- 판 = 파일 sha256 앞 16자(M01 Store 와 같음) · 판이 다르면 `StaleTemplate`.
|
||||
- `_initial/` 은 프로젝트 첫 복사 때만 씀 — 그 뒤 어떤 길로도 쓰지 않음(초기화는 읽기만).
|
||||
- 권한은 부르는 쪽(라우터) 몫 — 여기는 자리와 파일만.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
from config import config_system
|
||||
|
||||
LAYERS = ("system", "company", "personal", "project")
|
||||
KINDS = ("table", "drawing")
|
||||
TEMPLATES_DIRNAME = "templates"
|
||||
INITIAL_DIRNAME = "_initial"
|
||||
MANIFEST_NAME = "manifest.json"
|
||||
|
||||
#: 시스템 층 뿌리 — 시험이 임시 폴더로 바꿈.
|
||||
SYSTEM_ROOT = config_system.PROJECT_ROOT / "resources" / "master_template"
|
||||
|
||||
|
||||
class StaleTemplate(Exception):
|
||||
"""저장하려는 판이 파일의 지금 판과 다름 — 라우터가 409 로 답함."""
|
||||
|
||||
def __init__(self, current: str | None) -> None:
|
||||
super().__init__("양식이 그새 바뀌었습니다.")
|
||||
self.current = current
|
||||
|
||||
|
||||
# ── 자리 ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def storage_root() -> Path:
|
||||
"""`storage/` 실경로 — 부를 때마다 설정을 읽음(시험이 바꿈)."""
|
||||
return Path(config_system.STORAGE_BASE_DIR).resolve()
|
||||
|
||||
|
||||
def _inside(root: Path, path: Path) -> Path:
|
||||
path = path.resolve()
|
||||
if root.resolve() not in (path, *path.parents):
|
||||
raise ValueError("양식 자리가 저장소 루트를 벗어났습니다.")
|
||||
return path
|
||||
|
||||
|
||||
def _id(value: Any, what: str) -> str:
|
||||
text = str(value)
|
||||
if not text.isdigit():
|
||||
raise ValueError(f"{what} 식별자가 올바르지 않습니다.")
|
||||
return text
|
||||
|
||||
|
||||
def system_dir() -> Path:
|
||||
return Path(SYSTEM_ROOT)
|
||||
|
||||
|
||||
def company_dir(company_id: int | str) -> Path:
|
||||
root = storage_root()
|
||||
return _inside(root, root / _id(company_id, "회사") / TEMPLATES_DIRNAME)
|
||||
|
||||
|
||||
def personal_dir(company_id: int | str, user_id: int | str) -> Path:
|
||||
root = storage_root()
|
||||
return _inside(
|
||||
root, root / _id(company_id, "회사") / _id(user_id, "사용자") / TEMPLATES_DIRNAME
|
||||
)
|
||||
|
||||
|
||||
def project_dir(project_root: str | Path) -> Path:
|
||||
"""프로젝트 작업본 자리 — `project_root` 는 `resolve_stored_project_path` 결과."""
|
||||
return Path(project_root) / TEMPLATES_DIRNAME
|
||||
|
||||
|
||||
def initial_dir(project_root: str | Path) -> Path:
|
||||
return project_dir(project_root) / INITIAL_DIRNAME
|
||||
|
||||
|
||||
def check_kind(kind: str) -> str:
|
||||
if kind not in KINDS:
|
||||
raise ValueError(f"양식 종류는 {', '.join(KINDS)} 중 하나입니다.")
|
||||
return kind
|
||||
|
||||
|
||||
def check_name(name: str) -> str:
|
||||
text = str(name or "").strip()
|
||||
bad = not text or text != name or text.startswith((".", "_")) or len(text) > 80
|
||||
if bad or any(ch in text for ch in '/\\:*?"<>|') or ".." in text:
|
||||
raise ValueError("양식 이름이 올바르지 않습니다.")
|
||||
return text
|
||||
|
||||
|
||||
def template_path(layer_dir: str | Path, kind: str, name: str) -> Path:
|
||||
base = Path(layer_dir)
|
||||
return _inside(base, base / check_kind(kind) / f"{check_name(name)}.json")
|
||||
|
||||
|
||||
# ── 읽기 · 쓰기 ───────────────────────────────────────
|
||||
|
||||
|
||||
def version_of(path: str | Path) -> str | None:
|
||||
path = Path(path)
|
||||
if not path.is_file():
|
||||
return None
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()[:16]
|
||||
|
||||
|
||||
def _modified(path: Path) -> str:
|
||||
return datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds")
|
||||
|
||||
|
||||
def list_templates(layer_dir: str | Path) -> list[dict[str, Any]]:
|
||||
"""층 하나의 양식 목록 `[{종류, 이름, 판, 수정일}]` — 종류 · 이름 차례."""
|
||||
base = Path(layer_dir)
|
||||
rows: list[dict[str, Any]] = []
|
||||
for kind in KINDS:
|
||||
folder = base / kind
|
||||
if not folder.is_dir():
|
||||
continue
|
||||
for path in sorted(folder.glob("*.json")):
|
||||
if path.name.startswith((".", "_")):
|
||||
continue
|
||||
rows.append(
|
||||
{
|
||||
"종류": kind,
|
||||
"이름": path.stem,
|
||||
"판": version_of(path),
|
||||
"수정일": _modified(path),
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def read_template(layer_dir: str | Path, kind: str, name: str) -> dict[str, Any] | None:
|
||||
"""`{종류, 이름, 판, 문서}` — 없으면 None."""
|
||||
path = template_path(layer_dir, kind, name)
|
||||
if not path.is_file():
|
||||
return None
|
||||
raw = path.read_bytes()
|
||||
return {
|
||||
"종류": kind,
|
||||
"이름": name,
|
||||
"판": hashlib.sha256(raw).hexdigest()[:16],
|
||||
"문서": json.loads(raw.decode("utf-8")),
|
||||
}
|
||||
|
||||
|
||||
def write_template(
|
||||
layer_dir: str | Path,
|
||||
kind: str,
|
||||
name: str,
|
||||
document: dict[str, Any],
|
||||
*,
|
||||
version: str | None = None,
|
||||
check_version: bool = False,
|
||||
) -> str:
|
||||
"""원자 쓰기 → 새 판. `check_version` 이면 `version` 이 지금 판과 같아야 함(새 파일은 None)."""
|
||||
if not isinstance(document, dict):
|
||||
raise ValueError("양식 문서는 JSON 객체여야 합니다.")
|
||||
path = template_path(layer_dir, kind, name)
|
||||
if check_version:
|
||||
current = version_of(path)
|
||||
if current != (version or None):
|
||||
raise StaleTemplate(current)
|
||||
atomic_write_json(path, document)
|
||||
return version_of(path) or ""
|
||||
|
||||
|
||||
def delete_template(layer_dir: str | Path, kind: str, name: str) -> bool:
|
||||
path = template_path(layer_dir, kind, name)
|
||||
if not path.is_file():
|
||||
return False
|
||||
path.unlink()
|
||||
manifest = read_manifest(layer_dir)
|
||||
if manifest.pop(f"{kind}/{name}", None) is not None:
|
||||
_write_manifest(layer_dir, manifest)
|
||||
return True
|
||||
|
||||
|
||||
# ── manifest ──────────────────────────────────────────
|
||||
|
||||
|
||||
def read_manifest(layer_dir: str | Path) -> dict[str, Any]:
|
||||
path = Path(layer_dir) / MANIFEST_NAME
|
||||
if not path.is_file():
|
||||
return {}
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
return data.get("양식", {}) if isinstance(data, dict) else {}
|
||||
|
||||
|
||||
def _write_manifest(layer_dir: str | Path, entries: dict[str, Any]) -> None:
|
||||
atomic_write_json(Path(layer_dir) / MANIFEST_NAME, {"양식": entries})
|
||||
|
||||
|
||||
def _stamp(layer: str, name: str, version: str | None, **extra: Any) -> dict[str, Any]:
|
||||
entry = {
|
||||
"층": layer,
|
||||
"이름": name,
|
||||
"판": version,
|
||||
"적용일": datetime.now().isoformat(timespec="seconds"),
|
||||
}
|
||||
entry.update({key: value for key, value in extra.items() if value is not None})
|
||||
return entry
|
||||
|
||||
|
||||
# ── 복사 ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def copy_templates(
|
||||
source_dir: str | Path,
|
||||
target_dir: str | Path,
|
||||
*,
|
||||
source_layer: str,
|
||||
kind: str | None = None,
|
||||
name: str | None = None,
|
||||
only_missing: bool = False,
|
||||
source_ref: dict[str, Any] | None = None,
|
||||
) -> list[str]:
|
||||
"""`source_dir` 층의 양식을 `target_dir` 로 복사 → 복사한 `종류/이름` 목록.
|
||||
|
||||
`kind`·`name` 으로 좁힘 · `only_missing` 이면 이미 있는 파일은 건너뜀(더하기만).
|
||||
manifest 에 출처(층 · 이름 · 판 · `source_ref`)를 적음. 출처 manifest 의 원래 출처는
|
||||
적지 않음 — 「어디서 가져왔나」 한 단계만.
|
||||
"""
|
||||
if source_layer not in LAYERS:
|
||||
raise ValueError("출처 층이 올바르지 않습니다.")
|
||||
rows = list_templates(source_dir)
|
||||
if kind is not None:
|
||||
rows = [row for row in rows if row["종류"] == check_kind(kind)]
|
||||
if name is not None:
|
||||
rows = [row for row in rows if row["이름"] == check_name(name)]
|
||||
manifest = read_manifest(target_dir)
|
||||
copied: list[str] = []
|
||||
for row in rows:
|
||||
source = template_path(source_dir, row["종류"], row["이름"])
|
||||
target = template_path(target_dir, row["종류"], row["이름"])
|
||||
if only_missing and target.exists():
|
||||
continue
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = target.with_name(f".{target.name}.tmp")
|
||||
shutil.copyfile(source, temporary)
|
||||
temporary.replace(target)
|
||||
key = f"{row['종류']}/{row['이름']}"
|
||||
manifest[key] = _stamp(source_layer, row["이름"], row["판"], **(source_ref or {}))
|
||||
copied.append(key)
|
||||
if copied:
|
||||
_write_manifest(target_dir, manifest)
|
||||
return copied
|
||||
|
||||
|
||||
def seed_project(project_root: str | Path, *, only_missing: bool = False) -> dict[str, list[str]]:
|
||||
"""시스템 양식 전부를 프로젝트 작업본과 `_initial/` 로 복사.
|
||||
|
||||
프로젝트 만들 때 · 옛 프로젝트 넣기(`only_missing`) 둘 다 이 길.
|
||||
|
||||
`_initial/` 은 늘 없는 것만 더함 — 이미 있으면 절대 덮지 않음.
|
||||
"""
|
||||
return {
|
||||
"작업본": copy_templates(
|
||||
system_dir(),
|
||||
project_dir(project_root),
|
||||
source_layer="system",
|
||||
only_missing=only_missing,
|
||||
),
|
||||
"초기": copy_templates(
|
||||
system_dir(), initial_dir(project_root), source_layer="system", only_missing=True
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def reset_project(
|
||||
project_root: str | Path, *, kind: str | None = None, name: str | None = None
|
||||
) -> list[str]:
|
||||
"""[초기화] — `_initial/` 을 작업본에 덮어씀(`_initial/` 은 읽기만)."""
|
||||
initial = initial_dir(project_root)
|
||||
rows = list_templates(initial)
|
||||
if kind is not None:
|
||||
rows = [row for row in rows if row["종류"] == check_kind(kind)]
|
||||
if name is not None:
|
||||
rows = [row for row in rows if row["이름"] == check_name(name)]
|
||||
initial_manifest = read_manifest(initial)
|
||||
manifest = read_manifest(project_dir(project_root))
|
||||
done: list[str] = []
|
||||
for row in rows:
|
||||
key = f"{row['종류']}/{row['이름']}"
|
||||
target = template_path(project_dir(project_root), row["종류"], row["이름"])
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = target.with_name(f".{target.name}.tmp")
|
||||
shutil.copyfile(template_path(initial, row["종류"], row["이름"]), temporary)
|
||||
temporary.replace(target)
|
||||
manifest[key] = initial_manifest.get(key) or _stamp("system", row["이름"], row["판"])
|
||||
done.append(key)
|
||||
if done:
|
||||
_write_manifest(project_dir(project_root), manifest)
|
||||
return done
|
||||
Reference in New Issue
Block a user