Aislo/Z01_MasterData/Z01_MasterData_BasePrices.py

"""Z01 기초단가 다섯 — 노임·기계·자재·유가·요율을 kind 마다 **한 표**로(2026-09-15 사용자 지시 · 브레인 계약).

표 수준(한 번): columns · total · editable · locked{열: 왜 못 고치나} · formula{열: 식} · notice(요율만)
줄 수준(해당할 때만): `@id`(자료 제 열쇠) · `@overrides`(고친 줄) · `@formula`(계산값 줄)
`@id` — 줄 차례가 아니라 자료 열쇠(브레인 승인): 노임 `{dataset_id}/{occupation_code}` · 기계 machine_code ·
  자재 item_code · 유가 `{scope}/{변수}[/{sido_code}]`(원본 값 그대로 · 열 이름은 이름표 것) · 요율 `{변수}[/{목록}/{구간칸=값;…}]`(구간 칸 이름 차례 · 값·식 칸 뺌)
원본 파일 자리는 이름표 파일이 가리키는 곳(`Z01_MasterData_Tables`) — 원본은 읽기만, 고친 값은 덮개(`Z01_MasterData_Overrides`).
"""

from __future__ import annotations

import json
from datetime import date
from decimal import Decimal
from functools import lru_cache
from pathlib import Path
from typing import Any

from Z01_MasterData import Z01_MasterData_BasePrices_Machine as machine
from Z01_MasterData import Z01_MasterData_Overrides as overrides
from Z01_MasterData import Z01_MasterData_Tables as tables

KINDS = ("labor", "machine", "material", "oil", "rate")
SOURCES_DIR = (
    tables.RESOURCES / "data_master_sources"
)  # 출처표(데스크탑 서브) — 어디서 받고 최신이 무엇인지
_KEY = "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음"
_SOURCE = "공표 원문 칸 — 원본 갱신으로만 바뀜"
_RATE_EDITABLE = (
    "rate_percent",
    "base_amount_krw",
    "minimum_estimated_amount_krw",
    "minimum_total_construction_amount_krw",
    "rate_from_2033_percent",
    "manager_thresholds.default_estimated_amount_krw",
    "manager_thresholds.civil_main_work_estimated_amount_krw",
)
_SCOPE_NOTICE = (
    "여기서 고친 값은 **새로 만드는 프로젝트부터** 쓰임 — 이미 만든 프로젝트 금액은 안 바뀜"
    "(프로젝트는 만들 때 뜬 사본을 봄)"
)
_RATE_NOTICE = "법이 정한 값 — 고치면 원본 갱신 때 구간 이름이 바뀌어 고친 값이 주인을 잃기 쉬움(주인 없음으로 남음)"


def _source(file_id: str) -> Path:
    labels = tables.load_labels()
    entry = next((f for f in labels["files"] if f["file_id"] == file_id), None)
    path = tables._file_path(entry) if entry else None
    if path is None:
        raise FileNotFoundError(f"이름표에 원본 자리가 없음: {file_id}")
    return path


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


def _doc(file_id: str) -> tuple[dict[str, Any], str]:
    path = _source(file_id)
    return _read(str(path), path.stat().st_mtime_ns), path.name


def edition_of(doc: dict[str, Any]) -> str | None:
    """판 기준일 — 자료마다 자리가 달라 **여기 한 곳**에서 뽑음(`effective_date` · 파생본은 `derived_from.effective_date`)."""
    derived = doc.get("derived_from")
    return doc.get("effective_date") or (
        derived.get("effective_date") if isinstance(derived, dict) else None
    )


def _date(text: Any) -> date | None:
    try:
        return date.fromisoformat(str(text))
    except ValueError:
        return None


