"""M02 내 기계(PLAN 821-1 · 요구 795 · 요구서_796 3 · 7 장) — 개인 층 기계 줄 목록 · 서버 단독.

저장 = 층 뿌리 `내기계.json` {줄: [줄], 다음: 수} — 줄 = M01 기계 줄 꼴
    {키 `EQP{사용자}-####`, 원문번호 「개인 {키}」, 원문 · 구분 · 출처 = 개인, 상세구분 · 이름, 규격,
     단위, 값, 비고, 방식 바로|품셈|직접, 시간당?(직접 {노무비, 재료비, 경비}), 넣은날, 고친날, 본뜬?}
    값 칸 이름 = M01 기계 줄과 같음 · 바로 = 손료계수 한 칸 · 품셈 = 비율 다섯 → 계수 넷(저장 때 서버가
    계수 짝 `master_machine_coef` 로 다시 셈) · 직접 = 시간당 값만.
셈 색인 `whole()` = 지금 층(`store.root()`) 줄을 M01 위에 얹은 `skc.Copy`(키 앞 두 글자 EQ ·
    원문번호로도 찾음) — 바로 = 상각비계수 ← 손료계수 · 정비 · 관리 0(식은 M01 기계와 같은 GC 식) ·
    내 기계 없는 층 = M01 그대로. 단가산출 사본 · [단가 갱신] · 「단가새값」(`M02_Sangeun_Copy`) ·
    기계 찾기(`M02_Machine_Find`) · 시간당(`M02_Machine_Hourly`) · 사본 모으기(`M02_Item_Prices.collect`)가
    이 색인을 읽음 — 층을 모르는 길(`/api/m02/logic-lib/machine-hourly` 등)은 M01 만.
식 고름(채운 칸) = 연료 · 조종원 둘 다 GC000268 · 연료만 GC000272 · 조종원만 GC000276 · 없음 GC000273.
전력 칸은 담기만(M01 기계경비 식에 전력 줄 없음).
권한(본인 개인 층 · 맡은 회사 관리자)은 길(`M02_MyMachine_Router`) 몫.
"""

from __future__ import annotations

import json
import re
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
from M01_MasterData import M01_MasterData_Store as m01
from M02_MasterTemplete import M02_MasterTemplete_Store as store

FILE = "내기계.json"
PREFIX, NO = "EQP", "개인 "
FIND = re.compile(rf"{NO}({PREFIX}\d+-\d+)")  # 문서 안 기계 입력 「개인 EQP…」
EASY, BOOK, DIRECT = "바로", "품셈", "직접"
COSTS = ("노무비", "재료비", "경비")
RATIOS = ("내용시간", "연간표준가동시간", "상각비율", "정비비율", "연간관리비율")
COEFS = ("상각비계수", "정비비계수", "관리비계수")
NUMS = ("가격", *RATIOS, *COEFS, "손료계수", "주연료", "잡재료율", "조종원", "전력")
UNITS = {
    "가격": "천원", "내용시간": "시간", "연간표준가동시간": "시간", "상각비율": "", "정비비율": "",
    "연간관리비율": "", **{k: "10^-7/시간" for k in (*COEFS, "손료계수")}, "주연료": "ℓ/hr",
    "주연료종류": "", "잡재료율": "주연료의 %", "조종원": "인/일", "조종원직종": "", "전력": "kWh/hr",
}  # fmt: skip
LOGIC = {(True, True): "GC000268", (True, False): "GC000272",
         (False, True): "GC000276", (False, False): "GC000273"}  # fmt: skip


def mine(key: Any) -> bool:
    return str(key or "").startswith(PREFIX)


def _now() -> str:
    return datetime.now().isoformat(timespec="seconds")


def _bad(text: str, status: int = 400) -> store.StoreError:
    return store.StoreError(status, text)


# ── 저장 ──────────────────────────────────────────────────────────────
def _path(folder: Path | None = None) -> Path:
    return Path(folder or store.root()) / FILE


def _read(folder: Path | None = None) -> dict[str, Any]:
    try:
        got = json.loads(_path(folder).read_bytes())
    except (OSError, ValueError):
        got = {}
    rows = [r for r in got.get("줄") or [] if isinstance(r, dict) and mine(r.get("키"))]
    return {"줄": rows, "다음": int(got.get("다음") or 1)}


def rows(folder: Path | None = None) -> list[dict]:
    """지금 층 내 기계 줄(저장 꼴)."""
    return _read(folder)["줄"]


