From 6a2925a9ecba2adc5364bdfbe6482437a1ddd071 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 22 Sep 2026 03:10:31 +0900 Subject: [PATCH] auto: 2026-09-22 03:10 (EOMSANGDON-HOME) --- M01_MasterData/M01_MasterData_Router.py | 4 + M01_MasterData/M01_MasterData_Store.py | 42 +++- M01_MasterData/M01_MasterData_Store_Cache.py | 234 +++++++++++++++++++ M01_MasterData/M01_MasterData_Store_Combo.py | 7 +- M01_MasterData/M01_MasterData_Store_Make.py | 4 +- M01_MasterData/M01_MasterData_UI_Page.ts | 7 +- M01_MasterData/M01_MasterData_UI_Side.ts | 12 +- M01_MasterData/M01_MasterData_UI_Style.css | 13 ++ 8 files changed, 305 insertions(+), 18 deletions(-) create mode 100644 M01_MasterData/M01_MasterData_Store_Cache.py diff --git a/M01_MasterData/M01_MasterData_Router.py b/M01_MasterData/M01_MasterData_Router.py index 2c212d81..191ed674 100644 --- a/M01_MasterData/M01_MasterData_Router.py +++ b/M01_MasterData/M01_MasterData_Router.py @@ -5,6 +5,7 @@ from __future__ import annotations +import sys from decimal import Decimal from typing import Any, Literal @@ -17,6 +18,9 @@ from M01_MasterData import M01_MasterData_Store_Make as make router = APIRouter(prefix="/api/m01", tags=["M01 MasterData"]) +if "pytest" not in sys.modules: # 시험은 사본 폴더를 써서 정본을 미리 읽을 까닭이 없음 + store.warm() # 서버가 뜰 때 뒤에서 마스터를 한 번 읽어 둠 — 첫 목록을 기다리지 않게 + class CalcBody(BaseModel): key: str diff --git a/M01_MasterData/M01_MasterData_Store.py b/M01_MasterData/M01_MasterData_Store.py index 58fddb8a..cf8d28e0 100644 --- a/M01_MasterData/M01_MasterData_Store.py +++ b/M01_MasterData/M01_MasterData_Store.py @@ -24,6 +24,7 @@ import master_copy as mcp # noqa: E402,F401 (`_Store_Make` 가 `store.mcp` 로 import master_keys as mk # noqa: E402 import master_text as mtx # noqa: E402 +from M01_MasterData import M01_MasterData_Store_Cache as cache # noqa: E402 from M01_MasterData.M01_MasterData_Store_Shape import dump, keep_shape # noqa: E402 FOLDER: Path = cm.MASTER # 시험은 사본 폴더로 바꿈 @@ -56,6 +57,22 @@ def read(file: str) -> tuple[dict, str]: return json.loads(raw, parse_float=Decimal, parse_int=Decimal), version_of(raw) +# ── 한 번 읽어 두기(`_Store_Cache`) ──────────────────────────────────── +def loaded() -> tuple[dict[str, dict], object]: + """(파일 묶음, 색인) — **읽기 전용**. 고칠 것이면 `cm.load` 로 새로 읽을 것.""" + return cache.loaded(FOLDER) + + +def blocks() -> dict[str, list[str]]: + """로직 키마다 막힘 까닭 — 미리 세어 둔 값(목록에서 검사를 돌리지 않음).""" + return cache.blocks(FOLDER) + + +def warm() -> None: + """서버가 뜰 때 뒤에서 미리 읽어 둠 — 첫 목록을 기다리지 않게.""" + cache.warm(FOLDER) + + def items_key(data: dict) -> str: return "표" if data.get("그룹") in cm.mf.TABLE_GROUPS else "줄" @@ -318,9 +335,10 @@ def _logic_files(files: dict[str, dict]): def logics(sub: str, detail: str, q: str, blocked: int | None, owner: str = "") -> list[dict]: - """로직 목록 — 구분 · 상세구분 · 소유 · 찾기로 거름(장 차례대로 · 파일을 가로지름).""" - files = cm.load(folder=FOLDER) - whole = cm.mf.Master(files) + """로직 목록 — 구분 · 상세구분 · 소유 · 찾기로 거름(장 차례대로 · 파일을 가로지름). + 막힘은 미리 세어 둔 값(`blocks`) — 목록에서 줄마다 검사를 돌리지 않음.""" + files, _ = loaded() + why_of = blocks() out = [] for name, data in sorted(_logic_files(files), key=lambda f: f[1].get("차례") or 0): for row in data.get("줄") or []: @@ -330,7 +348,7 @@ def logics(sub: str, detail: str, q: str, blocked: int | None, owner: str = "") continue if not _hit(row, q): continue - reasons = cm.mf.check_logic(whole, row)[0] + reasons = why_of.get(str(row.get("키")), []) if blocked is not None and bool(reasons) != bool(blocked): continue out.append( @@ -358,11 +376,10 @@ def logics(sub: str, detail: str, q: str, blocked: int | None, owner: str = "") def logic(key: str) -> dict: """로직 하나 — 키는 전체에서 하나.""" - files = cm.load(folder=FOLDER) + files, whole = loaded() for name, data in _logic_files(files): for row in data.get("줄") or []: if str(row.get("키")) == key: - whole = cm.mf.Master(files) reasons = cm.mf.check_logic(whole, row)[0] _, version = read(name) return { @@ -427,7 +444,7 @@ def elements(group: str, q: str, limit: int) -> dict: if group not in cm.mf.GROUPS: raise StoreError(404, f"없는 그룹 「{group}」") hits = [] - for name, data in cm.load(folder=FOLDER).items(): + for name, data in loaded()[0].items(): if data.get("그룹") != group: continue for row in data.get(items_key(data)) or []: @@ -491,7 +508,7 @@ def materials(sub: str, detail: str, spec: str, region: str, unit: str, limit: i """재료 고르기 조건 안 후보 목록 — 화면·테스트 컨테이너가 같이 씀. `unit` = 호표 줄 단위 — 주면 그 단위 줄만(단위가 다르면 다른 물건). `기본` = 시험 계산이 쓸 줄(대표 줄이 없을 때의 첫 줄) · 그 줄을 `items` 맨 앞에 둠.""" - whole = cm.master(folder=FOLDER) + whole = loaded()[1] cond = {"구분": sub, "상세구분": detail or None, "규격": spec or None} env = {cm.mf.REGION: region} if region else {} keys = cm.mf.candidates(whole, cond, env, unit or None) @@ -648,8 +665,11 @@ def save(batch: list[dict]) -> list[dict]: for part in batch: _apply(after[part["file"]], part.get("changes") or [], part["file"], book) mk.finish(book) - old = set(cm.check_saved(before, names, mk.load_book(book_path))) - new = [x for x in cm.check_saved(after, names, book) if x not in old] + # 저장이 흔드는 줄만 검사(고친 줄 · 그 줄을 부르던 줄 · 그 요소를 보던 줄) — + # 마스터 전체 검사는 `check_master` 몫 + narrowed = cache.narrow(FOLDER, batch, before, after) + old = set(cache.check_saved(before, names, mk.load_book(book_path), narrowed)) + new = [x for x in cache.check_saved(after, names, book, narrowed) if x not in old] if new: raise StoreError(422, {"errors": new}) if book["다음"] != issued: @@ -664,4 +684,6 @@ def save(batch: list[dict]) -> list[dict]: tmp.write_text(text, encoding="utf-8", newline="\n") os.replace(tmp, FOLDER / name) out.append({"file": name, "version": version_of(text.encode("utf-8"))}) + # 저장 뒤에는 그 파일을 바로 다시 읽게 · 흔들린 자리를 알려 줘 그 줄만 다시 세게 + cache.stale(dict.fromkeys(names), narrowed[2]) return out diff --git a/M01_MasterData/M01_MasterData_Store_Cache.py b/M01_MasterData/M01_MasterData_Store_Cache.py new file mode 100644 index 00000000..c704de76 --- /dev/null +++ b/M01_MasterData/M01_MasterData_Store_Cache.py @@ -0,0 +1,234 @@ +"""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 +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` 가 고른 것) + + +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 + + +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() + + +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) + + +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("그룹") != "로직"} + 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) + + +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 diff --git a/M01_MasterData/M01_MasterData_Store_Combo.py b/M01_MasterData/M01_MasterData_Store_Combo.py index e8ae61bb..c363639f 100644 --- a/M01_MasterData/M01_MasterData_Store_Combo.py +++ b/M01_MasterData/M01_MasterData_Store_Combo.py @@ -103,8 +103,7 @@ def _brief(whole, row: dict) -> list[dict]: def combos(sub: str = "", detail: str = "", q: str = "", owner: str = "") -> list[dict]: """조합 목록 — 구분 · 상세구분 · 소유 · 찾기로 거름.""" - files = cm.load(folder=store.FOLDER) - whole = cm.mf.Master(files) + files, whole = store.loaded() out = [] for _, row in mcb.rows(files): if (sub and row.get("구분") != sub) or (detail and row.get("상세구분") != detail): @@ -130,7 +129,7 @@ def combo(key: str) -> dict: """조합 하나 — 담은 로직에 이름·단위를 붙여 줌.""" data, version = _read() row = _find(data, key) - whole = cm.mf.Master(cm.load(folder=store.FOLDER)) + whole = store.loaded()[1] problems = mcb.check_row(whole, row) return { "file": FILE, @@ -144,7 +143,7 @@ def combo(key: str) -> dict: def combos_of_logic(key: str) -> dict: """「이 로직을 쓰는 조합」 — 로직 화면이 거꾸로 볼 때(소요량 「용도」 와 같은 결).""" - files = cm.load(folder=store.FOLDER) + files, _ = store.loaded() return {"key": key, "combos": mcb.using(files, key)} diff --git a/M01_MasterData/M01_MasterData_Store_Make.py b/M01_MasterData/M01_MasterData_Store_Make.py index 7fdf6ebe..76dc319d 100644 --- a/M01_MasterData/M01_MasterData_Store_Make.py +++ b/M01_MasterData/M01_MasterData_Store_Make.py @@ -70,7 +70,7 @@ def _all_sections(book: str) -> list[dict]: def _counts() -> dict[tuple[str, str], list[int]]: """(구분, 절 번호) → [표 수, 로직 수] — 아래 절까지 세도록 절 번호의 모든 앞마디에 얹음.""" out: dict[tuple[str, str], list[int]] = {} - for data in cm.load(folder=store.FOLDER).values(): + for data in store.loaded()[0].values(): group = data.get("그룹") at = 0 if group in cm.mf.TABLE_GROUPS else 1 if group == "로직" else None if at is None: @@ -236,7 +236,7 @@ def logic_edit(key: str, version: str, logic: dict, owner: str = "") -> dict: def logic_formula(key: str) -> dict: """로직 한 줄을 별칭 꼴 식 묶음으로 펴 보임 — 되돌릴 원문은 `별칭` 에 그대로(정본은 안 바뀜).""" got = store.logic(key) - master = cm.mf.Master(cm.load(folder=store.FOLDER)) + master = store.loaded()[1] return { "file": got["file"], "version": got["version"], diff --git a/M01_MasterData/M01_MasterData_UI_Page.ts b/M01_MasterData/M01_MasterData_UI_Page.ts index 08f2e216..dd12bd71 100644 --- a/M01_MasterData/M01_MasterData_UI_Page.ts +++ b/M01_MasterData/M01_MasterData_UI_Page.ts @@ -80,7 +80,12 @@ function buildPage(): HTMLElement { const comboHost = el("div", { className: "m01-master__logic", attrs: { hidden: "" } }); const elementView = el("div", { className: "m01-master__elements", - children: [head, notice, body], + children: [ + head, + notice, + el("p", { className: "m01-master__hint", text: L("M01_WideHint") }), + body, + ], }); const showNotice = (nodes: (HTMLElement | string)[]): void => { diff --git a/M01_MasterData/M01_MasterData_UI_Side.ts b/M01_MasterData/M01_MasterData_UI_Side.ts index 4d586ae9..e379f5b4 100644 --- a/M01_MasterData/M01_MasterData_UI_Side.ts +++ b/M01_MasterData/M01_MasterData_UI_Side.ts @@ -379,6 +379,16 @@ export function buildSide( ); }; + /** 처음 펼칠 때 열 줄 — 재료는 차례 첫 파일(자재품목) */ + const firstRun = (group: Group): string => { + if (group !== "재료") return `${group}|`; + const first = [...(lists.get(group) ?? [])].sort((a, b) => order(a.file) - order(b.file))[0]; + return first + ? `${group} +${first.file}` + : ""; + }; + const refresh = async (group: string): Promise => { const list = await fetchFiles(group); if (list[0] && (group === "인력" || group === "기계")) { @@ -427,7 +437,7 @@ export function buildSide( if (!root.classList.contains("is-collapsed")) return; // 닫을 때는 그대로 둠 restoring = true; try { - runs.get(lastRun.get(group) ?? (group === "재료" ? "" : `${group}|`))?.(); + runs.get(lastRun.get(group) ?? firstRun(group))?.(); } finally { restoring = false; } diff --git a/M01_MasterData/M01_MasterData_UI_Style.css b/M01_MasterData/M01_MasterData_UI_Style.css index 7da36a04..2df3b62e 100644 --- a/M01_MasterData/M01_MasterData_UI_Style.css +++ b/M01_MasterData/M01_MasterData_UI_Style.css @@ -139,6 +139,19 @@ margin: 0; } +.m01-master__hint { + display: none; + margin: 0; + color: var(--color-text-muted); + font-size: var(--text-body-sm); +} + +@media (max-width: 600px) { + .m01-master__hint { + display: block; + } +} + .m01-master__body { display: flex; flex-direction: column;