Aislo/M02_MasterTemplete/M02_Hopyo_Judge.py

"""M02 엑셀 불러오기 판정 · 검산(PLAN 785-5 · 요구 642 · 설계 642-A 2-6 ⑤ · 2-9) — 서버 단독 · 읽기만.

판정 = 1차 줄 판(`M02_Hopyo_Row.rows`) + 단산 줄 2차 블록 풀이(`Clause.basis` 후보 → `Solve.parts` →
`Solve.block`) · 목록(`listing`) · 호표 하나(`hopyo`) · 블록 하나(`block`).
블록 풀이 = 파생 저장본(HP 폴더 `_derived/hopyo_judge_{키}` · 입력 지문 = HP + M01 파일 내용 + 풀이 판 ·
지문 다를 때만 다시 셈 · GET 은 안 지움 · 올림 뒤 한 번 뒤에서 `ensure` · 셈 중 = `셈중` · 판정 None) ·
블록 판정 읽기 = `judged`(Marks 칠) · [끝내기] = `forget`.
줄 판 줄에 엑셀 칸(이름 · 규격 · 단위 · 단가) · 블록에 엑셀 금액 · 절에 몫(785-6 화면이 읽음).
검산(`check`) = 고른 것마다 — 기초 = 입력값(`inputs` · 엔진이 쓴 칸 모두 · 지역 · 자재지역 · 조종원직종
비움) · 입력필요 · 엑셀 단위 줄과 품 대조(`Solve.pair`) · 금액(엔진 한 번 · 입력 필요면 셈 안 함) /
그 밖 = 「확인」 · 수량 = 엑셀 × 몫 × 배(`Row.quantity` · 사용자 배 우선).
저장 없음 · 지어 넣은 값 없음(입력 필요 칸은 비워 둠 · 안 쓰는 칸 = 기초 기본값).
"""

from __future__ import annotations

import shutil
import threading
from collections import Counter
from decimal import Decimal
from pathlib import Path
from typing import Any

from common_util import common_util_derived_store as derived
from M02_MasterTemplete import M02_Hopyo_Bind as hb
from M02_MasterTemplete import M02_Hopyo_Build as hbuild
from M02_MasterTemplete import M02_Hopyo_Clause as clause
from M02_MasterTemplete import M02_Hopyo_Ledger as ledger
from M02_MasterTemplete import M02_Hopyo_Row as hrow
from M02_MasterTemplete import M02_Hopyo_Sangeun as sangeun
from M02_MasterTemplete import M02_Hopyo_Solve as solve
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store
from M02_MasterTemplete import M02_Labor_Doc as labor
from M02_MasterTemplete import M02_Labor_Ref as refs
from M02_MasterTemplete import M02_MasterTemplete_Store as store

mf, ma = hb.mf, hb.ma
EXCEL = ("이름", "규격", "단위", "단가")
CHECKED = "확인"
KIND = "hopyo_judge_"  # 파생 저장본 종류 = 이것 + HP 키(HP 폴더 `_derived/`)
VERSION = "785-5.1"  # 풀이 꼴이 바뀌면 올림
_LOCK, _RUNNING = threading.Lock(), set()


def master() -> mf.Master:
    return hb.m01.cm.master()


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


def _blocks(book: dict[str, Any]) -> dict[int, dict[str, Any]]:
    return {int(b["번"]): b for b in book.get("단산") or [] if b.get("번") is not None}


def _unit(book: dict[str, Any], no: int) -> dict[str, Any]:
    got = next((h for h in book.get("호표") or [] if int(h["번"]) == int(no)), None)
    if got is None:
        raise _bad(f"없는 호표 「제{no}호표」", 404)
    return got


def _made(folder: Path, key: str, to: Path) -> tuple[dict[str, str], dict[str, str]]:
    """(만든 SC {본뜬 키: SC}, 만든 SK {본뜬 키: SK}) — 만들 층 · 이 파일 층 것만."""
    source = refs.where(folder)
    return hbuild.confirmed(to, source), sangeun.made(to, key, source)


# ── 기초 입력값 ───────────────────────────────────────────────────────
def inputs(m: mf.Master, code: str, given: dict[str, Any] | None) -> tuple[dict, list[str]]:
    """(저장 입력값, 입력필요) — 고른 값(그 기초 칸 · 빈 값 뺌) + 안 쓰는 칸 = 기초 기본값 ·
    쓰는데 빈 칸 = 입력필요(비워 둠) · 지역 · 자재지역 · 조종원직종 = 비움(프로젝트가 정함)."""
    row = m.logic(code)
    names = [s["이름"] for s in hb.specs_of(row)]
    have = {
        k: v
        for k, v in (given or {}).items()
        if k in names and k not in hb.SKIP and v not in (None, "")
    }
    need = [n for n in names if n in hb._needed(m, code, row, have) and n not in hb.SKIP]
    base = ma.defaults(m, row)
    idle = [n for n in names if n not in have and n not in need and n not in hb.SKIP]
    full = {n: hb.plain(base[n]) for n in idle if base.get(n) is not None}
    return {**full, **{k: hb.plain(v) for k, v in have.items()}}, need


