Aislo/M02_MasterTemplete/M02_Sangeun_FromHopyo.py

"""M02 불러오기 단산 블록 → 가져온 식 단가산출(PLAN 793-3b · 요구 646 · 647 · 652) — 서버 단독.

올림 때 경쟁사 산근 시트(`M02_San_Book.sheets` · STmate 숨은 G 열 포함)를
`실무자료/산근/<HP>.json` 에 둠
(원본 엑셀은 안 둠 · HP 지울 때 같이 지움) → [가져온 식으로 열기] = 그 블록을 793-2a 내보내기
(`M02_San_Emit` · 이음 = 1차 줄 판정) → 793-2 만들기(`M02_Sangeun_Imported.create` · 개인 층 ·
엑셀 단가 그대로 · 경쟁사 로직 인정 — 근거 기초 제안 없음 · 809-1 · 요구 703) ·
같은 내용 SK 가 있으면 그것(다시씀).
블록 번 = 판정 블록 번(STmate = 코드 「D00102」 숫자 · EST = 단가산출 블록 번).
STmate 블록 끝전(비목버림 | 합버림_큰나머지)은 파일 설정 — 참조 평가기로 둘 다 셈해
엑셀 계와 맞는 것.
원파일(.estx · .stc · 810-2 `M02_Doc_FromNative`) = 같은 자리 칸 `원파일`(둘 꼴) — 있으면 원파일 우선
(그 블록 번이 원파일에 있을 때) · 엑셀 올림 때 같은 공사 원파일이 다른 HP 에 있으면 붙임 · 원파일 올림 =
같은 공사 엑셀 HP 에 붙임(없으면 원파일로 새 HP · `upload_native`).
"""

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

from common_util.common_util_json import atomic_write_json
from common_util.common_util_time import utc_now_text
from M02_MasterTemplete import M02_Doc_FromNative as fn
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_San_Emit as em

DIRNAME = "산근"
RELOAD = "가져온 식 자료 없음 — 엑셀을 다시 올림"
RULES = ("비목버림", "합버림_큰나머지")
MAX_INNER = 6  # 안쪽 산근 깊이 상한


def _path(folder: str | Path, key: str) -> Path:
    return hopyo_store._dir(folder) / DIRNAME / f"{key}.json"


def unreadable(data: bytes) -> str | None:
    """읽기 불가 사유(797-1a) — 빈 파일 · DRM 암호(파일 머리 `DRMONE`) · 아니면 None."""
    if not data:
        return "빈 파일"
    if b"DRMONE" in data[:16]:
        return "DRM 암호 파일 — 암호 풀린 사본을 올림"
    return None


def keep(folder: str | Path, key: str, name: str, data: bytes) -> str | None:
    """올린 엑셀 → 산근 시트 둠(STmate · EST 꼴만 · 못 읽으면 안 둠 — 올리기는 됨).
    돌려줌 = 못 읽은 사유(빈 파일 · DRM 암호 · 형식 오류) · 둠 · 산근 꼴 아님 = None."""
    from M02_MasterTemplete import M02_San_Book as book

    if why := unreadable(data):
        return why
    try:
        sheets = book.sheets(name, data)
    except Exception as e:  # noqa: BLE001 — 산근은 덤(가져온 식 열기만 막힘) · 사유는 돌려줌
        return f"엑셀 형식 오류 — {type(e).__name__}: {str(e)[:80]}"
    if not book.kind_of(sheets):
        return None
    path = _path(folder, key)
    path.parent.mkdir(parents=True, exist_ok=True)
    got = {"파일": name, "시트": sheets}
    native = next((n for n in _natives(folder) if fn.same(n, sheets)), None)
    atomic_write_json(path, got | ({"원파일": native} if native else {}))
    return None


# ── 원파일(810-2) ──────────────────────────────────────────────────────
def _saved(folder: str | Path) -> list[tuple[Path, dict]]:
    out = []
    for path in sorted((hopyo_store._dir(folder) / DIRNAME).glob("HP-*.json")):
        try:
            out.append((path, json.loads(path.read_bytes())))
        except (OSError, ValueError):
            continue
    return out


def _natives(folder: str | Path) -> list[dict]:
    return [d["원파일"] for _, d in _saved(folder) if isinstance(d.get("원파일"), dict)]


