eomsangdonandClaude Opus 5
26d9092a9b
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn
276 lines
12 KiB
Python
276 lines
12 KiB
Python
"""B09 원가계산서 · 제비율 요율표 탭 API (PLAN 12장 · 랩탑 메인).
|
||
|
||
⚠ 옛 `/estimation/cost`(수기 입력)는 그대로 둔다 — 새 탭은 **프로젝트 값**으로 선다:
|
||
직접비 = 예산내역서(`/estimation/bill`) 합계 · 기준 입력 = `estimation.cost_sheet` 저장값 ·
|
||
폐기물처리비 = B08 「임목폐기물 처리」 톤 × 수동 처리단가(분리발주 여부도 B08 산출 조건).
|
||
⚠ 화면이 값을 만들지 않는다 — 서식 줄·금액·산식은 전부 여기서 낸다.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import math
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
from uuid import UUID
|
||
|
||
from fastapi import APIRouter
|
||
from fastapi.responses import JSONResponse
|
||
|
||
from B09_Estimation.B09_Estimation_CostSheet import (
|
||
AMOUNT_FIELDS,
|
||
CHOICE_FIELDS,
|
||
FIELD_HINTS,
|
||
SETTINGS_KEY,
|
||
WORK_TYPE_FIELDS,
|
||
clean_settings,
|
||
cost_input,
|
||
field_options,
|
||
sheet_rows,
|
||
status_line,
|
||
)
|
||
from B09_Estimation.B09_Estimation_CostSheet_Forms import FORM_LABELS, form_rows
|
||
from B09_Estimation.B09_Estimation_Engine_Cost import calculate_cost
|
||
from B09_Estimation.B09_Estimation_RateOverride import apply_overrides, clean_overrides
|
||
from B09_Estimation.B09_Estimation_Rates import RateLookupError, load_rate_dataset
|
||
from B09_Estimation.B09_Estimation_RateTable import rate_sections
|
||
from B09_Estimation.B09_Estimation_Rounding import OutputPlace, round_at
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/projects", tags=["B09 Estimation — Cost Sheet"])
|
||
|
||
TREE_WASTE_ITEM = "임목폐기물 처리"
|
||
|
||
|
||
async def _root(project_id: UUID) -> str | None:
|
||
from B09_Estimation.B09_Estimation_Router import _project_root_of
|
||
|
||
return await _project_root_of(project_id)
|
||
|
||
|
||
def _waste(bill: dict[str, Any], quantity: dict[str, Any]) -> tuple[Decimal, bool, str]:
|
||
"""(금액, 분리발주, 사유) — 톤은 내역서 제외 줄에서, 단가는 B08 수동 입력에서."""
|
||
tons = next(
|
||
(
|
||
Decimal(str(row.get("quantity") or 0))
|
||
for row in bill.get("excluded") or []
|
||
if row.get("name") == TREE_WASTE_ITEM
|
||
),
|
||
Decimal(0),
|
||
)
|
||
price = quantity.get("tree_waste_unit_price_krw_per_ton")
|
||
separate = bool(quantity.get("waste_separate_order"))
|
||
if tons <= 0:
|
||
return Decimal(0), separate, "임목폐기물 톤이 아직 안 섬(수량산출 산출 조건)"
|
||
if not price:
|
||
return Decimal(0), separate, f"임목폐기물 {tons:,.2f}톤 — 처리단가 없음(수동 입력 전)"
|
||
amount = Decimal(math.floor(tons * Decimal(str(price))))
|
||
return amount, separate, f"임목폐기물 {tons:,.2f}톤 × {price:,.0f}원 — ⚠ 수동 단가(미확정)"
|
||
|
||
|
||
def cost_from_bill(root: str, bill: dict[str, Any], direct: dict[str, Decimal]):
|
||
"""(엔진 입력, 원가계산 결과, 폐기물 사유) — 직접비만 갈아 끼우면 계약·기성도 같은 길로 섬.
|
||
|
||
기준 입력·요율 덮어쓰기·폐기물 톤은 프로젝트 저장값 그대로. `RateLookupError` 는 부른 쪽이 받음.
|
||
"""
|
||
from common_util.common_util_project_settings import estimation_settings, quantity_settings
|
||
|
||
stored = dict(estimation_settings(root).get(SETTINGS_KEY) or {})
|
||
waste, separate, waste_note = _waste(bill, quantity_settings(root))
|
||
overrides = tuple(estimation_settings(root).get(OVERRIDES_KEY) or ())
|
||
data = cost_input(direct, stored, waste, separate, overrides)
|
||
return data, calculate_cost(data), waste_note
|
||
|
||
|
||
@router.get("/{project_id}/estimation/cost-sheet")
|
||
async def get_cost_sheet(project_id: UUID, form: str = "general") -> JSONResponse:
|
||
"""원가계산서 한 장 — 서식 줄 · 기준 입력 선택지 · 상태줄.
|
||
|
||
`form` — 일반형식만 금액을 냄. 수공·실적일반·실적수공은 **서식 차례만**(실적형 임도 미적용 ·
|
||
수공 세부 경비 법정 요율 없음 줄은 막고 사유 · 0 원으로 안 채움).
|
||
"""
|
||
from B09_Estimation.B09_Estimation_Router import get_bill
|
||
from common_util.common_util_project_settings import estimation_settings
|
||
|
||
root = await _root(project_id)
|
||
if root is None:
|
||
return JSONResponse(
|
||
status_code=404,
|
||
content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
||
)
|
||
response = await get_bill(project_id)
|
||
bill = json.loads(bytes(response.body).decode("utf-8"))
|
||
if response.status_code != 200 or "summary" not in bill:
|
||
# 내역서가 안 서면 원가계산서도 못 섬 — 사유를 그대로 넘긴다.
|
||
return JSONResponse(status_code=response.status_code or 502, content=bill)
|
||
summary = bill["summary"]
|
||
place = OutputPlace.RESOURCE_SUMMARY
|
||
direct = {
|
||
part: round_at(Decimal(str(summary.get(f"direct_{part}_krw") or 0)), place)
|
||
for part in ("material", "labor", "expense")
|
||
}
|
||
stored = dict(estimation_settings(root).get(SETTINGS_KEY) or {})
|
||
overrides = tuple(estimation_settings(root).get(OVERRIDES_KEY) or ())
|
||
try:
|
||
data, result, waste_note = cost_from_bill(root, bill, direct)
|
||
dataset = load_rate_dataset(data.rate_file_name)
|
||
except RateLookupError as error:
|
||
return JSONResponse(status_code=422, content={"status": "error", "message": str(error)})
|
||
return JSONResponse(
|
||
content={
|
||
"status": "success",
|
||
"form": "일반",
|
||
"rows": sheet_rows(result, data),
|
||
"status_line": status_line(result, data),
|
||
"settings": {
|
||
**{key: str(getattr(data, key)) for key, _ in (*WORK_TYPE_FIELDS, *CHOICE_FIELDS)},
|
||
**{key: str(getattr(data, key)) for key, _ in AMOUNT_FIELDS},
|
||
"duration_days": data.duration_days,
|
||
},
|
||
"stored": stored,
|
||
"fields": {
|
||
"work_types": [{"key": k, "label": label} for k, label in WORK_TYPE_FIELDS],
|
||
"choices": [{"key": k, "label": label} for k, label in CHOICE_FIELDS],
|
||
"hints": FIELD_HINTS,
|
||
"amounts": [{"key": k, "label": label} for k, label in AMOUNT_FIELDS],
|
||
"options": field_options(dataset),
|
||
},
|
||
"waste_note": waste_note,
|
||
"bill_missing_count": len(summary.get("missing") or []),
|
||
"bill_unconfirmed_count": int(summary.get("unconfirmed_count") or 0),
|
||
"rate_version": result.rate_version,
|
||
"notes": result.notes,
|
||
"rate_override_count": len(overrides),
|
||
"form_key": "general",
|
||
"form_labels": FORM_LABELS,
|
||
**(_other_form(form) if form != "general" else {}),
|
||
}
|
||
)
|
||
|
||
|
||
@router.put("/{project_id}/estimation/cost-sheet")
|
||
async def put_cost_sheet(project_id: UUID, body: dict[str, Any]) -> JSONResponse:
|
||
"""기준 입력 저장 — `estimation.cost_sheet` 한 칸만 통째로 갈아 끼움."""
|
||
from common_util.common_util_project_settings import save_section
|
||
|
||
root = await _root(project_id)
|
||
if root is None:
|
||
return JSONResponse(
|
||
status_code=404,
|
||
content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
||
)
|
||
cleaned = clean_settings(body, load_rate_dataset())
|
||
try:
|
||
save_section(root, "estimation", {SETTINGS_KEY: cleaned}, replace_keys=(SETTINGS_KEY,))
|
||
except Exception:
|
||
logger.exception("B09 원가계산서 기준 저장 실패: project_id=%s", project_id)
|
||
return JSONResponse(
|
||
status_code=500,
|
||
content={"status": "error", "message": "원가계산서 기준을 저장하지 못했습니다."},
|
||
)
|
||
return JSONResponse(content={"status": "success", "stored": cleaned})
|
||
|
||
|
||
#: 프로젝트 요율 덮어쓰기 저장 자리 — `estimation` 구획 안 한 칸(마스터 요율 파일은 안 건드림).
|
||
OVERRIDES_KEY = "rate_overrides"
|
||
|
||
|
||
def _other_form(form: str) -> dict[str, Any]:
|
||
"""나머지 세 형식 — 금액 없는 서식 줄로 갈아 끼움(기준 입력판 자료는 그대로)."""
|
||
if form not in FORM_LABELS:
|
||
return {"form_error": f"모르는 형식입니다: {form}"}
|
||
return {
|
||
"form_key": form,
|
||
"form": FORM_LABELS[form],
|
||
"rows": form_rows(form),
|
||
"status_line": {},
|
||
"form_note": (
|
||
"금액 미산출 — 이 형식의 요율표(수공 세부 경비·표준시장단가 제비율)가 없어"
|
||
" 서식 차례만 보임"
|
||
),
|
||
}
|
||
|
||
|
||
@router.get("/{project_id}/estimation/rate-table")
|
||
async def get_rate_table(project_id: UUID) -> JSONResponse:
|
||
"""제비율 요율표 — 마스터 요율에 **이 프로젝트의 덮어쓰기**를 얹어 DFM 첫 탭 차례로.
|
||
|
||
칸마다 기본값(마스터)·지금 값·주소를 실어 화면이 갈라 보이고 고칠 수 있게 한다.
|
||
"""
|
||
from common_util.common_util_project_settings import estimation_settings
|
||
|
||
root = await _root(project_id)
|
||
master = load_rate_dataset()
|
||
overrides = list((estimation_settings(root) if root else {}).get(OVERRIDES_KEY) or [])
|
||
try:
|
||
current = apply_overrides(master, overrides)
|
||
except RateLookupError as error:
|
||
# 판이 바뀌어 주소가 안 맞는 덮어쓰기 — 조용히 버리지 않고 알림(마스터 값으로 보임).
|
||
return JSONResponse(
|
||
content={
|
||
"status": "success",
|
||
"rate_version": master.version_stamp,
|
||
"sections": rate_sections(master),
|
||
"overrides": overrides,
|
||
"override_error": str(error),
|
||
}
|
||
)
|
||
return JSONResponse(
|
||
content={
|
||
"status": "success",
|
||
"rate_version": master.version_stamp,
|
||
"sections": rate_sections(current, master),
|
||
"overrides": overrides,
|
||
"override_error": "",
|
||
}
|
||
)
|
||
|
||
|
||
@router.put("/{project_id}/estimation/rate-table")
|
||
async def put_rate_table(project_id: UUID, body: dict[str, Any]) -> JSONResponse:
|
||
"""덮어쓰기 저장 — 목록을 **통째로** 갈아 끼움(빈 목록 = 전부 기본값으로 되돌림).
|
||
|
||
⚠ 한 줄이라도 사유가 비거나 주소가 틀리면 **아무것도 안 저장**하고 까닭을 돌려줌.
|
||
"""
|
||
from datetime import datetime
|
||
|
||
from common_util.common_util_project_settings import estimation_settings, save_section
|
||
|
||
root = await _root(project_id)
|
||
if root is None:
|
||
return JSONResponse(
|
||
status_code=404,
|
||
content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
||
)
|
||
kept, errors = clean_overrides(body.get("overrides"), load_rate_dataset())
|
||
if errors:
|
||
return JSONResponse(
|
||
status_code=400,
|
||
content={"status": "error", "message": " · ".join(errors), "errors": errors},
|
||
)
|
||
previous = {
|
||
(o.get("variable"), o.get("table"), str(sorted((o.get("match") or {}).items()))): o
|
||
for o in estimation_settings(root).get(OVERRIDES_KEY) or []
|
||
}
|
||
now = datetime.now().isoformat(timespec="seconds")
|
||
for entry in kept:
|
||
# 값·사유가 그대로면 입력 시각을 지킴 — 「언제 고쳤나」가 저장할 때마다 바뀌지 않게.
|
||
before = previous.get(
|
||
(entry["variable"], entry["table"], str(sorted(entry["match"].items())))
|
||
)
|
||
same = before and (before.get("rate_percent"), before.get("reason")) == (
|
||
entry["rate_percent"],
|
||
entry["reason"],
|
||
)
|
||
entry["entered_at"] = before.get("entered_at") if same else now
|
||
try:
|
||
save_section(root, "estimation", {OVERRIDES_KEY: kept}, replace_keys=(OVERRIDES_KEY,))
|
||
except Exception:
|
||
logger.exception("B09 요율 덮어쓰기 저장 실패: project_id=%s", project_id)
|
||
return JSONResponse(
|
||
status_code=500,
|
||
content={"status": "error", "message": "요율 덮어쓰기를 저장하지 못했습니다."},
|
||
)
|
||
return JSONResponse(content={"status": "success", "overrides": kept})
|