Aislo/M02_MasterTemplete/M02_Sangeun_Copy.py

"""M02 단가산출(SK) 로직 사본(PLAN 795-1 · 요구 655 · 656 · 요구서_655 0 · 1 장) — 서버 단독.

사본 = 저장 때 서버가 찍는 M01 닫힌 묶음(`M02_Item_Prices.collect` 다시 씀) — 문서 칸 `사본`:
    {판, 시점, 로직 [M01 줄], 단가판, 단가시점, 단가 [M01 줄], 빠진?}
    로직 = 바탕 로직 · 더한줄 로직 · 가져온 식 이음 · 하위 로직 · 찾기 표 · 그 밖 요소(요율 등)
    단가 = 자원 단가 줄(`RES` 노임 · 자재 · 기계 · 유가전력 · 환율) = 단가 기준 시점 묶음.
셈 = M01 색인 위에 사본 줄을 얹은 `Copy`(키 · 원문번호 찾기만 사본 먼저 · M01 색인 안 고침) —
    `M02_Sangeun_Calc._base_of`(바탕 줄) · `run`(엔진 · 하위 로직 · 표 · 단가)이 `over` 로 부름 ·
    M01 에서 지워진 키 = 사본의 마지막 줄로 셈(표지 `지운키`) ·
    사본 없는 SK(옛 SK · 바탕 참조) = 옛 길.
사본은 셈 · 저장 첫머리 `settle` 이 정함 — 저장본에 사본이 있으면 그것(화면이 보낸 것 버림) ·
    없으면 받은 사본(층 복사 · 새로 · 판 다시 셈) · 그것도 없으면 저장 때 새로 찍음.
할일 칸 `사본할일`(화면 캐시 · [저장] 때 적용):
    단가갱신 = 단가 줄만 지금 M01 값(지워진 키 = 그대로) ·
    재생성 = 지금 M01 로 새로 찍음 — 덮개 · 식덮개 · 변수줄 · 상수 · 표값 · 까닭은 cid 열쇠라
    새 사본에 그대로 얹힘 · 못 얹은 칸 = 짝잃음.
    저장 때 = 사본에 없는 새 줄(더한줄 로직 더함 · 기초 바꿈)만 지금 M01 에서 더함.
길(`M02_Sangeun_Router`): POST …/sangeun/copy/{price|regen} 몸 {문서, 입력값?} →
    {문서(사본할일), 전, 후, 바뀐자원 [{키, 이름, 전, 후, 지워짐?}] · 못얹은 [{cid, 까닭}]}
    (저장 없음).
셈 답 덧칸 `사본` = {시점, 단가시점, M01새판, 지운키, 단가새값} · 같은 내용 지문 몫 = `same_part`.
"""

from __future__ import annotations

import hashlib
import json
from datetime import datetime
from decimal import Decimal
from typing import Any

from M02_MasterTemplete import M02_Item_Prices as prices
from M02_MasterTemplete import M02_Sangeun_Doc as sd
from M02_MasterTemplete import M02_Sangeun_Imported as imp

KEY = "사본"
JOB = "사본할일"
PRICE, REGEN = "단가갱신", "재생성"
RES = ("LB", "MT", "EQ", "MO", "FX")  # 자원 단가 = 단가 기준 시점 묶음
_MASTER = prices.mf.Master


def whole_() -> Any:
    return prices.m01.loaded()[1]


def _fp(rows: Any) -> str:
    raw = json.dumps(prices.to_json(rows), ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]


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


def has(doc: Any) -> bool:
    copy = doc.get(KEY) if isinstance(doc, dict) else None
    return isinstance(copy, dict) and isinstance(copy.get("로직"), list)


# ── 셈 색인 ──────────────────────────────────────────────────────────
class Copy:
    """M01 색인 위 사본 줄 — `get` · `logic` · `value` · `label` = 엔진 `Master` 한 벌
    (사본 줄 먼저) · 그 밖 칸(자재 색인 · 찾기 답)은 속 M01 것을 읽음."""

    get, label, value, logic = _MASTER.get, _MASTER.label, _MASTER.value, _MASTER.logic

    def __init__(self, whole: Any, rows: list[dict], tag: Any = None) -> None:
        self._whole, self._tag, self.twice = whole, tag, whole.twice
        self.index, self.numbers = dict(whole.index), dict(whole.numbers)
        for row in rows:
            key = str(row.get("키"))
            tid = key[:2]
            self._own(self.index, whole.index, tid)[key] = row
            if row.get("원문번호"):
                self._own(self.numbers, whole.numbers, tid)[str(row["원문번호"])] = row

    @staticmethod
    def _own(mine: dict, theirs: dict, tid: str) -> dict:
        if mine.get(tid) is theirs.get(tid):
            mine[tid] = dict(theirs.get(tid) or {})
        return mine[tid]

    def __getattr__(self, name: str) -> Any:
        return getattr(self._whole, name)

    def __copy__(self) -> Copy:  # 수식 고침(796-2b · `master_equation_edit.swap`) — 얕은 사본
        out = object.__new__(Copy)
        out.__dict__.update(self.__dict__)
        return out


