Aislo/B09_Estimation/B09_Estimation_Storage.py

"""B09 원가계산 — 프로젝트 저장 (PLAN 9-2).

**기준자료는 파일, 프로젝트가 채택한 단가는 사본**이다 (사용자 확정 2026-09-07).

  - 기준자료(요율·품셈·노임·자재) = `resources/data_cost_input_value/` 의 버전 파일.
    갱신해도 **옛 프로젝트 결과가 안 바뀌어야** 한다.
  - 그래서 프로젝트는 **그때 쓴 단가를 사본으로** 영구저장소
    `<project_root>/B09_Estimation/v1/` 에 남긴다. 3년 뒤 열어도 같은 금액이 나온다.
    상용 프로그램도 같은 방식이다 — STmate 는 단가표(`COSTN`)·제비율표(`RATE`)를
    **프로젝트 파일 안에 동봉**한다.

`dataset_version` 은 **세 쪽**을 다 적는다 — `dataset_id` · `effective_date` · `sha256`.
파일명만 적으면 같은 날짜로 재생성된 파일과 구분이 안 된다.

⚠ **차수 슬롯** — 실무 STC 는 단가를 `TAMT1~4`(당초·1차·2차·3차 변경) 네 벌로 들고
있고, 파일마다 살아 있는 슬롯이 다르다(계류보전 변경본은 3차). 지금 신설 설계에는
안 걸리지만 저장 구조를 정하는 중이라 **차수 자리를 비워 둔다**.
"""

from __future__ import annotations

import hashlib
import json
import os
from dataclasses import dataclass
from decimal import Decimal
from typing import Any

from B09_Estimation.B09_Estimation_PriceBook import (
    DEFAULT_SLOT_NAMES,
    PriceBook,
    PriceDetail,
    PriceKind,
    PriceTitle,
)

#: 영구저장소 안 B09 몫. `common_util_storage.PROJECT_STORAGE_LAYOUT_V2` 와 같은 자리.
SNAPSHOT_SUBPATH = ("B09_Estimation", "v1")
SNAPSHOT_FILE = "price_book.json"
MANIFEST_FILE = "_manifest.json"
SNAPSHOT_SCHEMA_VERSION = "1.0"

#: 차수 — 당초(1)만 쓴다. 변경설계가 붙으면 2·3·4 로 늘어난다.
DEFAULT_REVISION = 1


class SnapshotError(OSError):
    """스냅샷 읽기·쓰기 실패."""


@dataclass(frozen=True)
class DatasetVersion:
    """어느 판 기준자료로 계산했나 — 세 쪽을 다 적는다."""

    dataset_id: str
    effective_date: str
    sha256: str

    def as_dict(self) -> dict[str, str]:
        return {
            "dataset_id": self.dataset_id,
            "effective_date": self.effective_date,
            "sha256": self.sha256,
        }

    @classmethod
    def from_dict(cls, raw: dict[str, str]) -> DatasetVersion:
        return cls(
            dataset_id=raw.get("dataset_id", ""),
            effective_date=raw.get("effective_date", ""),
            sha256=raw.get("sha256", ""),
        )


def snapshot_dir(project_root: str) -> str:
    path = os.path.abspath(os.path.join(project_root, *SNAPSHOT_SUBPATH))
    root = os.path.abspath(project_root)
    if os.path.commonpath((root, path)) != root:
        raise SnapshotError("스냅샷 경로가 프로젝트 루트를 벗어났습니다.")
    return path