def money(m: mf.Master, code: str, given: dict[str, Any]) -> dict[str, Any]:
    """기초 한 줄 셈(일위대가 로직 줄과 같은 길 `Labor_Doc.run_line` · 끝수 · 비운 지역 칸 = 자동값 ·
    참고 셈만) → {노무비, 재료비, 경비, 계} · 못 셈 = {계 None, 까닭}."""
    got = labor.run_line(m, code, given)
    if got.get("결과") is not None:
        return {"결과": hb.plain(got["결과"])}
    if got.get("기준") is None:
        return {"계": None, "까닭": got.get("까닭")}
    return {k: hb.plain(v) for k, v in got["기준"].items()}


# ── 판정 ─────────────────────────────────────────────────────────────
def _row(line: dict[str, Any], one: dict[str, Any]) -> dict[str, Any]:
    return {**one, **{k: line.get(k) for k in EXCEL}}


def _solve(m: mf.Master, blk: dict, line: dict | None, same: list[dict]) -> dict[str, Any]:
    """블록 풀이 — 조항 색인 후보 → 절 → 검산 · 판정 + 근거 · 엑셀 금액 · 절 몫."""
    basis = clause.basis(blk)
    parts = solve.parts(blk, basis["후보"])
    got = solve.block(m, blk, parts, {"줄": line, "같은줄": same})
    for sec, part in zip(got["절"], parts, strict=True):
        share = part.get("몫")
        sec["몫"] = share.get("값") if isinstance(share, dict) else share
        sec["번"] = part.get("번")
    got["근거"] = {k: basis[k] for k in ("표지", "글", "출처", "후보", "뺌")}
    got["금액"] = blk.get("금액")
    return got


# ── 블록 풀이 파생 저장본(올림 뒤 한 번 · 뒤에서) ─────────────────────────
def _stamp(key: str, book: dict[str, Any]) -> str:
    """입력 지문 — HP(지문 · 판 · 올린때) + M01 첫 층 파일 내용 + 풀이 판."""
    folder = Path(hb.m01.FOLDER)
    files = {
        p.name: derived.file_digest(p)
        for p in sorted(folder.glob("*.json"))
        if not p.name.startswith("_")
    }
    head = (VERSION, key, book.get("지문"), book.get("판"), book.get("올린때"))
    return derived.derived_key(*head, files)


def _caller(book: dict[str, Any], no: int) -> tuple[dict | None, list[dict]]:
    """블록을 부르는 첫 줄 · 그 호표 다른 줄(B1 · B6) — 일위대가 밖 = (None, [])."""
    line = sangeun.calls(book)[1].get(no)
    if not line:
        return None, []
    unit = _unit(book, line["_호표"])
    same = [x for x in unit.get("줄") or [] if x.get("차례") != line.get("차례")]
    return {k: v for k, v in line.items() if k != "_호표"}, same


def solved(folder: Path, key: str) -> dict[str, Any] | None:
    """저장본 `{번 글: 풀이}` — 지문 다르거나 없으면 None(GET 은 안 지움)."""
    book = hopyo_store.read(folder, key)
    got = derived.read_derived(hopyo_store._dir(folder), KIND + key, _stamp(key, book))
    return got["value"] if got else None


def solve_all(folder: Path, key: str) -> dict[str, Any]:
    """블록 모두 풀어 저장본으로(부르는 첫 줄 기준)."""
    book = hopyo_store.read(folder, key)
    m, out = master(), {}
    for no, blk in sorted(_blocks(book).items()):
        out[str(no)] = _solve(m, blk, *_caller(book, no))
    root = hopyo_store._dir(folder)
    derived.write_derived(root, KIND + key, _stamp(key, book), out, keep=1)
    return out


def _work(folder: Path, key: str) -> None:
    try:
        solve_all(folder, key)
    except Exception:  # noqa: BLE001 — 뒤 셈 실패 = 다음 읽기에서 다시(파생 저장본)
        pass
    finally:
        with _LOCK:
            _RUNNING.discard(f"{folder}|{key}")