def _rows(copy: dict) -> list[dict]:
    raw = json.dumps(prices.to_json([*copy["로직"], *(copy.get("단가") or [])]), ensure_ascii=False)
    return json.loads(raw, parse_float=Decimal, parse_int=Decimal)


_CACHE: dict[tuple, Copy] = {}
_SEEN: dict[tuple, dict] = {}
_LIVE: list[Any] = [None]  # 지금 M01 색인 — 바뀌면(저장 · 다시 읽음) 메모 비움(옛 색인 안 붙잡음)


def _forget(whole: Any) -> None:
    while isinstance(whole, Copy):
        whole = whole._whole
    if whole is not _LIVE[0]:
        _CACHE.clear()
        _SEEN.clear()
        _LIVE[0] = whole


def over(doc: Any, whole: Any) -> Any:
    """사본 있는 문서 = 사본 색인 · 없으면 `whole` 그대로(옛 길)."""
    if not has(doc):
        return whole
    copy = doc[KEY]
    tag = (copy.get("판"), copy.get("단가판"))
    if isinstance(whole, Copy) and whole._tag == tag:
        return whole
    _forget(whole)
    got = _CACHE.get((id(whole), *tag))
    if got is None or got._whole is not whole:
        if len(_CACHE) > 32:  # ponytail: 통째 비움 · 쓰는 SK 가 많아 느려지면 LRU 로
            _CACHE.clear()
        got = _CACHE[(id(whole), *tag)] = Copy(whole, _rows(copy), tag)
    return got


# ── 찍기 · 단가 갱신 ──────────────────────────────────────────────────
def _split(rows: list[dict]) -> dict[str, Any]:
    logic = [r for r in rows if str(r.get("키"))[:2] not in RES]
    cost = [r for r in rows if str(r.get("키"))[:2] in RES]
    when = _now()
    return {
        "판": _fp(logic),
        "시점": when,
        "로직": logic,
        "단가판": _fp(cost),
        "단가시점": when,
        "단가": cost,
    }


def stamp(doc: dict) -> dict | None:
    """지금 M01 로 새 사본 — 바탕 로직 · 더한줄 로직 · 고정 줄 이음 · 가져온 식 줄이 닿는
    닫힌 묶음 · 바탕 참조 · 닿을 것 없음 = None."""
    base = doc.get("바탕") or {}
    if "참조" in base:
        return None
    keys, links = reach(doc.get("더한줄"))
    keys = [str(base["로직"]), *keys] if base.get("로직") else keys
    extra = [row] if (row := imp.row_of(doc)) else []
    return snap(keys, links, extra)


def reach(adds: Any) -> tuple[list[str], list[str]]:
    """더한줄이 닿는 (로직 키, 요소 키) — 로직 줄 · 인력 · 재료 · 기계 요소 · 고정 줄 이음."""
    adds = [x for x in adds or [] if isinstance(x, dict)]
    keys = [str(x["로직"]) for x in adds if x.get("종류") == "로직" and x.get("로직")]
    links = [str(x["요소"]) for x in adds if x.get("종류") in sd.ADD_PART and x.get("요소")]
    links += [str(v) for x in adds for v in ((x.get("고정") or {}).get("이음") or {}).values() if v]
    return keys, links


def snap(keys: list[str], links: list[str], extra: list[dict] | None = None) -> dict | None:
    """키 · 요소 · 마스터 밖 줄이 닿는 M01 닫힌 묶음 → 사본(닿을 것 없음 = None)."""
    if not (keys or links or extra):
        return None
    got = prices.collect(keys, links, extra)
    lost = got.pop("_빠진", None)
    keys = [str(r.get("키")) for data in got.values() for r in data[prices.m01.items_key(data)]]
    whole = whole_()  # 줄 = 셈 색인 것(파일 묶음과 같은 판)
    rows = [whole.index[k[:2]][k] for k in keys if k in whole.index.get(k[:2], {})]
    out = _split(prices.to_json(rows))
    return out | ({"빠진": lost} if lost else {})


