"""M02 층 목록 요약 색인(826-1 · 요구 804 · 조사 804 안 A) — 층 목록이 항목 파일을 다시 열지 않게. 항목 기록 `{자국, 판, 바탕?, 기댐?, 요약?}`(열쇠 「종류/이름」): 자국 파일 고친 때 ns : 크기 — 같으면 기억한 판(목록 표시만 · 저장 · 지움 409 는 그대로 `layers.version_of` 파일 내용) 요약 서고 줄 `M02_Labor_Ref.summary`(쓰는곳수 뺌) — 자국 · 바탕 · 기댐이 모두 같을 때만 씀 바탕 셈 코드 내용 · M01 폴더 · 층 곁 파일(내 기계 등 — 항목 폴더 · manifest 뺌) 자국 지문 기댐 셈 때 읽은 참조 파일(SC · SK) 자국(`refs.SEEN`) 쓰는곳수 = 요청마다 쓰는 곳 파일을 한 번 훑어 거꾸로 셈(`refs._ids_of` 기억 · 파일당 stat 한 번). 자리 = 프로세스 메모리 + storage `_derived/m02_layer_index/`(파생 저장본 · 층마다 한 파일 · 참조 이름표 기억 `ids.json`) — 없거나 깨지면 다시 지음 · 지워도 됨 · 다른 PC · 옛 코드는 안 읽음. """ from __future__ import annotations import hashlib import threading from functools import lru_cache from pathlib import Path from typing import Any from common_util.common_util_derived_store import derived_dir, derived_key, file_stamp, read_derived from common_util.common_util_json import atomic_write_json from config import config_system from M02_MasterTemplete import M02_Labor_Ref as refs from M02_MasterTemplete import M02_MasterTemplete_Store as store from M02_MasterTemplete import M02_Template_Layers as layers KIND = "m02_layer_index" SHAPE = 1 # 기록 꼴이 바뀌면 올림(옛 기록 = 다시 셈) IDS = "ids" CODE_DIRS = ("M02_MasterTemplete", "M01_MasterData", "resources/master_data/scripts", "common_util") _LOCK = threading.RLock() _MEM: dict[str, dict[str, Any]] = {} # 층 열쇠 → {항목: 기록} _IDS_LOADED: list[bool] = [] @lru_cache(maxsize=1) def _code() -> str: """셈 코드 내용 지문 — 프로세스에 한 번(코드가 바뀌면 서버가 다시 뜸 · 같은 코드 = 같은 값).""" digest = hashlib.blake2b(digest_size=16) for sub in CODE_DIRS: for path in sorted((config_system.PROJECT_ROOT / sub).glob("*.py")): digest.update(path.name.encode()) digest.update(path.read_bytes()) return digest.hexdigest() def _stamps(folder: Path, skip: tuple[str, ...] = ()) -> list[tuple[str, str | None]]: """폴더 안 파일 자국 — 첫 단 이름이 `skip` · `_` · `.` 머리인 것은 뺌.""" if not folder.is_dir(): return [] out = [] for path in sorted(folder.rglob("*")): rel = path.relative_to(folder).parts if rel[0] in skip or any(p.startswith(("_", ".")) for p in rel) or not path.is_file(): continue out.append(("/".join(rel), file_stamp(path))) return out def _ground(folder: Path) -> str: """바탕 지문 — 셈 코드 · M01 폴더 · 이 층 · 지금 층 기본 자리의 곁 파일.""" from M01_MasterData import M01_MasterData_Store as m01 skip = (*layers.KINDS, layers.MANIFEST_NAME) side = [_stamps(f, skip) for f in dict.fromkeys((folder, store.root()))] return derived_key(SHAPE, _code(), str(m01.FOLDER), _stamps(Path(m01.FOLDER)), side) # ── 기록 자리 ───────────────────────────────────────────────────────── def _book(folder: Path) -> tuple[str, dict[str, Any]]: key = derived_key(SHAPE, str(folder.resolve())) if key not in _MEM: got = read_derived(layers.storage_root(), KIND, key) value = got.get("value") if got else None _MEM[key] = value if isinstance(value, dict) else {} return key, _MEM[key] def _save(key: str, value: Any) -> None: path = derived_dir(layers.storage_root(), KIND) / f"{key}.json" try: atomic_write_json(path, {"key": key, "value": value}) except OSError: # 파생 저장본 — 못 써도 다음에 다시 셈 pass # ── 목록 · 판 ───────────────────────────────────────────────────────── def list_rows(folder: str | Path) -> list[dict[str, Any]]: """`layers.list_templates` 와 같은 줄 — 판은 자국이 같으면 기억한 것(파일 안 엶).""" folder = Path(folder) with _LOCK: key, recs = _book(folder) dirty: list[str] = [] def version(path: Path) -> str | None: name = f"{path.parent.name}/{path.stem}" stamp = file_stamp(path) # 읽기 전 자국 — 그새 바뀌면 다음에 다시 셈 rec = recs.get(name) if stamp and isinstance(rec, dict) and rec.get("자국") == stamp and "판" in rec: return rec["판"] recs[name] = {"자국": stamp, "판": layers.version_of(path)} dirty.append(name) return recs[name]["판"] rows = layers.list_templates(folder, version) have = {f"{row['종류']}/{row['이름']}" for row in rows} gone = [name for name in recs if name not in have] for name in gone: del recs[name] if dirty or gone: _save(key, recs) return rows # ── 서고 요약 ───────────────────────────────────────────────────────── def _good(rec: Any, stamp: str | None, ground: str) -> bool: if not (stamp and isinstance(rec, dict) and rec.get("자국") == stamp): return False deps = rec.get("기댐") if rec.get("바탕") != ground or not isinstance(rec.get("요약"), dict): return False return isinstance(deps, dict) and all(file_stamp(Path(p)) == s for p, s in deps.items()) def _fresh(folder: Path, kind: str, name: str, rec: Any, stamp: str | None, whole: Any) -> dict: """그 항목만 다시 읽어 셈 — 참조는 읽기 직전 자국을 기댐에 적음.""" from M02_MasterTemplete.M02_Labor_Expand import Expand path = folder / kind / f"{name}.json" deps: dict[str, str | None] = {} def note(p: Path) -> None: deps.setdefault(str(p), file_stamp(p)) token = refs.SEEN.set(note) try: # 셈마다 새 펼침 — 다른 항목 셈 기억을 나눠 쓰면 기댐이 빠짐 out = refs.summary(folder, kind, name, Expand(whole), []) finally: refs.SEEN.reset(token) out.pop("쓰는곳수", None) same = isinstance(rec, dict) and rec.get("자국") == stamp and "판" in rec return { "자국": stamp, "판": rec["판"] if same else layers.version_of(path), "기댐": deps, "요약": out, } def summaries(folder: str | Path, rows: list[dict[str, Any]]) -> None: """`refs.summaries` 와 같은 칸 · 차례 — 기록이 맞으면 파일 안 엶 · 다르면 그 항목만 셈.""" folder = Path(folder) lib = [row for row in rows if row["종류"] in layers.LIBRARY_KINDS] if not lib: return with _LOCK: key, recs = _book(folder) ground, whole, dirty = _ground(folder), None, False for row in lib: name = f"{row['종류']}/{row['이름']}" path = folder / row["종류"] / f"{row['이름']}.json" stamp = file_stamp(path) if not _good(recs.get(name), stamp, ground): if whole is None: from M01_MasterData import M01_MasterData_Store as m01 whole = m01.loaded()[1] rec = _fresh(folder, row["종류"], row["이름"], recs.get(name), stamp, whole) recs[name] = {**rec, "바탕": ground} dirty = True row.update(recs[name]["요약"]) if dirty: _save(key, recs) sk = [row for row in lib if row["종류"] == "sangeun" and "고친줄수" in row] if sk: uses, (layer, owner) = _uses(), refs.where(folder) for row in sk: rid = refs.ref_id({"층": layer, "주인": owner, "키": row["이름"]}) row["쓰는곳수"] = uses.get(rid, 0) # ── 쓰는 곳 수 ──────────────────────────────────────────────────────── def _uses() -> dict[str, int]: """참조 이름표 → 담은 파일 수(`refs.users` 개수와 같음) — 쓰는 곳 파일 한 번 훑음.""" _load_ids() before = {path: hit[0] for path, hit in refs._IDS.items()} files = [path for _what, _at, path in refs.use_files()] count: dict[str, int] = {} for path in files: got = refs._ids_of(path) for rid in got[1] if got else (): count[rid] = count.get(rid, 0) + 1 if any(before.get(path) != refs._IDS[path][0] for path in files if path in refs._IDS): _save_ids() return count def _load_ids() -> None: """참조 이름표 기억을 파생 저장본에서 채움 — 프로세스에 한 번 · 자국은 `_ids_of` 가 다시 봄.""" if _IDS_LOADED: return _IDS_LOADED.append(True) got = read_derived(layers.storage_root(), KIND, IDS) for text, one in ((got or {}).get("value") or {}).items(): try: mtime, size, name, ids = one refs._IDS.setdefault(Path(text), ((int(mtime), int(size)), (str(name), frozenset(ids)))) except (TypeError, ValueError): continue def _save_ids() -> None: """지금 기억을 디스크 것에 덧씀(다른 서버 몫 유지) · 없어진 파일 뺌.""" got = read_derived(layers.storage_root(), KIND, IDS) value = dict((got or {}).get("value") or {}) for path, ((mtime, size), (name, ids)) in list(refs._IDS.items()): value[str(path)] = [mtime, size, name, sorted(ids)] value = {text: one for text, one in value.items() if Path(text).is_file()} _save(IDS, value)