eomsangdonandClaude Opus 5
3a36a0372c
- 화면: 왼쪽 로직 목록(원문·장·이름 찾기·막힘) / 가운데 머리·받을 값·호표·중간 값·덧줄·끝수 / 오른쪽 시험 계산 - 고친 것은 캐시에 쌓고 [저장] 한 번에 · [고친 것 버리기] · 줄 더하기·지우기 · 로직 새로 만들기·지우기 - 진입은 mountM01Logic(host) — 공용 진입 파일(sub_laptop_3 몫)이 한 줄로 붙임 - 서버: 로직 하나에 단가 자동(prices) · 요소 찾기(GET /elements) · 저장 전 시험 계산(calc 에 row·file) - 계약 문서 갱신 · 시험 1개 더함(9개 통과) Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PAdA5ThqmtVSsbzjusk1cJ
354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""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:
|
|
whole = cm.mf.Master(files)
|
|
reasons = cm.mf.check_logic(whole, book, row)[0]
|
|
_, version = read(name)
|
|
return {
|
|
"file": name,
|
|
"version": version,
|
|
"logic": row,
|
|
"blocked": bool(reasons),
|
|
"reasons": reasons,
|
|
"prices": _prices(whole, row),
|
|
}
|
|
raise StoreError(404, f"없는 로직 「{book}:{key}」")
|
|
|
|
|
|
def _brief(ref: str, row: dict) -> dict:
|
|
return {"ref": ref, **{k: row.get(k) for k in ("이름", "규격", "단위", "값", "값칸")}}
|
|
|
|
|
|
def _prices(whole, row: dict) -> dict:
|
|
"""호표 요소 → 요소 줄 요약(단가 자동 칸) · `{이름}` 낀 열쇠·하위 로직은 계산 때 정해짐."""
|
|
out = {}
|
|
for item in row.get("호표") or []:
|
|
ref = str(item.get("요소", ""))
|
|
if item.get("종류") != "로직" and "{" not in ref:
|
|
try:
|
|
out[ref] = _brief(ref, whole.get(ref))
|
|
except cm.mf.FormulaError:
|
|
out[ref] = None
|
|
return out
|
|
|
|
|
|
def elements(group: str, q: str, limit: int) -> dict:
|
|
"""요소 찾기 창 — 그룹 전체에서 이름·열쇠 찾기 · `ref` 는 식에 그대로 넣는 `그룹:원문:열쇠`."""
|
|
if group not in cm.mf.GROUPS:
|
|
raise StoreError(404, f"없는 그룹 「{group}」")
|
|
hits = []
|
|
for name, data in cm.load(folder=FOLDER).items():
|
|
if data.get("그룹") != group:
|
|
continue
|
|
for row in data.get(items_key(data)) or []:
|
|
if _hit(row, q):
|
|
ref = f"{group}:{data.get('원문')}:{row.get('열쇠')}"
|
|
hits.append({**_brief(ref, row), "file": name})
|
|
return {"total": len(hits), "items": hits[: min(max(limit, 1), 200)]}
|
|
|
|
|
|
# ── 시험 계산 ──────────────────────────────────────────────────────────
|
|
def _swap(files: dict, book: str, key: str, row: dict, file: str | None) -> None:
|
|
"""저장 전 시험 — 메모리 사본에서 그 로직을 고친 줄로 바꿈(없으면 `file` 에 더함)."""
|
|
row = _dec(row)
|
|
for name, data in _logic_files(files):
|
|
if data.get("원문") == book:
|
|
items = data.get("줄") or []
|
|
for i, old in enumerate(items):
|
|
if str(old.get("열쇠")) == key:
|
|
items[i] = row
|
|
return
|
|
if file not in files or files[file].get("그룹") != "로직":
|
|
raise StoreError(404, f"없는 로직 파일 「{file}」")
|
|
files[file].setdefault("줄", []).append(row)
|
|
|
|
|
|
def calc(book: str, key: str, inputs: dict, row: dict | None = None, file: str | None = None):
|
|
given = {k: _dec(v) for k, v in inputs.items()}
|
|
files = cm.load(folder=FOLDER)
|
|
if row is not None:
|
|
_swap(files, book, key, row, file)
|
|
key = str(row.get("열쇠", key))
|
|
try:
|
|
result = cm.calc(files, 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")
|