eomsangdonandClaude Opus 5
ca881fbb18
- 공종 수량(코드=수량)을 일위대가에 곱해 직접비 3분할 생성, ⑤ 원가계산서 밑수로 투입 — 뭉치지 않고 재료·노무·경비 성분 그대로 전달 - 단가 없는 공종은 0 으로 안 때우고 `missing_unit_prices` 로 화면 노출 - 계산 원천을 화면에 표시 (「수량 원천: 손입력 / 직접 입력」) - ⑤ 표에 찍히는 값은 자원 집계표 규칙(반올림)으로 절단 — 소수점 유출 제거 - 일위대가 요약에 총액 분포(최소·중앙·최대)·의심 저가 목록 추가 검증: pytest 136 통과, 공용 브라우저(5174) 실측 — 수량 입력 후 재계산 시 재료비 10,478,189 · 순공사원가 84,454,045 · 총공사비 113,652,213, 미매칭 공종 1건 화면 표시 확인 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
249 lines
11 KiB
Python
249 lines
11 KiB
Python
"""B09 원가계산 라우터 — ⑤ 공사원가계산서 계산 결과를 화면에 낸다.
|
|
|
|
지금은 **무상태 계산 엔드포인트**다. 순공사비를 받아 원가계산서 한 장을 돌려주고,
|
|
저장은 하지 않는다. 프로젝트 저장(채택 단가 스냅샷 `B09_Estimation/v1/`)은 PLAN 9-2
|
|
항목으로 뒤에 붙인다.
|
|
|
|
화면이 「비목 · 금액 · 요율 · 산출근거」 네 칸을 다 보이므로 (PLAN 8-13) 줄마다 그 넷을
|
|
그대로 실어 보낸다. 안전관리비는 A·B 두 줄이 나란히 오고 `note` 에 채택 표시가 붙는다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import replace as dataclass_replace
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from B09_Estimation.B09_Estimation_Engine_Cost import (
|
|
CostInput,
|
|
CostResult,
|
|
calculate_cost,
|
|
proposed_profit_adjustment,
|
|
)
|
|
from B09_Estimation.B09_Estimation_PriceBook import PriceBookError
|
|
from B09_Estimation.B09_Estimation_Rates import RateLookupError
|
|
from B09_Estimation.B09_Estimation_Statutory import STATUTORY_ITEMS
|
|
from B09_Estimation.B09_Estimation_Rounding import OutputPlace, round_at
|
|
from B09_Estimation.B09_Estimation_UnitPrice import (
|
|
build_summary,
|
|
cached_build,
|
|
detail_of,
|
|
direct_cost_from_quantities,
|
|
list_unit_prices,
|
|
)
|
|
from common_util.common_util_workflow_state import complete_stage
|
|
from config.config_db import get_db_pool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["B09 Estimation"])
|
|
|
|
|
|
class CostRequest(BaseModel):
|
|
"""원가계산 입력 — 금액은 원 단위."""
|
|
|
|
direct_material_krw: Decimal = Field(default=Decimal(0), ge=0)
|
|
direct_labor_krw: Decimal = Field(default=Decimal(0), ge=0)
|
|
direct_expense_krw: Decimal = Field(default=Decimal(0), ge=0)
|
|
indirect_material_krw: Decimal = Field(default=Decimal(0), ge=0)
|
|
|
|
work_type_indirect_labor: str = "civil"
|
|
work_type_safety: str = "civil"
|
|
duration_days: int = Field(default=183, ge=1)
|
|
pension_year: int = 2026
|
|
|
|
owner_supplied_material_krw: Decimal = Field(default=Decimal(0), ge=0)
|
|
procurement_fee_krw: Decimal = Field(default=Decimal(0), ge=0)
|
|
include_fee_in_owner_material_total: bool = True
|
|
owner_supplied_for_safety_krw: Decimal | None = None
|
|
owner_supplied_includes_vat: bool = True
|
|
|
|
estimated_price_krw: Decimal | None = None
|
|
profit_adjustment_krw: Decimal = Field(default=Decimal(0), ge=0)
|
|
waste_disposal_krw: Decimal = Field(default=Decimal(0), ge=0)
|
|
|
|
environment_work_type: str = "civil_road"
|
|
equipment_guarantee_work_type: str = "civil_general"
|
|
subcontract_guarantee_variant: str = "integrated_civil_or_industrial"
|
|
|
|
rate_file_name: str = "rates_2026.json"
|
|
|
|
#: 공종별 수량 `{공종코드: 수량}`. 주면 **직접비 3분할을 여기서 만들어** 쓴다.
|
|
#: ⚠ 일위대가 합계를 뭉쳐 넣지 않는다 — 밑수가 항목마다 갈린다(PLAN 8-9 규칙 2).
|
|
#: 지금 원천은 **손입력**이고, B08 인계(9번)가 나오면 **원천만 바꿔 끼운다**.
|
|
quantities: dict[str, Decimal] | None = None
|
|
|
|
#: 목표 도급공사비 — 주면 「필요한 이윤 조정액」을 **보여만 준다**.
|
|
#: ★ 법대로(PLAN 8-10) — 프로그램이 스스로 이윤을 깎지 않는다.
|
|
target_contract_amount_krw: Decimal | None = None
|
|
|
|
def to_engine_input(self) -> CostInput:
|
|
return CostInput(
|
|
direct_material_krw=self.direct_material_krw,
|
|
direct_labor_krw=self.direct_labor_krw,
|
|
direct_expense_krw=self.direct_expense_krw,
|
|
indirect_material_krw=self.indirect_material_krw,
|
|
work_type_indirect_labor=self.work_type_indirect_labor,
|
|
work_type_safety=self.work_type_safety,
|
|
duration_days=self.duration_days,
|
|
pension_year=self.pension_year,
|
|
owner_supplied_material_krw=self.owner_supplied_material_krw,
|
|
procurement_fee_krw=self.procurement_fee_krw,
|
|
include_fee_in_owner_material_total=self.include_fee_in_owner_material_total,
|
|
owner_supplied_for_safety_krw=self.owner_supplied_for_safety_krw,
|
|
owner_supplied_includes_vat=self.owner_supplied_includes_vat,
|
|
estimated_price_krw=self.estimated_price_krw,
|
|
profit_adjustment_krw=self.profit_adjustment_krw,
|
|
waste_disposal_krw=self.waste_disposal_krw,
|
|
environment_work_type=self.environment_work_type,
|
|
equipment_guarantee_work_type=self.equipment_guarantee_work_type,
|
|
subcontract_guarantee_variant=self.subcontract_guarantee_variant,
|
|
rate_file_name=self.rate_file_name,
|
|
)
|
|
|
|
|
|
def _serialize(result: CostResult) -> dict[str, Any]:
|
|
"""계산 결과를 화면이 그대로 그릴 수 있는 모양으로 편다."""
|
|
return {
|
|
"lines": [
|
|
{
|
|
"key": line.key,
|
|
"name": line.name,
|
|
"base_label": line.base_label,
|
|
"base_amount_krw": str(line.base_amount_krw),
|
|
"rate_percent": (None if line.rate_percent is None else str(line.rate_percent)),
|
|
"flat_amount_krw": str(line.flat_amount_krw),
|
|
"amount_krw": str(line.amount_krw),
|
|
"formula_text": line.formula_text,
|
|
"note": line.note,
|
|
}
|
|
for line in result.lines
|
|
],
|
|
"totals": {key: str(value) for key, value in result.totals.items()},
|
|
"rate_version": result.rate_version,
|
|
"notes": result.notes,
|
|
}
|
|
|
|
|
|
@router.post("/{project_id}/estimation/cost")
|
|
async def compute_cost(project_id: UUID, payload: CostRequest) -> JSONResponse:
|
|
"""공사원가계산서 한 장을 계산해 돌려준다 (저장 없음)."""
|
|
direct_source = "manual"
|
|
missing_unit_prices: list[str] = []
|
|
try:
|
|
data = payload.to_engine_input()
|
|
if payload.quantities:
|
|
# 수량이 오면 **일위대가에서 직접비 3분할을 만들어** 갈아 끼운다.
|
|
breakdown = direct_cost_from_quantities(payload.quantities)
|
|
# ⑤ 표에 찍히는 자리라 **자원 집계표 규칙(반올림)** 으로 자른다 —
|
|
# 안 자르면 원가계산서에 소수점이 그대로 흘러나온다.
|
|
summary = OutputPlace.RESOURCE_SUMMARY
|
|
data = dataclass_replace(
|
|
data,
|
|
direct_material_krw=round_at(breakdown.material, summary),
|
|
direct_labor_krw=round_at(breakdown.labor, summary),
|
|
direct_expense_krw=round_at(breakdown.expense, summary),
|
|
)
|
|
direct_source = "quantities"
|
|
missing_unit_prices = breakdown.missing
|
|
result = calculate_cost(data)
|
|
except RateLookupError as error:
|
|
# 요율 구간을 못 고른 경우 — 기본값으로 때우지 않고 그대로 알린다.
|
|
logger.warning("B09 원가계산 요율 조회 실패: project_id=%s, %s", project_id, error)
|
|
return JSONResponse(status_code=422, content={"status": "error", "message": str(error)})
|
|
except Exception:
|
|
logger.exception("B09 원가계산 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "원가계산에 실패했습니다."},
|
|
)
|
|
|
|
body = _serialize(result)
|
|
# 어느 값으로 계산했는지 화면이 알아야 한다 — 안 보이면 나중에 못 가른다.
|
|
body["direct_cost_source"] = direct_source
|
|
# 수량은 있는데 단가가 없는 공종 — **화면에 반드시 보인다**.
|
|
body["missing_unit_prices"] = missing_unit_prices
|
|
if payload.target_contract_amount_krw is not None:
|
|
# 필요액을 **보여만 준다**. 적용은 설계자가 `profit_adjustment_krw` 로 명시해야 한다.
|
|
body["suggested_profit_adjustment_krw"] = str(
|
|
proposed_profit_adjustment(result, payload.target_contract_amount_krw)
|
|
)
|
|
return JSONResponse(content={"status": "success", **body})
|
|
|
|
|
|
@router.get("/{project_id}/estimation/items")
|
|
async def list_items(project_id: UUID) -> JSONResponse:
|
|
"""비목 정의 목록 — 화면이 무엇을 켜고 끌 수 있는지 알기 위한 것."""
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"items": [
|
|
{"key": item.key, "name": item.name, "base_label": item.base_label}
|
|
for item in STATUTORY_ITEMS
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}/estimation/unit-prices")
|
|
async def list_unit_price_titles(project_id: UUID) -> JSONResponse:
|
|
"""일위대가 **목록표** — 「무엇이 있나」 한 줄씩 + 산출 요약.
|
|
|
|
요약을 같이 보내는 까닭은 사용자가 **「무엇이 안 선 상태인가」를 화면에서**
|
|
알아야 하기 때문이다(자재 카탈로그 미확보로 구조물 계열이 안 섬).
|
|
"""
|
|
try:
|
|
build = cached_build()
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"summary": build_summary(build),
|
|
"rows": list_unit_prices(build),
|
|
}
|
|
)
|
|
except Exception:
|
|
logger.exception("B09 일위대가 목록 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "일위대가 목록을 못 만들었습니다."},
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}/estimation/unit-prices/{code}")
|
|
async def get_unit_price_detail(project_id: UUID, code: str) -> JSONResponse:
|
|
"""일위대가 **본표** — 「무엇으로 이루어졌나」. 줄마다 원천·파고들기 표시가 붙는다."""
|
|
try:
|
|
return JSONResponse(content={"status": "success", **detail_of(cached_build(), code)})
|
|
except PriceBookError as error:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": str(error)})
|
|
except Exception:
|
|
logger.exception("B09 일위대가 본표 실패: project_id=%s, code=%s", project_id, code)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "일위대가 본표를 못 만들었습니다."},
|
|
)
|
|
|
|
|
|
@router.post("/{project_id}/estimation/confirm")
|
|
async def confirm_estimation(project_id: UUID) -> JSONResponse:
|
|
"""원가계산 단계 확정 — 워크플로 stage 6(ESTIMATION)을 COMPLETE 로 전이한다."""
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
try:
|
|
async with connection.cursor() as cursor:
|
|
await complete_stage(cursor, str(project_id), 6)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
logger.exception("B09 원가계산 확정 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "원가계산 단계 확정에 실패했습니다."},
|
|
)
|
|
return JSONResponse(content={"status": "success", "project_id": str(project_id)})
|