eomsangdonandClaude Opus 5
827054f8b1
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn
281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""Z01 마스터 데이터 — `resources/` 마스터 JSON 을 **갈래 → 파일 → 표 → 줄** 로 가름(읽기 전용).
|
|
|
|
마스터 = 로직 + 기초값(2026-09-15 사용자 방향 · bottom-up). 가르는 잣대 = 「값이 바뀌는 것 = 기초값 · 규칙·표 = 로직」(브레인).
|
|
표 가르기(브레인 승인):
|
|
· `variables` 칸은 변수 하나 = 표 하나 — 변수 안에 목록이 둘 이상이면 한 표에 잇고 `@part` 열에 목록 이름
|
|
· 나머지는 줄 모음마다 표 하나 — 딕셔너리 목록 · 같은 길이 **수** 열(글 목록은 길이가 같아도 짝짓지 않음) ·
|
|
딕셔너리의 딕셔너리(`@key` 열) · 값 딕셔너리(`@key`·`@value` 줄)
|
|
· 설명(note·source·policy …)은 표가 아님 — 줄 안의 설명 칸은 그대로 둠
|
|
한글 이름·단위는 **이름표 파일**(`resources/data_master_labels/*.json` · 데스크탑 서브 몫)에서만 — 없으면 영문 key.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
RESOURCES = Path(__file__).resolve().parent.parent / "resources"
|
|
LABEL_DIR = RESOURCES / "data_master_labels"
|
|
DEFAULT_PAGE_SIZE = 50
|
|
MAX_PAGE_SIZE = 500
|
|
|
|
# 갈래 나눔표(브레인 2026-09-15) — 차례가 화면 차례. 여기 없는 파일(강우 IDF 캐시 등)은 안 보임.
|
|
GROUPS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
|
("logic", (
|
|
"pum_forest", "pum_const", # 품셈 원문
|
|
"work_item_master", # 공종 축
|
|
"coef", "material_surcharge", "formwork_reuse", "rebar_complexity", "timber_structure_class",
|
|
"masonry_class", "masonry_slope", "masonry_back_length", "stone_kind", "revetment_sabang",
|
|
"structure_unit_observed", # 계수
|
|
"work_item_mapping", "aliases", "resource_catalog_ext", # 잇는 표
|
|
)),
|
|
("base", (
|
|
"labor_const", "labor_mfg", "mach_base", "machine_operating", "mat_price_public",
|
|
"oil", "oil_regional", "fx", "rates",
|
|
)),
|
|
("byproduct", ("basis_missing", "form_undetermined", "resource_axis", "unmatched")), # 정본 아님 — 기록
|
|
("seed", ("masonry_wet",)),
|
|
) # fmt: skip
|
|
|
|
# 파일 머리 — 표가 아님
|
|
_HEAD_KEYS = {
|
|
"source", "sources", "policy", "derived_from", "source_dataset_version", "source_master_file",
|
|
"dataset_version", "stats", "master", "parse_audit", "variables",
|
|
} # fmt: skip
|
|
_NOTE_KEYS = {"note", "notes", "source", "sources", "quote", "why", "policy"}
|
|
_HIDDEN_KEYS = {"sha256", "generated_at", "raw_row_index", "sort_order"}
|
|
_DATE_SUFFIX = re.compile(r"_\d{4}(?:-\d{2}-\d{2})?$")
|
|
|
|
|
|
def _is_note(key: str) -> bool:
|
|
return key in _NOTE_KEYS or key.endswith("_note")
|
|
|
|
|
|
def _scalar(v: Any) -> bool:
|
|
return not isinstance(v, (dict, list))
|
|
|
|
|
|
def _flat(d: dict, prefix: str = "") -> dict[str, Any]:
|
|
"""줄 하나 — 안쪽 딕셔너리는 `a.b` 열로 펼치고 목록은 값 그대로."""
|
|
out: dict[str, Any] = {}
|
|
for k, v in d.items():
|
|
if isinstance(v, dict) and v:
|
|
out.update(_flat(v, f"{prefix}{k}."))
|
|
else:
|
|
out[f"{prefix}{k}"] = v
|
|
return out
|
|
|
|
|
|
def _list_rows(items: list) -> list[dict[str, Any]]:
|
|
return [_flat(x) if isinstance(x, dict) else {"@value": x} for x in items]
|
|
|
|
|
|
def _numeric(v: Any) -> bool:
|
|
if isinstance(v, list):
|
|
return all(_numeric(x) and not isinstance(x, list) for x in v)
|
|
return v is None or (isinstance(v, (int, float)) and not isinstance(v, bool))
|
|
|
|
|
|
def _columnar(d: dict) -> dict[str, list] | None:
|
|
"""같은 길이 수 열(2개 이상) — 안쪽 한 겹 딕셔너리(`columns`)는 끌어올림."""
|
|
cols: dict[str, list] = {}
|
|
for k, v in d.items():
|
|
if _is_note(k):
|
|
continue
|
|
if isinstance(v, dict):
|
|
for ck, cv in v.items():
|
|
if _is_note(ck):
|
|
continue
|
|
if ck in cols or not isinstance(cv, list):
|
|
return None
|
|
cols[ck] = cv
|
|
elif isinstance(v, list) and k not in cols:
|
|
cols[k] = v
|
|
else:
|
|
return None
|
|
lengths = {len(v) for v in cols.values()}
|
|
if len(cols) < 2 or len(lengths) != 1 or 0 in lengths:
|
|
return None
|
|
if not all(_numeric(x) for v in cols.values() for x in v):
|
|
return None
|
|
return cols
|
|
|
|
|
|
def _tableish(d: dict) -> bool:
|
|
"""제 표로 떼어 낼 딕셔너리 — 같은 길이 수 열이거나 그런 것을 품음. 나머지는 한 줄로 펼침."""
|
|
return _columnar(d) is not None or any(
|
|
isinstance(v, dict) and _tableish(v) for k, v in d.items() if not _is_note(k)
|
|
)
|
|
|
|
|
|
def _dict_rows(d: dict) -> list[dict[str, Any]]:
|
|
"""목록 없는 딕셔너리 — 안쪽 딕셔너리는 키마다 한 줄(`@key` + 펼친 칸), 값은 키·값 줄."""
|
|
body = {k: v for k, v in d.items() if not _is_note(k)}
|
|
kv = [{"@key": k, "@value": v} for k, v in body.items() if not isinstance(v, dict)]
|
|
return kv + [{"@key": k, **_flat(v)} for k, v in body.items() if isinstance(v, dict)]
|
|
|
|
|
|
def _variable_rows(v: Any) -> list[dict[str, Any]]:
|
|
if isinstance(v, list):
|
|
return _list_rows(v)
|
|
lists = [(k, x) for k, x in v.items() if isinstance(x, list) and not _is_note(k)]
|
|
if not lists:
|
|
cols = _columnar(v)
|
|
return [dict(zip(cols, r)) for r in zip(*cols.values())] if cols else _dict_rows(v)
|
|
consts = _flat(
|
|
{k: x for k, x in v.items() if not _is_note(k) and (_scalar(x) or isinstance(x, dict))}
|
|
)
|
|
# `key` = 어느 열이 키인지 적은 머리 — 줄 값이 아님
|
|
consts = {k: x for k, x in consts.items() if k != "key" and not _is_note(k.rsplit(".", 1)[-1])}
|
|
return [
|
|
{**consts, **({"@part": k} if len(lists) > 1 else {}), **row}
|
|
for k, items in lists
|
|
for row in _list_rows(items)
|
|
]
|
|
|
|
|
|
def _collect(node: Any, path: str, out: list[tuple[str, list[dict[str, Any]]]]) -> None:
|
|
if isinstance(node, list):
|
|
out.append((path, _list_rows(node)))
|
|
return
|
|
cols = _columnar(node)
|
|
if cols:
|
|
out.append((path, [dict(zip(cols, r)) for r in zip(*cols.values())]))
|
|
return
|
|
body = {k: v for k, v in node.items() if not _is_note(k)}
|
|
lists = {k: v for k, v in body.items() if isinstance(v, list)}
|
|
if lists: # 목록마다 표 하나 · 곁의 값은 줄마다 붙임
|
|
consts = {k: v for k, v in body.items() if _scalar(v)}
|
|
for k, v in lists.items():
|
|
out.append((f"{path}/{k}", [{**consts, **r} for r in _list_rows(v)]))
|
|
for k, v in body.items():
|
|
if isinstance(v, dict):
|
|
_collect(v, f"{path}/{k}", out)
|
|
return
|
|
split = [k for k, v in body.items() if isinstance(v, dict) and _tableish(v)]
|
|
rest = {k: v for k, v in body.items() if k not in split}
|
|
if rest:
|
|
out.append((path, _dict_rows(rest)))
|
|
for k in split:
|
|
_collect(body[k], f"{path}/{k}", out)
|
|
|
|
|
|
def file_id(path: Path) -> str:
|
|
return _DATE_SUFFIX.sub("", path.stem)
|
|
|
|
|
|
def master_files() -> dict[str, Path]:
|
|
"""갈래표에 든 id → 파일(같은 id 가 여러 판이면 이름 차례로 끝 판)."""
|
|
wanted = {i for _, ids in GROUPS for i in ids}
|
|
found: dict[str, Path] = {}
|
|
for p in sorted(
|
|
[*RESOURCES.glob("data_*/*.json"), *RESOURCES.glob("library_structure/*.json")]
|
|
):
|
|
if file_id(p) in wanted and p.parent != LABEL_DIR:
|
|
found[file_id(p)] = p
|
|
return found
|
|
|
|
|
|
@lru_cache(maxsize=64)
|
|
def _tables_of(path: str, _mtime_ns: int) -> tuple[tuple[str, tuple[dict[str, Any], ...]], ...]:
|
|
data = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
out: list[tuple[str, list[dict[str, Any]]]] = []
|
|
for name, v in (data.get("variables") or {}).items():
|
|
out.append((f"variables/{name}", _variable_rows(v)))
|
|
for k, v in data.items():
|
|
if k not in _HEAD_KEYS and not _is_note(k) and not _scalar(v):
|
|
_collect(v, k, out)
|
|
return tuple((tid, tuple(rows)) for tid, rows in out)
|
|
|
|
|
|
def tables_of(path: Path) -> dict[str, tuple[dict[str, Any], ...]]:
|
|
return dict(_tables_of(str(path), path.stat().st_mtime_ns))
|
|
|
|
|
|
def load_labels() -> dict[str, dict[str, Any]]:
|
|
"""이름표 파일 모두를 한 벌로 — 부를 때마다 읽음(작은 파일 · 고치면 바로 보임)."""
|
|
merged: dict[str, dict[str, Any]] = {"groups": {}, "files": {}, "tables": {}, "columns": {}}
|
|
for p in sorted(LABEL_DIR.glob("*.json")) if LABEL_DIR.is_dir() else []:
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
for section, entries in merged.items():
|
|
entries.update(data.get(section) or {})
|
|
return merged
|
|
|
|
|
|
def _text(entry: Any, key: str = "label") -> str:
|
|
if isinstance(entry, dict):
|
|
return str(entry.get(key) or "")
|
|
return str(entry or "") if key == "label" else ""
|
|
|
|
|
|
def tree() -> dict[str, Any]:
|
|
labels = load_labels()
|
|
files = master_files()
|
|
groups = []
|
|
for gkey, ids in GROUPS:
|
|
entries = []
|
|
for fid in ids:
|
|
if fid not in files:
|
|
continue
|
|
entries.append(
|
|
{
|
|
"id": fid,
|
|
"label": _text(labels["files"].get(fid)) or fid,
|
|
"key": files[fid].name,
|
|
"tables": [
|
|
{
|
|
"id": tid,
|
|
"label": _text(labels["tables"].get(f"{fid}/{tid}"))
|
|
or tid.rsplit("/", 1)[-1],
|
|
"key": tid.rsplit("/", 1)[-1],
|
|
"row_count": len(rows),
|
|
}
|
|
for tid, rows in tables_of(files[fid]).items()
|
|
],
|
|
}
|
|
)
|
|
groups.append(
|
|
{"key": gkey, "label": _text(labels["groups"].get(gkey)) or gkey, "files": entries}
|
|
)
|
|
return {"groups": groups}
|
|
|
|
|
|
def rows(
|
|
fid: str, tid: str, page: int = 1, size: int = DEFAULT_PAGE_SIZE, q: str = ""
|
|
) -> dict[str, Any] | None:
|
|
"""표 줄 한 쪽 — 없는 파일·표는 None. 검색은 줄 값 글자에 든 것(대소문자 무시)."""
|
|
path = master_files().get(fid)
|
|
table = tables_of(path).get(tid) if path else None
|
|
if table is None:
|
|
return None
|
|
columns: dict[str, None] = {}
|
|
for r in table:
|
|
columns.update(dict.fromkeys(r))
|
|
needle = q.strip().lower()
|
|
hits = (
|
|
[r for r in table if needle in json.dumps(r, ensure_ascii=False).lower()]
|
|
if needle
|
|
else list(table)
|
|
)
|
|
size = max(1, min(size, MAX_PAGE_SIZE))
|
|
start = (max(page, 1) - 1) * size
|
|
col_labels = load_labels()["columns"]
|
|
out_cols = []
|
|
for key in columns:
|
|
entry = col_labels.get(f"{fid}/{key}") or col_labels.get(key)
|
|
hidden = entry.get("hidden") if isinstance(entry, dict) and "hidden" in entry else None
|
|
out_cols.append(
|
|
{
|
|
"key": key,
|
|
"label": _text(entry) or key,
|
|
"unit": _text(entry, "unit"),
|
|
"hidden": bool(hidden)
|
|
if hidden is not None
|
|
else key.rsplit(".", 1)[-1] in _HIDDEN_KEYS,
|
|
}
|
|
)
|
|
return {"columns": out_cols, "rows": hits[start : start + size], "total": len(hits)}
|