Merge remote-tracking branch 'origin/main_laptop_1' into sub_laptop_1
This commit is contained in:
@@ -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)}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user