def sources(kind: str) -> list[dict[str, Any]]:
    """표 수준 출처 — kind 하나에 판이 둘인 곳(노임·유가·기계)이 있어 목록.

    `our_edition` 은 출처표 글자가 아니라 **실제 파일**에서 읽음(글자가 굳으면 또 뒤처짐) ·
    `outdated` 판정은 서버만(최신 공표일 > 우리 판 · 둘 중 하나라도 모르면 None).
    """
    found = sorted(SOURCES_DIR.glob("sources_*.json")) if SOURCES_DIR.is_dir() else []
    if not found:
        return []
    listing = _read(str(found[-1]), found[-1].stat().st_mtime_ns)
    out = []
    for entry in listing.get("sources") or []:
        if entry.get("kind") != kind:
            continue
        editions = []
        for name in entry.get("files") or []:
            path = (tables.ROOT / name).resolve()
            if path.is_file() and path.is_relative_to(tables.RESOURCES):
                editions.append(_date(edition_of(_read(str(path), path.stat().st_mtime_ns))))
        ours = min((e for e in editions if e), default=None)
        latest = _date(entry.get("latest_published"))
        out.append(
            {
                "source_id": entry.get("source_id"),
                "name": entry.get("name_ko") or entry.get("source_id"),
                "publisher": entry.get("publisher") or "",
                "where": entry.get("where_to_get") or "없음",  # 안 지어냄(브레인)
                "cycle": entry.get("cycle") or "",
                "latest_published": latest.isoformat() if latest else None,
                "our_edition": ours.isoformat() if ours else None,
                "checked_at": entry.get("checked_on") or None,
                "outdated": None if ours is None or latest is None else latest > ours,
            }
        )
    return out


def _records(doc: dict[str, Any]) -> list[dict[str, Any]]:
    return next(
        v["records"] for v in doc["variables"].values() if isinstance(v.get("records"), list)
    )


def _labor() -> list[dict[str, Any]]:
    rows = []
    for file_id in ("labor_const", "labor_mfg"):
        doc, name = _doc(file_id)
        dataset = doc.get("dataset_id") or file_id
        for r in _records(doc):
            # 기준일 — 두 판(건설 · 제조) 기준일이 달라 한 표에 섞임 → 칸으로 드러냄
            row = {
                "@id": f"{dataset}/{r['occupation_code']}",
                "@source": name,
                "source": dataset,
                "effective_date": edition_of(doc),
                **r,
            }
            row.setdefault("daily_wage_krw", None)  # 미공표 직종 — 사람이 넣을 수 있음
            rows.append(row)
    return rows


def _material() -> list[dict[str, Any]]:
    doc, name = _doc("mat_price_public")
    date = edition_of(doc)
    return [
        {"@id": str(r["item_code"]), "@source": name, "effective_date": date, **r}
        for r in _records(doc)
    ]


def _oil() -> list[dict[str, Any]]:
    rows = []
    for file_id in ("oil", "oil_regional"):
        doc, name = _doc(file_id)
        for variable, v in doc["variables"].items():
            head = {k: x for k, x in v.items() if k not in ("records", "value", "scope")}
            common = {
                "source": v.get("scope"),
                "fuel": variable,
                "effective_date": edition_of(doc),  # 전국 · 지역 판 기준일이 다름
                **head,
            }
            if "records" not in v:
                rows.append(
                    {
                        "@id": f"{v.get('scope')}/{variable}",
                        "@source": name,
                        **common,
                        "price_krw_per_l": v.get("value"),
                    }
                )
            for r in v.get("records") or []:
                rows.append(
                    {
                        "@id": f"{v.get('scope')}/{variable}/{r['sido_code']}",
                        "@source": name,
                        **common,
                        "region_code": r["sido_code"],
                        "region_name": r["sido_name"],
                        "price_krw_per_l": r["value"],
                    }
                )
    return rows


def _flat(d: dict[str, Any], prefix: str = "") -> dict[str, Any]:
    out: dict[str, Any] = {}
    for k, v in d.items():
        if isinstance(v, dict):
            out.update(_flat(v, f"{prefix}{k}."))
        elif not isinstance(v, list):
            out[f"{prefix}{k}"] = v
    return out


