eomsangdonandClaude Opus 5
ef7dc80198
- 저장(`POST /save`) 6.8초 → 0.8초 — 고친 줄과 그 줄을 부르던 줄만 검사 (안 보는 줄이 부르는 키는 목록이 미리 세어 둔 것을 씀 · 돌고 도는 길은 전체 그림으로) - 로직 줄만 고쳤으면 품셈재료 연결 · 인력 준용 검사(`check_links`)도 건너뜀 - 로직 밖 파일(요소 · 표)이 섞인 저장은 예전처럼 전부 검사 — 전체 검사는 `check_master` 몫 - 켠 뒤 첫 목록 3.4초 → 0.03초 — M01 라우터가 뜰 때 뒤에서 한 번 읽어 둠(`cache.warm`) - 저장 전 검사(`check_saved`)를 M01 쪽(`_Store_Cache`)으로 옮김 — 좁히는 값이 있는 자리 · `check_master` 가 700줄에 닿음 - 시험 `test_m01_cache.py` 에 넷 더함 — 없는 로직 키 · 없는 요소 키 · 돌고 도는 참조 · 다른 파일에서 그 줄을 부르던 줄이 깨짐(모두 422) · 미리 읽은 뒤 첫 목록 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JgUhN55Z3BCYhUJTjwTNfj
688 lines
27 KiB
Python
688 lines
27 KiB
Python
"""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 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 # 시험은 사본 폴더로 바꿈
|
|
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)
|
|
|
|
|
|
# ── 한 번 읽어 두기(`_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 "줄"
|
|
|
|
|
|
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]:
|
|
"""로직 목록 — 구분 · 상세구분 · 소유 · 찾기로 거름(장 차례대로 · 파일을 가로지름).
|
|
막힘은 미리 세어 둔 값(`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 []:
|
|
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 = why_of.get(str(row.get("키")), [])
|
|
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, whole = loaded()
|
|
for name, data in _logic_files(files):
|
|
for row in data.get("줄") or []:
|
|
if str(row.get("키")) == key:
|
|
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 loaded()[0].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 = 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)
|
|
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)
|
|
# 고친 줄과 그 줄을 부르던 줄만 검사 — 마스터 전체 검사는 `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:
|
|
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"))})
|
|
cache.stale(dict.fromkeys(names)) # 저장 뒤에는 그 파일을 바로 다시 읽게
|
|
return out
|