def upload_native(folder: str | Path, name: str, data: bytes, who: Any) -> dict[str, Any]:
    """원파일 올림 — 같은 공사 엑셀 HP 에 원파일을 붙임(`붙음` 키들 · 첫 HP 답) · 없으면 원파일로
    새 HP(EST = 일위대가 호표 · STmate = 호표 없음 · 산근만). 같은 파일 다시 = 그 HP(`이어서`)."""
    mark = hopyo_store.fingerprint(data)
    if same := next((r for r in hopyo_store.rows(folder) if r.get("지문") == mark), None):
        return {**same, "이어서": True, "원파일": True}
    if why := unreadable(data):
        raise store.StoreError(400, why)
    try:
        native = fn.read(name, data)
    except fn.NativeError as e:
        raise store.StoreError(400, str(e)) from e
    joined = []
    for path, got in _saved(folder):
        if got.get("시트") and fn.same(native, got["시트"]):
            atomic_write_json(path, {**got, "원파일": native})
            joined.append(path.stem)
    rows = {r["키"]: r for r in hopyo_store.rows(folder)}
    if joined and joined[0] in rows:
        return {**rows[joined[0]], "원파일": True, "붙음": joined}
    row = hopyo_store.save(folder, name, fn.hopyo(native), who, None, mark)
    path = _path(folder, row["키"])
    path.parent.mkdir(parents=True, exist_ok=True)
    atomic_write_json(path, {"파일": name, "원파일": native})
    return {**row, "원파일": True}


def has(folder: str | Path, key: str) -> bool:
    return _path(folder, key).is_file()


def _rule(blk: dict, nodes: list, prices: dict) -> str:
    """STmate 블록 끝전 — 엑셀 계(꼬리 · 없으면 머리)와 맞는 쪽 · 둘 다 아니면 비목버림."""
    from M02_MasterTemplete import M02_San_Eval as ev

    tails = [t for t in blk["꼬리"] if t["A"].startswith("계") and not t["A"].startswith("계약")]
    want = (tails or [blk["머리"]])[0].get("C")
    for rule in RULES:
        try:
            got = ev.stmate_block(blk, nodes, prices, rule)["계"]["T"]
        except Exception:  # noqa: BLE001 — 못 셈 = 다음 갈래
            continue
        if want is not None and abs(float(got) - float(want)) < 0.5:
            return rule
    return RULES[0]


def emit(folder: str | Path, key: str, no: int) -> dict[str, Any]:
    """HP 단산 블록 하나 → `{로직, 원본}`(793-2a 꼴 · 이음 = M01 키)."""
    from M02_MasterTemplete import M02_San_Book as book
    from M02_MasterTemplete import M02_San_Emit as em
    from M02_MasterTemplete import M02_San_Est as est
    from M02_MasterTemplete import M02_San_Stmate as stm

    hopyo_store.read(folder, key)  # 없는 HP = 404
    try:
        saved = json.loads(_path(folder, key).read_bytes())
    except (OSError, ValueError) as e:
        raise store.StoreError(409, RELOAD) from e
    native = saved.get("원파일")
    if isinstance(native, dict) and int(no) in fn.numbers(native):  # 원파일 우선(810-2)
        return fn.emit(native, int(no))
    if not saved.get("시트"):
        raise store.StoreError(404, f"없는 단산{no}")
    sheets, name = saved["시트"], str(saved.get("파일") or "")
    if book.kind_of(sheets) == "STmate":
        blk = next((b for b in book.stmate_blocks(sheets) if int(b["코드"][1:]) == int(no)), None)
        if blk is None:
            raise store.StoreError(404, f"없는 단산{no}")
        prices = book.stmate_prices(sheets)
        nodes = stm.parse_block(blk["몸"])
        return em.stmate(blk, nodes, prices, _rule(blk, nodes, prices), True, name)
    prices, total = book.est_prices(sheets)
    blk = next((b for b in book.est_blocks(sheets) if int(b["번"]) == int(no)), None)
    if blk is None or blk["번"] not in total:
        raise store.StoreError(404, f"없는 단산{no}")
    return em.est(blk, est.parse_block(blk["글"], prices), total[blk["번"]], True, name)