def _rate() -> list[dict[str, Any]]:
    doc, name = _doc("rates")
    rows = []
    for variable, v in doc["variables"].items():
        rows.append(
            {
                "@id": variable,
                "@source": name,
                "variable": variable,
                "part": "",
                "effective_date": edition_of(doc),
                **_flat(v),
            }
        )
        for part, items in v.items():
            if not (isinstance(items, list) and items and all(isinstance(x, dict) for x in items)):
                continue
            for item in items:
                bracket = ";".join(
                    f"{k}={item[k]}"
                    for k in sorted(item)
                    if k not in _RATE_EDITABLE
                    and k != "formula"
                    and not isinstance(item[k], (dict, list))
                )
                rows.append(
                    {
                        "@id": f"{variable}/{part}/{bracket}",
                        "@source": name,
                        "variable": variable,
                        "part": part,
                        "effective_date": edition_of(doc),
                        "base": v.get("base"),
                        **item,
                    }
                )
    return rows


_BUILDERS = {"labor": _labor, "material": _material, "oil": _oil, "rate": _rate}


def base_rows(kind: str) -> list[dict[str, Any]]:
    """덮개 얹기 전 원본 줄 — 기계는 입력 칸까지(계산 칸은 덮개를 얹은 뒤 셈)."""
    if kind == "machine":
        doc, name = _doc("mach_base")
        return machine.base_rows(name, edition_of(doc))
    return _BUILDERS[kind]()


def _machine_inputs() -> machine.Inputs:
    """기계가 끌어 쓰는 밑값 — **덮개를 얹은** 노임(건설)·전국평균 유가 + 그 판들의 기준일."""
    labor = [r for r in overrides.overlay("labor", _labor()) if r["source"] == "labor_const"]
    oil = [
        r for r in overrides.overlay("oil", _oil()) if r["@id"] == f"national_average/{r['fuel']}"
    ]
    return machine.Inputs(
        wages={
            str(r["occupation_code"]): None
            if r["daily_wage_krw"] is None
            else Decimal(str(r["daily_wage_krw"]))
            for r in labor
        },
        wage_date=labor[0]["effective_date"] if labor else None,
        oil={r["fuel"]: Decimal(str(r["price_krw_per_l"])) for r in oil},
        oil_dates={r["fuel"]: r.get("date") or r["effective_date"] for r in oil},
        operating_date=edition_of(_doc("pum_const")[0]),
    )


def rows(kind: str) -> list[dict[str, Any]]:
    """덮개를 얹은 줄 — 기계는 그 위에서 시간당 단가를 다시 셈."""
    out = overrides.overlay(kind, base_rows(kind))
    if kind == "machine":
        inputs = _machine_inputs()
        out = [machine.compute(r, inputs) for r in out]
    return out


def _columns(kind: str, all_rows: list[dict[str, Any]]) -> list[str]:
    keys: dict[str, None] = {}
    for r in all_rows:
        keys.update(dict.fromkeys(k for k in r if not k.startswith("@")))
    return list(keys)


class SortError(ValueError):
    """모르는 열로 세우라 함 — 조용히 원본 차례로 돌아가지 않음(브레인 ④)."""


def _order_key(value: Any) -> tuple:
    """섞인 칸도 세움 — 수 먼저, 그 뒤 글자."""
    if isinstance(value, bool):
        return (1, str(value))
    if isinstance(value, (int, float)):
        return (0, float(value))
    if isinstance(value, (dict, list)):
        return (1, json.dumps(value, ensure_ascii=False, default=str))
    return (1, str(value))