def refresh(copy: dict, whole: Any) -> tuple[dict, list[dict]]:
    """[단가 갱신] — 단가 줄만 지금 M01 값 · (새 사본, 바뀐자원) · 지워진 키 = 마지막 값 그대로."""
    rows, changed = [], []
    for row in copy.get("단가") or []:
        key = str(row.get("키"))
        live = whole.index.get(key[:2], {}).get(key)
        if live is None:
            rows.append(row)
            changed.append({"키": key, "이름": row.get("이름"), "지워짐": True})
            continue
        new = prices.to_json(live)
        if _fp(new) != _fp(row):
            changed.append(
                {"키": key, "이름": new.get("이름"), "전": row.get("값"), "후": new.get("값")}
            )
        rows.append(new)
    return {**copy, "단가": rows, "단가판": _fp(rows), "단가시점": _now()}, changed


def grow(copy: dict, fresh: dict | None) -> dict:
    """저장 때 — 사본에 없는 키 줄만 새 사본(`fresh` = 지금 M01)에서 더함(있는 줄 = 그대로)."""
    have = {str(r.get("키")) for r in [*copy["로직"], *(copy.get("단가") or [])]}
    new = [
        r
        for r in [*(fresh or {}).get("로직", []), *(fresh or {}).get("단가", [])]
        if str(r.get("키")) not in have
    ]
    if not new:
        return copy
    logic = [*copy["로직"], *(r for r in new if str(r.get("키"))[:2] not in RES)]
    cost = [*(copy.get("단가") or []), *(r for r in new if str(r.get("키"))[:2] in RES)]
    return {**copy, "로직": logic, "판": _fp(logic), "단가": cost, "단가판": _fp(cost)}


def rejoin(copy: dict, adds: list) -> dict:
    """[다시 잇기](796-2e) — 표지 낀 더한 기초가 닿는 닫힌 묶음을 지금 M01 줄로 갈음(단가 줄은 더하기만)."""
    keys = [
        str(x["로직"]) for x in adds if isinstance(x, dict) and x.get("다시잇기") and x.get("로직")
    ]
    fresh = snap(keys, [])
    if not fresh:
        return copy
    new = {str(r.get("키")): r for r in fresh["로직"]}
    logic = [new.pop(str(r.get("키")), r) for r in copy["로직"]] + list(new.values())
    return grow({**copy, "로직": logic, "판": _fp(logic)}, fresh)


def taken(raw: Any) -> dict | None:
    """받은 사본(층 복사 · 새로) — 줄 꼴만 보고 판은 서버가 다시 셈."""
    if not has({KEY: raw}):
        return None
    logic = [r for r in raw["로직"] if isinstance(r, dict) and str(r.get("키"))[:2] not in RES]
    cost = [r for r in raw.get("단가") or [] if isinstance(r, dict) and str(r.get("키"))[:2] in RES]
    keep = {k: str(raw.get(k) or "") for k in ("시점", "단가시점")}
    return {**keep, "로직": logic, "판": _fp(logic), "단가": cost, "단가판": _fp(cost)}


def _stored(key: Any) -> dict | None:
    from M02_MasterTemplete import M02_MasterTemplete_Store as store

    if not sd.KEY.match(str(key or "")):
        return None
    try:
        got = json.loads((store.root() / sd.KIND / f"{key}.json").read_bytes())
    except (OSError, ValueError):
        return None
    return got[KEY] if has(got) else None


def settle(doc: Any, save: bool = False) -> Any:
    """셈 · 저장 첫머리 — 서버가 정한 사본으로 갈음(할일 적용 · 저장 때 새 줄 더함 · 할일 칸 뗌)."""
    if not isinstance(doc, dict):
        return doc
    job = doc.get(JOB)  # 모르는 할일 = 없음
    out = {k: v for k, v in doc.items() if k not in (KEY, JOB)}
    if "참조" in (doc.get("바탕") or {}):
        return out
    copy = _stored(doc.get("키")) or taken(doc.get(KEY))
    if job == REGEN or (copy is None and save):
        copy = stamp(out)
    elif job == PRICE and copy:
        copy = refresh(copy, whole_())[0]
    if copy and any(isinstance(x, dict) and x.get("다시잇기") for x in out.get("더한줄") or []):
        copy = rejoin(copy, out["더한줄"])
    if save and copy:
        copy = grow(copy, stamp(out))
    out = _formula_job(out, job)
    return {**out, KEY: copy} if copy else out