def _sha256_of(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _title_to_dict(title: PriceTitle) -> dict[str, Any]:
    return {
        "code": title.code,
        "kind": title.kind.value,
        "name": title.name,
        "spec": title.spec,
        "unit": title.unit,
        "slots": [None if v is None else str(v) for v in title.slots],
        "slot_pages": list(title.slot_pages),
        "adopted_slot": title.adopted_slot,
    }


def _title_from_dict(raw: dict[str, Any]) -> PriceTitle:
    return PriceTitle(
        code=raw["code"],
        kind=PriceKind(raw["kind"]),
        name=raw.get("name", ""),
        spec=raw.get("spec", ""),
        unit=raw.get("unit", ""),
        slots=[None if v is None else Decimal(str(v)) for v in raw.get("slots", [])],
        slot_pages=list(raw.get("slot_pages", [])),
        adopted_slot=int(raw.get("adopted_slot", 6)),
    )


def _detail_to_dict(detail: PriceDetail) -> dict[str, Any]:
    return {
        "parent_code": detail.parent_code,
        "ref_code": detail.ref_code,
        "quantity": str(detail.quantity),
        "note": detail.note,
        "percent_of_parent": (
            None if detail.percent_of_parent is None else str(detail.percent_of_parent)
        ),
        # ⚠ 비율 줄 셋을 **다 적는다** — 하나라도 빠뜨리면 저장했다 다시 읽을 때 그 줄이
        #   조용히 사라져 단가가 낮아진다(제잡비가 그 자리였다).
        "percent_of_labor": (
            None if detail.percent_of_labor is None else str(detail.percent_of_labor)
        ),
        "percent_of_labor_target": detail.percent_of_labor_target,
        "percent_of_material": (
            None if detail.percent_of_material is None else str(detail.percent_of_material)
        ),
        "output": None if detail.output is None else str(detail.output),
    }


def _detail_from_dict(raw: dict[str, Any]) -> PriceDetail:
    percent = raw.get("percent_of_parent")
    labor_percent = raw.get("percent_of_labor")
    material_percent = raw.get("percent_of_material")
    return PriceDetail(
        parent_code=raw["parent_code"],
        ref_code=raw["ref_code"],
        quantity=Decimal(str(raw.get("quantity", "0"))),
        note=raw.get("note", ""),
        percent_of_parent=None if percent is None else Decimal(str(percent)),
        percent_of_labor=None if labor_percent is None else Decimal(str(labor_percent)),
        percent_of_labor_target=str(raw.get("percent_of_labor_target") or "expense"),
        percent_of_material=(None if material_percent is None else Decimal(str(material_percent))),
        output=None if raw.get("output") is None else Decimal(str(raw["output"])),
    )


def save_price_book(
    project_root: str,
    book: PriceBook,
    *,
    dataset_versions: list[DatasetVersion],
    revision: int = DEFAULT_REVISION,
) -> str:
    """프로젝트가 채택한 단가를 사본으로 남긴다. 저장한 파일 경로를 돌려준다.

    기준자료가 갱신돼도 이 사본이 있어 **옛 프로젝트 결과가 안 바뀐다**.
    """
    directory = snapshot_dir(project_root)
    os.makedirs(directory, exist_ok=True)

    payload = {
        "schema_version": SNAPSHOT_SCHEMA_VERSION,
        "revision": revision,
        "slot_names": list(book.slot_names),
        "titles": [_title_to_dict(t) for t in book.titles.values()],
        "details": [_detail_to_dict(d) for rows in book.details.values() for d in rows],
    }
    text = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)

    path = os.path.join(directory, SNAPSHOT_FILE)
    with open(path, "w", encoding="utf-8", newline="\n") as handle:
        handle.write(text)

    manifest = {
        "schema_version": SNAPSHOT_SCHEMA_VERSION,
        "revision": revision,
        "files": [
            {
                "file": SNAPSHOT_FILE,
                "sha256": _sha256_of(text),
                "size_bytes": len(text.encode("utf-8")),
            }
        ],
        # 어느 판 기준자료로 계산했나 — 세 쪽을 다 적는다(9-2).
        "dataset_versions": [v.as_dict() for v in dataset_versions],
    }
    with open(
        os.path.join(directory, MANIFEST_FILE), "w", encoding="utf-8", newline="\n"
    ) as handle:
        json.dump(manifest, handle, ensure_ascii=False, indent=2, sort_keys=True)

    return path


def load_price_book(project_root: str) -> tuple[PriceBook, list[DatasetVersion]]:
    """사본을 그대로 되읽는다. 파일이 손상됐으면 조용히 넘기지 않고 멈춘다."""
    directory = snapshot_dir(project_root)
    path = os.path.join(directory, SNAPSHOT_FILE)
    if not os.path.exists(path):
        raise SnapshotError(f"단가 스냅샷이 없습니다: {path}")

    with open(path, encoding="utf-8") as handle:
        text = handle.read()
    payload = json.loads(text)

    manifest_path = os.path.join(directory, MANIFEST_FILE)
    versions: list[DatasetVersion] = []
    if os.path.exists(manifest_path):
        with open(manifest_path, encoding="utf-8") as handle:
            manifest = json.load(handle)
        recorded = next(
            (f.get("sha256") for f in manifest.get("files", []) if f.get("file") == SNAPSHOT_FILE),
            None,
        )
        if recorded and recorded != _sha256_of(text):
            raise SnapshotError(
                f"단가 스냅샷 지문이 매니페스트와 다릅니다 — 파일이 바뀌었습니다: {path}"
            )
        versions = [DatasetVersion.from_dict(v) for v in manifest.get("dataset_versions", [])]

    book = PriceBook(slot_names=tuple(payload.get("slot_names", DEFAULT_SLOT_NAMES)))
    for raw in payload.get("titles", []):
        book.add_title(_title_from_dict(raw))
    for raw in payload.get("details", []):
        book.add_detail(_detail_from_dict(raw))
    return book, versions