# ── 셈 색인 ──────────────────────────────────────────────────────────
def engine(row: dict) -> dict:
    """셈 꼴 — 바로 = 상각비계수 ← 손료계수 · 정비 · 관리 0 · 빈 연료 · 조종원 칸 = 0 · 수 = Decimal."""
    vals = dict(row.get("값") or {})
    if row.get("방식") == EASY:
        vals |= {"상각비계수": vals.get("손료계수") or 0, "정비비계수": 0, "관리비계수": 0}
    if row.get("방식") != DIRECT:
        vals = {"주연료": 0, "잡재료율": 0, "조종원": 0, "주연료종류": "", "조종원직종": ""} | vals
    raw = json.dumps({**row, "값": vals}, ensure_ascii=False)
    return json.loads(raw, parse_float=Decimal, parse_int=Decimal)


_MEMO: dict[str, tuple] = {}


def whole(base: Any = None) -> Any:
    """M01 색인 + 지금 층 내 기계 줄 — 파일 · M01 이 그대로면 같은 객체(셈 메모가 붙음)."""
    from M02_MasterTemplete import M02_Sangeun_Copy as skc

    base = m01.loaded()[1] if base is None else base
    path = _path()
    try:
        st = path.stat()
    except OSError:
        return base
    sig = (st.st_mtime_ns, st.st_size)
    got = _MEMO.get(str(path))
    if got and got[0] == sig and got[1] is base:
        return got[2]
    own = [engine(r) for r in rows(path.parent)]
    out = skc.Copy(base, own, ("내기계", sig)) if own else base
    if len(_MEMO) > 16:  # ponytail: 통째 비움 · 층이 많아지면 LRU 로
        _MEMO.clear()
    _MEMO[str(path)] = (sig, base, out)
    return out


def added(index: Any) -> list[dict]:
    """`whole()` 이 얹은 내 기계 셈 줄(사본 모으기 `collect` 가 마스터 밖 줄로 담음)."""
    return [r for k, r in (getattr(index, "index", {}).get("EQ") or {}).items() if mine(k)]


def keys_in(doc: Any) -> list[str]:
    """문서 안 기계 입력 「개인 EQP…」 → 내 기계 키(줄 쌓기 `로직(GC…, 기계='개인 …')` · 사본 찍기)."""
    return sorted(set(FIND.findall(json.dumps(doc, ensure_ascii=False, default=str))))


def logic_key(row: dict) -> str:
    vals = row.get("값") or {}
    return LOGIC[(float(vals.get("주연료") or 0) > 0, float(vals.get("조종원") or 0) > 0)]


def logic_for(code: str) -> dict | None:
    """내 기계 원문번호 → 시간당 경비 식(채운 칸으로 고름) · 내 기계 아님 = None."""
    index = whole()
    row = (index.numbers.get("EQ") or {}).get(code)
    if row is None or not mine(row.get("키")) or row.get("방식") == DIRECT:
        return None
    return index.logic(logic_key(row))


