Aislo/M02_MasterTemplete/M02_Copy_Migrate.py

"""옛 SK · SC · 일위대가 → 로직 사본 찍어 넣기 — 한 번 돌리는 도구(PLAN 795-3 · 요구 655 · 656).

대상 = 사본 없는 마스터 · 회사 · 개인 층 문서(`sangeun` · `precombo` · `labor`) — 지금 M01 로 사본을
찍고(`M02_Sangeun_Copy.stamp` · `M02_Labor_Copy.stamp`) 전후 금액 대조:
    SK = `calc_doc` 기준(노 · 재 · 경 · 계) · SC · 일위대가 = `Labor_Doc.calc` 기준합 · 반영합.
같음 = 파일에 `사본` 칸만 더해 씀(그 밖 칸 그대로) · 다름 · 못 셈 = 안 씀(목록) ·
닿을 것 없음(바탕 참조 SK · 참조 · 손 단가 줄만) = 그대로.
프로젝트 층 일위대가 = 세기 · 대조만(795-4 · 쓰지 않음) — 옛 SP 참조(프로젝트 층은 굳힌
    사본만 읽음 · 마스터 SP 를 바로 가리키는 옛 꼴) = 「못셈(옛 SP 참조)」 갈래.
B09 사본(795-4 · 보기만) = 구조물 사본 · 서고 사본 — 만든 때 그대로 셈(옮길 것 없음 ·
    상태 「그대로」) ·
    `재생성` = 지금 원본으로 다시 가져오면 금액(바뀌면 `재생성다름`) · 못 셈 = 까닭.
되돌릴 길 = 쓰기 전 파일 그대로 `--backup` 폴더에(층 자리 그대로) · `--restore` = 다시 덮음.

    ./venv/Scripts/python.exe -m M02_MasterTemplete.M02_Copy_Migrate             # 보기만(표)
    ./venv/Scripts/python.exe -m M02_MasterTemplete.M02_Copy_Migrate --apply --backup tmp/795_3
    ./venv/Scripts/python.exe -m M02_MasterTemplete.M02_Copy_Migrate --restore tmp/795_3
"""

from __future__ import annotations

import json
import shutil
import sys
from collections import Counter
from pathlib import Path
from typing import Any

from common_util.common_util_json import atomic_write_json
from M02_MasterTemplete import M02_Labor_Copy as lc
from M02_MasterTemplete import M02_Sangeun_Copy as skc
from M02_MasterTemplete import M02_Template_Layers as layers

KINDS = ("sangeun", "precombo", "labor")


def _anchors() -> dict[str, Path]:
    return {"system": layers.system_dir(), "storage": layers.storage_root()}


def _files() -> list[tuple[str, str, Path, bool]]:
    """(어디, 종류, 파일, 프로젝트?) — 층 문서 + 프로젝트 층 일위대가."""
    out = [(w, k, p, False) for k in KINDS for w, p in layers.layer_files(k)]
    root = layers.storage_root()
    for p in sorted(root.glob("*/*/*/templates/labor/[!.]*.json")):
        out.append((f"프로젝트 {p.parts[-4]}", "labor", p, True))
    return out


def _round(value: Any) -> Any:
    if isinstance(value, dict):
        return {k: _round(v) for k, v in value.items()}
    return None if value is None else round(float(value), 6)


def _money(kind: str, doc: dict, whole: Any) -> Any:
    """그 문서 금액 — `whole` = 셈 색인(지금 M01 | 사본 덮개)."""
    from M02_MasterTemplete import M02_Labor_Doc as labor
    from M02_MasterTemplete import M02_Sangeun_Calc as calc

    if kind == "sangeun":
        return _round(calc.run(doc, whole, None, line=False).get("기준"))
    got = labor.calc(doc, None if kind == "labor" else {}, whole=whole)
    return _round({k: got.get(k) for k in ("기준합", "반영합")})


def _clean(kind: str, doc: dict, key: str) -> dict:
    from M02_MasterTemplete import M02_Labor_Doc as labor
    from M02_MasterTemplete import M02_Sangeun_Doc as sd

    plain = {k: v for k, v in doc.items() if k not in (skc.KEY, skc.JOB)}
    if kind == "sangeun":
        return sd.clean(plain, None, skc.whole_())
    return labor.clean({**plain, "열": doc.get("열") or key})


def check(kind: str, path: Path) -> dict[str, Any]:
    """문서 하나 — {상태: 있음 | 필요없음 | 같음 | 다름 | 못셈, 전, 후, 사본?, 까닭?}."""
    from M02_MasterTemplete import M02_MasterTemplete_Store as store

    doc = json.loads(path.read_bytes())
    if skc.has(doc):
        return {"상태": "있음"}
    folder = path.parent.parent
    try:
        with store.at(folder):
            clean = _clean(kind, doc, path.stem)
            copy = skc.stamp(clean) if kind == "sangeun" else lc.stamp(clean)
            if copy is None:
                return {"상태": "필요없음"}
            live = skc.whole_()
            before = _money(kind, clean, live)
            after = _money(kind, clean, skc.over({skc.KEY: copy}, live))
    except Exception as e:  # noqa: BLE001 — 못 셈 = 목록(통째로 멈추지 않음)
        return {"상태": "못셈", "까닭": str(e) or type(e).__name__}
    row = {"상태": "같음" if before == after else "다름", "전": before, "후": after}
    return {**row, "사본": copy} if before == after else {**row, "까닭": "전후 금액 다름 — 안 옮김"}


