Aislo/B09_Estimation/B09_Estimation_Refresh.py

"""B09 [업데이트] 한 곳(PLAN 703-3) — 단가 책 다시 굳힘 + 원본이 바뀐 구조물 사본 [재적용]을 함께.

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

from __future__ import annotations

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


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"],
        "계수": diff["factor"],
        "환율": diff["fx"],
        "사본": 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}