auto: 2026-09-22 03:10 (EOMSANGDON-HOME)

This commit is contained in:
2026-09-22 03:10:31 +09:00
parent 323e1d2c3d
commit 6a2925a9ec
8 changed files with 305 additions and 18 deletions
+4
View File
@@ -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
+32 -10
View File
@@ -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
@@ -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
+3 -4
View File
@@ -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)}
+2 -2
View File
@@ -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"],
+6 -1
View File
@@ -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 => {
+11 -1
View File
@@ -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<FileInfo[]> => {
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;
}
@@ -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;