def _sorted(rows: list[dict[str, Any]], column: str, desc: bool) -> list[dict[str, Any]]:
    """⚠ 빈 칸은 오름·내림 **어느 쪽이든 끝** — 미공표 노임 14 · 손료계수 없는 기계 226 이
    첫 쪽을 통째로 채우면 표를 못 씀(브레인 ②)."""
    filled = [r for r in rows if r.get(column) not in (None, "")]
    empty = [r for r in rows if r.get(column) in (None, "")]
    return sorted(filled, key=lambda r: _order_key(r[column]), reverse=desc) + empty


def notice(kind: str) -> list[str]:
    """표 수준 알림 — 화면이 따로 가지면 문구가 두 벌이 됨(브레인 ③)."""
    return [_SCOPE_NOTICE] + ([_RATE_NOTICE] if kind == "rate" else [])


def spec(kind: str, columns: list[str]) -> dict[str, Any]:
    """표 수준 — editable · locked · formula."""
    if kind == "machine":
        return machine.spec(columns)
    editable = {
        "labor": ["daily_wage_krw"],
        "material": ["price_krw"],
        "oil": ["price_krw_per_l"],
        "rate": [c for c in _RATE_EDITABLE if c in columns],
    }[kind]
    keys = {"labor": {"source", "occupation_code"}, "material": {"item_code"}}.get(kind, set())
    if kind == "oil":
        keys = {"source", "fuel", "region_code"}
    if kind == "rate":
        locked = {
            c: "규칙·구간 칸(로직) — 원본 갱신으로만 바뀜" for c in columns if c not in editable
        }
    else:
        locked = {c: _KEY if c in keys else _SOURCE for c in columns if c not in editable}
    return {
        "editable": editable,
        "locked": locked,
        "formula": {},
    }


def row_label(kind: str, row: dict[str, Any] | None) -> str:
    if row is None:
        return ""
    if kind == "labor":
        return str(row.get("occupation_name") or "")
    if kind == "machine":
        return f"{row.get('machine_name') or ''} {row.get('specification') or ''}".strip()
    if kind == "material":
        return str(row.get("specification") or row.get("classification_name") or "")
    if kind == "oil":
        return f"{row.get('source_product_name') or row.get('fuel')} {row.get('region_name') or '전국평균'}"
    named = (tables.load_labels().get("value_overrides") or {}).get(
        f"rates::variables/{row['variable']}"
    ) or {}
    bracket = row["@id"].split("/", 2)[2].replace(";", " · ") if row["part"] else ""
    return f"{named.get('name_ko') or row['variable']} {bracket}".strip()


def column_meta(kind: str, key: str) -> dict[str, Any]:
    """열 이름 — 이름표 `merged_tables[kind]` → `columns[열key]` → 영문 key(이름은 이름표에만 · 브레인 ②)."""
    labels = tables.load_labels()
    merged = next(
        (t for t in labels.get("merged_tables") or [] if t.get("key") in (kind, f"{kind}s")), {}
    )
    named = next((c for c in merged.get("columns") or [] if c.get("key") == key), None)
    named = named or labels["columns"].get(key) or {}
    return {
        "key": key,
        "label": named.get("name_ko") or key,
        "unit": named.get("unit") or "",
        "hidden": named.get("visible") is False,
    }


#: 뜻 칸에서 뺄 자리 — 판·공표 날짜는 갱신 때마다 바뀌어 「뜻 바뀜」과 구별이 안 됨
_NOT_MEANING = ("effective_date", "date", "notice_datetime")


def meaning_of(kind: str, row: dict[str, Any], editable: list[str]) -> dict[str, Any]:
    """그 줄의 **값이 아닌 글자 칸**(이름·규격·단위·인도조건 …) — 이름표가 숨김 아니라 한 열만.

    갱신으로 같은 코드에 다른 기준이 붙으면(노임 국가유산직종 개편 · 자재 KS 개정) 값만으로는 못 잡음.
    """
    out = {}
    for key, value in row.items():
        if key.startswith("@") or key in editable or key in _NOT_MEANING or key.endswith("_date"):
            continue
        if isinstance(value, str) and value.strip() and not column_meta(kind, key)["hidden"]:
            out[key] = value
    return out


