Aislo/M01_MasterData/M01_MasterData_Store_Cache.py

"""M01 마스터 한 번 읽어 두기 — 목록 길이 요청마다 마스터를 새로 읽지 않게.

`M01_MasterData_Store.loaded` · `.blocks` 가 씀(읽기 전용 길만).
자국(파일 시각·크기)이 바뀐 파일만 다시 읽고, 로직 막힘도 미리 세어 둠 —
목록은 검사를 돌리지 않고 세어 둔 값을 씀. 고치는 길(`calc` · `text` · `save`)은
사본을 만지므로 그대로 `cm.load` 로 새로 읽음.
저장 전 검사(`check_saved`)도 여기 — 세어 둔 부르기 그림으로 고친 줄 둘레만 봄(`narrow`).
"""

from __future__ import annotations

import json
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

_FRESH = threading.RLock()
_READ: dict[str, tuple[tuple, dict]] = {}  # 파일 이름 → (자국, 읽은 내용)
_INDEX: dict = {}  # 폴더 · 자국 · files · whole
_BLOCK: dict[str, tuple[tuple, dict]] = {}  # 로직 파일 → (자국, {키: (막힘 까닭, 부르는 로직)})
_BASE: 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)
    return out


def _logic_files(files: dict[str, dict]):
    for name, data in files.items():
        if data.get("그룹") == "로직":
            yield name, data


def forget() -> None:
    """읽어 둔 것을 버림 — 파일을 고친 자리(저장)에서 부름."""
    with _FRESH:
        _READ.clear()
        _INDEX.clear()
        _BLOCK.clear()
        _BASE.clear()


def stale(names) -> None:
    """그 파일만 다시 읽게 함 — 저장처럼 스스로 고친 자리에서 부름
    (자국은 시각이 같은 틱이면 못 가릴 수 있으므로 고친 쪽이 알려 줌)."""
    with _FRESH:
        for name in names:
            was = _READ.pop(name, (None, {}))[1]
            _BLOCK.pop(name, None)
            if was.get("그룹") != "로직":
                _BASE.clear()  # 요소·표가 바뀌면 모든 로직을 다시 셈
        _INDEX.pop("자국", None)


def loaded(folder: Path) -> tuple[dict[str, dict], object]:
    """(파일 묶음, 색인) — **읽기 전용**. 고칠 것이면 `cm.load` 로 새로 읽을 것."""
    with _FRESH:
        if _INDEX.get("폴더") != folder:  # 시험은 사본 폴더로 바꿈
            forget()
            _INDEX["폴더"] = folder
        now = _stamps(folder)
        for gone in [name for name in _READ if name not in now]:
            del _READ[gone]
        for name, stamp in now.items():
            if _READ.get(name, (None,))[0] != stamp:
                raw = (folder / name).read_bytes()
                _READ[name] = (stamp, json.loads(raw, parse_float=Decimal, parse_int=Decimal))
        if _INDEX.get("자국") != now:
            files = {name: _READ[name][1] for name in now}
            _INDEX.update(자국=now, files=files, whole=cm.mf.Master(files))
        return _INDEX["files"], _INDEX["whole"]


def blocks(folder: Path) -> dict[str, list[str]]:
    """로직 키마다 막힘 까닭 — 미리 세어 둔 값. 요소 파일이 바뀌면 모든 로직을,
    로직 파일은 바뀐 그 파일과 그 로직을 부르던 줄만 다시 셈."""
    with _FRESH:
        files, whole = loaded(folder)
        stamps = _INDEX["자국"]
        base = {n: s for n, s in stamps.items() if files[n].get("그룹") != "로직"}
        if _BASE != base:  # 인력 · 재료 · 표가 바뀌면 바탕이 달라짐
            _BLOCK.clear()
            _BASE.clear()
            _BASE.update(base)
        moved, fresh = set(), set()  # 바뀐 로직 파일의 키(옛것 · 새것)
        names = {name for name, _ in _logic_files(files)}
        for gone in [name for name in _BLOCK if name not in names]:  # 파일째 사라진 것
            moved |= set(_BLOCK.pop(gone)[1])
        for name, data in _logic_files(files):
            if _BLOCK.get(name, (None,))[0] == stamps[name]:
                continue
            got = {str(r.get("키")): cm.mf.check_logic(whole, r) for r in data.get("줄") or []}
            moved |= set(_BLOCK.get(name, (None, {}))[1]) | set(got)
            _BLOCK[name] = (stamps[name], got)
            fresh.add(name)
        for name, data in _logic_files(files):  # 다른 파일에서 그 로직을 부르던 줄
            if name in fresh:
                continue
            got = _BLOCK[name][1]
            rows = {str(r.get("키")): r for r in data.get("줄") or []}
            for key in [k for k, (_, calls) in got.items() if calls & moved]:
                got[key] = cm.mf.check_logic(whole, rows[key])
        return {key: why for _, got in _BLOCK.values() for key, (why, _) in got.items()}


def graph(folder: Path) -> dict[str, set]:
    """로직 키마다 부르는 로직 키 — 미리 세어 둔 것(저장 때 검사를 좁히는 데 씀)."""
    with _FRESH:
        blocks(folder)
        return {key: calls for _, got in _BLOCK.values() for key, (_, calls) in got.items()}


def narrow(folder: Path, batch: list[dict], before: dict, after: dict):
    """(검사할 로직 키, 안 보는 줄이 부르는 키) — 저장이 건드린 줄과 그 줄을 부르던 줄만.

    로직 밖 파일(요소 · 표)이 섞이면 어느 로직이 흔들릴지 모르므로 None — 전부 검사.
    """
    if any(after[part["file"]].get("그룹") != "로직" for part in batch):
        return None
    calls, touched, gone = graph(folder), set(), set()
    for part in batch:
        was, now = [
            {str(r.get("키")) for r in side[part["file"]].get("줄") or []}
            for side in (before, after)
        ]
        touched |= now - was  # 더한 줄(키는 저장 때 대장이 냄)
        gone |= was - now
        touched |= {str(c.get("key")) for c in part.get("changes") or [] if c.get("op") != "add"}
    only = touched | {key for key, called in calls.items() if called & touched}
    return only, {key: called for key, called in calls.items() if key not in gone}


def check_saved(files: dict, changed: list[str], book: dict, narrowed=None) -> list[str]:
    """저장 전 검사(M01) — 고친 파일의 틀 + 로직의 변수 · 키 + 조합.

    `files` = 폴더 전부(고친 뒤) · `book` = 키 대장 · `narrowed` = `narrow` 가 준 좁힘
    (None 이면 전부 검사 — 마스터 전체 검사는 `check_master` 몫).
    """
    whole = cm.mf.Master(files)
    out = [x for name in changed for x in cm.check_form(name, files[name])]
    return out + cm.check_logics(files, whole, book, narrowed) + cm.mcb.check_all(files, whole)


def warm(folder: Path) -> threading.Thread:
    """서버가 뜰 때 뒤에서 한 번 읽어 둠 — 켠 뒤 첫 목록이 3초 걸리던 것을 없앰."""

    def run() -> None:
        try:
            blocks(folder)
        except Exception:  # 마스터가 깨져 있어도 서버 시동을 막지 않음 — 목록에서 다시 드러남
            pass

    thread = threading.Thread(target=run, name="m01-warm", daemon=True)
    thread.start()
    return thread