Files
Aislo/B09_Estimation/B09_Estimation_Router_Execution.py
T
eomsangdon 2ab3a65807 @
feat(b09): 실행예산 차액 보정 뜻을 화면에 — 기본보정은 「안 몰고 남김(뜻 확인 대기)」

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn
@
2026-09-14 05:47:36 +09:00

97 lines
3.8 KiB
Python

"""B09 실행예산 단계 탭 API — 설계 내역 → 실행예산 (PLAN 12장 · 랩탑 메인).
⚠ 설계 내역·단가표를 **읽기만** 한다(복사본에 중기 실행단가·실행수량을 얹음). 저장은
`estimation.execution` 한 칸 — 설계 내역·계약·원가계산서 저장본은 안 건드린다.
"""
from __future__ import annotations
import json
import logging
from typing import Any
from uuid import UUID
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from B09_Estimation.B09_Estimation_Execution import (
CORRECTION_HINTS,
CORRECTIONS,
CUT_UNITS,
SETTINGS_KEY,
clean_settings,
execution_bill,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B09 Estimation — Execution"])
_NOT_FOUND = {"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."}
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)
@router.get("/{project_id}/estimation/execution")
async def get_execution(project_id: UUID) -> JSONResponse:
"""실행예산 한 장 — 설계 줄 옆에 실행수량·실행단가·실행금액 · 중기 실행단가 표 · 합계."""
from B09_Estimation.B09_Estimation_Router import _build_for, 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=_NOT_FOUND)
response = await get_bill(project_id)
bill = json.loads(bytes(response.body).decode("utf-8"))
if response.status_code != 200 or "rows" not in bill:
return JSONResponse(status_code=response.status_code or 502, content=bill)
stored, _ = clean_settings(dict(estimation_settings(root).get(SETTINGS_KEY) or {}))
# 조립본은 캐시 공유본 — `execution_bill` 이 단가표 복사본에만 중기 실행단가를 얹음(설계 불변).
result = execution_bill(bill["rows"], stored, build=await _build_for(project_id))
return JSONResponse(
content={
"status": "success",
**result,
"settings": stored,
"fields": {
"cut_units": list(CUT_UNITS),
"corrections": [
{"key": k, "label": label, "hint": CORRECTION_HINTS[k]}
for k, label in CORRECTIONS
],
},
"bill_missing_count": len((bill.get("summary") or {}).get("missing") or []),
"limit_note": (
"실행예산 표본 0건 — 구조가 서는지까지만 확인됨 · 성분 가르기·기본보정은 확인 대기"
),
}
)
@router.put("/{project_id}/estimation/execution")
async def put_execution(project_id: UUID, body: dict[str, Any]) -> JSONResponse:
"""중기 실행단가 입력·실행수량 저장 — 틀린 칸이 있으면 아무것도 안 저장."""
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=_NOT_FOUND)
cleaned, errors = clean_settings(body)
if errors:
return JSONResponse(
status_code=400,
content={"status": "error", "message": " · ".join(errors), "errors": errors},
)
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", "settings": cleaned})