feat(826-2): M01 기계 = 한 대 한 파일 · M01 저장 길 줄이기
- 기계.json 한 파일 → 폴더 기계/(EQ 키마다 파일 + _머리.json · 차례 = 키 차례) · 옮긴 빌더 build_기계폴더_826_2 (655 대 머리 · 줄 · 차례 대조 뒤 옛 파일 지움) - 읽기 도우미 master_machine_files — cm.load · Store_Cache · Store.read 가 옛 「기계.json」 한 표 꼴로 지어 줌(API · 화면 · 색인 그대로) · 쓰기 = 바뀐 대 파일만 - 기계.json 직접 열던 곳 도우미로: M02_San_Basis_Index · M02_Logic_Standard 지문 · build_용도 · master_keys.adopt · _틀 · _갱신절차 - 저장 몸 M01_MasterData_Store_Save — 전 = 기억 · 후 = 고친 파일만 얕은 사본 · 틀은 고친 줄만 · 조합 · 일위대가 문서는 로직 줄이 빠질 때만 · 재료 찾기 기억 이어받음 · 저장 뒤 쓴 파일만 기억에 갈아 끼움 - 마스터 사본 시험 22 개 = 기계 폴더도 복사 · 821_2 · 815_2 글 대조 = 대 파일 하나 대조(옛 한 파일 글 대조 낡음) - 새 시험 test_826_2 — 폴더 = 한 표 · 한 대 한 파일 · 새 · 지움 · 409 · 옛 길과 같은 오류 스무 견본 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -28,7 +28,7 @@ import master_text as mtx # noqa: E402
|
||||
from M01_MasterData import M01_MasterData_Store_Cache as cache # noqa: E402
|
||||
from M01_MasterData.M01_MasterData_Store_Find import hit as _hit # noqa: E402
|
||||
from M01_MasterData.M01_MasterData_Store_Find import squash as _squash # noqa: E402
|
||||
from M01_MasterData.M01_MasterData_Store_Shape import dump, keep_shape # noqa: E402
|
||||
from M01_MasterData.M01_MasterData_Store_Shape import dump # noqa: E402,F401
|
||||
|
||||
_COPY = os.environ.get("AISLO_MASTER_DIR") # 화면 저장 시험 사본 폴더(정함 13) · 없으면 정본
|
||||
FOLDER: Path = Path(_COPY).resolve() if _COPY else cm.MASTER # 시험은 사본 폴더로 바꿈
|
||||
@@ -47,7 +47,8 @@ class StoreError(Exception):
|
||||
# ── 읽기 ──────────────────────────────────────────────────────────────
|
||||
def _path(file: str) -> Path:
|
||||
path = FOLDER / file
|
||||
if "/" in file or "\\" in file or file.startswith("_") or not path.is_file():
|
||||
there = path.is_file() or (file == cm.mmf.NAME and cm.mmf.here(FOLDER)) # 기계 = 폴더(826-2)
|
||||
if "/" in file or "\\" in file or file.startswith("_") or not there:
|
||||
raise StoreError(404, f"없는 파일 「{file}」")
|
||||
return path
|
||||
|
||||
@@ -57,10 +58,19 @@ def version_of(raw: bytes) -> str:
|
||||
|
||||
|
||||
def read(file: str) -> tuple[dict, str]:
|
||||
raw = _path(file).read_bytes()
|
||||
_path(file)
|
||||
if file == cm.mmf.NAME:
|
||||
return cm.mmf.table(FOLDER)
|
||||
raw = (FOLDER / file).read_bytes()
|
||||
return json.loads(raw, parse_float=Decimal, parse_int=Decimal), version_of(raw)
|
||||
|
||||
|
||||
def version(file: str) -> str:
|
||||
"""판만(읽어 풀지 않음 — 저장 409 확인)."""
|
||||
_path(file)
|
||||
return version_of(cm.mmf.raw(FOLDER, file))
|
||||
|
||||
|
||||
# ── 한 번 읽어 두기(`_Store_Cache`) ────────────────────────────────────
|
||||
def loaded() -> tuple[dict[str, dict], object]:
|
||||
"""(파일 묶음, 색인) — **읽기 전용**. 고칠 것이면 `cm.load` 로 새로 읽을 것."""
|
||||
@@ -109,13 +119,15 @@ def files_of(group: str) -> list[dict]:
|
||||
if group not in cm.mf.GROUPS:
|
||||
raise StoreError(404, f"없는 그룹 「{group}」")
|
||||
out = []
|
||||
for path in [*FOLDER.glob(f"{group}.json"), *FOLDER.glob(f"{group}_*.json")]:
|
||||
data, version = read(path.name)
|
||||
for name in cm.mmf.names(FOLDER):
|
||||
if name != f"{group}.json" and not name.startswith(f"{group}_"):
|
||||
continue
|
||||
data, version = read(name)
|
||||
out.append(
|
||||
{
|
||||
"file": path.name,
|
||||
"file": name,
|
||||
"book": _book(data),
|
||||
"chapter": chapter_of(path.name, data),
|
||||
"chapter": chapter_of(name, data),
|
||||
"edition": data.get("판"),
|
||||
"rows": len(data.get(items_key(data)) or []),
|
||||
"version": version,
|
||||
@@ -659,37 +671,7 @@ def _apply(data: dict, changes: list[dict], file: str, book: dict) -> None:
|
||||
|
||||
|
||||
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 = 고칠 사본
|
||||
book_path = FOLDER / mk.BOOK.name
|
||||
book = mk.load_book(book_path)
|
||||
issued = dict(book["다음"])
|
||||
names = [part["file"] for part in batch]
|
||||
for part in batch:
|
||||
_apply(after[part["file"]], part.get("changes") or [], part["file"], book)
|
||||
mk.finish(book)
|
||||
# 저장이 흔드는 줄만 검사(고친 줄 · 그 줄을 부르던 줄 · 그 요소를 보던 줄) —
|
||||
# 마스터 전체 검사는 `check_master` 몫
|
||||
narrowed = cache.narrow(FOLDER, batch, before, after)
|
||||
old = set(cache.check_saved(before, names, mk.load_book(book_path), narrowed))
|
||||
new = [x for x in cache.check_saved(after, names, book, narrowed) if x not in old]
|
||||
if new:
|
||||
raise StoreError(422, {"errors": new})
|
||||
if book["다음"] != issued:
|
||||
mk.save_book(book, book_path)
|
||||
out = []
|
||||
for name in dict.fromkeys(names):
|
||||
old_text = (FOLDER / name).read_bytes().decode("utf-8")
|
||||
text = keep_shape(old_text, before[name], after[name], items_key(after[name])) or 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"))})
|
||||
# 저장 뒤에는 그 파일을 바로 다시 읽게 · 흔들린 자리를 알려 줘 그 줄만 다시 세게
|
||||
cache.stale(dict.fromkeys(names), narrowed[2])
|
||||
return out
|
||||
"""몸은 `M01_MasterData_Store_Save`(826-2 — 기억에서 전 · 후 · 바뀐 것만 검사)."""
|
||||
from M01_MasterData import M01_MasterData_Store_Save as saver
|
||||
|
||||
return saver.save(batch)
|
||||
|
||||
@@ -57,10 +57,12 @@ def _now(folder: Path) -> dict[str, tuple]:
|
||||
def _stamps(folder: Path) -> dict[str, tuple]:
|
||||
"""첫 층 JSON 자국 — {이름: (시각, 크기)} · 밑줄 파일은 `cm.load` 가 안 읽으므로 뺌."""
|
||||
out = {}
|
||||
for path in sorted(folder.glob("*.json")):
|
||||
if not path.name.startswith("_"):
|
||||
stat = path.stat()
|
||||
out[path.name] = (stat.st_mtime_ns, stat.st_size)
|
||||
for name in cm.mmf.names(folder):
|
||||
if name == cm.mmf.NAME and cm.mmf.here(folder): # 기계 폴더(826-2) — 대마다 자국
|
||||
out[name] = cm.mmf.stamp(folder)
|
||||
elif not name.startswith("_"):
|
||||
stat = (folder / name).stat()
|
||||
out[name] = (stat.st_mtime_ns, stat.st_size)
|
||||
return out
|
||||
|
||||
|
||||
@@ -127,6 +129,29 @@ def stale(names, marks=()) -> None:
|
||||
held.clear()
|
||||
|
||||
|
||||
def adopt(folder: Path, files: dict[str, dict], whole, names, marks=()) -> None:
|
||||
"""저장 뒤 — 쓴 파일(`names`)만 기억에 갈아 끼움(다시 읽지 않음 · `files` · `whole` = 저장한 뒤
|
||||
모습). 흔들린 자리는 `stale` 처럼 적어 둠. 그 사이 다른 파일도 바뀌었으면 `stale` 로."""
|
||||
with _FRESH:
|
||||
now = _stamps(folder)
|
||||
rest = [n for n in now if n not in names]
|
||||
if (
|
||||
_INDEX.get("폴더") != folder
|
||||
or set(now) != set(_READ) | set(names)
|
||||
or any(_READ[n][0] != now[n] for n in rest)
|
||||
):
|
||||
return stale(names, marks)
|
||||
for name in names:
|
||||
was = _READ.get(name, (None, {}))[1]
|
||||
_READ[name] = (now[name], files[name])
|
||||
_BLOCK.pop(name, None)
|
||||
if was.get("그룹") != "로직":
|
||||
_MARKED.setdefault(name, set()).update(marks)
|
||||
_INDEX.update(자국=now, files={n: _READ[n][1] for n in now}, whole=whole)
|
||||
if (held := _STEADY.get()) is not None:
|
||||
held.clear()
|
||||
|
||||
|
||||
def loaded(folder: Path) -> tuple[dict[str, dict], object]:
|
||||
"""(파일 묶음, 색인) — **읽기 전용**. 고칠 것이면 `cm.load` 로 새로 읽을 것."""
|
||||
with _FRESH:
|
||||
@@ -138,6 +163,9 @@ def loaded(folder: Path) -> tuple[dict[str, dict], object]:
|
||||
del _READ[gone]
|
||||
for name, stamp in now.items():
|
||||
if _READ.get(name, (None,))[0] != stamp:
|
||||
if name == cm.mmf.NAME and cm.mmf.here(folder): # 바뀐 대만 다시 읽음
|
||||
_READ[name] = (stamp, cm.mmf.table(folder, shared=True)[0])
|
||||
continue
|
||||
raw = (folder / name).read_bytes()
|
||||
_READ[name] = (stamp, json.loads(raw, parse_float=Decimal, parse_int=Decimal))
|
||||
if _INDEX.get("자국") != now:
|
||||
@@ -260,20 +288,25 @@ def _labor_files() -> list[tuple[str, Path]]:
|
||||
]
|
||||
|
||||
|
||||
def labor_errors(whole) -> list[str]:
|
||||
def labor_errors(whole, keys=None) -> list[str]:
|
||||
"""일위대가 · 사전조합 문서가 담은 로직 마스터에 없음(39-1d · 50-3) —
|
||||
로직 지우기를 막음 · 어디서 쓰는지 알림 · 참조 줄(로직 없음)은 건너뜀. 전부터 깨진 것은
|
||||
`save` 가 고치기 전 검사와 견주어 뺌."""
|
||||
`save` 가 고치기 전 검사와 견주어 뺌. `keys` = 그 로직 키만 봄(빈 묶음이면 파일을 안 엶)."""
|
||||
out = []
|
||||
if keys is not None and not keys:
|
||||
return out
|
||||
for where, path in _labor_files():
|
||||
try:
|
||||
doc = json.loads(path.read_bytes())
|
||||
raw = path.read_bytes()
|
||||
if keys is not None and not any(k.encode() in raw for k in keys):
|
||||
continue
|
||||
doc = json.loads(raw)
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
name = (doc.get("일위대가번") if isinstance(doc, dict) else None) or path.stem
|
||||
for one in (doc.get("담은로직") if isinstance(doc, dict) else None) or []:
|
||||
key = str(one.get("로직") or "") if isinstance(one, dict) else ""
|
||||
if not key:
|
||||
if not key or (keys is not None and key not in keys):
|
||||
continue
|
||||
try:
|
||||
whole.logic(key)
|
||||
|
||||
@@ -322,9 +322,8 @@ def save(batch: list[dict], confirmed: bool = False) -> dict:
|
||||
|
||||
def _keep_same(part: dict) -> None:
|
||||
"""고친 줄의 값이 같은 칸 = 파일 값 그대로(화면 왕복 「9.0」 → 「9」 막음 · 그 칸만 바뀜)."""
|
||||
try:
|
||||
data = store.read(str(part.get("file") or ""))[0]
|
||||
except store.StoreError:
|
||||
data = store.loaded()[0].get(str(part.get("file") or "")) # 읽기만 — 기억(826-2)
|
||||
if data is None:
|
||||
return
|
||||
rows = {str(r.get("키")): r for r in data.get(store.items_key(data)) or []}
|
||||
for ch in part.get("changes") or []:
|
||||
@@ -335,10 +334,7 @@ def _keep_same(part: dict) -> None:
|
||||
|
||||
|
||||
def _file_keys(file: str) -> list[str]:
|
||||
try:
|
||||
data = store.read(file)[0]
|
||||
except store.StoreError:
|
||||
return []
|
||||
data = store.loaded()[0].get(file) or {} # 읽기만 — 기억(826-2 · 저장 뒤엔 갈아 끼운 것)
|
||||
return [str(r.get("키")) for r in data.get(store.items_key(data)) or []]
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
"""M01 저장 — `M01_MasterData_Store.save` 의 몸(826-2 · 저장 길 줄이기).
|
||||
|
||||
전 = 기억(`Store_Cache.loaded`) · 후 = 고친 파일만 얕은 사본(줄 목록만 새로 · 줄은 바꿔 끼움만) —
|
||||
폴더를 다시 읽지 않음. 검사 = 저장이 흔드는 것만: 로직은 `Store_Cache.narrow` · 틀은 고친 줄만 ·
|
||||
연결 · 준용은 요소를 고쳤을 때 · 조합은 조합 파일을 고쳤거나 로직 줄이 빠질 때 · 일위대가 · 사전조합
|
||||
문서(모든 층 파일)는 로직 줄이 빠질 때 그 키만 · 키 겹침 · 대장 대조 · 돌고 도는 참조는 전체(가벼움).
|
||||
전 검사는 기억 색인(재료 고르기 찾기 기억이 데워져 있음) · 후 색인은 그 기억을 이어받음
|
||||
(자재품목 줄이 바뀌면 그 줄이 걸리는 찾기만 버림). 저장 뒤 = 쓴 파일만 기억에 갈아 끼움.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from M01_MasterData import M01_MasterData_Store as store
|
||||
from M01_MasterData import M01_MasterData_Store_Cache as cache
|
||||
from M01_MasterData.M01_MasterData_Store_Shape import dump, keep_shape
|
||||
|
||||
cm, mk = store.cm, store.mk
|
||||
mmf = cm.mmf
|
||||
|
||||
|
||||
def _copy(data: dict) -> dict:
|
||||
"""고칠 사본 — 머리는 그대로 나누고 줄 목록만 새로(`_apply` 는 줄을 바꿔 끼우기만 함)."""
|
||||
key = store.items_key(data)
|
||||
return {**data, key: list(data.get(key) or [])}
|
||||
|
||||
|
||||
def _rows(data: dict) -> dict[str, dict]:
|
||||
return {str(r.get("키")): r for r in data.get(store.items_key(data)) or []}
|
||||
|
||||
|
||||
def _only(name: str, data: dict, keys: set[str]) -> dict:
|
||||
"""틀 검사 몫 — 줄마다 보는 검사라 고친 줄만(조합 파일은 이름 겹침을 보므로 통째)."""
|
||||
if name == cm.mcb.FILE:
|
||||
return data
|
||||
key = store.items_key(data)
|
||||
return {**data, key: [r for r in data.get(key) or [] if str(r.get("키")) in keys]}
|
||||
|
||||
|
||||
def _index(files: dict, whole, moved: list[dict]):
|
||||
"""후 색인 — 전 색인의 찾기 기억을 이어받음. `moved` = 바뀐 자재품목 줄(전 · 후)."""
|
||||
out = cm.mf.Master(files)
|
||||
mp = cm.mf.mp
|
||||
mm = mp.mm
|
||||
if not moved:
|
||||
for name in ("_found", "_market", "_by_class"):
|
||||
if hasattr(whole, name):
|
||||
setattr(out, name, getattr(whole, name))
|
||||
return out
|
||||
found = getattr(whole, "_found", None) or {}
|
||||
texts = [
|
||||
(mm.plain(r.get("이름")).replace("×", "x"), mm.plain_spec(r.get("규격")), r) for r in moved
|
||||
]
|
||||
|
||||
def hit(memo_key) -> bool: # `master_pick._search` 의 거름과 같음 — 걸리면 답이 바뀔 수 있음
|
||||
term, kind, detail = memo_key[:3]
|
||||
words = mp._words(term)[0]
|
||||
return any(
|
||||
all(w in name or w in spec for w in words)
|
||||
and (not kind or row.get("구분") == kind)
|
||||
and (not detail or row.get("상세구분") == detail)
|
||||
for name, spec, row in texts
|
||||
)
|
||||
|
||||
out._found = {k: v for k, v in found.items() if not hit(k)}
|
||||
return out
|
||||
|
||||
|
||||
def _errors(files, whole, names, book, narrowed, keys, gone) -> list[str]:
|
||||
out = [x for name in names for x in cm.check_form(name, _only(name, files[name], keys[name]))]
|
||||
out += cm.check_logics(files, whole, book, narrowed[:2])
|
||||
if narrowed[2]: # 요소·표를 고쳤으면 연결 · 준용도 봄(로직 좁힘과 따로)
|
||||
out += cm.check_links(whole)
|
||||
if cm.mcb.FILE in names or gone:
|
||||
out += cm.mcb.check_all(files, whole)
|
||||
return out
|
||||
|
||||
|
||||
def save(batch: list[dict]) -> list[dict]:
|
||||
with store._LOCK:
|
||||
stale = [
|
||||
p.get("file", "") for p in batch if store.version(p.get("file", "")) != p.get("version")
|
||||
]
|
||||
if stale:
|
||||
raise store.StoreError(409, {"stale": stale})
|
||||
before, whole = store.loaded()
|
||||
names = list(dict.fromkeys(part["file"] for part in batch))
|
||||
after = {**before, **{name: _copy(before[name]) for name in names}}
|
||||
book_path = store.FOLDER / mk.BOOK.name
|
||||
book = mk.load_book(book_path)
|
||||
old_book = {"키": dict(book["키"]), "다음": dict(book["다음"])} # `issue` 는 새 키만 더함
|
||||
issued = dict(book["다음"])
|
||||
for part in batch:
|
||||
store._apply(after[part["file"]], part.get("changes") or [], part["file"], book)
|
||||
mk.finish(book)
|
||||
keys, gone, moved = {}, set(), []
|
||||
for name in names:
|
||||
was, now = _rows(before[name]), _rows(after[name])
|
||||
keys[name] = set(now) - set(was)
|
||||
keys[name] |= {
|
||||
str(c.get("key"))
|
||||
for p in batch
|
||||
if p["file"] == name
|
||||
for c in p.get("changes") or []
|
||||
if c.get("op") != "add"
|
||||
}
|
||||
if after[name].get("그룹") == "로직":
|
||||
gone |= set(was) - set(now)
|
||||
moved += [s[k] for k in keys[name] for s in (was, now) if k[:2] == "MT" and k in s]
|
||||
# 저장이 흔드는 줄만 검사(고친 줄 · 그 줄을 부르던 줄 · 그 요소를 보던 줄) —
|
||||
# 마스터 전체 검사는 `check_master` 몫
|
||||
narrowed = cache.narrow(store.FOLDER, batch, before, after)
|
||||
old = set(_errors(before, whole, names, old_book, narrowed, keys, gone))
|
||||
fresh = _index(after, whole, moved)
|
||||
new = _errors(after, fresh, names, book, narrowed, keys, gone)
|
||||
# 일위대가 · 사전조합 문서가 담은 로직이 빠짐 — 빠지기 전엔 있던 키라 전 검사는 없음
|
||||
new = [x for x in new + cache.labor_errors(fresh, gone) if x not in old]
|
||||
if new:
|
||||
raise store.StoreError(422, {"errors": new})
|
||||
if book["다음"] != issued:
|
||||
mk.save_book(book, book_path)
|
||||
out = [{"file": name, "version": _write(name, before[name], after[name])} for name in names]
|
||||
# 저장 뒤 — 쓴 파일만 기억에 갈아 끼우고 흔들린 자리를 알려 줘 그 줄만 다시 세게
|
||||
cache.adopt(store.FOLDER, after, fresh, names, narrowed[2])
|
||||
return out
|
||||
|
||||
|
||||
def _write(name: str, before: dict, after: dict) -> str:
|
||||
if name == mmf.NAME and mmf.here(store.FOLDER): # 기계 = 바뀐 대 파일만(826-2)
|
||||
mmf.write(store.FOLDER, after, before)
|
||||
return store.version(name)
|
||||
path = store.FOLDER / name
|
||||
old_text = path.read_bytes().decode("utf-8")
|
||||
key = store.items_key(after)
|
||||
text = keep_shape(old_text, before, after, key) or dump(after)
|
||||
tmp = store.FOLDER / f".{name}.tmp"
|
||||
tmp.write_text(text, encoding="utf-8", newline="\n")
|
||||
os.replace(tmp, path)
|
||||
return store.version_of(text.encode("utf-8"))
|
||||
Reference in New Issue
Block a user