eomsangdonandClaude Opus 5.5
c7cbef86a0
- 기계.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>
330 lines
14 KiB
Python
330 lines
14 KiB
Python
"""M01 마스터 한 번 읽어 두기 — 목록 길이 요청마다 마스터를 새로 읽지 않게.
|
|
|
|
`M01_MasterData_Store.loaded` · `.blocks` 가 씀(읽기 전용 길만).
|
|
자국(파일 시각·크기)이 바뀐 파일만 다시 읽고, 로직 막힘도 미리 세어 둠 —
|
|
목록은 검사를 돌리지 않고 세어 둔 값을 씀. 고치는 길(`calc` · `text` · `save`)은
|
|
사본을 만지므로 그대로 `cm.load` 로 새로 읽음.
|
|
저장 전 검사(`check_saved`)도 여기 — 어느 로직이 어느 요소·표·로직을 보는지 세어 두고
|
|
그것으로 저장이 흔드는 줄만 검사함(`narrow`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import contextvars
|
|
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
|
|
import master_material as mm # noqa: E402 (자재품목 테이블ID — 재료 고르기는 그 묶음을 훑음)
|
|
|
|
_FRESH = threading.RLock()
|
|
_READ: dict[str, tuple[tuple, dict]] = {} # 파일 이름 → (자국, 읽은 내용)
|
|
_INDEX: dict = {} # 폴더 · 자국 · files · whole
|
|
_BLOCK: dict[str, tuple[tuple, dict]] = {} # 로직 파일 → (자국, {키: `_check` 가 센 것})
|
|
_BASE: dict[str, tuple] = {} # 막힘을 셀 때 바탕이던, 로직 아닌 파일의 자국
|
|
_MARKED: dict[str, set] = {} # 우리가 고친 요소·표 파일 → 흔들린 자리(`narrow` 가 고른 것)
|
|
#: `steady()` 범위 — 폴더 → 자국(범위 안에선 자국 한 번 · `to_thread` 도 같은 묶음)
|
|
_STEADY: contextvars.ContextVar[dict | None] = contextvars.ContextVar("m01_steady", default=None)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def steady():
|
|
"""요청 한 번처럼 마스터가 안 바뀌는 범위 — `loaded` 가 파일 자국을 한 번만 셈."""
|
|
token = _STEADY.set({}) if _STEADY.get() is None else None
|
|
try:
|
|
yield
|
|
finally:
|
|
if token is not None:
|
|
_STEADY.reset(token)
|
|
|
|
|
|
def _now(folder: Path) -> dict[str, tuple]:
|
|
held = _STEADY.get()
|
|
if held is None:
|
|
return _stamps(folder)
|
|
if folder not in held:
|
|
held[folder] = _stamps(folder)
|
|
return held[folder]
|
|
|
|
|
|
def _stamps(folder: Path) -> dict[str, tuple]:
|
|
"""첫 층 JSON 자국 — {이름: (시각, 크기)} · 밑줄 파일은 `cm.load` 가 안 읽으므로 뺌."""
|
|
out = {}
|
|
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
|
|
|
|
|
|
def _logic_files(files: dict[str, dict]):
|
|
for name, data in files.items():
|
|
if data.get("그룹") == "로직":
|
|
yield name, data
|
|
|
|
|
|
class _Watch:
|
|
"""`Master.get` 을 감싸 그 로직 줄이 본 요소·표를 적어 둠 — 저장 때 검사를 좁히는 지도."""
|
|
|
|
def __init__(self, whole):
|
|
self._whole, self.seen = whole, set()
|
|
|
|
def get(self, ref):
|
|
self.seen.add(str(ref))
|
|
return self._whole.get(ref)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._whole, name)
|
|
|
|
def __setattr__(self, name, value): # 속에 적어 둠 — 미리 갈라 둔 것을 줄마다 다시 안 짜게
|
|
if name in ("_whole", "seen"):
|
|
super().__setattr__(name, value)
|
|
else:
|
|
setattr(self._whole, name, value)
|
|
|
|
|
|
def _check(whole, row: dict) -> tuple[list[str], set, set]:
|
|
"""(막힘 까닭, 부르는 로직, 본 요소·표) — 본 것은 키 · 「ID:원문번호」 ·
|
|
재료 고르기가 있으면 자재품목 묶음 통째(「#MT」 · 후보를 그 안에서 고름)."""
|
|
watch = _Watch(whole)
|
|
why, calls = cm.mf.check_logic(watch, row)
|
|
if any(isinstance(h.get("요소"), dict) for h in row.get("호표") or []):
|
|
watch.seen.add("#" + mm.MARKET)
|
|
return why, calls, watch.seen
|
|
|
|
|
|
def forget() -> None:
|
|
"""읽어 둔 것을 버림 — 파일을 고친 자리(저장)에서 부름."""
|
|
with _FRESH:
|
|
_READ.clear()
|
|
_INDEX.clear()
|
|
_BLOCK.clear()
|
|
_BASE.clear()
|
|
_MARKED.clear()
|
|
if (held := _STEADY.get()) is not None:
|
|
held.clear()
|
|
|
|
|
|
def stale(names, marks=()) -> None:
|
|
"""그 파일만 다시 읽게 함 — 저장처럼 스스로 고친 자리에서 부름
|
|
(자국은 시각이 같은 틱이면 못 가릴 수 있으므로 고친 쪽이 알려 줌).
|
|
`marks` = 요소·표에서 흔들린 자리 — 그 자리를 보던 로직 줄만 다시 셈."""
|
|
with _FRESH:
|
|
for name in names:
|
|
was = _READ.pop(name, (None, {}))[1]
|
|
_BLOCK.pop(name, None)
|
|
if was.get("그룹") != "로직":
|
|
_MARKED.setdefault(name, set()).update(marks)
|
|
_INDEX.pop("자국", None)
|
|
if (held := _STEADY.get()) is not 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:
|
|
if _INDEX.get("폴더") != folder: # 시험은 사본 폴더로 바꿈
|
|
forget()
|
|
_INDEX["폴더"] = folder
|
|
now = _now(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:
|
|
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:
|
|
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("그룹") != "로직"}
|
|
marks = None # 요소·표에서 흔들린 자리 — None 이면 무엇이 바뀐지 모름(통째로 다시)
|
|
if _BASE != base: # 인력 · 재료 · 표가 바뀌면 바탕이 달라짐
|
|
off = {n for n in set(_BASE) | set(base) if _BASE.get(n) != base.get(n)}
|
|
marks = set()
|
|
for name in off:
|
|
if name not in _MARKED: # 다른 창이 고친 파일 — 어디가 흔들렸는지 모름
|
|
marks = None
|
|
break
|
|
marks |= _MARKED[name]
|
|
if marks is None:
|
|
_BLOCK.clear()
|
|
_BASE.clear()
|
|
_BASE.update(base)
|
|
_MARKED.clear()
|
|
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("키")): _check(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 []}
|
|
shaken = [
|
|
k
|
|
for k, (_, calls, seen) in got.items()
|
|
if calls & moved or (marks and seen & marks)
|
|
]
|
|
for key in shaken:
|
|
got[key] = _check(whole, rows[key])
|
|
return {key: why for _, got in _BLOCK.values() for key, (why, *_) in got.items()}
|
|
|
|
|
|
def counted(folder: Path) -> dict[str, tuple]:
|
|
"""로직 키마다 (막힘 까닭, 부르는 로직, 본 요소·표) — 미리 세어 둔 것."""
|
|
with _FRESH:
|
|
blocks(folder)
|
|
return {key: got for _, rows in _BLOCK.values() for key, got in rows.items()}
|
|
|
|
|
|
def _rows_of(data: dict) -> list:
|
|
return data.get("표") or data.get("줄") or []
|
|
|
|
|
|
def narrow(folder: Path, batch: list[dict], before: dict, after: dict):
|
|
"""(검사할 로직 키, 안 보는 줄이 부르는 키, 요소도 고쳤는지) — 저장이 흔드는 줄만.
|
|
|
|
로직 줄은 고친 줄과 **그 줄을 부르던 줄** · 요소·표 줄은 그 줄을 **보던 로직**
|
|
(키 · 「ID:원문번호」 · 자재품목은 묶음 통째 — `_Watch` 가 적어 둔 지도).
|
|
"""
|
|
counts = counted(folder)
|
|
calls = {key: got[1] for key, got in counts.items()}
|
|
touched, gone, marks = set(), set(), set()
|
|
for part in batch:
|
|
name = part["file"]
|
|
was, now = ({str(r.get("키")): r for r in _rows_of(side[name])} for side in (before, after))
|
|
keys = (set(now) - set(was)) | {
|
|
str(c.get("key")) for c in part.get("changes") or [] if c.get("op") != "add"
|
|
}
|
|
if after[name].get("그룹") == "로직":
|
|
touched |= keys
|
|
gone |= set(was) - set(now)
|
|
continue
|
|
for key in keys | (set(was) - set(now)): # 지운 줄도 — 가리키던 자리가 끊김
|
|
marks |= {key, "#" + key[:2]}
|
|
for side in (was, now):
|
|
number = (side.get(key) or {}).get("원문번호")
|
|
if number:
|
|
marks.add(f"{key[:2]}:{number}")
|
|
only = touched | {key for key, called in calls.items() if called & touched}
|
|
only |= {key for key, got in counts.items() if got[2] & marks}
|
|
return only, {key: called for key, called in calls.items() if key not in gone}, marks
|
|
|
|
|
|
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])]
|
|
out += cm.check_logics(files, whole, book, narrowed and narrowed[:2])
|
|
if narrowed and narrowed[2]: # 요소·표를 고쳤으면 연결 · 준용도 봄(로직 좁힘과 따로)
|
|
out += cm.check_links(whole)
|
|
return out + cm.mcb.check_all(files, whole) + labor_errors(whole)
|
|
|
|
|
|
def _labor_files() -> list[tuple[str, Path]]:
|
|
"""모든 층(마스터 · 회사 · 개인)의 일위대가 `labor/<열>.json` · 사전조합 `precombo/<키>.json`(51-11) —
|
|
(어디, 파일)."""
|
|
from M02_MasterTemplete import M02_Template_Layers as layers # M02 가 M01 을 부름 — 늦게
|
|
|
|
return [
|
|
(f"{what} {where}", path)
|
|
for kind, what in (("labor", "일위대가"), ("precombo", "사전조합"))
|
|
for where, path in layers.layer_files(kind)
|
|
]
|
|
|
|
|
|
def labor_errors(whole, keys=None) -> list[str]:
|
|
"""일위대가 · 사전조합 문서가 담은 로직 마스터에 없음(39-1d · 50-3) —
|
|
로직 지우기를 막음 · 어디서 쓰는지 알림 · 참조 줄(로직 없음)은 건너뜀. 전부터 깨진 것은
|
|
`save` 가 고치기 전 검사와 견주어 뺌. `keys` = 그 로직 키만 봄(빈 묶음이면 파일을 안 엶)."""
|
|
out = []
|
|
if keys is not None and not keys:
|
|
return out
|
|
for where, path in _labor_files():
|
|
try:
|
|
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 or (keys is not None and key not in keys):
|
|
continue
|
|
try:
|
|
whole.logic(key)
|
|
except cm.mf.FormulaError:
|
|
out.append(f"{where} 「{name}」 이 로직 「{key}」 을 씀 — 거기서 먼저 뺌")
|
|
return out
|
|
|
|
|
|
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
|