def _formula_job(out: dict, job: Any) -> dict:
    """수식 사본(802-4a · `M02_Doc_Sk`) 할일 — 재생성 = 파생으로(부름 새로 · 고친 문서 버림) ·
    단가갱신 = 따로 문서의 값 줄도 지금 M01."""
    from M02_MasterTemplete import M02_Doc_Copy as dc

    got = out.get(dc.KEY) if dc.has(out) else None
    if got is None or job not in (PRICE, REGEN):
        return out
    if job == REGEN:
        got = {k: v for k, v in got.items() if k not in ("문서", "끊음", "따로")} | {"부름": {}}
    elif got.get("따로"):
        got = dc.refresh(got)[0]
    return {**out, dc.KEY: got}


# ── 표지 ──────────────────────────────────────────────────────────────
def state(copy: dict, whole: Any) -> dict[str, Any]:
    """{M01새판, 지운키, 단가새값} — 사본 줄과 지금 M01 줄 대조(같은 M01 · 같은 판 = 메모)."""
    _forget(whole)
    tag = (id(whole), copy.get("판"), copy.get("단가판"))
    if tag in _SEEN and _SEEN[tag]["whole"] is whole:
        return _SEEN[tag]["답"]

    def live(row: dict) -> Any:
        key = str(row.get("키"))
        return whole.index.get(key[:2], {}).get(key)

    gone = [str(r.get("키")) for r in [*copy["로직"], *(copy.get("단가") or [])] if live(r) is None]
    logic = any(live(r) is None or _fp(live(r)) != _fp(r) for r in copy["로직"])
    cost = any(live(r) is not None and _fp(live(r)) != _fp(r) for r in copy.get("단가") or [])
    got = {"M01새판": logic, "지운키": gone, "단가새값": cost}
    _SEEN[tag] = {"whole": whole, "답": got}
    return got


def mark(ans: dict, doc: dict) -> dict:
    """셈 답 덧칸 `사본` — 사본 없는 문서 = 칸 없음."""
    if not has(doc):
        return ans
    copy = doc[KEY]
    head = {"시점": copy.get("시점"), "단가시점": copy.get("단가시점")}
    return {**ans, KEY: {**head, **state(copy, whole_())}}


def same_part(doc: Any) -> str | None:
    """같은 내용 지문 몫 — 지금 M01 과 같은 사본 = None(새로 찍을 문서와 같음) · 다르면 판."""
    if not has(doc):
        return None
    copy = doc[KEY]
    seen = state(copy, whole_())
    fresh = not (seen["M01새판"] or seen["지운키"] or seen["단가새값"])
    return None if fresh else f"{copy.get('판')}:{copy.get('단가판')}"


# ── [단가 갱신] · [재생성] 미리 셈(저장 없음) ──────────────────────────
def _money(ans: dict) -> dict | None:
    base = ans.get("기준")
    return {k: base.get(k) for k in ("노무비", "재료비", "경비", "계")} if base else None


def _lost(ans: dict) -> dict[str, dict]:
    rows = [*(ans.get("짝잃음") or []), *((ans.get("시트") or {}).get("짝잃음") or [])]
    return {str(x.get("cid")): x for x in rows if isinstance(x, dict)}


def preview(doc: Any, job: str, given: dict | None = None) -> dict[str, Any]:
    """{문서(할일 얹음), 전, 후, 바뀐자원 · 못얹은} — 재생성은 단가갱신을 품음."""
    from M02_MasterTemplete import M02_Sangeun_Calc as calc

    if not isinstance(doc, dict):
        raise sd.DocError("단가산출 문서가 묶음 아님")
    old = doc.get(JOB)
    new = REGEN if REGEN in (old, job) else job
    after_doc = {**doc, JOB: new}
    before = calc.calc_doc(doc, given, False)
    after = calc.calc_doc(after_doc, given, False)
    out: dict[str, Any] = {"문서": after_doc, "전": _money(before), "후": _money(after)}
    if cut := [
        str(x.get("이름") or x.get("로직")) for x in doc.get("더한줄") or [] if x.get("끊음")
    ]:
        out["끊은기초"] = cut  # 줄은 그대로 · 자원 단가만(요구 656-1)
    if job == PRICE:
        copy = _stored(doc.get("키")) or taken(doc.get(KEY))
        out["바뀐자원"] = refresh(copy, whole_())[1] if copy else []
    else:
        was = _lost(before)
        out["못얹은"] = [
            {"cid": c, "까닭": x.get("까닭") or "새 사본에 그 줄 없음"}
            for c, x in _lost(after).items()
            if c not in was
        ]
    return out