Aislo/M01_MasterData/M01_MasterData_Machine.py

"""M01 기계 화면(PLAN 821-2) — 양식 목록 · 시간당 경비(서버 값) · 쓰는 곳 · 비슷한 기계 찾기 ·
저장 전 기계 줄 손질(계수 서버 재계산 · 단위 · 출처 채움).

계수 = `master_machine_coef`(TS 짝 `common_util/common_util_machine_coef.ts`) ·
시간당 = 서버 기계경비식(`M02_Machine_Hourly.hourly` · 전국평균 · 기본 옵션) ·
마스터 파일은 읽기만 — 쓰기는 `Store_Make.save`.
"""

from __future__ import annotations

import json
import re
from decimal import Decimal
from typing import Any

import master_machine_coef as mc  # noqa: E402  (`store` 가 scripts 를 길에 둠)

from M01_MasterData import M01_MasterData_Store as store

FILE = "기계.json"
OWN_SOURCE = "자체"
_EQ = re.compile(r"EQ\d{6}")
_cache: dict[str, Any] = {"ver": None, "hourly": {}, "uses": None}


def _rows() -> list[dict]:
    return store.read(FILE)[0].get("줄") or []


def _fresh() -> None:
    """마스터를 다시 읽었으면(저장 뒤) 셈 보관을 비움."""
    ver = id(store.loaded()[1])
    if _cache["ver"] != ver:
        _cache.update(ver=ver, hourly={}, uses=None)


def _plain(v):
    if isinstance(v, Decimal):
        return int(v) if v == v.to_integral_value() else float(v)
    return v


# ── 양식 목록 ────────────────────────────────────────────────────────
def lists() -> dict:
    """연료(유가 전국평균) · 직종(인력) · 전력 단가 · 시간환산 · 원문 · 구분 — 양식 고르기."""
    files = store.loaded()[0]
    oil = files.get("재료_유가전력.json", {}).get("줄") or []
    fuels = [
        {
            "value": str(r["원문번호"]).split(":")[0],
            "text": str(r["원문번호"]).split(":")[0],
            "price": _plain(r.get("값")),
        }
        for r in oil
        if r.get("구분") == "유류" and str(r.get("원문번호", "")).endswith(":전국평균")
    ]
    power = next(
        (_plain(r.get("값")) for r in oil if r.get("원문번호") == "일반용갑I저압:전력량_평균"), 0
    )
    jobs = [
        {
            "value": str(r["원문번호"]),
            "text": f"{r['원문번호']} {r.get('이름') or ''}",
            "price": _plain(r.get("값")),
        }
        for r in files.get("인력.json", {}).get("줄") or []
        if isinstance(r.get("값"), (int, float, Decimal))
    ]
    head = store.read(FILE)[0]
    hour = Decimal(1) / Decimal(8)
    for t in files.get("계수_건설품셈_08장_건설기계.json", {}).get("표") or []:
        if t.get("키") == "CC000336":
            got = next((x for x in t.get("줄") or [] if x.get("구분") == "시간환산"), None)
            if got:
                hour = Decimal(got["분자"]) / Decimal(got["분모"])
    return {
        "fuels": fuels,
        "jobs": jobs,
        "power": power,
        "hour": float(hour),
        "books": list(head.get("원문") or {}),
        "groups": {k: (v or {}).get("원문") for k, v in (head.get("구분") or {}).items()},
    }


# ── 시간당 경비 ──────────────────────────────────────────────────────
def hourly(keys: list[str]) -> dict[str, dict]:
    """키마다 {노무비 · 재료비 · 경비 · 계} 또는 {reason} — 마스터 판이 같으면 한 번만 셈."""
    from M02_MasterTemplete import M02_Machine_Hourly as mh  # M02 가 M01 을 부름 — 늦게

    _fresh()
    out = {}
    for key in keys[:100]:
        if key not in _cache["hourly"]:
            got = mh.hourly(key)
            _cache["hourly"][key] = (
                {k: got[k] for k in ("노무비", "재료비", "경비", "계", "로직")}
                if got.get("ok")
                else {"reason": got.get("reason")}
            )
        out[key] = _cache["hourly"][key]
    return out


