diff --git a/Z01_MasterData/Z01_MasterData_Router.py b/Z01_MasterData/Z01_MasterData_Router.py new file mode 100644 index 00000000..e3e1e315 --- /dev/null +++ b/Z01_MasterData/Z01_MasterData_Router.py @@ -0,0 +1,29 @@ +"""Z01 마스터 데이터 라우터 — 읽기 전용(고치기는 다음 차례 · 2026-09-15 브레인). + + GET /api/master-data/tree 갈래 → 파일 → 표 + GET /api/master-data/rows?file=&table=&page=&size=&q= 표 줄(쪽 나누기 · 검색) +⚠ 권한은 등록하는 쪽(`main.py` · 랩탑 서브)이 `dependencies=[verify_session, require_system_admin]` 로 붙임. +""" + +from __future__ import annotations + +from fastapi import APIRouter, HTTPException + +from Z01_MasterData import Z01_MasterData_Tables as tables + +router = APIRouter(prefix="/api/master-data", tags=["Z01 MasterData"]) + + +@router.get("/tree") +def get_tree() -> dict: + return tables.tree() + + +@router.get("/rows") +def get_rows( + file: str, table: str, page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE, q: str = "" +) -> dict: + result = tables.rows(file, table, page=page, size=size, q=q) + if result is None: + raise HTTPException(status_code=404, detail="없는 파일이나 표입니다.") + return result diff --git a/Z01_MasterData/Z01_MasterData_Tables.py b/Z01_MasterData/Z01_MasterData_Tables.py new file mode 100644 index 00000000..0779b85f --- /dev/null +++ b/Z01_MasterData/Z01_MasterData_Tables.py @@ -0,0 +1,280 @@ +"""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)} diff --git a/resources/tester/test_z01_master_data.py b/resources/tester/test_z01_master_data.py new file mode 100644 index 00000000..3fbd0891 --- /dev/null +++ b/resources/tester/test_z01_master_data.py @@ -0,0 +1,173 @@ +"""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 · 생성시각. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from fastapi import FastAPI +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 + + +@pytest.fixture +def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient: + monkeypatch.setattr(tables, "LABEL_DIR", tmp_path / "labels") # 이름표 없음 — 영문 key 그대로 + app = FastAPI() + app.include_router(router_module.router) + return TestClient(app) + + +def _tree(client: TestClient) -> dict: + res = client.get("/api/master-data/tree") + assert res.status_code == 200, res.text + return res.json() + + +def _rows(client: TestClient, **params) -> dict: + res = client.get("/api/master-data/rows", params=params) + assert res.status_code == 200, res.text + return res.json() + + +def test_갈래_넷에_파일_31_강우_캐시는_없음(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) + + +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_자재_단가는_쪽으로_나눠_주고_검색으로_좁힘(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) + 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 + ) + 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) + 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: + hit = _rows(client, file="pum_forest", table="variables/pum", 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( + json.dumps( + { + "groups": {"base": "기초값"}, + "files": {"rates": "요율"}, + "tables": {"rates/variables/rate_vat": "부가가치세"}, + "columns": { + "base": {"label": "밑수"}, + "rates/rate_percent": {"label": "요율", "unit": "%"}, + "rate_percent": {"label": "딴 이름", "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"] + } + assert ( + goyong["rate_percent"]["label"] == "요율" and goyong["rate_percent"]["unit"] == "%" + ) # 파일 것이 이김 + assert goyong["base"]["label"] == "밑수" and goyong["base"]["unit"] == "" + + +def test_없는_파일_표는_404_경로_넘기도_404(client: TestClient) -> None: + for params in ( + {"file": "nope", "table": "rows"}, + {"file": "rates", "table": "nope"}, + {"file": "../../config/config_db", "table": "rows"}, + {"file": "002yr_01hr", "table": "features"}, # 강우 IDF 캐시 — 갈래표 밖 + {"file": "_manifest", "table": "files"}, + ): + assert client.get("/api/master-data/rows", params=params).status_code == 404, params