Aislo/Z01_MasterData/Z01_MasterData_Overrides.py

"""Z01 기초단가 덮개 — 사람이 고친 값만 담는 층(2026-09-15 브레인 틀 · 사용자 지시 「기초단가를 고칠 수 있게」).

  원본  resources/data_*/…                              읽기만 — 주기 갱신은 이 파일을 통째로 갈아끼움
  덮개  resources/data_master_override/{kind}.json      고친 것만 — 주기가 달라 kind 마다 한 파일
  읽기  원본 줄 위에 덮개를 얹음 · 줄 수준 `@overrides[열] = {original, value, state, current}`

state 셋(판정은 **여기 한 곳** — 화면이 따로 재지 않음):
  edited          고쳐짐 — 원본 그대로(덮개를 쓸 때 본 원본 = 지금 원본)
  source_changed  고쳐짐 — 원본이 바뀜 · ⚠ 덮개 값을 그대로 씀(자동으로 원본을 쓰면 금액이 말없이 바뀜 — 사람이 정함)
  orphan          주인 없음 — 갱신으로 그 줄·칸이 사라짐 · 버리지 않고 모아 보기에 남김
되돌리기(「새 원본 값으로 받기」 도 같음) = 값 null → 그 칸을 덮개에서 뺌.
"""

from __future__ import annotations

import json
import math
import threading
from datetime import datetime
from decimal import Decimal
from pathlib import Path
from typing import Any

from common_util.common_util_json import atomic_write_json

OVERRIDE_DIR = Path(__file__).resolve().parent.parent / "resources" / "data_master_override"
EDITED, SOURCE_CHANGED, ORPHAN = "edited", "source_changed", "orphan"
STATE_ORDER = (SOURCE_CHANGED, ORPHAN, EDITED)  # 모아 보기 차례 — 확인할 것이 맨 위
_LOCK = threading.Lock()  # ponytail: 한 프로세스 잠금 — 서버를 여럿 띄우면 파일 잠금이 필요


def _path(kind: str) -> Path:
    return OVERRIDE_DIR / f"{kind}.json"


def load(kind: str) -> dict[str, dict[str, dict[str, Any]]]:
    """`@id` → 열 → {value, original, source, at, by}. 파일이 없으면 빈 덮개."""
    path = _path(kind)
    if not path.is_file():
        return {}
    return json.loads(path.read_text(encoding="utf-8")).get("overrides") or {}


def same(a: Any, b: Any) -> bool:
    """값 비교 — 215907 과 215907.0 은 같음 · None 은 None 과만 같음."""
    if a is None or b is None:
        return a is None and b is None
    if isinstance(a, (int, float)) and isinstance(b, (int, float)):
        return Decimal(str(a)) == Decimal(str(b))
    return a == b


def check_value(value: Any) -> str | None:
    """넣을 수 있는 값인가 — 막는 까닭(없으면 None). null 은 되돌리기라 통과."""
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return "숫자만 넣을 수 있음"
    if not math.isfinite(value) or value < 0:
        return "0 이상 유한한 수만 넣을 수 있음"
    return None


def mark(original: Any, value: Any, current: Any) -> dict[str, Any]:
    state = EDITED if same(original, current) else SOURCE_CHANGED
    return {"original": original, "value": value, "state": state, "current": current}


def overlay(kind: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """원본 줄(바꾸지 않음) 위에 덮개를 얹은 **새 줄들**."""
    store = load(kind)
    out = []
    for row in rows:
        entries = store.get(row["@id"])
        if not entries:
            out.append(row)
            continue
        row = dict(row)
        marks = {}
        for column, entry in entries.items():
            if column not in row:
                continue  # 주인 없는 칸 — 모아 보기에만
            marks[column] = mark(entry.get("original"), entry.get("value"), row[column])
            row[column] = entry.get("value")
        if marks:
            row["@overrides"] = marks
        out.append(row)
    return out


def write(kind: str, base_row: dict[str, Any], values: dict[str, Any], by: Any) -> None:
    """고친 값을 덮개에 씀 — null 이거나 지금 원본과 같으면 그 칸을 뺌. 원본은 안 건드림."""
    row_id = base_row["@id"]
    with _LOCK:
        store = load(kind)
        entries = dict(store.get(row_id) or {})
        for column, value in values.items():
            current = base_row.get(column)
            if value is None or same(value, current):
                entries.pop(column, None)
                continue
            entries[column] = {
                "value": value,
                "original": current,  # 이 값을 보고 고침 — 갱신 뒤 원본과 달라지면 source_changed
                "source": base_row.get("@source", ""),
                "at": datetime.now().astimezone().isoformat(timespec="seconds"),
                "by": by,
            }
        if entries:
            store[row_id] = entries
        else:
            store.pop(row_id, None)
        _save(kind, store)


def _save(kind: str, store: dict[str, dict[str, dict[str, Any]]]) -> None:
    atomic_write_json(
        _path(kind),
        {
            "schema_version": "1.0",
            "kind": kind,
            "note": "사람이 고친 기초단가만 — 원본 파일은 그대로 · Z01 마스터 데이터 화면이 씀",
            "overrides": store,
        },
    )


def clear(kind: str, state: str, base_rows: list[dict[str, Any]]) -> int:
    """한 kind 의 그 state 덮개를 한꺼번에 뺌 — 품목 퇴출로 여럿이 주인을 잃는 갱신 뒤(브레인 ②). 뺀 수."""
    with _LOCK:
        doomed = {(i["row_id"], i["column"]) for i in items(kind, base_rows) if i["state"] == state}
        if not doomed:
            return 0
        store = load(kind)
        for row_id, column in doomed:
            store[row_id].pop(column, None)
            if not store[row_id]:
                del store[row_id]
        _save(kind, store)
        return len(doomed)


def items(kind: str, base_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """모아 보기 한 kind 몫 — 주인 없는 것까지 · 차례는 부르는 쪽이 STATE_ORDER 로."""
    by_id = {r["@id"]: r for r in base_rows}
    out = []
    for row_id, entries in load(kind).items():
        row = by_id.get(row_id)
        for column, entry in entries.items():
            if row is None or column not in row:
                current, state = None, ORPHAN
            else:
                current = row[column]
                state = mark(entry.get("original"), entry.get("value"), current)["state"]
            out.append(
                {
                    "kind": kind,
                    "row_id": row_id,
                    "column": column,
                    "original": entry.get("original"),
                    "current": current,
                    "value": entry.get("value"),
                    "state": state,
                }
            )
    return out