def _backup_path(path: Path, backup: Path) -> Path:
    for name, anchor in _anchors().items():
        if path.resolve().is_relative_to(anchor.resolve()):
            return backup / name / path.resolve().relative_to(anchor.resolve())
    raise ValueError(f"층 밖 파일 「{path}」")


OLD_SP = "위 층만 가리킴"  # 프로젝트 층 문서가 마스터 SP 를 바로 가리킴(굳힌 사본 아님)
LOCAL = ("입력값.json", "고른자재.json", "면적.json")  # 재생성 대조 때 사본 값 그대로


def _b09_files() -> list[tuple[str, str, Path]]:
    """(어디, 종류, 폴더) — 프로젝트 구조물 사본 · B09 서고 사본."""
    root = layers.storage_root()
    out = []
    for kind, pattern in (
        ("구조물사본", "*/*/*/templates/structures/*/manifest.json"),
        ("서고사본", "*/*/*/B09_Estimation/v1/library/*/manifest.json"),
    ):
        for p in sorted(root.glob(pattern)):
            at = p.parents[3] if kind == "구조물사본" else p.parents[4]
            out.append((f"B09 {at.name}", kind, p.parent))
    return out


def _copy_money(files: dict[str, Any]) -> Any:
    from M02_MasterTemplete import M02_Item_Prices as prices

    labor = files.get("일위대가.json")
    if not isinstance(labor, dict):
        return None
    got = prices.calc(
        labor,
        files.get("단가.json") or {},
        files.get("입력값.json") or {},
        files.get("고른자재.json") or {},
        files.get("면적.json"),
    )
    return _round({k: got.get(k) for k in ("기준합", "반영합")})


def b09_check(kind: str, folder: Path) -> dict[str, Any]:
    """B09 사본 하나(보기만) — {상태: 그대로 | 필요없음 | 못셈, 전, 재생성?, 재생성다름?, 까닭?}."""
    from B09_Estimation import B09_Estimation_Bill_Library as lib
    from M02_MasterTemplete.M02_Item_Source import build_source, read_json

    own = {p.name: read_json(p) for p in folder.glob("*.json")}
    manifest = own.get("manifest.json") or {}
    try:
        before = _copy_money(own)
    except Exception as e:  # noqa: BLE001 — 못 셈 = 목록
        return {"상태": "못셈", "까닭": str(e) or type(e).__name__}
    if before is None:
        return {"상태": "필요없음"}
    try:
        if kind == "서고사본":
            fresh, _ = lib._build(*lib._source_of(manifest))
            lib._apply(fresh, str(manifest.get("종류")), manifest.get("값") or {})
        else:
            fresh = build_source(manifest["원본"])
        after = _copy_money({**fresh, **{n: own[n] for n in LOCAL if n in own}})
    except Exception as e:  # noqa: BLE001 — 원본 없음 · 못 셈 = 재생성 대조만 빠짐
        return {"상태": "그대로", "전": before, "까닭": f"재생성 대조 못 함 — {e}"}
    return {"상태": "그대로", "전": before, "재생성": after, "재생성다름": after != before}


def migrate(apply: bool = False, backup: Path | None = None) -> list[dict[str, Any]]:
    """모든 층 — 문서마다 {어디, 종류, 키, 상태, 씀, …} · `apply` = 같음만 씀(프로젝트 층 빼고)."""
    if apply and backup is None:
        raise ValueError("--apply 는 --backup 폴더가 있어야 함")
    out = []
    for where_, kind, path, project in _files():
        got = check(kind, path)
        if project and got["상태"] == "못셈" and OLD_SP in got.get("까닭", ""):
            got["상태"] = "못셈(옛 SP 참조)"
        copy = got.pop("사본", None)
        row = {"어디": where_, "종류": kind, "키": path.stem, **got, "씀": False}
        if apply and copy and not project:
            keep = _backup_path(path, backup)
            keep.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(path, keep)
            atomic_write_json(path, {**json.loads(path.read_bytes()), skc.KEY: copy})
            row["씀"] = True
        out.append(row)
    for where_, kind, folder in _b09_files():  # 795-4 B09 사본 = 보기만
        out.append({"어디": where_, "종류": kind, "키": folder.name, **b09_check(kind, folder)})
        out[-1]["씀"] = False
    return out


def restore(backup: Path) -> list[str]:
    """`--backup` 폴더 파일을 제자리로 다시 덮음 — 덮은 파일 목록."""
    done = []
    for name, anchor in _anchors().items():
        for keep in sorted((backup / name).rglob("*.json")):
            target = anchor / keep.relative_to(backup / name)
            shutil.copy2(keep, target)
            done.append(str(target))
    return done


def table(rows: list[dict[str, Any]]) -> dict[str, dict[str, int]]:
    """층 갈래 × 종류 × 상태 개수."""
    count: Counter = Counter()
    for r in rows:
        count[(r["어디"].split(" ")[0], r["종류"], r["상태"])] += 1
    out: dict[str, dict[str, int]] = {}
    for (layer, kind, state), n in sorted(count.items()):
        out.setdefault(f"{layer} {kind}", {})[state] = n
    return out


if __name__ == "__main__":  # 브레인 · 사용자 승인 뒤만 --apply
    args = sys.argv[1:]
    if "--restore" in args:
        print("\n".join(restore(Path(args[args.index("--restore") + 1]))))
        raise SystemExit
    at = Path(args[args.index("--backup") + 1]) if "--backup" in args else None
    rows = migrate("--apply" in args, at)
    for one in rows:
        if one["상태"] not in ("있음", "필요없음", "같음") or one.get("재생성다름"):
            print(json.dumps(one, ensure_ascii=False, default=str))
    print(json.dumps(table(rows), ensure_ascii=False, indent=1))