feat(M01): 마스터 데이터 관리자 API 서버 쪽 — 읽기 · 시험 계산 · 저장

- M01_MasterData 라우터 + 파일 저장소 (/api/m01 · 시스템관리자 전용)
- 저장 = 판본 대조(409) → 적용 → 틀·로직 변수 검사(새 걸림 422) → 씀
- check_master.py 에 load(folder) · check_saved · calc 드러냄
- 계약 문서 장 이름 보정(공통3장)
- 시험 test_m01_api.py 8개 (임시 폴더 사본)

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PAdA5ThqmtVSsbzjusk1cJ

M01_MasterData/M01_MasterData_Store.py

@@ -0,0 +1,300 @@
+"""M01 마스터 데이터 파일 저장소 — `resources/master_data/` 첫 층 JSON 읽기·검사·쓰기.
+
+계약: `resources/master_data/_화면_계약.md` · 파일 모양: `_틀.md`
+검사·계산 엔진: `scripts/check_master.py`.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import sys
+import threading
+from decimal import Decimal
+from pathlib import Path
+
+SCRIPTS = Path(__file__).resolve().parents[1] / "resources/master_data/scripts"
+sys.path.insert(0, str(SCRIPTS))
+
+import check_master as cm  # noqa: E402
+
+FOLDER: Path = cm.MASTER  # 시험은 사본 폴더로 바꿈
+_CHAPTER = re.compile(r"_([^_]*\d+장)\.json$")  # 「공통3장」 · 「13장」
+# ponytail: 저장은 한 번에 하나(프로세스 안 잠금) · 서버를 여럿 띄우면 파일 잠금으로
+_LOCK = threading.Lock()
+
+
+class StoreError(Exception):
+    def __init__(self, status: int, detail):
+        super().__init__(detail)
+        self.status, self.detail = status, detail
+
+
+# ── 읽기 ──────────────────────────────────────────────────────────────
+def _path(file: str) -> Path:
+    path = FOLDER / file
+    if "/" in file or "\\" in file or file.startswith("_") or not path.is_file():
+        raise StoreError(404, f"없는 파일 「{file}」")
+    return path
+
+
+def version_of(raw: bytes) -> str:
+    return hashlib.sha256(raw).hexdigest()[:16]
+
+
+def read(file: str) -> tuple[dict, str]:
+    raw = _path(file).read_bytes()
+    return json.loads(raw, parse_float=Decimal, parse_int=Decimal), version_of(raw)
+
+
+def items_key(data: dict) -> str:
+    return "표" if data.get("그룹") in cm.mf.TABLE_GROUPS else "줄"
+
+
+def chapter_of(file: str) -> str:
+    m = _CHAPTER.search(file)
+    return m.group(1) if m else ""
+
+
+def _hit(row: dict, q: str) -> bool:
+    q = q.lower()
+    return not q or q in str(row.get("열쇠", "")).lower() or q in str(row.get("이름", "")).lower()
+
+
+def groups() -> list[dict]:
+    out = {g: {"group": g, "files": 0, "rows": 0} for g in cm.mf.GROUPS}
+    for data in cm.load(folder=FOLDER).values():
+        slot = out.get(data.get("그룹"))
+        if slot:
+            slot["files"] += 1
+            slot["rows"] += len(data.get(items_key(data)) or [])
+    return list(out.values())
+
+
+def files_of(group: str) -> list[dict]:
+    if group not in cm.mf.GROUPS:
+        raise StoreError(404, f"없는 그룹 「{group}」")
+    out = []
+    for path in sorted(FOLDER.glob(f"{group}_*.json")):
+        data, version = read(path.name)
+        out.append(
+            {
+                "file": path.name,
+                "book": data.get("원문"),
+                "chapter": chapter_of(path.name),
+                "edition": data.get("판"),
+                "rows": len(data.get(items_key(data)) or []),
+                "version": version,
+            }
+        )
+    return out
+
+
+def rows(file: str, page: int, size: int, q: str) -> dict:
+    data, version = read(file)
+    hits = [r for r in data.get(items_key(data)) or [] if _hit(r, q)]
+    page, size = max(page, 1), min(max(size, 1), 500)
+    return {
+        "file": file,
+        "version": version,
+        "total": len(hits),
+        "page": page,
+        "size": size,
+        "rows": hits[(page - 1) * size : page * size],
+    }
+
+
+def tables(file: str, q: str) -> dict:
+    data, version = read(file)
+    if items_key(data) != "표":
+        raise StoreError(400, f"표형 파일 아님 「{file}」")
+    heads = ("열쇠", "이름", "기준", "출처", "조건", "값칸")
+    return {
+        "file": file,
+        "version": version,
+        "tables": [
+            {**{k: t.get(k) for k in heads}, "count": len(t.get("줄") or [])}
+            for t in data["표"]
+            if _hit(t, q)
+        ],
+    }
+
+
+def _one(data: dict, key: str) -> dict:
+    for row in data.get(items_key(data)) or []:
+        if str(row.get("열쇠")) == key:
+            return row
+    raise StoreError(404, f"없는 열쇠 「{key}」")
+
+
+def table(file: str, key: str) -> dict:
+    data, version = read(file)
+    return {"file": file, "version": version, "table": _one(data, key)}
+
+
+def _logic_files(files: dict[str, dict]):
+    for name, data in files.items():
+        if data.get("그룹") == "로직":
+            yield name, data
+
+
+def logics(book: str, chapter: str, q: str, blocked: int | None) -> list[dict]:
+    files = cm.load(folder=FOLDER)
+    whole = cm.mf.Master(files)
+    out = []
+    for name, data in _logic_files(files):
+        if (book and data.get("원문") != book) or (chapter and chapter_of(name) != chapter):
+            continue
+        for row in data.get("줄") or []:
+            if not _hit(row, q):
+                continue
+            reasons = cm.mf.check_logic(whole, data.get("원문"), row)[0]
+            if blocked is not None and bool(reasons) != bool(blocked):
+                continue
+            out.append(
+                {
+                    "file": name,
+                    "book": data.get("원문"),
+                    "chapter": chapter_of(name),
+                    **{k: row.get(k) for k in ("열쇠", "이름", "결과단위", "출처")},
+                    "blocked": bool(reasons),
+                    "reasons": reasons,
+                }
+            )
+    return out
+
+
+def logic(book: str, key: str) -> dict:
+    files = cm.load(folder=FOLDER)
+    for name, data in _logic_files(files):
+        if data.get("원문") != book:
+            continue
+        for row in data.get("줄") or []:
+            if str(row.get("열쇠")) == key:
+                reasons = cm.mf.check_logic(cm.mf.Master(files), book, row)[0]
+                _, version = read(name)
+                return {
+                    "file": name,
+                    "version": version,
+                    "logic": row,
+                    "blocked": bool(reasons),
+                    "reasons": reasons,
+                }
+    raise StoreError(404, f"없는 로직 「{book}:{key}」")
+
+
+# ── 시험 계산 ──────────────────────────────────────────────────────────
+def calc(book: str, key: str, inputs: dict) -> dict:
+    given = {k: _dec(v) for k, v in inputs.items()}
+    try:
+        result = cm.calc(cm.load(folder=FOLDER), f"{book}:{key}", given)
+    except (cm.mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e:
+        return {"ok": False, "reason": str(e) or type(e).__name__}
+    if "결과" in result:
+        return {"ok": True, "result": result["결과"], "middle": result["중간"]}
+    return {
+        "ok": True,
+        "lines": result["줄"],
+        "sums": {k: result[k] for k in cm.mf.BUNDLE},
+        "middle": result["중간"],
+    }
+
+
+# ── 저장 ──────────────────────────────────────────────────────────────
+def _dec(v):
+    """화면이 보낸 수 → Decimal(파일을 읽은 모양과 같게)."""
+    if isinstance(v, bool) or v is None:
+        return v
+    if isinstance(v, (int, float)):
+        return Decimal(str(v))
+    if isinstance(v, list):
+        return [_dec(x) for x in v]
+    if isinstance(v, dict):
+        return {k: _dec(x) for k, x in v.items()}
+    return v
+
+
+def _apply(data: dict, changes: list[dict], file: str) -> None:
+    items = data.setdefault(items_key(data), [])
+    for ch in changes:
+        op, key, row = ch.get("op"), ch.get("key"), _dec(ch.get("row"))
+        if op not in ("edit", "add", "delete"):
+            raise StoreError(400, f"모르는 op 「{op}」")
+        if op != "add":
+            at = next((i for i, r in enumerate(items) if str(r.get("열쇠")) == key), None)
+            if at is None:
+                raise StoreError(404, f"{file} · 없는 열쇠 「{key}」")
+        if op != "delete" and not (isinstance(row, dict) and "열쇠" in row):
+            raise StoreError(400, f"{file} · row 에 열쇠 없음")
+        new_key = str(row["열쇠"]) if op != "delete" else None
+        taken = {str(r.get("열쇠")) for i, r in enumerate(items) if op == "add" or i != at}
+        if new_key in taken:
+            raise StoreError(400, f"{file} · 열쇠 겹침 「{new_key}」")
+        if op == "add":
+            items.append(row)
+        elif op == "edit":
+            items[at] = row
+        else:
+            del items[at]
+
+
+def save(batch: list[dict]) -> list[dict]:
+    with _LOCK:
+        stale = [p.get("file", "") for p in batch if read(p.get("file", ""))[1] != p.get("version")]
+        if stale:
+            raise StoreError(409, {"stale": stale})
+        before, after = cm.load(folder=FOLDER), cm.load(folder=FOLDER)  # after = 고칠 사본
+        names = [part["file"] for part in batch]
+        for part in batch:
+            _apply(after[part["file"]], part.get("changes") or [], part["file"])
+        old = set(cm.check_saved(before, names))
+        new = [x for x in cm.check_saved(after, names) if x not in old]
+        if new:
+            raise StoreError(422, {"errors": new})
+        out = []
+        for name in dict.fromkeys(names):
+            text = dump(after[name])
+            tmp = FOLDER / f".{name}.tmp"
+            tmp.write_text(text, encoding="utf-8", newline="\n")
+            os.replace(tmp, FOLDER / name)
+            out.append({"file": name, "version": version_of(text.encode("utf-8"))})
+        return out
+
+
+# ── 쓰기 모양 ─────────────────────────────────────────────────────────
+def _atom(v) -> str:
+    if isinstance(v, Decimal):
+        return str(v)
+    return json.dumps(v, ensure_ascii=False)
+
+
+def _flat(v) -> str:
+    if isinstance(v, dict):
+        return (
+            "{ " + ", ".join(f"{_atom(k)}: {_flat(x)}" for k, x in v.items()) + " }" if v else "{}"
+        )
+    if isinstance(v, list):
+        return "[" + ", ".join(_flat(x) for x in v) + "]"
+    return _atom(v)
+
+
+def dump(v, indent: int = 0, width: int = 160) -> str:
+    """UTF-8 · 들여쓰기 2칸 · 한 줄에 들어가는 묶음은 한 줄로."""
+    flat = _flat(v)
+    if indent and len(flat) + indent <= width or not isinstance(v, (dict, list)) or not v:
+        return flat + ("" if indent else "\n")
+    pad = " " * (indent + 2)
+    if isinstance(v, dict):
+        body = [f"{pad}{_atom(k)}: {dump(x, indent + 2, width)}" for k, x in v.items()]
+        text = "{\n" + ",\n".join(body) + "\n" + " " * indent + "}"
+    else:
+        text = (
+            "[\n"
+            + ",\n".join(pad + dump(x, indent + 2, width) for x in v)
+            + "\n"
+            + " " * indent
+            + "]"
+        )
+    return text + ("" if indent else "\n")