"""M01 마스터 데이터 파일 저장소 — `resources/master_data/` 첫 층 JSON 읽기·검사·쓰기. 계약: `resources/master_data/_화면_계약.md` · 파일 모양: `_틀.md` 검사·계산 엔진: `scripts/check_master.py`. """ from __future__ import annotations import hashlib import json import os import re 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_combo as mcb # noqa: E402,F401 (`_Store_Combo` 가 `store.mcb` 로 씀) 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.M01_MasterData_Store_Shape import dump, keep_shape # noqa: E402 FOLDER: Path = cm.MASTER # 시험은 사본 폴더로 바꿈 DRAFT = "_초안.json" # 저장 안 한 새 로직을 얹는 메모리 자리 — 파일로 쓰지 않음 _CHAPTER = re.compile(r"_(\d\d)장_(.+)\.json$") # 「_03장_토공사.json」 # ponytail: 저장은 한 번에 하나(프로세스 안 잠금) · 서버를 여럿 띄우면 파일 잠금으로 _LOCK = threading.Lock() class StoreError(Exception): def __init__(self, status: int, detail): super().__init__(detail) self.status, self.detail = status, detail # ── 읽기 ────────────────────────────────────────────────────────────── def _path(file: str) -> Path: path = FOLDER / file if "/" in file or "\\" in file or file.startswith("_") or not path.is_file(): raise StoreError(404, f"없는 파일 「{file}」") return path def version_of(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest()[:16] def read(file: str) -> tuple[dict, str]: raw = _path(file).read_bytes() return json.loads(raw, parse_float=Decimal, parse_int=Decimal), version_of(raw) def items_key(data: dict) -> str: return "표" if data.get("그룹") in cm.mf.TABLE_GROUPS else "줄" def chapter_of(file: str, data: dict | None = None) -> str: """「공통 03장 토공사」 · 「13장 구조물」 — 부문은 파일 머리.""" m = _CHAPTER.search(file) if not m: return "" division = (data or {}).get("부문") or "" return f"{division} {m.group(1)}장 {m.group(2).replace('_', ' ')}".strip() def _hit(row: dict, q: str) -> bool: q = q.lower() return not q or any( q in str(row.get(k, "")).lower() for k in ("키", "원문번호", "이름", "옛이름") ) def _book(data: dict) -> str | None: """머리 원문 이름 — 한 테이블 파일(기계)은 머리 「원문」 이 갈래 묶음이라 이름이 묶음 안에.""" return data.get("원문") if isinstance(data.get("원문"), str) else None def groups() -> list[dict]: out = {g: {"group": g, "files": 0, "rows": 0} for g in cm.mf.GROUPS} for data in cm.load(folder=FOLDER).values(): slot = out.get(data.get("그룹")) if slot: slot["files"] += 1 slot["rows"] += len(data.get(items_key(data)) or []) return list(out.values()) def files_of(group: str) -> list[dict]: if group not in cm.mf.GROUPS: raise StoreError(404, f"없는 그룹 「{group}」") out = [] for path in [*FOLDER.glob(f"{group}.json"), *FOLDER.glob(f"{group}_*.json")]: data, version = read(path.name) out.append( { "file": path.name, "book": _book(data), "chapter": chapter_of(path.name, data), "edition": data.get("판"), "rows": len(data.get(items_key(data)) or []), "version": version, "order": data.get("차례") or 0, } ) return sorted(out, key=lambda f: (f["order"], f["file"])) _REF_KEY = re.compile(r"^[A-Z]{2}\d{6}$") _REF_FILE = { "LB": "인력.json", "MT": "재료_자재품목.json", "MP": "재료_품셈재료.json", "EQ": "기계.json", } def _ref_keys(row: dict) -> list[str]: """줄이 가리키는 키 — 준용 · 연결.""" put = (row.get("준용"), row.get("연결")) return [v for v in put if isinstance(v, str) and _REF_KEY.match(v)] def refs_of(rows_: list[dict]) -> dict[str, str]: """참조 칸에 키 옆에 보일 이름 — {키: 「이름 규격」}.""" want = {k for r in rows_ for k in _ref_keys(r)} out: dict[str, str] = {} for file in {_REF_FILE[k[:2]] for k in want if k[:2] in _REF_FILE}: for r in read(file)[0].get("줄") or []: if r.get("키") in want: out[r["키"]] = f"{r.get('이름') or ''} {r.get('규격') or ''}".strip() return out _STATES = ("공표", "산정", "미공표", "미확보", "추정") # 인력 「상태」 거름 차례 def subs(file: str = "", group: str = "") -> dict: """하위 거름 목록 — 한 테이블 파일은 머리 묶음(인력 · 기계 모두 「구분」 → 「상세구분」 · 기계는 「원문」(건설품셈 · 산림품셈)도 머리에 있으나 「구분」 이 우선) · 장 그룹(소요량 · 계수 · 로직)은 장 파일에서 모음(구분 = 원문 + 부문 · 상세구분 = 장 · 차례대로).""" if group: if group not in cm.mf.GROUPS: raise StoreError(404, f"없는 그룹 「{group}」") found: dict[str, dict] = {} for info in files_of(group): name = mk.CHAPTER.match(info["file"]) if not name: continue data = read(info["file"])[0] kind = " ".join(x for x in (data.get("원문"), data.get("부문")) if x) slot = found.setdefault(kind, {"name": kind, "book": data.get("원문"), "details": []}) slot["details"].append(mk.detail_of(info["file"])) return {"file": "", "version": "", "slot": "구분", "subs": list(found.values())} data, version = read(file) slot = next((s for s in cm.MERGED.values() if isinstance(data.get(s), dict)), "") lines = data.get(items_key(data)) or [] has_state = any("상태" in r for r in lines) def details_of(name: str, head: dict) -> list: """머리에 상세구분이 없으면 줄에서 모음(재료_유가전력 — 지역 · 계약종별).""" if head.get("상세구분"): return head["상세구분"] return list( dict.fromkeys( r["상세구분"] for r in lines if r.get("구분") == name and r.get("상세구분") ) ) heads = data.get(slot) or {} if heads: subs_list = [ {"name": name, "book": head.get("원문"), "details": details_of(name, head)} for name, head in heads.items() ] else: # 머리에 구분 묶음이 등록 안 된 파일(재료_자재품목) — 줄에서 바로 모음 subs_list = [ {"name": name, "book": None, "details": details_of(name, {})} for name in dict.fromkeys(r["구분"] for r in lines if r.get("구분")) ] return { "file": file, "version": version, "slot": slot, "states": list(_STATES) if has_state else [], "subs": subs_list, } def rows( file: str, page: int, size: int, q: str, unlinked: bool = False, sub: str = "", detail: str = "", state: str = "", ) -> dict: data, version = read(file) hits = [r for r in data.get(items_key(data)) or [] if _hit(r, q)] if sub: # 인력·기계 구분(우선) · 기계 원문(책)도 받음 hits = [r for r in hits if sub in {r.get(s) for s in cm.MERGED.values()}] if detail: # 인력 상세구분 · 기계 상세구분(기종) hits = [r for r in hits if r.get("상세구분") == detail] if state: # 인력 상태 hits = [r for r in hits if r.get("상태") == state] if unlinked: # 품셈재료 — 아직 못 이은 줄 hits = [r for r in hits if not r.get("연결")] page, size = max(page, 1), min(max(size, 1), 500) shown = hits[(page - 1) * size : page * size] return { "file": file, "version": version, "total": len(hits), "page": page, "size": size, "rows": shown, "refs": refs_of(shown), } _TABLE_HEADS = ( "키", "원문번호", "구분", "상세구분", "이름", "기준", "출처", "비고", "조건", "값칸", ) def in_section(number, section: str) -> bool: """원문번호가 그 절(또는 그 아래 절)인지 — 「13-4」 는 「13-4-1」 을 품고 「13-40」 은 아님.""" number = str(number or "") return not section or number == section or number.startswith(section + "-") def tables( file: str = "", group: str = "", sub: str = "", detail: str = "", q: str = "", page: int = 1, size: int = 30, usage: str = "", section: str = "", ) -> dict: """표 목록 — 파일 하나 또는 그룹 전체(장 차례대로 · 쪽 나눔). 거름 = 구분 · 상세구분 · 찾기 · `usage`(용도.대상 하나) · `section`(절 번호 · 아래 절 포함).""" if file: names = [file] elif group in cm.mf.TABLE_GROUPS: names = [f["file"] for f in files_of(group)] else: raise StoreError(404, f"없는 표형 그룹 「{group}」") hits, version = [], "" for name in names: data, version = read(name) if items_key(data) != "표": raise StoreError(400, f"표형 파일 아님 「{name}」") for t in data["표"]: if sub and t.get("구분") != sub: continue if detail and t.get("상세구분") != detail: continue if usage and usage not in ((t.get("용도") or {}).get("대상") or []): continue if not in_section(t.get("원문번호"), section): continue if _hit(t, q): hits.append( { "file": name, **{k: t.get(k) for k in _TABLE_HEADS}, "count": len(t.get("줄") or []), } ) page, size = max(page, 1), min(max(size, 1), 500) return { "file": file, "version": version if file else "", "total": len(hits), "page": page, "size": size, "tables": hits[(page - 1) * size : page * size], } def _one(data: dict, key: str) -> dict: for row in data.get(items_key(data)) or []: if str(row.get("키")) == key: return row raise StoreError(404, f"없는 키 「{key}」") def table(file: str, key: str) -> dict: data, version = read(file) return {"file": file, "version": version, "table": _one(data, key)} def _logic_files(files: dict[str, dict]): for name, data in files.items(): if data.get("그룹") == "로직": yield name, data 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) out = [] for name, data in sorted(_logic_files(files), key=lambda f: f[1].get("차례") or 0): for row in data.get("줄") or []: if (sub and row.get("구분") != sub) or (detail and row.get("상세구분") != detail): continue if owner and row.get("소유") != owner: continue if not _hit(row, q): continue reasons = cm.mf.check_logic(whole, row)[0] if blocked is not None and bool(reasons) != bool(blocked): continue out.append( { "file": name, **{ k: row.get(k) for k in ( "키", "원문번호", "구분", "상세구분", "이름", "결과단위", "출처", "소유", ) }, "blocked": bool(reasons), "reasons": reasons, } ) return out def logic(key: str) -> dict: """로직 하나 — 키는 전체에서 하나.""" files = cm.load(folder=FOLDER) 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 { "file": name, "version": version, "logic": row, "blocked": bool(reasons), "reasons": reasons, "prices": _prices(whole, row), } raise StoreError(404, f"없는 로직 「{key}」") def _brief(ref: str, row: dict) -> dict: return { "ref": ref, **{ k: row.get(k) for k in ("원문번호", "구분", "상세구분", "이름", "규격", "단위", "값", "값칸") }, } def _pick_brief(whole, cond: dict, unit: str | None = None) -> dict | None: """고르기 조건 줄의 단가 미리보기 — 후보 수와 시험 계산이 쓸 줄(값 열 다섯 포함). `unit` = 호표 줄 단위 — 후보를 같은 단위 줄로만 거름.""" try: keys = cm.mf.candidates(whole, cond, {}, unit) brief = _market_brief(whole.get(cm.mf.pick(whole, cond, {}, unit))) except (cm.mf.FormulaError, KeyError, TypeError): return None return {**brief, "조건": cond, "후보": len(keys)} def _prices(whole, row: dict) -> dict: """호표 요소 → 요소 줄 요약(단가 자동 칸) · `{이름}` 낀 원문번호·하위 로직은 계산 때 정해짐. 품셈재료는 연결을 따라간 낮은 값과 출처 · 자재지역 후보는 계산 때 정해짐(값 None). 재료 고르기 조건 줄은 열쇠가 조건 글(`구분|상세구분|규격|대표`) · 값 = 시험 계산이 쓸 줄.""" out = {} for item in row.get("호표") or []: if isinstance(item.get("요소"), dict): out[cm.mf.cond_text(item["요소"])] = _pick_brief(whole, item["요소"], item.get("단위")) continue ref = str(item.get("요소", "")) if item.get("종류") != "로직" and "{" not in ref: try: out[ref] = _brief(ref, whole.get(ref)) except cm.mf.FormulaError: out[ref] = None continue if ref[:2] == cm.mf.LINKED: try: value, source = cm.mf.element(whole, ref, {}) except (cm.mf.FormulaError, KeyError, TypeError): value, source = None, None out[ref].update({"값": value, "출처": source}) return out def elements(group: str, q: str, limit: int) -> dict: """요소 찾기 창 — 그룹 전체에서 키·원문번호·이름 찾기 · `ref` 는 식에 그대로 넣는 키.""" if group not in cm.mf.GROUPS: raise StoreError(404, f"없는 그룹 「{group}」") hits = [] for name, data in cm.load(folder=FOLDER).items(): if data.get("그룹") != group: continue for row in data.get(items_key(data)) or []: if _hit(row, q): hits.append({**_brief(str(row.get("키")), row), "file": name}) return {"total": len(hits), "items": hits[: min(max(limit, 1), 200)]} _PICK = {"price": "재료_자재품목.json", "job": "인력.json"} def _squash(text) -> str: return "".join(str(text or "").lower().split()) def _lowest(row: dict) -> object: """자재품목 줄 — 값 열 다섯 가운데 낮은 값(시험 계산용) · 값 칸이 있는 줄은 그 값 그대로.""" if "값" in row: return row["값"] nums = [row[s] for s in cm.mf.SLOTS if row.get(s) is not None] return min(nums, default=None) def pick(kind: str, q: str, limit: int) -> dict: """고르기 창 — `price` 자재품목(관급 줄 포함 · ref = 키) · `job` 값 있는 공표 노임 직종 (미확보·추정 빼고 모든 구분)(ref = 키) · 이름·규격 낱말이 모두 들어간 줄.""" if kind not in _PICK: raise StoreError(404, f"없는 고르기 「{kind}」") words = [_squash(w) for w in q.split()] hits = [] for row in read(_PICK[kind])[0].get("줄") or []: if kind == "job" and ( row.get("값") is None or row.get("구분") == "미확보" or row.get("상태") == "추정" ): continue text = _squash(f"{row.get('이름')} {row.get('규격')} {' '.join(row.get('옛이름') or [])}") if all(w in text for w in words): hits.append( { "ref": str(row.get("키")), **{k: row.get(k) for k in ("원문번호", "구분", "이름", "규격", "단위")}, "값": _lowest(row), "관급": row.get("관급") is not None, } ) return {"total": len(hits), "items": hits[: min(max(limit, 1), 200)]} def _market_brief(row: dict) -> dict: """후보 한 줄 — 값 열 다섯을 다 주고 `값` 에 시험 계산이 쓸 낮은 값(없으면 null).""" return { "ref": str(row.get("키")), **{k: row.get(k) for k in ("구분", "상세구분", "이름", "규격", "단위")}, "값들": {s: row.get(s) for s in cm.mf.SLOTS}, "값": _lowest(row), "관급": row.get("관급") is not None, } def materials(sub: str, detail: str, spec: str, region: str, unit: str, limit: int) -> dict: """재료 고르기 조건 안 후보 목록 — 화면·테스트 컨테이너가 같이 씀. `unit` = 호표 줄 단위 — 주면 그 단위 줄만(단위가 다르면 다른 물건). `기본` = 시험 계산이 쓸 줄(대표 줄이 없을 때의 첫 줄) · 그 줄을 `items` 맨 앞에 둠.""" whole = cm.master(folder=FOLDER) 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) first = cm.mf.pick(whole, cond, env, unit or None) if keys else None keys = [first] + [k for k in keys if k != first] if first else keys return { "total": len(keys), "기본": first, "items": [_market_brief(whole.get(k)) for k in keys[: min(max(limit, 1), 200)]], } # ── 시험 계산 ────────────────────────────────────────────────────────── def _swap(files: dict, key: str, row: dict, file: str | None) -> str: """저장 전 시험 — 메모리 사본에서 그 로직을 고친 줄로 바꿈(없으면 `file` 에 더함) · 돌려받는 것 = 계산할 키.""" row = _dec(row) for name, data in _logic_files(files): items = data.get("줄") or [] for i, old in enumerate(items): if str(old.get("키")) == key: items[i] = {**row, "키": key} return key if file is None: # 저장 안 한 새 초안 — 메모리에만 얹고 셈(만들기 모달의 시험 계산) file = DRAFT files[file] = {"그룹": "로직", "원문": "자체", "줄": []} if file not in files or files[file].get("그룹") != "로직": raise StoreError(404, f"없는 로직 파일 「{file}」") new = mk.table_id("로직", files[file].get("원문")) + "999999" # 아직 번호 안 받은 새 로직 files[file].setdefault("줄", []).append({**row, "키": new}) return new def calc(key: str, inputs: dict, row: dict | None = None, file: str | None = None): given = {k: _dec(v) for k, v in inputs.items()} files = cm.load(folder=FOLDER) if row is not None: key = _swap(files, str(row.get("키") or key), row, file) try: result = cm.calc(files, key, given) except (cm.mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e: return {"ok": False, "reason": str(e) or type(e).__name__} if "결과" in result: return {"ok": True, "result": result["결과"], "middle": result["중간"]} return { "ok": True, "lines": result["줄"], "sums": {k: result[k] for k in cm.mf.BUNDLE}, "middle": result["중간"], } def text(key: str, inputs: dict, row: dict | None = None, file: str | None = None): """사람이 읽는 식 줄 — 비목 묶음마다 「이름 : 단가 * 수량식 = 금액」 과 소계(`master_text`).""" given = {k: _dec(v) for k, v in inputs.items()} files = cm.load(folder=FOLDER) if row is not None: key = _swap(files, str(row.get("키") or key), row, file) try: return mtx.lines(files, key, given) except (cm.mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e: return {"ok": False, "reason": str(e) or type(e).__name__} # ── 저장 ────────────────────────────────────────────────────────────── def _dec(v): """화면이 보낸 수 → Decimal(파일을 읽은 모양과 같게).""" if isinstance(v, bool) or v is None: return v if isinstance(v, (int, float)): return Decimal(str(v)) if isinstance(v, list): return [_dec(x) for x in v] if isinstance(v, dict): return {k: _dec(x) for k, x in v.items()} return v def _chapter_cols(file: str, data: dict) -> dict: """장 파일에 더하는 줄의 갈래 — 구분 = 원문 + 부문 · 상세구분 = 「NN장 장 제목」.""" if not mk.CHAPTER.match(file): return {} return { "구분": " ".join(x for x in (data.get("원문"), data.get("부문")) if x), "상세구분": mk.detail_of(file), } def _follow(row: dict, items: list[dict], file: str) -> dict: """인력 줄의 준용 = 그 직종의 값을 다시 읽어 값 칸에 채움(화면이 보낸 값을 믿지 않음).""" ref = row.get("준용") if file != "인력.json" or not isinstance(ref, str) or not _REF_KEY.match(ref): return row src = next((r for r in items if r.get("키") == ref), None) if src is None or src.get("값") is None: return row return {**row, "값": src["값"]} def ledger_label(book: dict, tid: str, number: str, row: dict) -> str: """대장 열쇠 — 보통은 원문번호 · 자체 로직(GX)은 한 절에 여럿이라 「번호 이름」 으로 하나씩.""" if tid != "GX" or not number: # 원문번호 없는 줄은 빈 열쇠 — 대장이 늘 새 번호를 줌 return number base = f"{number} {row.get('이름') or ''}".strip() index, label, at = mk.lookup(book), base, 1 while (tid, label) in index: at += 1 label = f"{base} ({at})" return label def _apply(data: dict, changes: list[dict], file: str, book: dict) -> None: """고침 · 더함(키는 대장의 다음 번호 · 원문번호 없으면 빈 글) · 지움 — 키는 바꾸지 않음.""" items = data.setdefault(items_key(data), []) tid = mk.table_id(data.get("그룹"), _book(data)) for ch in changes: op, key, row = ch.get("op"), ch.get("key"), _dec(ch.get("row")) if op not in ("edit", "add", "delete"): raise StoreError(400, f"모르는 op 「{op}」") if op != "add": at = next((i for i, r in enumerate(items) if str(r.get("키")) == key), None) if at is None: raise StoreError(404, f"{file} · 없는 키 「{key}」") if op != "delete" and not isinstance(row, dict): raise StoreError(400, f"{file} · row 없음") if op != "delete": row = _follow(row, items, file) if op == "add": number = str(row.get("원문번호") or "") cols = _chapter_cols(file, data) drop = ("키", "원문번호", *cols) rest = {k: v for k, v in row.items() if k not in drop} key = mk.issue(book, tid, ledger_label(book, tid, number, row), file) # 자재품목처럼 줄에 원문번호를 두지 않는 테이블은 키만 — 번호는 대장에만 head = {"키": key} if tid in mk.NO_NUMBER else {"키": key, "원문번호": number} items.append({**head, **cols, **rest}) elif op == "edit": if str(row.get("키")) != key: raise StoreError(400, f"{file} · 키는 바꾸지 않음 「{key}」") items[at] = row else: del items[at] def save(batch: list[dict]) -> list[dict]: with _LOCK: stale = [p.get("file", "") for p in batch if read(p.get("file", ""))[1] != p.get("version")] if stale: raise StoreError(409, {"stale": stale}) before, after = cm.load(folder=FOLDER), cm.load(folder=FOLDER) # after = 고칠 사본 book_path = FOLDER / mk.BOOK.name book = mk.load_book(book_path) issued = dict(book["다음"]) names = [part["file"] for part in batch] 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] if new: raise StoreError(422, {"errors": new}) if book["다음"] != issued: mk.save_book(book, book_path) out = [] for name in dict.fromkeys(names): old_text = (FOLDER / name).read_bytes().decode("utf-8") text = keep_shape(old_text, before[name], after[name], items_key(after[name])) or dump( after[name] ) tmp = FOLDER / f".{name}.tmp" 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"))}) return out