Aislo/Z01_MasterData/Z01_MasterData_WorkItems.py

"""Z01 공종 축 — 산림·건설 두 kind 를 기초단가와 **같은 계약**으로(2026-09-17 브레인 · PLAN_공종축 8-4).

줄 = **품셈 표가 붙은 마디**(`axis_policy.json` `flat_list_filter: pum_tables > 0` — 잎으로 거르면 부모에 붙은 표가 안 뜸) ·
  갈래(`variant_keys`)는 칸으로(편 수 = `stats.units`).
이름 = 뿌리부터 전체 경로(「토공 › 토사깍기 › 기계」 · 건설은 부문부터) — 잎 이름만으론 겹침.
계산 규칙 = 표 수준 `rules`(부모 마디 choose_one · sum_steps) + 줄 칸 `rule`·`step_weight`
  — 빠지면 발파암(절취 0.1 + 깍기 0.9 + 집토 1) 금액이 틀어짐.
`@id` = 불변 열쇠(FW-·CW-) — 없는 줄은 목차 코드로 서고 알림에 드러냄.
⚠ 읽기만 — 고칠 칸 없음(원문 값 풀기 8-7 뒤). 원본 `work_item_code`(FP-)는 B08·B09 가 씀 — 안 건드림.
"""

from __future__ import annotations

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

from Z01_MasterData import Z01_MasterData_Tables as tables

GROUP = "work_item"
#: kind → 원본 품셈 dataset_id — 공종 축 파일 `dataset_version.dataset_id` 로 가림(파일 이름에 기대지 않음)
SOURCE_OF = {
    "work_item_forest": "pum_forest",
    "work_item_const": "pum_const",
}  # 이름표 `work_item_axis` 와 같은 이름
FOLDER = tables.RESOURCES / "data_work_item_master"
#: 불변 열쇠 칸 — ⚠ 데스크탑 서브 자료가 오면 칸 이름을 맞출 것(여기 한 곳)
KEY_FIELD = "work_item_key"
PATH_JOIN = " › "
_LOCKED = "공종 축 뼈대 — 품셈 원문으로만 바뀜(값 고치기는 원문 값 풀기 뒤)"


@lru_cache(maxsize=8)
def _read(path: str, _mtime_ns: int) -> dict[str, Any]:
    return json.loads(Path(path).read_text(encoding="utf-8"))


def doc(kind: str) -> tuple[dict[str, Any], str] | None:
    """그 품셈의 공종 축 끝 판(문서 · 파일 이름) — 없으면 None(건설은 아직 없음)."""
    found = []
    for path in sorted(FOLDER.glob("*work_item_master_*.json")):  # 건설은 `const_` 앞꼬리
        d = _read(str(path), path.stat().st_mtime_ns)
        if (d.get("dataset_version") or {}).get("dataset_id") == SOURCE_OF[kind]:
            found.append((d, path.name))
    return max(found, key=lambda each: each[0].get("effective_date") or "", default=None)


#: 자료가 있는 품셈만 kind 로 섬 — 빈 상자는 이름만 뜨고 안 열림 · 건설은 자료가 오면 서버 재시작 뒤 저절로
KINDS = tuple(kind for kind in SOURCE_OF if doc(kind))


def _key(item: dict[str, Any]) -> str:
    return item.get(KEY_FIELD) or item["work_item_code"]


def _tree(d: dict[str, Any]) -> tuple[list[dict], dict[str, dict], dict[str, list[dict]]]:
    items = d.get("work_items") or []
    children: dict[str, list[dict]] = {}
    for item in items:
        if item.get("parent_code"):
            children.setdefault(item["parent_code"], []).append(item)
    return items, {i["work_item_code"]: i for i in items}, children


def _path(item: dict[str, Any] | None, by: dict[str, dict]) -> str:
    names = []
    while item:
        names.append(item["name"])
        item = by.get(item.get("parent_code"))
    return PATH_JOIN.join(reversed(names))


def base_rows(kind: str) -> list[dict[str, Any]]:
    found = doc(kind)
    if found is None:
        return []
    d, name = found
    items, by, children = _tree(d)
    rows = []
    for item in items:
        if not item.get("tables"):
            continue
        parent = by.get(item.get("parent_code")) or {}
        weight = next(
            (s["weight"] for s in parent.get("steps") or [] if s["code"] == item["work_item_code"]),
            None,
        )
        rows.append(
            {
                "@id": _key(item),
                "@source": name,
                "work_item_code": item["work_item_code"],  # 목차 코드 — 열쇠 아님, 칸으로만
                "number": item["number"],
                "name": _path(item, by),
                "variant_keys": item.get("variant_keys") or [],
                "pum_tables": [t["pum_table_id"] for t in item["tables"]],
                "rule": parent.get("parent_mode"),
                "step_weight": weight,
                # 「임도가 쓰는 것」 — 연결고리 표가 확정으로 이은 줄(판정은 그 표 · 감추지 않고 표시만)
                "link": LINKED if _key(item) in linked_keys() else "",
                "effective_date": d.get("pum_edition") or d.get("effective_date"),
            }
        )
    return rows


LINK_FOLDER = tables.RESOURCES / "data_work_item_link"
LINKED = "연결됨"


