diff --git a/Z01_MasterData/Z01_MasterData_Tables.py b/Z01_MasterData/Z01_MasterData_Tables.py index 0779b85f..1d544f90 100644 --- a/Z01_MasterData/Z01_MasterData_Tables.py +++ b/Z01_MasterData/Z01_MasterData_Tables.py @@ -1,280 +1,163 @@ """Z01 마스터 데이터 — `resources/` 마스터 JSON 을 **갈래 → 파일 → 표 → 줄** 로 가름(읽기 전용). -마스터 = 로직 + 기초값(2026-09-15 사용자 방향 · bottom-up). 가르는 잣대 = 「값이 바뀌는 것 = 기초값 · 규칙·표 = 로직」(브레인). -표 가르기(브레인 승인): - · `variables` 칸은 변수 하나 = 표 하나 — 변수 안에 목록이 둘 이상이면 한 표에 잇고 `@part` 열에 목록 이름 - · 나머지는 줄 모음마다 표 하나 — 딕셔너리 목록 · 같은 길이 **수** 열(글 목록은 길이가 같아도 짝짓지 않음) · - 딕셔너리의 딕셔너리(`@key` 열) · 값 딕셔너리(`@key`·`@value` 줄) - · 설명(note·source·policy …)은 표가 아님 — 줄 안의 설명 칸은 그대로 둠 -한글 이름·단위는 **이름표 파일**(`resources/data_master_labels/*.json` · 데스크탑 서브 몫)에서만 — 없으면 영문 key. +마스터 = 로직 + 기초값(2026-09-15 사용자 방향 · bottom-up). +**이름표 파일**(`resources/data_master_labels/labels_*.json` · 데스크탑 서브)이 파일 목록·갈래·한글 이름·단위·보임의 자리 — +여기는 그 목록의 파일을 읽어 표를 찾고 줄을 냄. 갈래표를 코드에 또 두지 않음(두 벌 금지). +표 = **화면이 표로 그릴 마디** — 이름표 생성기와 같은 잣대(브레인 「91 이 정본」): + · 딕셔너리 목록(비지 않음) → 줄 = 딕셔너리 그대로({열key: 값}) + · 딕셔너리 둘 이상이 모인 딕셔너리(열이 비슷함) → 줄마다 `@key` 열(그 줄의 이름) + 값 딕셔너리 +⚠ 표 id 가 이름표와 한 글자라도 다르면 이름이 통째로 안 붙음 — `test_z01_master_data.py` 가 양쪽으로 셈. """ 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" +ROOT = Path(__file__).resolve().parent.parent +RESOURCES = ROOT / "resources" LABEL_DIR = RESOURCES / "data_master_labels" DEFAULT_PAGE_SIZE = 50 MAX_PAGE_SIZE = 500 +ROW_KEY = "@key" -# 갈래 나눔표(브레인 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", +# 파일 머리 — 표도 값 묶음도 아님(이름표 생성기와 같은 목록) +META_KEYS = { + "schema_version", "dataset_id", "effective_date", "generated_at", "publication_date", "survey_month", + "pum_edition", "dataset_version", "source_master_file", "source_dataset_version", } # 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 _is_table(node: Any) -> str | None: + if isinstance(node, list) and node and all(isinstance(r, dict) for r in node): + return "list" + if isinstance(node, dict): + values = list(node.values()) + dicts = [v for v in values if isinstance(v, dict)] + if ( + len(values) >= 2 + and len(dicts) == len(values) + and not any( + isinstance(v.get("records"), list) or isinstance(v.get("rows"), list) for v in dicts + ) + ): + keysets = {frozenset(v) for v in dicts} + if len(keysets) <= max(2, len(dicts) // 3 + 1): + return "map" + return None -def _scalar(v: Any) -> bool: - return not isinstance(v, (dict, list)) +def _has_table(node: Any, depth: int = 0) -> bool: + if _is_table(node): + return True + if depth > 5 or not isinstance(node, dict): + return False + return any(_has_table(v, depth + 1) for v in node.values()) -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 +def _collect(node: Any, path: list[str], out: dict[str, list], depth: int = 0) -> None: + shape = _is_table(node) + if shape == "list": + out["/".join(path)] = node + elif shape == "map": + out["/".join(path)] = [{ROW_KEY: k, **v} for k, v in node.items()] + elif depth <= 5 and isinstance(node, dict) and any(_has_table(v) for v in node.values()): + for key, value in node.items(): + _collect(value, [*path, key], out, depth + 1) @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: str, _mtime_ns: int) -> dict[str, list[dict[str, Any]]]: + doc = json.loads(Path(path).read_text(encoding="utf-8")) + out: dict[str, list[dict[str, Any]]] = {} + for key, value in doc.items(): + if key not in META_KEYS: + _collect(value, [key], out) + return out -def tables_of(path: Path) -> dict[str, tuple[dict[str, Any], ...]]: - return dict(_tables_of(str(path), path.stat().st_mtime_ns)) +def tables_of(path: Path) -> dict[str, list[dict[str, Any]]]: + """표 id(이름표의 표 key · `/` 로 이은 자리) → 줄.""" + return _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 +@lru_cache(maxsize=4) +def _labels_at(path: str, _mtime_ns: int) -> dict[str, Any]: + return json.loads(Path(path).read_text(encoding="utf-8")) -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 load_labels() -> dict[str, Any]: + """이름표 끝 판(이름 차례) — 없으면 빈 이름표(파일도 안 보임).""" + found = sorted(LABEL_DIR.glob("labels_*.json")) if LABEL_DIR.is_dir() else [] + if not found: + return {"kinds": {}, "files": [], "columns": {}, "column_overrides": {}} + return _labels_at(str(found[-1]), found[-1].stat().st_mtime_ns) + + +def _file_path(entry: dict[str, Any]) -> Path | None: + path = (ROOT / str(entry.get("path") or "")).resolve() + return path if path.is_file() and path.is_relative_to(RESOURCES) else None 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} + groups = { + k: {"key": k, "label": v.get("name_ko") or k, "files": []} + for k, v in labels["kinds"].items() + } + for entry in labels["files"]: + path = _file_path(entry) + if path is None: + continue + named = {t["key"]: t.get("name_ko") or "" for t in entry.get("tables") or []} + kind = entry.get("kind") or "" + group = groups.setdefault(kind, {"key": kind, "label": kind, "files": []}) + group["files"].append( + { + "id": entry["file_id"], + "label": entry.get("name_ko") or entry["file_id"], + "key": path.name, + "tables": [ + {"id": tid, "label": named.get(tid) or tid, "key": tid, "row_count": len(rows)} + for tid, rows in tables_of(path).items() + ], + } ) - return {"groups": groups} + return {"groups": [g for g in groups.values() if g["files"]]} 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) + labels = load_labels() + entry = next((f for f in labels["files"] if f["file_id"] == fid), None) + path = _file_path(entry) if entry else None table = tables_of(path).get(tid) if path else None if table is None: return None - columns: dict[str, None] = {} + keys: dict[str, None] = {} for r in table: - columns.update(dict.fromkeys(r)) + keys.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) + else 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( + columns = [] + for key in keys: # 찾는 차례: 「파일id/열key」 → 「열key」 → 영문 key + named = labels["column_overrides"].get(f"{fid}/{key}") or labels["columns"].get(key) or {} + columns.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, + "label": named.get("name_ko") or key, + "unit": named.get("unit") or "", + "hidden": named.get("visible") is False, } ) - return {"columns": out_cols, "rows": hits[start : start + size], "total": len(hits)} + return {"columns": columns, "rows": hits[start : start + size], "total": len(hits)} diff --git a/resources/tester/test_z01_master_data.py b/resources/tester/test_z01_master_data.py index 3fbd0891..8be5f053 100644 --- a/resources/tester/test_z01_master_data.py +++ b/resources/tester/test_z01_master_data.py @@ -1,9 +1,8 @@ """Z01 마스터 데이터 — 읽기 API(갈래 → 파일 → 표 · 표 줄 쪽 나누기·검색) · 2026-09-15 브레인 새 판. -약속(브레인 승인): 갈래 넷 로직 17 · 기초값 9 · 부산물 4 · 씨앗 1 = 31 파일 · 강우 IDF 캐시는 뺌 · -variables 칸은 변수 하나 = 표 하나 · 나머지는 줄 모음(목록·같은 길이 수 열)마다 표 하나 · 설명(note·source·policy)은 표 아님 · -한글 이름은 이름표 파일(`resources/data_master_labels/`)에서만 — 찾는 차례 「파일id/열key」 → 「열key」 → 영문 key · -unit 도 이름표에서만(자료에서 지어내지 않음) · 숨김 = 내부 id · sha · 생성시각. +약속(브레인): 파일 목록·갈래·한글 이름·단위·보임은 **이름표 파일**(`resources/data_master_labels/`)에서만 · +표 = 화면이 표로 그릴 마디(이름표 91 이 정본) · ⭐ 트리 표 id 와 이름표 표 id 를 **양쪽으로** 세어 둘 다 0 · +열 이름 찾는 차례 「파일id/열key」 → 「열key」 → 영문 key · 줄 한 줄 = {열key: 값} · file·table 은 트리의 id. """ from __future__ import annotations @@ -18,25 +17,13 @@ from fastapi.testclient import TestClient from Z01_MasterData import Z01_MasterData_Router as router_module from Z01_MasterData import Z01_MasterData_Tables as tables -GROUPS = { - "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 +LABELS = json.loads( + (tables.LABEL_DIR / "labels_2026-01-01.json").read_text(encoding="utf-8") +) # 서브 이름표 판 @pytest.fixture -def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient: - monkeypatch.setattr(tables, "LABEL_DIR", tmp_path / "labels") # 이름표 없음 — 영문 key 그대로 +def client() -> TestClient: app = FastAPI() app.include_router(router_module.router) return TestClient(app) @@ -54,120 +41,135 @@ def _rows(client: TestClient, **params) -> dict: return res.json() -def test_갈래_넷에_파일_31_강우_캐시는_없음(client: TestClient) -> None: +def _files(client: TestClient) -> dict[str, dict]: + return {f["id"]: f for g in _tree(client)["groups"] for f in g["files"]} + + +def test_갈래는_이름표_차례_파일_34_강우_캐시는_없음(client: TestClient) -> None: groups = _tree(client)["groups"] - assert [g["key"] for g in groups] == ["logic", "base", "byproduct", "seed"] - for g in groups: - assert {f["id"] for f in g["files"]} == GROUPS[g["key"]], g["key"] - assert g["label"] == g["key"] # 이름표가 없으면 영문 key - for f in g["files"]: - assert f["label"] == f["id"] and f["key"].endswith(".json") and f["tables"], f["id"] - ids = [f["id"] for g in groups for f in g["files"]] - assert len(ids) == 31 and not any("idf" in i or "rainfall" in i for i in ids) + assert [g["key"] for g in groups] == [k for k in LABELS["kinds"]] + counts = {g["key"]: len(g["files"]) for g in groups} + assert counts == {"logic": 17, "base_value": 9, "byproduct": 4, "fingerprint": 3, "seed": 1} + assert {g["key"]: g["label"] for g in groups}["base_value"] == "기초값" + by_id = {f["file_id"]: f for f in LABELS["files"]} + for f in (f for g in groups for f in g["files"]): + assert ( + f["label"] == by_id[f["id"]]["name_ko"] + and f["key"] == Path(by_id[f["id"]]["path"]).name + ) + assert not [ + f for f in _files(client).values() if "rainfall" in f["key"] or f["key"][:3] == "002" + ] -def test_variables_는_변수_하나가_표_하나(client: TestClient) -> None: - files = {f["id"]: f for g in _tree(client)["groups"] for f in g["files"]} - rates = [t for t in files["rates"]["tables"] if t["id"].startswith("variables/")] - assert len(rates) == 19, [t["id"] for t in rates] - vat = next(t for t in rates if t["id"] == "variables/rate_vat") - assert ( - vat["key"] == "rate_vat" and vat["label"] == "rate_vat" and vat["row_count"] == 2 - ) # rate_percent · base - assert len([t for t in files["coef"]["tables"] if t["id"].startswith("variables/")]) == 5 - assert "processing_rules" in {t["id"] for t in files["rates"]["tables"]} # 변수 밖 규칙도 표 - # 한 변수에 목록이 둘이면 한 표에 잇고 어느 목록인지 열로 가름 - fuel = _rows(client, file="mach_base", table="variables/mach_fuel_rate", size=500) - assert fuel["total"] == 214 + 21 - assert {r["@part"] for r in fuel["rows"]} >= {"parsed_records"} +def test_트리_표_id_와_이름표_표_id_는_양쪽으로_같음(client: TestClient) -> None: + files = _files(client) + not_labelled, label_only = [], [] + for entry in LABELS["files"]: + tree_ids = {t["id"] for t in files[entry["file_id"]]["tables"]} + label_ids = {t["key"] for t in entry["tables"]} + not_labelled += [f"{entry['file_id']}::{i}" for i in sorted(tree_ids - label_ids)] + label_only += [f"{entry['file_id']}::{i}" for i in sorted(label_ids - tree_ids)] + assert not_labelled == [], not_labelled + assert label_only == [], label_only + assert sum(len(f["tables"]) for f in files.values()) == LABELS["counts"]["tables"] == 91 + for entry in LABELS["files"]: # 이름이 실제로 붙음 · 줄 수도 이름표 셈과 같음 + named = {t["key"]: t for t in entry["tables"]} + for t in files[entry["file_id"]]["tables"]: + assert t["label"] == named[t["id"]]["name_ko"], (entry["file_id"], t["id"]) + assert t["row_count"] == named[t["id"]]["rows"], (entry["file_id"], t["id"]) def test_자재_단가는_쪽으로_나눠_주고_검색으로_좁힘(client: TestClient) -> None: - first = _rows(client, file="mat_price_public", table="variables/mat_price", page=1, size=50) - second = _rows(client, file="mat_price_public", table="variables/mat_price", page=2, size=50) + table = "variables/mat_price/records" + first = _rows(client, file="mat_price_public", table=table, page=1, size=50) + second = _rows(client, file="mat_price_public", table=table, page=2, size=50) assert first["total"] == 6999 and len(first["rows"]) == 50 and first["rows"] != second["rows"] - assert {c["key"] for c in first["columns"]} >= {"item_code", "price_krw", "specification"} - assert "key" not in { - c["key"] for c in first["columns"] - } # 「어느 열이 키인지」 머리는 줄 값이 아님 - found = _rows( - client, file="mat_price_public", table="variables/mat_price", q="육각볼트", size=500 - ) + raw = json.loads( + Path( + tables.ROOT / "resources/data_cost_input_value/mat_price_public_2026-08-14.json" + ).read_text(encoding="utf-8") + )["variables"]["mat_price"]["records"] + assert first["rows"][0] == raw[0] and second["rows"][0] == raw[50] # 한 줄 = {열key: 값} 그대로 + found = _rows(client, file="mat_price_public", table=table, q="육각볼트", size=500) assert 0 < found["total"] < 6999 assert all("육각볼트" in json.dumps(r, ensure_ascii=False) for r in found["rows"]) - capped = _rows(client, file="mat_price_public", table="variables/mat_price", size=100000) + capped = _rows(client, file="mat_price_public", table=table, size=100000) assert len(capped["rows"]) == tables.MAX_PAGE_SIZE -def test_같은_길이_수_열은_줄로_세움(client: TestClient) -> None: - slope = _rows(client, file="masonry_slope", table="table/메쌓기") - assert slope["total"] == 5 and slope["rows"][0] == {"성토": 0.3, "절토": 0.25} - wedge = _rows(client, file="masonry_wet", table="tables/고임돌표") - assert wedge["total"] == 7 and wedge["rows"][0]["keys"] == 25 - # 글 목록 둘은 길이가 같아도 짝짓지 않음(자재 이름 ↔ 뒤진 자리) - files = {f["id"]: f for g in _tree(client)["groups"] for f in g["files"]} - ids = {t["id"] for t in files["material_surcharge"]["tables"]} - assert {"not_found/materials", "not_found/checked"} <= ids +def test_묶음_표는_줄마다_이름_열(client: TestClient) -> None: + slope = _rows(client, file="masonry_slope", table="table") + assert [r["@key"] for r in slope["rows"]] == ["메쌓기", "찰쌓기"] + assert slope["rows"][0]["성토"] == [0.3, 0.35, 0.4, 0.45, 0.5] + cols = {c["key"]: c for c in slope["columns"]} + assert cols["성토"]["label"] == "성토부 경사" and cols["성토"]["unit"] == "1:n" + fx = {c["key"]: c for c in _rows(client, file="fx", table="variables")["columns"]} + assert fx["value"]["label"] == "환율" and fx["value"]["unit"] == "원" # 「fx/value」 덮어쓰기 -def test_원문_표는_번호로_찾음(client: TestClient) -> None: - hit = _rows(client, file="pum_forest", table="variables/pum", q="F0155") +def test_숨김은_이름표의_보임에서(client: TestClient) -> None: + cols = {c["key"]: c for c in _rows(client, file="pum_forest", table="sources")["columns"]} + assert cols["sha256"]["hidden"] is True and cols["path"]["hidden"] is False + hit = _rows(client, file="pum_forest", table="variables/pum/tables", q="F0155") assert [r["table_id"] for r in hit["rows"]] == ["F0155"] -def test_내부_id_sha_생성시각은_숨김(client: TestClient) -> None: - axis = _rows(client, file="resource_axis", table="rows", size=1) - hidden = {c["key"]: c["hidden"] for c in axis["columns"]} - assert hidden["raw_row_index"] is True and hidden["resource_name"] is False - master = _rows(client, file="work_item_master", table="work_items", size=1) - assert {c["key"]: c["hidden"] for c in master["columns"]}["sort_order"] is True - assert all(c["unit"] == "" and c["label"] == c["key"] for c in axis["columns"]) # 이름표 없음 - - -def test_이름표_파일이_있으면_label_unit_을_붙임(client: TestClient, tmp_path: Path) -> None: - labels = tmp_path / "labels" - labels.mkdir() - (labels / "labels.json").write_text( +def test_찾는_차례와_없는_이름은_영문_key( + client: TestClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + (tmp_path / "labels_2026-01-01.json").write_text( json.dumps( { - "groups": {"base": "기초값"}, - "files": {"rates": "요율"}, - "tables": {"rates/variables/rate_vat": "부가가치세"}, + "kinds": {"base_value": {"name_ko": "기초값"}}, + "files": [ + { + "file_id": "rates", + "path": "resources/data_cost_input_value/rates_2026.json", + "kind": "base_value", + "name_ko": "", + "tables": [], + }, + {"file_id": "outside", "path": "config/config_db.py", "kind": "base_value"}, + ], "columns": { - "base": {"label": "밑수"}, - "rates/rate_percent": {"label": "요율", "unit": "%"}, - "rate_percent": {"label": "딴 이름", "unit": "딴 단위"}, + "rate_percent": {"name_ko": "딴 이름", "unit": "딴 단위", "visible": True}, + "grade": {"name_ko": "등급", "unit": "", "visible": False}, }, + "column_overrides": {"rates/rate_percent": {"name_ko": "요율", "unit": "%"}}, }, ensure_ascii=False, ), encoding="utf-8", ) - base = next(g for g in _tree(client)["groups"] if g["key"] == "base") - rates = next(f for f in base["files"] if f["id"] == "rates") - assert base["label"] == "기초값" and rates["label"] == "요율" - assert ( - next(t for t in rates["tables"] if t["id"] == "variables/rate_vat")["label"] == "부가가치세" - ) - cols = {c["key"]: c for c in _rows(client, file="rates", table="variables/rate_vat")["columns"]} - assert cols["@key"]["label"] == "@key" # 없는 것은 영문 key - vat = _rows(client, file="rates", table="variables/rate_sanjae") - assert {r["@key"] for r in vat["rows"]} == {"rate_percent", "base"} - goyong = { - c["key"]: c for c in _rows(client, file="rates", table="variables/rate_goyong")["columns"] + monkeypatch.setattr(tables, "LABEL_DIR", tmp_path) + groups = _tree(client)["groups"] + assert [f["id"] for g in groups for f in g["files"]] == ["rates"] # resources 밖은 안 읽음 + rates = groups[0]["files"][0] + assert rates["label"] == "rates" + goyong = next(t for t in rates["tables"] if t["id"] == "variables/rate_goyong/brackets") + assert goyong["label"] == goyong["id"] + cols = {c["key"]: c for c in _rows(client, file="rates", table=goyong["id"])["columns"]} + assert cols["rate_percent"]["label"] == "요율" and cols["rate_percent"]["unit"] == "%" + assert cols["grade"]["label"] == "등급" and cols["grade"]["hidden"] is True + assert cols["estimated_amount_bracket"] == { + "key": "estimated_amount_bracket", + "label": "estimated_amount_bracket", + "unit": "", + "hidden": False, } assert ( - goyong["rate_percent"]["label"] == "요율" and goyong["rate_percent"]["unit"] == "%" - ) # 파일 것이 이김 - assert goyong["base"]["label"] == "밑수" and goyong["base"]["unit"] == "" + client.get("/api/master-data/rows", params={"file": "outside", "table": "x"}).status_code + == 404 + ) -def test_없는_파일_표는_404_경로_넘기도_404(client: TestClient) -> None: +def test_없는_파일_표는_404(client: TestClient) -> None: for params in ( {"file": "nope", "table": "rows"}, {"file": "rates", "table": "nope"}, + {"file": "rates", "table": "variables/rate_sanjae"}, # 값 묶음 — 표 아님 {"file": "../../config/config_db", "table": "rows"}, - {"file": "002yr_01hr", "table": "features"}, # 강우 IDF 캐시 — 갈래표 밖 - {"file": "_manifest", "table": "files"}, + {"file": "002yr_01hr", "table": "features"}, # 강우 IDF 캐시 — 이름표 밖 ): assert client.get("/api/master-data/rows", params=params).status_code == 404, params