Aislo/Z01_MasterData/Z01_MasterData_Tables.py

"""Z01 마스터 데이터 — `resources/` 마스터 JSON 을 **갈래 → 파일 → 표 → 줄** 로 가름(읽기 전용).

마스터 = 로직 + 기초값(2026-09-15 사용자 방향 · bottom-up).
**이름표 파일**(`resources/data_master_labels/labels_*.json` · 데스크탑 서브)이 파일 목록·갈래·한글 이름·단위·보임의 자리 —
여기는 그 목록의 파일을 읽어 표를 찾고 줄을 냄. 갈래표를 코드에 또 두지 않음(두 벌 금지).
표 = **화면이 표로 그릴 마디** — 이름표 생성기와 같은 잣대(브레인 「91 이 정본」):
  · 딕셔너리 목록(비지 않음) → 줄 = 딕셔너리 그대로({열key: 값})
  · 딕셔너리 둘 이상이 모인 딕셔너리(열이 비슷함) → 줄마다 `@key` 열(그 줄의 이름) + 값 딕셔너리
  · 표 아닌 마디(값 묶음 — 산재 3.56% 같은 낱값)는 파일마다 표 하나 `@values`(이름 · 값) — 설명·방침은 뺌(브레인 ④)
⚠ 표 id 가 이름표와 한 글자라도 다르면 이름이 통째로 안 붙음 — `test_z01_master_data.py` 가 양쪽으로 셈.
"""

from __future__ import annotations

import json
from functools import lru_cache
from pathlib import Path
from typing import Any

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"
VALUES_TABLE = "@values"
_CODE_HIDDEN = {"@path"}  # 낱값 표의 영문 자리 — 이름표에 없을 때만

# 값 묶음 갈래 — 이름표 `value_groups[].role`(데스크탑 서브) · 낱말이 바뀌면 이 자리만 갈아끼움
ROLE_FIELD = "role"
SHOWN_ROLE = "value"  # 원가·수량에 드는 값 — 낱값 표에 보임
HIDDEN_ROLE = "note"  # 읽으라고 적은 글 — 안 보임
# 이름표가 안 가른(또는 모르는 낱말) 마디 — 조용히 빼지도 섞지도 않고 값으로 보이되 드러냄(브레인)
UNDECIDED_COLUMN = "@undecided"
UNDECIDED = "미판정"


# 파일 머리 — 표도 값 묶음도 아님(이름표 생성기와 같은 목록)
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


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 _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 _collect(
    node: Any, path: list[str], out: dict[str, list], values: dict, 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, values, depth + 1)
    else:
        values["/".join(path)] = node


@lru_cache(maxsize=64)
def _scan(path: str, _mtime_ns: int) -> tuple[dict[str, list[dict[str, Any]]], dict[str, Any]]:
    doc = json.loads(Path(path).read_text(encoding="utf-8"))
    out: dict[str, list[dict[str, Any]]] = {}
    values: dict[str, Any] = {}
    for key, value in doc.items():
        if key not in META_KEYS:
            _collect(value, [key], out, values)
    return out, values


def tables_of(path: Path) -> dict[str, list[dict[str, Any]]]:
    """표 id(이름표의 표 key · `/` 로 이은 자리) → 줄."""
    return _scan(str(path), path.stat().st_mtime_ns)[0]


def values_of(path: Path) -> dict[str, Any]:
    """값 묶음 자리(이름표의 값 묶음 key) → 값 — 설명·방침도 섞임."""
    return _scan(str(path), path.stat().st_mtime_ns)[1]


def _value_rows(entry: dict[str, Any], path: Path, labels: dict[str, Any]) -> list[dict[str, Any]]:
    """낱값 표 줄 — 이름 찾는 차례: 「파일id::key」 → 「끝 조각」 → 영문 key."""
    fid = entry["file_id"]
    roles = {g.get("key"): g.get(ROLE_FIELD) for g in entry.get("value_groups") or []}
    rows = []
    for key, value in values_of(path).items():
        role = roles.get(key)
        if role == HIDDEN_ROLE:
            continue
        named = (
            (labels.get("value_overrides") or {}).get(f"{fid}::{key}")
            or (labels.get("value_keys") or {}).get(key.rsplit("/", 1)[-1])
            or {}
        )
        row = {"@name": named.get("name_ko") or key, "@value": value, "@path": key}
        if role != SHOWN_ROLE:
            row[UNDECIDED_COLUMN] = UNDECIDED
        rows.append(row)
    return rows


@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 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()
    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": []})
        counts = {tid: len(rows) for tid, rows in tables_of(path).items()}
        value_count = len(_value_rows(entry, path, labels))
        if value_count:
            counts[VALUES_TABLE] = value_count
        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": n}
                    for tid, n in counts.items()
                ],
            }
        )
    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. 검색은 줄 값 글자에 든 것(대소문자 무시)."""
    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
    if path and tid == VALUES_TABLE:
        table = _value_rows(entry, path, labels) or None
    else:
        table = tables_of(path).get(tid) if path else None
    if table is None:
        return None
    own = next((t for t in entry.get("tables") or [] if t.get("key") == tid), {})
    own_columns = {c.get("key"): c for c in own.get("columns") or []}
    keys: dict[str, None] = {}
    for r in table:
        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 table
    )
    size = max(1, min(size, MAX_PAGE_SIZE))
    start = (max(page, 1) - 1) * size
    columns = []
    for key in keys:  # 찾는 차례: 그 표 열 목록(줄 이름 열 `@key` 는 표마다 제 이름) → 「파일id/열key」 → 「열key」 → 영문 key
        named = (
            own_columns.get(key)
            or labels["column_overrides"].get(f"{fid}/{key}")
            or labels["columns"].get(key)
            or {}
        )
        columns.append(
            {
                "key": key,
                "label": named.get("name_ko") or key,
                "unit": named.get("unit") or "",
                "hidden": named.get("visible") is False if named else key in _CODE_HIDDEN,
            }
        )
    return {"columns": columns, "rows": hits[start : start + size], "total": len(hits)}