def _start(folder: Path, key: str) -> None:
    tag = f"{folder}|{key}"
    with _LOCK:
        if tag in _RUNNING:
            return
        _RUNNING.add(tag)
    threading.Thread(target=_work, args=(Path(folder), key), daemon=True).start()


def ensure(folder: Path, key: str) -> bool:
    """저장본 있음 = 참 · 없으면 뒤 셈 시작(화면 안 막음 · 같은 파일 한 번) = 거짓."""
    if solved(folder, key) is not None:
        return True
    _start(folder, key)
    return False


def judged(folder: Path, key: str) -> dict[int, dict[str, Any]] | None:
    """블록 판정 `{번: {판정, 고를수}}` — 셈 중 = None(뒤 셈 시작 · Marks 칠이 부름)."""
    got = solved(folder, key)
    if got is None:
        _start(folder, key)
        return None
    return {int(no): {"판정": v["판정"], "고를수": v["고를수"]} for no, v in got.items()}


def forget(folder: Path, key: str) -> None:
    """[끝내기] — 그 HP 저장본 지움."""
    shutil.rmtree(derived.derived_dir(hopyo_store._dir(folder), KIND + key), ignore_errors=True)


def listing(folder: Path, key: str, to: Path) -> dict[str, Any]:
    """호표 목록(1차 판정 수 · 만든 SC) · 블록 목록(부름 · 표지 · 판정 · 고를수 · 만든 SK) ·
    `셈중` = 블록 풀이 뒤 셈 중(블록 판정 · 고를수 None)."""
    book = hopyo_store.read(folder, key)
    rows = hrow.rows(book, folder, ledger.doc_id(folder, key))
    sc, sk = _made(folder, key, to)
    units = []
    for h in book.get("호표") or []:
        no = int(h["번"])
        count = Counter(r["판정"] for r in rows.get(h["번"]) or [])
        head = {"번": no, **{k: h.get(k) for k in ("이름", "규격", "단위", "단가")}}
        units.append(
            {**head, "줄수": len(h.get("줄") or []), "판정": dict(count),
             "만든": sc.get(hbuild.mark(key, no))}
        )  # fmt: skip
    call, done = sangeun.callers(book), judged(folder, key)
    blocks = []
    for no, b in sorted(_blocks(book).items()):
        got = (done or {}).get(no) or {"판정": None, "고를수": None}
        blocks.append(
            {"번": no, "이름": b.get("이름"), "단위": b.get("단위"), "금액": b.get("금액"),
             "부름": call.get(no, []), "표지": clause.basis(b)["표지"], **got,
             "만든": sk.get(sangeun.mark(key, no))}
        )  # fmt: skip
    out = {"키": key, "파일": book.get("파일"), "셈중": done is None}
    return {**out, "호표": units, "블록": blocks}


def hopyo(folder: Path, key: str, no: int, to: Path) -> dict[str, Any]:
    """호표 하나 — 줄 판(+ 엑셀 칸) · 단산 줄 `풀이`(부르는 첫 줄이면 저장본)."""
    book = hopyo_store.read(folder, key)
    unit = _unit(book, no)
    rows = hrow.rows({**book, "호표": [unit]}, folder, ledger.doc_id(folder, key))[unit["번"]]
    lines, blocks, m = unit.get("줄") or [], _blocks(book), None
    cache, first = solved(folder, key) or {}, sangeun.calls(book)[1]
    out = []
    for r, line in zip(rows, lines, strict=True):
        r = _row(line, r)
        sub = r.get("번")
        if "단가산출" in r["갈래"] and sub is not None and int(sub) in blocks:
            head = first.get(int(sub)) or {}
            mine = head.get("_호표") == unit["번"] and head.get("차례") == line.get("차례")
            if mine and str(sub) in cache:
                r["풀이"] = cache[str(sub)]
            else:
                m = m or master()
                others = [x for x in lines if x is not line]
                r["풀이"] = _solve(m, blocks[int(sub)], line, others)
        out.append(r)
    sc, _ = _made(folder, key, to)
    head = {k: unit.get(k) for k in ("번", "이름", "규격", "단위", "단가")}
    return {**head, "만든": sc.get(hbuild.mark(key, no)), "줄": out}