def _inner(
    source: Path, key: str, logic: dict, native: bool = False
) -> list[tuple[dict, int | None, str | None]]:
    """안쪽 산근 줄 `[(호표 줄, 같은 엑셀 안쪽 블록 번, 못 잇는 까닭)]` — STmate = 줄 코드가 블록 코드 ·
    EST = 줄 이름 「단산N」(N 은 총괄표에 있는 블록). 비목 하나만 부른 줄은 그 비목이 그대로(옮김 없음)일 때만."""
    from M02_MasterTemplete import M02_San_Book as book

    saved = json.loads(_path(source, key).read_bytes())
    if native and isinstance(saved.get("원파일"), dict):
        return fn.inner(saved["원파일"], logic)
    sheets = saved["시트"]
    stmate = book.kind_of(sheets) == "STmate"
    have = {b["코드"]: int(b["코드"][1:]) for b in book.stmate_blocks(sheets)} if stmate else {}
    total = set() if stmate else set(book.est_prices(sheets)[1])
    out = []
    for line in logic["호표"]:
        res = line.get("자원") or {}
        if res.get("갈래") != "단가산출":
            continue
        if stmate:
            n = have.get(str(res.get("코드")))
        else:
            hit = re.fullmatch(r"단산(\d+)", str(res.get("이름")))
            n = int(hit.group(1)) if hit and int(hit.group(1)) in total else None
        if n is None:
            out.append((line, None, "같은 엑셀에 안쪽 블록 없음"))
        elif res.get("옮김") or (res.get("꼴") == "EST" and res.get("열쇠비목") not in em.COMP):
            out.append((line, n, "비목이 바뀐 부름"))
        else:
            out.append((line, n, None))
    return out


def _link_inner(source: Path, key: str, no: int, out: dict, trail: tuple) -> None:
    """산근 속 산근(797-1) — 안쪽 블록을 먼저 같은 길로 만들어 `참조` 로 이음(안쪽이 바뀌면 따라감) ·
    못 잇는 줄 = 엑셀 값묶음 얼림 + 표지."""
    from M02_MasterTemplete import M02_Labor_Ref as refs

    made: dict[int, dict | None] = {}
    native = bool(out["원본"].get("원파일"))
    for line, n, why in _inner(source, key, out["로직"], native):
        if n is not None and not why and (n == no or n in trail or len(trail) >= MAX_INNER):
            why = "안쪽 블록이 자기 자신을 부름"
        if n is not None and not why and n not in made:
            try:
                made[n] = make(source, key, n, "", (*trail, no))
            except Exception as e:  # noqa: BLE001 — 안쪽을 못 만듦 = 얼림(바깥은 만듦)
                made[n], why = None, f"안쪽 블록을 못 만듦 — {str(getattr(e, 'detail', e))[:60]}"
        if n is not None and not why and made[n] is None:
            why = "안쪽 블록을 못 만듦"
        if why:
            out["원본"]["표지"].append(
                f"{em.NOTE}안쪽 산근 「{line['이름']}」 이음 못 함({why}) · 엑셀 값 얼림"
            )
            continue
        layer, owner = refs.here()
        line["참조"] = {"층": layer, "주인": owner, "키": made[n]["이름"]}
        res = line["자원"]
        if res.get("비목"):
            line["참조비목"] = em.COMP[res["비목"]]


def make(source: Path, key: str, no: int, name: str = "", _trail: tuple = ()) -> dict[str, Any]:
    """지금 층(만들 층)에 가져온 식 SK — 같은 내용이 있으면 그 SK(`다시씀`) ·
    안쪽 산근 블록은 형제 SK 를 먼저 만들어 참조(797-1)."""
    from M02_MasterTemplete import M02_Sangeun_Imported as imp

    out = emit(source, key, no)
    out["원본"]["가져온때"] = utc_now_text()  # 원문 패널 「어디서 왔나」(807-1)
    _link_inner(source, key, no, out, _trail)
    try:  # 소요량 · 계수 후보 표지(797-2 · 사람 확인) — 못 셈 = 표지 없이 만듦
        from M02_MasterTemplete import M02_Sangeun_QtyLinks as qty

        out["원본"]["표지"] += qty.flags(out["로직"])
    except Exception:  # noqa: BLE001 — 역색인 짓기 실패가 가져온 식 만들기를 막지 않음
        pass
    title = str(name or "").strip() or str(out["로직"].get("이름") or f"단산{no}")
    try:
        got = imp.create(out["로직"], out["원본"], title)
        return {**got, "다시씀": False}
    except store.StoreError as e:
        same = (e.detail or {}).get("같은것") if isinstance(e.detail, dict) else None
        if e.status != 409 or not same:
            raise
        got = imp.create(out["로직"], out["원본"], title, use=same[0]["키"])
        return {**got, "다시씀": True}