def linked_keys() -> frozenset[str]:
    """연결고리 표(`work_item_link_*.json` 끝 판 · 코덱스)가 **확정**으로 이은 열쇠(산림·건설).

    ⚠ 목록을 코드에 안 둠 — 판정은 그 표 · 미판정은 안 이음(표의 `unconfirmed_action: do_not_link`).
    """
    found = sorted(LINK_FOLDER.glob("work_item_link_*.json")) if LINK_FOLDER.is_dir() else []
    if not found:
        return frozenset()
    links = _read(str(found[-1]), found[-1].stat().st_mtime_ns).get("links") or []
    return frozenset(
        key
        for link in links
        if link.get("status") == "확정"
        for key in (link.get("forest_key"), link.get("const_key"))
        if key
    )


def rules(kind: str) -> list[dict[str, Any]]:
    """부모 마디 계산 규칙 — choose_one(아래 중 하나) · sum_steps(아래를 무게대로 더해야 한 단위)."""
    found = doc(kind)
    if found is None:
        return []
    items, by, children = _tree(found[0])
    out = []
    for item in items:
        mode = item.get("parent_mode")
        if not mode:
            continue
        rule = {
            "@id": _key(item),
            "work_item_code": item["work_item_code"],
            "name": _path(item, by),
            "mode": mode,
            "children": [_key(c) for c in children.get(item["work_item_code"], [])],
        }
        if mode == "sum_steps":
            rule["steps"] = [
                {"@id": _key(by[s["code"]]), "weight": s["weight"]} for s in item["steps"]
            ]
            rule["basis"] = item.get("steps_basis")
        out.append(rule)
    return out


def spec(columns: list[str]) -> dict[str, Any]:
    return {"editable": [], "locked": dict.fromkeys(columns, _LOCKED), "formula": {}}


def notice(kind: str) -> list[str]:
    lines = ["공종 축은 읽기만 — 고칠 칸 없음(품셈 원문 값 풀기 뒤에 열림)"]
    found = doc(kind)
    if found and any(KEY_FIELD not in i for i in found[0].get("work_items") or []):
        lines.append(
            "불변 열쇠(FW-·CW-) 아직 없음 — 목차 코드(FP-)를 열쇠로 씀 · 품셈 개정으로 번호가 밀리면 딴 줄을 가리킬 수 있음"
        )
    return lines


#: 거르기 — **자료에 있는 가름만**(건설 부문 · 줄 구실 · 연결고리 표). 어느 줄을 거를지는 서버 한 곳 ·
#: 화면은 고른 값만 보냄 · 기본은 늘 「전부」(감추지 않음 — 사용자 「어디에 쓸지 모르니 건설 전체」 · 브레인).
FILTER_KEYS = ("link", "division", "axis_role")
ALL = ""


def _facets(kind: str) -> dict[str, dict[str, str]]:
    found = doc(kind)
    items = found[0].get("work_items") or [] if found else []
    linked = linked_keys()
    return {
        _key(i): {
            "link": LINKED if _key(i) in linked else ALL,
            "division": str(i.get("division") or ""),
            "axis_role": str(i.get("axis_role") or ""),
        }
        for i in items
    }


def _role_names() -> dict[str, str]:
    """줄 구실 이름 — `axis_policy.json`(데스크탑 서브 잣대) 한 곳에서."""
    path = FOLDER / "axis_policy.json"
    if not path.is_file():
        return {}
    roles = _read(str(path), path.stat().st_mtime_ns).get("roles") or {}
    return {key: str(role.get("name_ko") or key) for key, role in roles.items()}


def filters(kind: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """거르기 상자 — 줄이 둘 이상으로 갈리는 가름만 · 기본은 「전부」(안 거름)."""
    facets = _facets(kind)
    labels = {
        "link": ("연결", {LINKED: "연결된 것만"}),
        "division": ("부문", {}),
        "axis_role": ("줄 구실", _role_names()),
    }
    out = []
    for key in FILTER_KEYS:
        counts: dict[str, int] = {}
        for row in rows:
            value = facets.get(row["@id"], {}).get(key, ALL)
            counts[value] = counts.get(value, 0) + 1
        values = [v for v in counts if v != ALL]
        if len(counts) < 2:  # 다 같은 값이면 거를 것이 없음(연결은 「연결됨」 · 빈칸 둘로 갈림)
            continue
        label, names = labels[key]
        out.append(
            {
                "key": key,
                "label": label,
                "options": [{"value": ALL, "label": "전부", "rows": len(rows)}]
                + [{"value": v, "label": names.get(v, v), "rows": counts[v]} for v in values],
                "default": ALL,
            }
        )
    return out


def apply_filters(
    kind: str, rows: list[dict[str, Any]], picked: dict[str, str]
) -> list[dict[str, Any]]:
    facets = _facets(kind)
    for key, value in picked.items():
        if key in FILTER_KEYS and value != ALL:
            rows = [r for r in rows if facets.get(r["@id"], {}).get(key) == value]
    return rows


def extra(kind: str) -> dict[str, Any]:
    """표 수준 덧붙임 — 계산 규칙 · 줄 수 · 갈래 편 수(쪽 나누기와 무관하게 전체)."""
    rows = base_rows(kind)
    return {
        "filters": filters(kind, rows),
        "rules": rules(kind),
        "stats": {
            "work_items": len(rows),
            "units": sum(max(1, len(r["variant_keys"])) for r in rows),
        },
    }