def hourly(row: dict, index: Any) -> dict[str, Any]:
    """`M02_Machine_Hourly.hourly` 꼴 — 직접 = 그 값 · 아니면 고른 GC 식(입력 기본값 · 기계 = 원문번호)."""
    head = {"이름": row.get("이름"), "규격": row.get("규격"), "원문번호": row.get("원문번호")}
    if row.get("방식") == DIRECT:
        sums = {k: float((row.get("시간당") or {}).get(k) or 0) for k in COSTS}
        return {"ok": True, **head, "로직": None, "직접": True, **sums, "계": sum(sums.values())}
    import master_cut as mcut

    mf = m01.cm.mf
    logic = index.logic(logic_key(row))
    given = {i["이름"]: (i.get("고르기") or [""])[0] for i in logic.get("입력") or []}
    given["기계"] = str(row.get("원문번호"))
    try:
        got = mcut.engine(
            index, logic["키"], mf.run(index, logic["키"], mf.fill_empty(logic, given))
        )
    except (mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e:
        return {"ok": False, "reason": str(e) or type(e).__name__}
    sums = {k: float(got[k]) for k in (*COSTS, "계")}
    return {"ok": True, **head, "로직": logic["키"], **sums}


# ── 검사 ──────────────────────────────────────────────────────────────
def _num(value: Any, name: str) -> float | None:
    if value in (None, ""):
        return None
    try:
        got = float(value)
    except (TypeError, ValueError):
        raise _bad(f"「{name}」 은 수") from None
    if got < 0 or got != got:
        raise _bad(f"「{name}」 은 0 이상")
    return int(got) if got == int(got) else got


def _coef(vals: dict) -> dict:
    """품셈 방식 — 비율 다섯 → 계수 넷(계수 짝 `master_machine_coef` · 821-2)."""
    import master_machine_coef as mmc

    got = mmc.coef_of(vals)
    if got is None:
        raise _bad("품셈 방식 — 내용시간 · 연간표준가동시간은 0 보다 큼")
    return got


def clean(raw: Any) -> dict[str, Any]:
    """받은 줄 → 저장 꼴 몸(키 · 날 빼고) — 틀리면 400."""
    if not isinstance(raw, dict):
        raise _bad("내 기계 줄이 묶음 아님")
    name, spec = str(raw.get("이름") or "").strip(), str(raw.get("규격") or "").strip()
    if not name:
        raise _bad("이름 없음")
    how = raw.get("방식") or EASY
    if how not in (EASY, BOOK, DIRECT):
        raise _bad("방식은 바로 · 품셈 · 직접")
    given = raw.get("값") if isinstance(raw.get("값"), dict) else {}
    out: dict[str, Any] = {"이름": name, "규격": spec, "방식": how}
    if how == DIRECT:
        hour = raw.get("시간당") if isinstance(raw.get("시간당"), dict) else {}
        out["시간당"] = {k: _num(hour.get(k), k) or 0 for k in COSTS}
        vals: dict[str, Any] = {}
    else:
        vals = {k: got for k in NUMS if (got := _num(given.get(k), k)) is not None}
        need = ("가격", "손료계수") if how == EASY else ("가격", *RATIOS)
        if lost := [k for k in need if not vals.get(k)]:
            raise _bad(f"빈 칸 — {' · '.join(lost)}")
        drop = (*RATIOS, *COEFS) if how == EASY else ()
        vals = {k: v for k, v in vals.items() if k not in drop}
        if how == BOOK:
            vals |= _coef(vals)
        vals |= _kinds(given, vals)
    out["값"] = vals
    out["단위"] = {k: UNITS[k] for k in vals if k in UNITS}
    out["출처"] = {k: "개인" for k in vals}
    if note := str(raw.get("비고") or "").strip():
        out["비고"] = note
    return out


def _kinds(given: dict, vals: dict) -> dict:
    """주연료종류 = M01 유가 이름 · 조종원직종 = M01 인력 원문번호(쓸 때만 · 고르기 목록 밖 = 400)."""
    index, out = m01.loaded()[1], {}
    fuel, job = str(given.get("주연료종류") or ""), str(given.get("조종원직종") or "")
    if vals.get("주연료"):
        if f"{fuel}:전국평균" not in (index.numbers.get("MO") or {}):
            raise _bad(f"주연료종류 「{fuel}」 — 유가 목록에서 고름")
        out["주연료종류"] = fuel
    if vals.get("조종원"):
        if job not in (index.numbers.get("LB") or {}):
            raise _bad(f"조종원직종 「{job}」 — 인력 목록에서 고름")
        out["조종원직종"] = job
    return out


# ── 만들기 · 고치기 · 지우기 · 읽기 ─────────────────────────────────────
def _seed(key: str) -> dict:
    """「비슷한 기계에서 시작」 — M01 · 내 기계 줄 값 복사(품셈 비율 있으면 품셈 · 없으면 바로)."""
    try:
        src = whole().get(key)
    except m01.cm.mf.FormulaError:
        raise _bad(f"없는 기계 「{key}」", 404) from None
    if not str(src.get("키")).startswith("EQ"):
        raise _bad(f"「{key}」 는 기계 아님")
    if mine(key):
        got = next(r for r in rows() if r["키"] == key)
        return {k: got[k] for k in ("이름", "규격", "방식", "값", "시간당", "비고") if k in got}
    vals = json.loads(json.dumps(src.get("값") or {}, default=float))
    how = BOOK if all(vals.get(k) for k in RATIOS) else EASY
    return {"이름": src.get("이름"), "규격": src.get("규격"), "방식": how, "값": vals}


def _user() -> str:
    """개인 층(`storage/{회사}/{사용자}/templates`) 사용자 번호 — 그 밖 층 = 404."""
    from M02_MasterTemplete import M02_Template_Layers as layers

    here = store.root().resolve()
    user, company = here.parent.name, here.parent.parent.name
    if not (user.isdigit() and company.isdigit()) or here != layers.personal_dir(company, user):
        raise _bad("내 기계는 개인 층만", 404)
    return user


def create(raw: Any = None, base: str | None = None) -> dict[str, Any]:
    _user()
    seed = _seed(base) if base else {}
    raw = raw if isinstance(raw, dict) else {}
    body = clean({**seed, **raw, "값": {**seed.get("값", {}), **(raw.get("값") or {})}})
    data = _read()
    key = f"{PREFIX}{_user()}-{data['다음']:04d}"
    when = _now()
    row = {"키": key, "원문번호": f"{NO}{key}", "원문": "개인", "구분": "개인", "상세구분": body["이름"],
           **body, "넣은날": when, "고친날": when, **({"본뜬": base} if base else {})}  # fmt: skip
    atomic_write_json(_path(), {"줄": [*data["줄"], row], "다음": data["다음"] + 1})
    return answer(row)


def update(key: str, raw: Any) -> dict[str, Any]:
    _user()
    data = _read()
    old = next((r for r in data["줄"] if r["키"] == key), None)
    if old is None:
        raise _bad(f"없는 내 기계 「{key}」", 404)
    body = clean(raw)
    keep = {k: old[k] for k in ("키", "원문번호", "원문", "구분", "넣은날", "본뜬") if k in old}
    row = {**keep, "상세구분": body["이름"], **body, "고친날": _now()}
    atomic_write_json(_path(), {**data, "줄": [row if r["키"] == key else r for r in data["줄"]]})
    return answer(row)


def delete(key: str) -> dict[str, Any]:
    """지움 — 이미 만든 단가산출은 사본의 마지막 값 그대로(「지운키」)."""
    _user()
    data = _read()
    if not any(r["키"] == key for r in data["줄"]):
        raise _bad(f"없는 내 기계 「{key}」", 404)
    used = uses().get(key, [])
    atomic_write_json(_path(), {**data, "줄": [r for r in data["줄"] if r["키"] != key]})
    return {"지움": key, "쓰는곳": used}


def uses() -> dict[str, list[dict]]:
    """{내 기계 키: [{키, 이름}]} — 이 층 단가산출 가운데 사본 단가 줄에 그 키를 가진 것."""
    from M02_MasterTemplete import M02_Sangeun_Copy as skc
    from M02_MasterTemplete import M02_Sangeun_Doc as sd

    out: dict[str, list[dict]] = {}
    for path in sorted((store.root() / sd.KIND).glob("SK-*.json")):
        raw = path.read_bytes()
        if PREFIX.encode() not in raw:
            continue
        try:
            doc = json.loads(raw)
        except ValueError:
            continue
        for row in (doc.get(skc.KEY) or {}).get("단가") or []:
            if mine(row.get("키")):
                out.setdefault(str(row["키"]), []).append(
                    {"키": doc.get("키"), "이름": doc.get("이름")}
                )
    return out


def same(name: Any, spec: Any, skip: str = "") -> list[dict]:
    """같은 이름 · 규격 알림 — M01 기계 · 이 층 내 기계(표기 정규화 `M02_Machine_Find.norm`)."""
    from M02_MasterTemplete.M02_Machine_Find import norm

    want = (norm(name), norm(spec))
    return [
        {
            "키": r["키"],
            "이름": r.get("이름"),
            "규격": r.get("규격"),
            "출처": "개인" if mine(r["키"]) else "M01",
        }  # fmt: skip
        for r in (whole().index.get("EQ") or {}).values()
        if r["키"] != skip and (norm(r.get("이름")), norm(r.get("규격"))) == want
    ]


def answer(row: dict, used: list | None = None) -> dict[str, Any]:
    out = {
        "줄": row,
        "시간당": hourly(engine(row), whole()),
        "같은이름": same(row["이름"], row["규격"], row["키"]),
    }
    return out | ({"쓰는곳": used} if used is not None else {})


def one(key: str) -> dict[str, Any]:
    got = next((r for r in rows() if r["키"] == key), None)
    if got is None:
        raise _bad(f"없는 내 기계 「{key}」", 404)
    return answer(got, uses().get(key, []))


def listing() -> dict[str, Any]:
    """목록 — 줄마다 시간당 · 쓰는곳수(넣은 차례)."""
    index, used = whole(), uses()
    return {
        "줄": [
            {**r, "시간당": hourly(engine(r), index), "쓰는곳수": len(used.get(r["키"], []))}
            for r in rows()
        ]
    }