Aislo/resources/master_data/scripts/master_machine_files.py

"""M01 기계 = 폴더 `기계/` — 한 대가 파일 하나(826-2).

`기계/<키>.json` = 줄 하나 · `기계/_머리.json` = 머리(그룹 · 원문 · 구분) · 줄 차례 = 키 차례
(새 기계는 늘 다음 키라 끝에 붙음 · 옛 한 파일 차례와 같음).
읽는 쪽은 옛 한 파일 꼴 그대로 — `table()` 이 「기계.json」 {그룹, 원문, 구분, 줄} 로 지어 줌
(색인 · 계산식 · 화면 · API 는 그 이름을 그대로 씀). 쓰는 쪽은 바뀐 대만 — `write()`.
폴더가 없고 옛 `기계.json` 파일이 있으면 그 파일을 읽음(시험이 만든 작은 마스터).
"""

from __future__ import annotations

import hashlib
import json
import os
from decimal import Decimal
from pathlib import Path

NAME = "기계.json"  # 표 꼴 이름(메모리 · API)
DIR = "기계"
HEAD = "_머리.json"
_MEMO: dict[str, tuple] = {}  # 파일 경로 → (자국, 바이트, 읽은 줄)


def here(folder: Path) -> bool:
    return (Path(folder) / DIR / HEAD).is_file()


def names(folder: Path) -> list[str]:
    """첫 층 표 이름 — `*.json` 파일 + 기계 폴더(「기계.json」) · 이름 차례."""
    out = {p.name for p in Path(folder).glob("*.json")}
    return sorted(out | {NAME}) if here(folder) else sorted(out)


def _entries(folder: Path) -> list[os.DirEntry]:
    with os.scandir(Path(folder) / DIR) as it:
        return sorted(
            (e for e in it if e.name.endswith(".json") and not e.name.startswith(".")),
            key=lambda e: e.name,
        )


def stamp(folder: Path) -> tuple:
    """폴더 자국 — 파일마다 (이름, 시각, 크기)(Windows 는 목록이 자국을 같이 줌)."""
    return tuple((e.name, e.stat().st_mtime_ns, e.stat().st_size) for e in _entries(folder))


def _parse(raw: bytes):
    return json.loads(raw, parse_float=Decimal, parse_int=Decimal)


def _read(entry: os.DirEntry) -> tuple[bytes, object]:
    """(바이트, 읽은 것) — 자국이 같으면 기억한 것(읽은 것은 **읽기 전용**으로만 나눔)."""
    st = entry.stat()
    sig = (st.st_mtime_ns, st.st_size)
    got = _MEMO.get(entry.path)
    if got is None or got[0] != sig:
        raw = Path(entry.path).read_bytes()
        got = _MEMO[entry.path] = (sig, raw, _parse(raw))
    return got[1], got[2]


def _parts(folder: Path, shared: bool, parse=_parse) -> tuple[list[bytes], dict, list]:
    raws, head, rows = [], {}, []
    for entry in _entries(folder):
        raw, data = _read(entry)
        data = data if shared else parse(raw)
        raws.append(entry.name.encode() + b"\0" + raw)
        if entry.name == HEAD:
            head = data
        elif not entry.name.startswith("_"):
            rows.append(data)
    return raws, head, rows


def table(folder: Path, shared: bool = False, exact: bool = True) -> tuple[dict, str]:
    """(옛 한 파일 꼴, 판) — 판 = 머리 · 줄 파일 내용 지문. `shared` = 기억한 줄을 그대로 나눔
    (읽기 전용 기억 `Store_Cache` 만) · 아니면 새로 읽은 사본(고쳐도 됨) · `exact` 아니면 수 = int · float."""
    parse = _parse if exact else json.loads
    if not here(folder):
        raw = (Path(folder) / NAME).read_bytes()
        return parse(raw), hashlib.sha256(raw).hexdigest()[:16]
    raws, head, rows = _parts(folder, shared and exact, parse)
    return {**head, "줄": rows}, hashlib.sha256(b"\n".join(raws)).hexdigest()[:16]


def raw(folder: Path, name: str) -> bytes:
    """첫 층 표의 내용 바이트(지문용) — 기계 폴더는 파일들을 이은 것."""
    if name == NAME and here(folder):
        return b"\n".join(_parts(folder, True)[0])
    return (Path(folder) / name).read_bytes()


def read(folder: Path, name: str):
    """첫 층 표 하나(새로 읽은 사본 · 수 = Decimal)."""
    if name == NAME and here(folder):
        return table(folder)[0]
    return _parse((Path(folder) / name).read_bytes())


def text(row) -> str:
    """줄 파일 글 — 들여쓰기 2칸 · 수는 읽은 글 그대로(M01 저장 글 모양 `Store_Shape.dump`)."""
    import sys

    sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
    from M01_MasterData.M01_MasterData_Store_Shape import dump

    return dump(row, 0, width=0, expand=True)


def _put(path: Path, body: str) -> bool:
    data = body.encode("utf-8")
    if path.is_file() and path.read_bytes() == data:
        return False
    tmp = path.with_name(f".{path.name}.tmp")
    tmp.write_bytes(data)
    os.replace(tmp, path)
    return True


def write(folder: Path, data: dict, old: dict | None = None) -> list[str]:
    """표 꼴 → 폴더 — 바뀐 대만 씀 · 빠진 대는 지움 · 머리가 바뀌면 머리도. `old` = 고치기 전 표
    (없으면 폴더 전부와 견줌). 돌려줌 = 쓰거나 지운 파일 이름."""
    base = Path(folder) / DIR
    base.mkdir(exist_ok=True)
    rows = {f"{r['키']}.json": r for r in data.get("줄") or []}
    if len(rows) != len(data.get("줄") or []):
        raise ValueError("기계 키 겹침 — 한 대 한 파일")
    head = {k: v for k, v in data.items() if k != "줄"}
    was = None if old is None else {f"{r['키']}.json": r for r in old.get("줄") or []}
    done = []
    if (old is None or {k: v for k, v in old.items() if k != "줄"} != head) and _put(
        base / HEAD, text(head)
    ):
        done.append(HEAD)
    for name, row in rows.items():
        if (was is None or was.get(name) != row) and _put(base / name, text(row)):
            done.append(name)
    gone = set(was) - set(rows) if was is not None else {e.name for e in _entries(folder)}
    for name in sorted(gone - set(rows) - {HEAD}):
        (base / name).unlink(missing_ok=True)
        done.append(name)
    return done