def block(folder: Path, key: str, no: int, to: Path) -> dict[str, Any]:
    """단산 블록 하나 — 풀이(저장본 › 셈) + 이름 · 단위 · 부름 · 만든(부르는 첫 줄로 B1 · B6)."""
    book = hopyo_store.read(folder, key)
    blocks = _blocks(book)
    if no not in blocks:
        raise _bad(f"없는 블록 「단산{no}」", 404)
    line, same = _caller(book, no)
    got = (solved(folder, key) or {}).get(str(no)) or _solve(master(), blocks[no], line, same)
    _, sk = _made(folder, key, to)
    blk = blocks[no]
    head = {"이름": blk.get("이름"), "단위": blk.get("단위") or (line or {}).get("단위")}
    head |= {"부름": sangeun.callers(book).get(no, []), "만든": sk.get(sangeun.mark(key, no))}
    return {**got, **head}


# ── 검산 ─────────────────────────────────────────────────────────────
def _want(m: mf.Master, pick: dict[str, Any]) -> Any:
    """고른 것 단위 — 기초 = 결과단위(「원/㎥」 → ㎥ · 「원/자재 단위」 → 자재) · 그 밖 = 고른 것 단위."""
    if pick.get("갈래") == "기초" and pick.get("키"):
        unit = str(m.logic(pick["키"]).get("결과단위") or "")
        unit = unit.split("/", 1)[1] if unit.startswith("원/") else unit
        return unit.removesuffix(" 단위").strip()  # 「원/자재 단위」 = 줄 단위 그대로
    return pick.get("단위")


def quantity(m: mf.Master, line: dict, pick: dict) -> dict[str, Any]:
    """줄 수량 `{엑셀, 몫, 배, 갈래, 우리}` — 사용자 배(단위 환산 값)가 있으면 그것."""
    got = hrow.quantity(line, _want(m, pick), pick.get("몫"))
    times = pick.get("배")
    if times is not None and line.get("수량") is not None:
        ours = Decimal(str(line["수량"])) * Decimal(str(pick.get("몫") or 1))
        got |= {"배": times, "갈래": "사용자", "우리": hrow._plain(ours * Decimal(str(times)))}
    return got


def one_pick(m: mf.Master, pick: dict[str, Any], items: list[dict], share: Any) -> dict:
    """고른 것 하나 검산."""
    out = {"갈래": pick["갈래"], "키": pick.get("키")}
    if pick["갈래"] != "기초":
        return {**out, "판정": CHECKED}
    code = str(pick.get("키") or "")
    try:
        m.logic(code)
    except Exception as e:  # noqa: BLE001 — 없는 키
        raise _bad(f"없는 기초 「{code}」") from e
    full, need = inputs(m, code, pick.get("입력값"))
    rows = solve.pair(hb.quantities(m, code, full), solve._usable(items), share) if items else []
    if need:
        verdict = solve.NEED
    elif any(r["판정"] == solve.DIFF for r in rows):
        verdict = solve.DIFF
    else:
        verdict = solve.SAME if any(r["판정"] == solve.SAME for r in rows) else CHECKED
    got = money(m, code, full) if not need else {"계": None, "까닭": f"입력 필요 {len(need)}"}
    return {**out, "입력값": full, "입력필요": need, "줄": rows, "판정": verdict, "금액": got}


def check(
    folder: Path, key: str, hno: int | None, at: int | None, bno: int | None, picks: list[dict]
) -> dict[str, Any]:
    """`{고른 [...], 수량}` — 호표 줄(`hno` · `at`) · 블록(`bno` · 줄이 단산이면 그 블록) · 고른 기초는
    맡은 절(`Solve.parts`) 단위 줄 · 몫과 대조."""
    book = hopyo_store.read(folder, key)
    blocks, line = _blocks(book), None
    if hno is not None:
        lines = _unit(book, hno).get("줄") or []
        if at is None or not 0 <= at < len(lines):
            raise _bad(f"없는 줄 「제{hno}호표 {at}」")
        line = lines[at]
        ref = line.get("참조") or {}
        if bno is None and ref.get("종류") == "단산" and ref.get("번") is not None:
            bno = int(ref["번"])
    if bno is not None and bno not in blocks:
        raise _bad(f"없는 블록 「단산{bno}」", 404)
    blk = blocks[bno] if bno is not None else {}
    parts = solve.parts(blk, clause.basis(blk)["후보"]) if blk else []
    m = master()
    out = []
    for i, pick in enumerate(picks):
        part = next((x for x in parts if pick.get("키") in x["후보"]), None) or {}
        items = part.get("항목")  # 그 기초가 맡은 절 단위 줄(없으면 블록 통째)
        items = (blk.get("항목") or []) if items is None else items
        share = part.get("몫")
        share = pick.get("몫") or (share.get("값") if isinstance(share, dict) else share) or 1
        out.append({"차례": i, **one_pick(m, pick, items, share)})
    qty = quantity(m, line, picks[0]) if line is not None and picks else None
    return {"고른": out, "수량": qty}