Aislo/M02_MasterTemplete/M02_Sangeun_FromHopyo.py

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

올림 때 경쟁사 산근 시트(`M02_San_Book.sheets` · STmate 숨은 G 열 포함)와 그 엑셀 읽기기 둘 꼴
(812-1 `M02_San_Xlsx` → `M02_Doc_FromNative.pack` · 칸 `엑셀`)을 `실무자료/산근/<HP>.json` 에 둠
(원본 엑셀은 안 둠 · HP 지울 때 같이 지움) → [가져온 식으로 열기] = 그 블록을 원파일과 같은 길
(`M02_Doc_FromNative.emit` · 원본 `엑셀` 참) → 793-2 만들기(`M02_Sangeun_Imported.create` · 개인 층 ·
엑셀 단가 그대로 · 경쟁사 로직 인정 — 근거 기초 제안 없음 · 809-1 · 요구 703) ·
같은 내용 SK 가 있으면 그것(다시씀).
블록 번 = 판정 블록 번(STmate = 코드 「D00102」 숫자 · EST = 단가산출 블록 번).
STmate 블록 끝전 = 파일 투표(`M02_San_Xlsx_Stmate.vote`).
원파일(.estx · .stc · 810-2 `M02_Doc_FromNative`) = 같은 자리 칸 `원파일`(둘 꼴) — 있으면 원파일 우선
(그 블록 번이 원파일에 있을 때) · 엑셀 올림 때 같은 공사 원파일이 다른 HP 에 있으면 붙임 · 원파일 올림 =
같은 공사 엑셀 HP 에 붙임(없으면 원파일로 새 HP · `upload_native`).
"""

from __future__ import annotations

import json
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 = "가져온 식 자료 없음 — 엑셀을 다시 올림"
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
    try:
        excel = _excel(name, sheets)
    except Exception as e:  # noqa: BLE001 — 산근 읽기 실패 = 사유(올리기는 됨)
        return f"산근 시트 읽기 오류 — {type(e).__name__}: {str(e)[:80]}"
    path = _path(folder, key)
    path.parent.mkdir(parents=True, exist_ok=True)
    got = {"파일": name, "시트": sheets, "엑셀": excel}
    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


def _excel(name: str, sheets: dict) -> dict:
    """산근 시트 → 원파일과 같은 둘 꼴(812-1 엑셀 읽기기 `M02_San_Xlsx` → `M02_Doc_FromNative.pack`)."""
    from M02_MasterTemplete import M02_San_Xlsx as sx

    return fn.pack(sx.read_sheets(sheets), name)


def _excel_of(path: Path, saved: dict) -> dict:
    """저장 HP 의 엑셀 둘 꼴 — 옛 저장(시트만)은 한 번 지어 그 자리에 둠."""
    if not isinstance(saved.get("엑셀"), dict):
        saved["엑셀"] = _excel(str(saved.get("파일") or ""), saved["시트"])
        atomic_write_json(path, saved)
    return saved["엑셀"]


# ── 원파일(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 emit(folder: str | Path, key: str, no: int) -> dict[str, Any]:
    """HP 단산 블록 하나 → `{로직, 원본}`(원파일 우선 · 아니면 엑셀 읽기기 같은 길 · 이음 = M01 키)."""
    hopyo_store.read(folder, key)  # 없는 HP = 404
    path = _path(folder, key)
    try:
        saved = json.loads(path.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}")
    excel = _excel_of(path, saved)
    if int(no) not in fn.numbers(excel):
        raise store.StoreError(404, f"없는 단산{no}")
    return fn.emit(excel, int(no))


def _inner(
    source: Path, key: str, logic: dict, native: bool = False
) -> list[tuple[dict, int | None, str | None]]:
    """안쪽 산근 줄 `[(호표 줄, 같은 파일 안쪽 블록 번, 못 잇는 까닭)]` — 원파일 · 엑셀 둘 꼴 같은 길
    (`M02_Doc_FromNative.inner` · 줄 자원 코드 = 블록 코드)."""
    path = _path(source, key)
    saved = json.loads(path.read_bytes())
    if native and isinstance(saved.get("원파일"), dict):
        return fn.inner(saved["원파일"], logic)
    return fn.inner(_excel_of(path, saved), logic)


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}