# ── 쓰는 곳 ─────────────────────────────────────────────────────────
def _logic_index() -> tuple[dict[str, list[str]], dict[str, list[str]]]:
    """(원문번호 → 로직 「키 이름」 · 기계 키 → 로직) — 기계 고르기 · 식 안 EQ 키."""
    _fresh()
    if _cache["uses"] is None:
        by_code: dict[str, list[str]] = {}
        by_key: dict[str, list[str]] = {}
        for data in store.loaded()[0].values():
            if data.get("그룹") != "로직":
                continue
            for row in data.get("줄") or []:
                label = f"로직 {row.get('키')} {row.get('이름') or ''}".strip()
                for i in row.get("입력") or []:
                    if i.get("이름") == "기계":
                        for c in i.get("고르기") or []:
                            by_code.setdefault(str(c), []).append(label)
                text = json.dumps(row, ensure_ascii=False, default=str)
                for k in set(_EQ.findall(text)):
                    by_key.setdefault(k, []).append(label)
        _cache["uses"] = (by_code, by_key)
    return _cache["uses"]


def uses(key: str) -> list[str]:
    """그 기계를 쓰는 곳 — 로직(기계 고르기 · 식) · M02 단가산출(사본 · 줄에 키)."""
    from M02_MasterTemplete import M02_Template_Layers as layers  # M02 가 M01 을 부름 — 늦게

    row = next((r for r in _rows() if r.get("키") == key), None)
    if row is None:
        raise store.StoreError(404, f"없는 기계 「{key}」")
    by_code, by_key = _logic_index()
    code = str(row.get("원문번호") or "")
    out = list(dict.fromkeys([*(by_code.get(code, []) if code else []), *by_key.get(key, [])]))
    mark = f'"{key}'.encode()
    for where, path in layers.layer_files("sangeun"):
        try:
            raw = path.read_bytes()
        except OSError:
            continue
        if mark in raw:
            try:
                name = json.loads(raw).get("이름") or path.stem
            except ValueError:
                name = path.stem
            out.append(f"M02 단가산출 {where} {name}")
    return out


# ── 비슷한 기계 찾기 · 한 줄 ─────────────────────────────────────────
def find(q: str, limit: int = 30) -> list[dict]:
    words = [w for w in q.lower().split() if w]
    out = []
    for r in _rows():
        hay = " ".join(str(r.get(k) or "") for k in ("이름", "규격", "별칭", "상세구분", "키"))
        hay = hay.lower()
        if all(w in hay for w in words):
            out.append(_brief(r))
            if len(out) >= limit:
                break
    return out


def _brief(r: dict) -> dict:
    keep = ("키", "원문번호", "원문", "구분", "상세구분", "이름", "규격", "값", "별칭", "비고")
    return {k: r[k] for k in keep if k in r}


def row(key: str) -> dict:
    r = next((x for x in _rows() if x.get("키") == key), None)
    if r is None:
        raise store.StoreError(404, f"없는 기계 「{key}」")
    return {"version": store.read(FILE)[1], "row": r}


# ── 저장 전 손질(`Store_Make.save` 가 부름) ──────────────────────────
def _units() -> dict[str, str]:
    out: dict[str, str] = {}
    for r in _rows():
        for k, u in (r.get("단위") or {}).items():
            out.setdefault(k, u)
    return out


def _same_inputs(a: dict, b: dict) -> bool:
    return all(
        float(a[k]) == float(b[k]) if isinstance(b.get(k), (int, float, Decimal)) else False
        for k in mc.INPUTS
    )


def fix(part: dict) -> None:
    """기계 줄 고침 · 더함 — 계수는 서버가 짝으로 다시 셈(품셈 입력이 저장 값과 같으면
    저장 값 그대로) · 단위 · 출처 묶음 = 값 칸과 같은 칸
    (있던 칸 그대로 · 새 칸 = 마스터 단위 · 출처 「자체」)."""
    if part.get("file") != FILE:
        return
    stored = {str(r.get("키")): r for r in _rows()}
    units = _units()
    for ch in part.get("changes") or []:
        row = ch.get("row")
        if (
            ch.get("op") == "delete"
            or not isinstance(row, dict)
            or not isinstance(row.get("값"), dict)
        ):
            continue
        old = stored.get(str(ch.get("key") or "")) if ch.get("op") == "edit" else None
        values = row["값"]
        old_values = (old or {}).get("값") or {}
        if all(isinstance(values.get(k), (int, float)) for k in mc.INPUTS):
            if old and _same_inputs(values, old_values):
                values.update({k: old_values[k] for k in mc.COEFS if k in old_values})
            elif (got := mc.coef_of(values)) is not None:
                values.update(got)
        was_unit = (old or {}).get("단위") or {}
        was_src = (old or {}).get("출처") or {}
        row["단위"] = {k: was_unit.get(k, units.get(k, "")) for k in values}
        row["출처"] = {k: was_src.get(k, OWN_SOURCE) for k in values}