Aislo/M01_MasterData/M01_MasterData_Store_Save.py

"""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"))