Aislo/B09_Estimation/B09_Estimation_Refresh.py

"""B09 [단가 갱신] · [재생성](PLAN 795-4 · 옛 [업데이트] 703-3) — 사본 로직은 만든 때 그대로 ·
[단가 갱신](기본 · 841-1) = 프로젝트 사본 자원 단가 줄(`B09_Estimation_Copy_Prices`) + 자재 단가 책만
    지금 M01 값 · 미리 보기 `price_preview` = 바뀐 자원 · 총괄 전 → 후 ·
[재생성](사용자가 따로) = 아래 옛 [업데이트] — 단가 책 다시 굳힘 + 원본이 바뀐 구조물 사본 [재적용].

미리 보기(`preview`) = 바뀌는 단가 줄(새 줄 + 바뀐 줄 · 지금 수요) · 다시 가져올 구조물 사본
    (연결된 것 중 원본이 바뀐 것만 · 원본 없음 = 그대로) · 지킬 고친 칸(덮개 칸 수 · 702-4 —
    재적용 · 새 판 뒤에도 남음) · 노임 판(굳힌 판 → 지금 M01 판 · 옛 책 = 굳힌 날)
적용(`reapply_changed` → 부르는 쪽이 수요 다시 → `prices.refresh_book`) — 사본 먼저(수요가 바뀜) ·
    덮개(사용자 값)는 안 건드림 · 사본 하나가 막혀도 나머지는 함(막힌 것 = 알림).
M02 단가산출(SK · 719-2) — 구조물 · 서고 사본의 굳힌 SK 를 원본 층에서 다시 풀어 굳힘(같은 굳힌 키 ·
    같은 cover → B09 덮개 · 입력값 · 고른 자재 그대로) · 서고 사본도 원본판이 다르면 다시 굳힘 ·
    미리 보기 `단가산출` = 바뀌는 SK 줄 [{사본, 이름, 키, 판(새 · 바뀜 · 빠짐)}].
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from B09_Estimation import B09_Estimation_Bill_Library as library
from B09_Estimation import B09_Estimation_Prices as prices
from B09_Estimation.B09_Estimation_Calc_Labor import COPY
from M02_MasterTemplete import M02_Item_Copy as item_copy
from M02_MasterTemplete import M02_Item_Sangeun as item_sk
from M02_MasterTemplete.M02_Item_Snapshot import MANIFEST_NAME, linked_snapshots, structures_dir
from M02_MasterTemplete.M02_Item_Source import LABOR_FILE, build_source, read_json

REGEN = "재생성"  # POST …/estimation/prices · preview 몸 `job`(795-4)


def _value(row: dict[str, Any] | None) -> Any:
    return None if row is None else row.get("price", row.get("prices"))


def price_preview(
    old: dict[str, Any] | None,
    new: dict[str, Any],
    before: dict,
    after: dict,
    cells: int,
    copies: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """[단가 갱신] 미리 보기(795-4 · 841-1) — {할일, 바뀐자원 [{키, 이름, 전, 후}](사본 자원 줄 +
    자재 단가 책), 전, 후(총괄 grand · 빈 줄 있으면 None), 대가 {전, 후}(일위대가 블록 합), 고친칸,
    노임판} · 사본(로직)은 안 바뀜."""
    was = json.loads(json.dumps((old or {}).get("rows") or {}, ensure_ascii=False))
    now = json.loads(json.dumps(new["rows"], ensure_ascii=False))
    rows = [
        {"키": key, "이름": row.get("name") or "", "전": _value(was.get(key)), "후": _value(row)}
        for key, row in now.items()
        if was.get(key) != row
    ]

    def grand(sheets: dict) -> Any:
        return (((sheets.get("total") or {}).get("subtotals")) or {}).get("grand")

    def daega(sheets: dict) -> float:
        return sum(float(b.get("sum") or 0) for b in sheets.get("daega_blocks") or [])

    seen = {r["키"] for r in copies or []}
    return {
        "할일": "단가갱신",
        "바뀐자원": [*(copies or []), *(r for r in rows if r["키"] not in seen)],
        "전": grand(before),
        "후": grand(after),
        "대가": {"전": daega(before), "후": daega(after)},
        "고친칸": cells,
        "노임판": {
            "전": edition_text((old or {}).get("labor_edition")),
            "후": edition_text(new.get("labor_edition")),
        },
    }


def changed_copies(root: str | Path) -> list[dict[str, Any]]:
    """연결된 구조물 사본 중 원본(층 · 마스터)이 바뀐 것 `[{폴더, 이름}]`."""
    linked = {one["폴더"] for one in linked_snapshots(root).values()}
    return [
        {"폴더": it["폴더"], "이름": it["이름"] or it["폴더"]}
        for it in item_copy.list_items(root)
        if it["폴더"] in linked and it["마스터바뀜"]
    ]


def _sks(labor: Any) -> dict[str, dict]:
    copies = (labor or {}).get("참조사본") or {}
    return {rid: doc for rid, doc in copies.items() if item_sk.is_sangeun(doc)}


def sk_lines(old: Any, new: Any, where: str) -> list[dict[str, str]]:
    """사본 하나 전후 일위대가.json → 바뀌는 SK 줄(굳힌 줄 · 원본 SK 문서가 다름)."""
    was, now = _sks(old), _sks(new)
    out = []
    for rid in sorted(set(was) | set(now)):
        a, b = was.get(rid) or {}, now.get(rid) or {}
        if all(a.get(k) == b.get(k) for k in (item_sk.ROW, "원본")):
            continue
        mark = "새" if not a else "빠짐" if not b else "바뀜"
        out.append({"사본": where, "이름": (b or a).get("이름") or "", "키": rid.rsplit("|")[-1],
                    "판": mark})  # fmt: skip
    return out


def sk_changes(root: str | Path, copies: list[dict], libs: list[tuple]) -> list[dict[str, str]]:
    """다시 가져올 구조물 사본(`changed_copies`) · 서고 사본(`library.changed`)의 바뀌는 SK 줄."""
    out = []
    for one in copies:
        folder = structures_dir(root) / one["폴더"]
        try:
            fresh = build_source((read_json(folder / MANIFEST_NAME) or {})["원본"])
        except (LookupError, ValueError, KeyError):
            continue
        out += sk_lines(read_json(folder / LABOR_FILE), fresh.get(LABOR_FILE), one["이름"])
    for folder, manifest, files, origin in libs:
        name = str(origin.get("name") or folder.name)
        out += sk_lines(read_json(folder / LABOR_FILE), files.get(LABOR_FILE), name)
    return out


def cover_cells(cover: dict[str, Any] | None) -> int:
    """고친 칸 수 — 덮개 칸(품 · 수량 · 단가) · 복사본 표지 뺌."""
    return sum(
        len(cell) for lines in (cover or {}).values() for cid, cell in lines.items() if cid != COPY
    )


def edition_text(editions: dict[str, str] | None) -> str:
    return " · ".join(f"{kind} {day[:7]}" for kind, day in sorted((editions or {}).items()))


def preview(
    root: str | Path, book: dict[str, Any] | None, keys: list[str], cover: dict | None
) -> dict[str, Any]:
    diff = prices.book_diff(book, prices.refresh_book(keys, book))
    copies, libs = changed_copies(root), library.changed(root)
    lib_names = [{"폴더": f.name, "이름": str(o.get("name") or f.name)} for f, _m, _f, o in libs]
    before = (book or {}).get("labor_edition")
    made = str((book or {}).get("made") or "")[:10]
    return {
        "단가줄": diff["new"] + diff["changed"],
        "새줄": diff["new"],
        "바뀐줄": diff["changed"],
        "사본": copies + lib_names,
        "단가산출": sk_changes(root, copies, libs),
        "고친칸": cover_cells(cover),
        "노임판": {
            "전": edition_text(before) if before else (f"굳힌 날 {made}" if made else "안 굳힘"),
            "후": edition_text(prices.labor_editions(keys)),
        },
    }


def reapply_changed(root: str | Path, who: dict[str, Any] | None = None) -> dict[str, list]:
    """원본이 바뀐 연결 사본 재적용 · 서고 사본 다시 굳힘 —
    `{다시: [이름], 막힘: [이름 — 까닭]}`."""
    done, stuck = [], []
    for one in library.changed(root):
        name = str(one[3].get("name") or one[0].name)
        try:
            library.refresh_copy(root, one)
            done.append(name)
        except (LookupError, ValueError, KeyError, OSError) as error:
            stuck.append(f"{name} — {error}")
    for one in changed_copies(root):
        try:
            item_copy.reapply(root, one["폴더"], who)
            done.append(one["이름"])
        except (LookupError, ValueError, KeyError, OSError) as error:
            stuck.append(f"{one['이름']} — {error}")
    return {"다시": done, "막힘": stuck}