Aislo/M02_MasterTemplete/M02_Doc_Sk_Path.py

"""M02 단가산출(SK) 수식 길 — 읽기 · 미리 셈 · 저장(PLAN 802-4a 4 · 요구 664) — 서버 단독.

M01 수식 문서 길과 같은 꼴: 읽기 {문서, 셈} · 미리 셈 {문서} → {셈} · 저장 {문서} → {문서, 셈}.
몸 칸 — `문서`(이 SK 수식 문서 · 키 = SK 키) · `끊음` {키: 문서}(이 사본에 굳힌 부름 · 고칠 수 있음 ·
    없는 키 = 다시 이음) · `할일` 단가갱신 | 재생성 · `판`(저장 · 그 SK 파일 판).
셈 = 서버 정본(`M02_Doc_Copy.run`) — 화면이 보낸 값 · 사본은 안 받음(문서 · 끊음 글만).
저장 = 그 SK 저장(`store.write` — 층 쓰기 권한 · 판 409 · 옛 칸 · 로직 사본 그대로) ·
    문서 고침 = 따로(이 문서가 정본) · 재생성 = 지금 M01 로 다시 찍어 파생으로(고친 문서 · 끊음 버림) ·
    단가갱신 = 값 줄만 지금 M01(문서 그대로).
가져온 식 SK(엑셀) = 802-4b 몫(400) · 바탕 참조 SK = 값 사본 없음(400).
"""

from __future__ import annotations

import json
from typing import Any

from M02_MasterTemplete import M02_Doc_Copy as dc
from M02_MasterTemplete import M02_Doc_Screen as screen
from M02_MasterTemplete import M02_Doc_Sk as dsk
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Sangeun_Copy as skc
from M02_MasterTemplete import M02_Sangeun_Doc as sd

MAX = 2_000_000  # 문서 글 크기(바이트) 상한
JOBS = (skc.PRICE, skc.REGEN)


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


def _file(key: str) -> dict[str, Any]:
    if not sd.KEY.match(str(key or "")):
        raise _bad("키(SK-######)")
    return store.read(sd.KIND, key)


def _settled(doc: dict, job: Any = None) -> tuple[dict, Any]:
    """옛 셈 길 첫머리(`calc_doc`)와 같게 — (정리한 문서, 사본 색인)."""
    from M01_MasterData import M01_MasterData_Store as m01
    from M02_MasterTemplete import M02_Sangeun_Calc as calc

    raw = {**doc, skc.JOB: job} if job else doc
    whole = skc.over(doc := skc.settle(raw), m01.loaded()[1])
    try:
        return sd.clean(doc, None, whole), whole
    except sd.DocError as e:
        raise calc._fail(e) from e


def current(doc: dict, job: Any = None) -> dict:
    """그 SK 의 지금 수식 사본(할일 적용 · 저장 없음) — 따로 = 저장한 문서 · 파생 = 옛 칸으로 지음."""
    from M02_MasterTemplete import M02_Sangeun_Calc as calc

    if not dsk.own(doc):
        if doc.get("가져온"):
            raise _bad("가져온 식 단가산출 — 수식 문서는 엑셀 불러오기 몫(802-4b)")
        raise _bad("바탕 참조 단가산출 — 수식 사본 없음")
    clean, whole = _settled(doc, job)
    if not dc.has(clean):  # 옮기기 전 · 새 SK — 저장 때 지을 사본을 미리
        seed = dsk.settle(clean, whole, job, doc).get(dsk.KEY)
        if seed is None:
            raise _bad("수식 사본을 지을 수 없음 — 바탕 로직 · 줄 없음")
        clean = {**clean, dsk.KEY: seed}
    box: list = []
    try:
        calc.run(clean, whole, catch=box)
    except sd.DocError as e:
        raise calc._fail(e) from e
    got = dsk.copy_of(clean, box[-1] if box else None)
    return got if got is not None else clean[dsk.KEY]