def public(row: dict[str, Any]) -> dict[str, Any]:
    """내보낼 줄 — 안쪽 칸(`@source`)은 뺌."""
    return {k: v for k, v in row.items() if k != "@source"}


def table(
    kind: str,
    page: int = 1,
    size: int = tables.DEFAULT_PAGE_SIZE,
    q: str = "",
    sort: str = "",
    desc: bool = False,
) -> dict[str, Any]:
    all_rows = rows(kind)
    columns = _columns(kind, all_rows)
    needle = q.strip().lower()
    hits = (
        [
            r
            for r in all_rows
            if needle in json.dumps(public(r), ensure_ascii=False, default=str).lower()
        ]
        if needle
        else all_rows
    )
    if sort:  # ⚠ 거른 뒤에 세움 — 거꾸로 하면 차례가 깨짐(브레인 ③)
        if sort not in columns:
            raise SortError(f"없는 열로 세울 수 없음: {sort}")
        hits = _sorted(hits, sort, desc)
    size = max(1, min(size, tables.MAX_PAGE_SIZE))
    start = (max(page, 1) - 1) * size
    return {
        "columns": [column_meta(kind, c) for c in columns],
        # 못 고치는 칸(계산값)도 **세우기는 됨** — 화면은 이 목록으로 눌릴 제목을 가림
        "sortable": list(columns),
        "rows": [public(r) for r in hits[start : start + size]],
        "total": len(hits),
        **spec(kind, columns),
        "notice": notice(kind),
        "source": sources(kind),
    }


def edit(kind: str, row_id: str, values: dict[str, Any], by: Any) -> tuple[int, dict[str, Any]]:
    """고치기 — (상태 코드, 고친 뒤 그 줄 또는 막은 까닭)."""
    base = next((r for r in base_rows(kind) if r["@id"] == row_id), None)
    if base is None:
        return 404, {"message": f"없는 줄: {row_id}"}
    info = spec(kind, _columns(kind, rows(kind)))
    for column, value in values.items():
        if column not in info["editable"] or column not in base:
            reason = info["locked"].get(column) or "이 줄에 없는 칸"
            return 400, {"message": f"못 고치는 칸 {column} — {reason}"}
        problem = overrides.check_value(value)
        if problem:
            return 400, {"message": f"{column}: {problem}"}
    overrides.write(kind, base, values, by, meaning_of(kind, base, info["editable"]))
    return 200, public(next(r for r in rows(kind) if r["@id"] == row_id))


def override_items(
    kind: str = "", state: str = "", page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE
) -> dict[str, Any]:
    """모아 보기 — 서버가 정렬(원본 바뀜 → 주인 없음 → 고쳐짐) · 거름(kind·state) · 쪽 · kind 별 state 수."""
    out = []
    counts: dict[str, dict[str, int]] = {}
    for each in KINDS:  # 수는 늘 다섯 다 셈 — 거름은 줄에만
        base = base_rows(each)
        by_id = {r["@id"]: r for r in base}
        for item in overrides.items(each, base):
            counts.setdefault(each, {}).setdefault(item["state"], 0)
            counts[each][item["state"]] += 1
            if (kind and each != kind) or (state and item["state"] != state):
                continue
            item["row_label"] = row_label(each, by_id.get(item["row_id"]))
            meta = column_meta(each, item["column"])
            item["column_label"] = meta["label"]
            item["unit"] = meta["unit"]
            out.append(item)
    out.sort(key=lambda i: overrides.STATE_ORDER.index(i["state"]))
    size = max(1, min(size, tables.MAX_PAGE_SIZE))
    start = (max(page, 1) - 1) * size
    return {"items": out[start : start + size], "total": len(out), "counts": counts}


def clear_overrides(kind: str, state: str) -> int:
    return overrides.clear(kind, state, base_rows(kind))