feat(B09): 단가 계층·단수 자리·프로젝트 스냅샷 (PLAN 9-2·9-3·9-4)
**단수 처리를 「출력 위치」에 바인딩** (`B09_Estimation_Rounding.py`) - `단수처리_규칙.md` §1 — 같은 수량×단가라도 **내역서 본체는 절사, 집계표는 반올림**. 함수가 항목이 아니라 **표 종류**에 붙음. 자리를 안 대고 부르는 길을 안 둠(기본값 없음). - ⚠ **집계표와 본체의 합이 어긋나는 것이 정상** — 「합계가 1원 틀린다」에 계산을 고치지 않도록 `summary_vs_body_gap()` 으로 차이를 값으로 드러내고 화면 문구를 상수로 둠. - 계산은 전정밀, 자르는 것은 표 그리는 자리에서. 값은 원문에서 복사 안 하고 문서를 가리킴. **단가 계층 — 표를 세 벌 만들지 않음** (`B09_Estimation_PriceBook.py`) - 「제목 + 상세」 한 쌍 + 종류 구분. 층은 `S 중기취득가 → X 시간당사용료 → B 일위대가 → D 단가산출`, 바닥은 `M 자재`·`L 노임`, `W 일식·견적`은 **단가 0**(무대처리, 8-7 ㉡). - 금액은 어느 층이든 **재료·노무·경비 3분할**, 합계 = 셋의 합. - **단가 슬롯 6개** — 번호 고정, **이름은 프로젝트 설정**. 기본 채택 6번(`JUKNM=6`). 슬롯별 **근거 쪽수**(STC `PG_` 열) 자리도 둠. - 안전장치: 채택 슬롯이 비면 **0 으로 안 때우고 멈춤** · 없는 코드는 `unmatched_codes()` 로 **목록으로 냄** · 참조 순환 감지. **프로젝트 스냅샷** (`B09_Estimation_Storage.py`) - 채택 단가를 `<project_root>/B09_Estimation/v1/` 에 사본으로. 기준자료가 갱신돼도 **옛 프로젝트 결과가 안 바뀜**. `dataset_version` 은 `dataset_id`+`effective_date`+ `sha256` **세 쪽**. 사본이 바뀌면 지문 불일치로 멈춤. 차수(당초·1~3차) 자리 비워 둠. 자체검증 — 신규 13건 포함 `pytest tmp/tests/ -q` **85 passed** · ruff 통과 · 파일 최대 534줄. 중기 실측값(`X00005` 96,829 = 노 55,700 + 재 18,001 + 경 23,128)으로 3분할 유지·층 쌓임 확인. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,212 @@
|
||||
"""B09 원가계산 — 단가 계층 (PLAN 9-3 · 9-4).
|
||||
|
||||
**표를 세 벌 만들지 않는다.** 「제목 한 줄 + 상세 여러 줄」 한 쌍을 두고 **종류로만** 가른다.
|
||||
상용 프로그램 둘(STmate `COSTN`/`BOQ11`, EST Plus `*Title`/`*Main`)이 같은 모양이었고,
|
||||
실무 시트 이름도 `일위대가목록표 / 일위대가표` 처럼 짝을 이룬다.
|
||||
|
||||
실제 층은 넷이다 (2026-09-07 STC `COSTN` 186행 실측, PLAN 9-3):
|
||||
|
||||
S 중기 취득가(천원) → X 시간당 중기사용료 → B 일위대가 → D 단가산출
|
||||
↑ L 노임 · M 자재를 참조
|
||||
|
||||
금액은 **어느 층이든 재료·노무·경비 3분할**이고 `합계 = 재료 + 노무 + 경비` 다
|
||||
(ESTX 9.1만 건 전건 통과).
|
||||
|
||||
단가 원천은 **슬롯 6개**다 (PLAN 9-4). 번호는 고정, **이름은 프로젝트 설정**이다 —
|
||||
설계사무소마다 다르다(영월만 「유통 물가·거래 가격등·기타 단가」). 기본 채택은 **6번**
|
||||
(STmate `JUKNM=6`, Ini 보유 6파일 전건 일치).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from decimal import Decimal
|
||||
from enum import Enum
|
||||
|
||||
_ZERO = Decimal(0)
|
||||
|
||||
#: 단가 원천 슬롯 수. 번호 고정.
|
||||
PRICE_SLOT_COUNT = 6
|
||||
#: 기본 채택 슬롯(1-based). STmate `JUKNM=6` = 「적용 단가」.
|
||||
DEFAULT_ADOPTED_SLOT = 6
|
||||
|
||||
#: 슬롯 이름 기본값 — **프로젝트 설정으로 덮어쓴다**.
|
||||
#: TODO(미결 PLAN 9-4): 사무소마다 1~5 이름이 달라 확정 기본값이 아님. 프로젝트가 고름.
|
||||
DEFAULT_SLOT_NAMES: tuple[str, ...] = (
|
||||
"조달가격",
|
||||
"물가정보",
|
||||
"물가자료",
|
||||
"적산정보",
|
||||
"견적단가",
|
||||
"적용 단가",
|
||||
)
|
||||
|
||||
|
||||
class PriceKind(str, Enum):
|
||||
"""단가 항목의 종류. STC `COSTN.CODE` 앞글자와 1:1 (PLAN 9-3)."""
|
||||
|
||||
MATERIAL = "material" # M — 자재 카탈로그 (재료비만)
|
||||
LABOR = "labor" # L — 노임 카탈로그 (노무비만)
|
||||
MACHINE_BASE = "machine_base" # S — 중기 취득가 (천원 단위, 경비만)
|
||||
MACHINE_HOURLY = "machine_hourly" # X — 시간당 중기사용료 (3분할)
|
||||
UNIT_PRICE = "unit_price" # B — 일위대가 (3분할)
|
||||
PRICE_BASIS = "price_basis" # D — 단가산출 (3분할)
|
||||
LUMPSUM = "lumpsum" # W — 일식·견적 (무대처리 등, 단가 0)
|
||||
|
||||
|
||||
#: 카탈로그 층 — 상세를 갖지 않고 값이 바로 있는 종류.
|
||||
CATALOG_KINDS = frozenset({PriceKind.MATERIAL, PriceKind.LABOR, PriceKind.MACHINE_BASE})
|
||||
|
||||
|
||||
class PriceBookError(LookupError):
|
||||
"""단가 조립이 성립하지 않는 경우. 0 으로 때우지 않고 멈춘다."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Money3:
|
||||
"""금액 3분할 — 재료·노무·경비. 합계는 셋의 합이다."""
|
||||
|
||||
material: Decimal = _ZERO
|
||||
labor: Decimal = _ZERO
|
||||
expense: Decimal = _ZERO
|
||||
|
||||
@property
|
||||
def total(self) -> Decimal:
|
||||
return self.material + self.labor + self.expense
|
||||
|
||||
def __add__(self, other: Money3) -> Money3:
|
||||
return Money3(
|
||||
self.material + other.material,
|
||||
self.labor + other.labor,
|
||||
self.expense + other.expense,
|
||||
)
|
||||
|
||||
def scaled(self, factor: Decimal) -> Money3:
|
||||
return Money3(self.material * factor, self.labor * factor, self.expense * factor)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PriceTitle:
|
||||
"""제목 줄 — 「무엇이 있나」 한 줄. 실무 시트의 `…목록표` 에 해당."""
|
||||
|
||||
code: str
|
||||
kind: PriceKind
|
||||
name: str
|
||||
spec: str = ""
|
||||
unit: str = ""
|
||||
|
||||
#: 원천 슬롯 6개. 값이 없는 슬롯은 None (그 출처에 안 실린 자재).
|
||||
slots: list[Decimal | None] = field(default_factory=lambda: [None] * PRICE_SLOT_COUNT)
|
||||
#: 슬롯별 근거 쪽수 — 실무 내역서가 「물가정보 몇 쪽」을 남긴다(STC `PG_` 열).
|
||||
slot_pages: list[str | None] = field(default_factory=lambda: [None] * PRICE_SLOT_COUNT)
|
||||
#: 채택 슬롯(1-based).
|
||||
adopted_slot: int = DEFAULT_ADOPTED_SLOT
|
||||
|
||||
def adopted_price(self) -> Decimal:
|
||||
"""채택 슬롯의 단가. 비어 있으면 0 으로 때우지 않고 멈춘다."""
|
||||
if not 1 <= self.adopted_slot <= PRICE_SLOT_COUNT:
|
||||
raise PriceBookError(
|
||||
f"{self.code}: 채택 슬롯 번호가 범위 밖입니다 ({self.adopted_slot})"
|
||||
)
|
||||
value = self.slots[self.adopted_slot - 1]
|
||||
if value is None:
|
||||
raise PriceBookError(
|
||||
f"{self.code} ({self.name}): 채택 슬롯 {self.adopted_slot} 에 단가가 없습니다. "
|
||||
"유료 물가지를 안 봤다면 6번(적용 단가)에 직접 넣으십시오."
|
||||
)
|
||||
return value
|
||||
|
||||
def catalog_money(self) -> Money3:
|
||||
"""카탈로그 층의 3분할 — 종류가 성분을 정한다.
|
||||
|
||||
자재는 재료비만, 노임은 노무비만, 중기 취득가는 경비만 갖는다(STC 실측).
|
||||
"""
|
||||
price = self.adopted_price()
|
||||
if self.kind is PriceKind.MATERIAL:
|
||||
return Money3(material=price)
|
||||
if self.kind is PriceKind.LABOR:
|
||||
return Money3(labor=price)
|
||||
if self.kind is PriceKind.MACHINE_BASE:
|
||||
return Money3(expense=price)
|
||||
raise PriceBookError(f"{self.code}: 카탈로그 종류가 아닙니다 ({self.kind})")
|
||||
|
||||
|
||||
@dataclass
|
||||
class PriceDetail:
|
||||
"""상세 줄 — 「그것이 무엇으로 이루어졌나」 한 줄.
|
||||
|
||||
`ref_code` 가 **원천 참조**다. 어느 층을 가리키는지가 그 코드의 종류로 드러난다
|
||||
(ESTX `LinkIndex` 와 같은 축).
|
||||
"""
|
||||
|
||||
parent_code: str
|
||||
ref_code: str
|
||||
quantity: Decimal
|
||||
note: str = ""
|
||||
#: 비율 행(공구손료 등) — 참조 단가의 %로 계산하는 줄.
|
||||
percent_of_parent: Decimal | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class PriceBook:
|
||||
"""제목 + 상세 한 벌. 종류로만 갈린다."""
|
||||
|
||||
titles: dict[str, PriceTitle] = field(default_factory=dict)
|
||||
details: dict[str, list[PriceDetail]] = field(default_factory=dict)
|
||||
#: 슬롯 이름 — 프로젝트 설정.
|
||||
slot_names: tuple[str, ...] = DEFAULT_SLOT_NAMES
|
||||
|
||||
def add_title(self, title: PriceTitle) -> None:
|
||||
if title.code in self.titles:
|
||||
raise PriceBookError(f"코드가 겹칩니다: {title.code}")
|
||||
self.titles[title.code] = title
|
||||
|
||||
def add_detail(self, detail: PriceDetail) -> None:
|
||||
self.details.setdefault(detail.parent_code, []).append(detail)
|
||||
|
||||
def title(self, code: str) -> PriceTitle:
|
||||
try:
|
||||
return self.titles[code]
|
||||
except KeyError as exc:
|
||||
raise PriceBookError(f"단가표에 없는 코드입니다: {code}") from exc
|
||||
|
||||
def resolve(self, code: str, _seen: tuple[str, ...] = ()) -> Money3:
|
||||
"""그 항목의 단가를 3분할로 조립한다.
|
||||
|
||||
카탈로그 층(M·L·S)은 값이 바로 있고, 그 위 층(X·B·D)은 상세 줄을 재귀로 더한다.
|
||||
`W`(일식·견적)는 **단가가 0** 이다 — 무대처리처럼 품에 이미 포함된 줄 (PLAN 8-7 ㉡).
|
||||
"""
|
||||
if code in _seen:
|
||||
raise PriceBookError(f"단가 참조가 돌고 있습니다: {' → '.join((*_seen, code))}")
|
||||
|
||||
title = self.title(code)
|
||||
if title.kind is PriceKind.LUMPSUM:
|
||||
return Money3()
|
||||
if title.kind in CATALOG_KINDS:
|
||||
return title.catalog_money()
|
||||
|
||||
rows = self.details.get(code)
|
||||
if not rows:
|
||||
raise PriceBookError(f"{code} ({title.name}): 상세 줄이 없어 단가를 조립할 수 없습니다")
|
||||
|
||||
total = Money3()
|
||||
for row in rows:
|
||||
child = self.resolve(row.ref_code, (*_seen, code))
|
||||
if row.percent_of_parent is not None:
|
||||
# 비율 행 — 지금까지 쌓인 값의 %로 붙는다(공구손료 등).
|
||||
total = total + total.scaled(row.percent_of_parent / Decimal(100))
|
||||
continue
|
||||
total = total + child.scaled(row.quantity)
|
||||
return total
|
||||
|
||||
def unmatched_codes(self) -> list[str]:
|
||||
"""상세가 가리키는데 제목이 없는 코드 — **빈칸으로 두지 않고 목록으로 낸다**.
|
||||
|
||||
문자열 매칭 실패를 조용히 0 원으로 넘기지 않기 위한 자리 (PLAN 8-6).
|
||||
"""
|
||||
missing: list[str] = []
|
||||
for rows in self.details.values():
|
||||
for row in rows:
|
||||
if row.ref_code not in self.titles and row.ref_code not in missing:
|
||||
missing.append(row.ref_code)
|
||||
return missing
|
||||
@@ -0,0 +1,69 @@
|
||||
"""B09 원가계산 — 단수 처리는 **출력 위치**에 붙는다.
|
||||
|
||||
`resources/knowledge/technical_info/01_임도/05_원가정보/단수처리_규칙.md` §1 관측:
|
||||
같은 「수량 × 단가」라도 **내역서 본체는 절사(ROUNDDOWN), 집계표는 반올림(ROUND)** 이다.
|
||||
즉 단수 함수는 **항목이 아니라 표 종류에 바인딩**된다.
|
||||
|
||||
그래서 이 모듈이 있는 자리 —
|
||||
- **계산 함수 안에서 자르지 않는다.** 계산은 전정밀 `Decimal` 로 내고,
|
||||
**표를 그리는 자리에서** 이 모듈의 함수로 자른다.
|
||||
- ⚠ **집계표(반올림)와 본체(절사)를 더하면 합이 1원 단위로 어긋나는 것이 정상**이다.
|
||||
나중에 「합계가 안 맞는다」는 지적이 반드시 나오는데, **그때 계산을 고치면 안 된다.**
|
||||
어긋남 자체가 규칙이다.
|
||||
|
||||
값은 원문에서 복사하지 않는다 — 자릿수·함수만 여기 두고 근거는 위 문서를 가리킨다.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from decimal import ROUND_CEILING, ROUND_FLOOR, ROUND_HALF_UP, Decimal
|
||||
from enum import Enum
|
||||
|
||||
_ONE = Decimal(1)
|
||||
_THOUSAND = Decimal(1000)
|
||||
|
||||
|
||||
class OutputPlace(str, Enum):
|
||||
"""숫자가 찍히는 자리. 자리마다 단수 함수가 다르다."""
|
||||
|
||||
#: 설계내역서 행 금액(수량×단가) — 절사
|
||||
BOQ_ROW = "boq_row"
|
||||
#: 제경비 각 항목(밑수×율) — 행별 절사
|
||||
OVERHEAD_ROW = "overhead_row"
|
||||
#: 조달수수료(관급×율) — 절사
|
||||
PROCUREMENT_FEE = "procurement_fee"
|
||||
#: 자원 집계표(재료·노무·경비·중기) — **반올림**
|
||||
RESOURCE_SUMMARY = "resource_summary"
|
||||
#: 관급자재대 총액 — **천원 올림**
|
||||
OWNER_MATERIAL_TOTAL = "owner_material_total"
|
||||
|
||||
|
||||
def round_at(value: Decimal, place: OutputPlace) -> Decimal:
|
||||
"""그 자리의 규칙대로 자른다.
|
||||
|
||||
자리를 안 대고 부르는 길을 두지 않는다 — 기본값을 두면 어느 자리인지 모른 채
|
||||
아무 함수나 쓰게 된다.
|
||||
"""
|
||||
if place in (OutputPlace.BOQ_ROW, OutputPlace.OVERHEAD_ROW, OutputPlace.PROCUREMENT_FEE):
|
||||
return value.quantize(_ONE, rounding=ROUND_FLOOR)
|
||||
if place is OutputPlace.RESOURCE_SUMMARY:
|
||||
return value.quantize(_ONE, rounding=ROUND_HALF_UP)
|
||||
if place is OutputPlace.OWNER_MATERIAL_TOTAL:
|
||||
return (value / _THOUSAND).quantize(_ONE, rounding=ROUND_CEILING) * _THOUSAND
|
||||
raise ValueError(f"단수 처리 자리를 모릅니다: {place}")
|
||||
|
||||
|
||||
#: 집계표와 본체를 나란히 보일 때 화면 비고에 다는 문구.
|
||||
#: 「합이 1원 안 맞는다」는 지적에 계산을 고치지 않게 하려는 것이다.
|
||||
SUMMARY_MISMATCH_NOTE = (
|
||||
"집계표는 반올림, 내역서 본체는 절사 — 두 표의 합이 원 단위로 어긋나는 것은 정상입니다."
|
||||
)
|
||||
|
||||
|
||||
def summary_vs_body_gap(summary_total: Decimal, body_total: Decimal) -> Decimal:
|
||||
"""집계표 합계와 내역서 본체 합계의 차이.
|
||||
|
||||
**0 이 아닌 것이 정상**이다. 화면에 그 차이를 숨기지 않고 보여, 설계자가
|
||||
「어긋남이 규칙임」을 알고 넘어가게 한다.
|
||||
"""
|
||||
return summary_total - body_total
|
||||
@@ -0,0 +1,215 @@
|
||||
"""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)
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _detail_from_dict(raw: dict[str, Any]) -> PriceDetail:
|
||||
percent = raw.get("percent_of_parent")
|
||||
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)),
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user