def answer(copy: dict) -> dict[str, Any]:
    """{문서, 부름, 끊음, 셈, 사본 표지, 연결 기록} — 글 · 수는 JSON 꼴."""
    got = dc.run(copy["문서"], copy)
    head = {k: copy.get(k) for k in ("시점", "단가시점")}
    mark = {**head, "따로": bool(copy.get("따로")), **dc.state(copy)}
    body = {
        "문서": copy["문서"],
        "부름": copy.get("부름") or {},
        "끊음": copy.get("끊음") or {},
        "셈": got,
        "사본": mark,
        "연결": screen.record(copy),  # 화면 캐시(802-4b-2)
    }
    return json.loads(json.dumps(dc.ml._norm(body), ensure_ascii=False, default=str))


def read(key: str) -> dict[str, Any]:
    """읽기 — {문서, 부름, 끊음, 셈, 사본, 판}."""
    got = _file(key)
    return {**answer(current(got["문서"])), "판": got["판"]}


# ── 몸 검사 ───────────────────────────────────────────────────────────
def _doc(raw: Any, key: str, where: str) -> dict:
    if not isinstance(raw, dict):
        raise _bad(f"{where} · 묶음 아님")
    if len(json.dumps(raw, ensure_ascii=False, default=str).encode("utf-8")) > MAX:
        raise _bad(f"{where} · 너무 큼")
    for name in ("변수", "식", "금액"):
        if not isinstance(raw.get(name, []), list):
            raise _bad(f"{where} · 「{name}」 목록 아님")
    return {**raw, "키": key}


def edited(copy: dict, body: dict, key: str) -> dict:
    """화면 고침 얹은 사본 — 문서 · 끊음 글만 받음(값 · 부름은 서버 것)."""
    if body.get("할일") not in (None, "", *JOBS):
        raise _bad("할일은 단가갱신 · 재생성")
    if body.get("할일") == skc.REGEN:
        return copy  # 재생성 = 고친 것 버림(그 사본이 이미 새로 찍힌 것)
    out = {**copy}
    if body.get("문서") is not None:
        out["문서"] = _doc(body["문서"], key, "문서")
    cut = body.get("끊음")
    if cut is not None:
        if not isinstance(cut, dict):
            raise _bad("끊음 · {키: 문서} 묶음")
        keys = set(copy.get("부름") or {}) | set(copy.get("끊음") or {})
        if bad := sorted(k for k in cut if k not in keys):
            raise _bad(f"끊음 · 이 사본이 부르지 않는 단가산출 {bad}")
        out["끊음"] = {k: _doc(v, k, f"끊음 「{k}」") for k, v in cut.items()}
    calls = [
        *dc.calls(out["문서"]),
        *(c for d in out.get("끊음", {}).values() for c in dc.calls(d)),
    ]
    whole = dc.over(out)
    out["부름"] = dc.closure(whole, calls, copy.get("부름"))
    out["따로"] = True
    return out


def preview(key: str, body: dict) -> dict[str, Any]:
    """미리 셈(저장 없음) — {셈, 전, 후, 바뀐자원?, 버림?}."""
    doc = _file(key)["문서"]
    before = current(doc)
    job = body.get("할일") or None
    after = edited(current(doc, job) if job else before, body, key)
    out = answer(after)
    out |= {"전": dc.money(dc.run(before["문서"], before)), "후": dc.money(out["셈"])}
    if job == skc.PRICE:
        out["바뀐자원"] = dc.refresh(before)[1]
    if job == skc.REGEN:
        out["버림"] = {
            "문서고침": bool(before.get("따로")),
            "끊음": sorted(before.get("끊음") or {}),
        }
    return json.loads(json.dumps(dc.ml._norm(out), ensure_ascii=False, default=str))


def save(key: str, body: dict) -> dict[str, Any]:
    """저장(층 쓰기) — 서버가 다시 셈한 사본을 그 SK 에 씀 · {문서, 부름, 끊음, 셈, 사본, 판}."""
    got = _file(key)
    if str(body.get("판") or "") != got["판"]:
        raise _bad({"stale": [key], "판": got["판"]}, 409)
    job = body.get("할일") or None
    raw = {**got["문서"], **({skc.JOB: job} if job else {})}
    token = None
    if job != skc.REGEN and (body.get("문서") is not None or body.get("끊음") is not None):
        mine = edited(current(got["문서"], job), body, key)
        token = dsk.SAVING.set({"키": key, "사본": mine})
    try:
        store.write(sd.KIND, key, got["판"], raw)
    finally:
        if token is not None:
            dsk.SAVING.reset(token)
    return read(key)