eomsangdonandClaude Opus 5
9bf739fa92
사용자 확정 12번의 마지막 둘. ③(자재값 출처 둘 다)·⑮(유가 전국/지역)와 한 벌. - 서식은 실무 시트(영월 기번6) 그대로 — 원천 5칸(단가+쪽수) + 적용(단가+출처) + 비고. 원천 이름은 코드에 안 박고 PriceBook.slot_names 를 씀(사무소마다 다름, 9-4 미결). - ⚠ 확정 ③ — 슬롯마다 출처를 남길 자리를 둠(쪽수·업체명·날짜). 값이 없는 원천은 0 이 아니라 빈칸. 0 이면 「0원짜리 견적」으로 읽힘. 견적이 국계법 시행령 §9 의 4순위라는 사실을 표가 스스로 밝힘. - ⚠ 확정 ⑮ — 유가 전국/지역 고르는 칸을 냄. 다만 지역값이 아직 자료에 없어 「고를 수 없음 + 까닭」으로 드러냄. 없는 값을 지어내지 않음. - ⚠ 환율및기초자료 인건비 칸은 운전사 셋만. 직종 118개를 다 실으면 노무비목록표와 같은 표가 두 벌이 됨 — 실무 시트도 셋뿐임. - 조종원 시간당이 실무·교본과 다른 사실(× 16/12 × 25/20)을 표 비고에 그대로 냄. 라우터 조회 추가(GET .../estimation/price-sources). 시험 5건 추가, 267건 통과. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
366 lines
16 KiB
Python
366 lines
16 KiB
Python
"""B09 원가계산 라우터 — ⑤ 공사원가계산서 계산 결과를 화면에 낸다.
|
|
|
|
지금은 **무상태 계산 엔드포인트**다. 순공사비를 받아 원가계산서 한 장을 돌려주고,
|
|
저장은 하지 않는다. 프로젝트 저장(채택 단가 스냅샷 `B09_Estimation/v1/`)은 PLAN 9-2
|
|
항목으로 뒤에 붙인다.
|
|
|
|
화면이 「비목 · 금액 · 요율 · 산출근거」 네 칸을 다 보이므로 (PLAN 8-13) 줄마다 그 넷을
|
|
그대로 실어 보낸다. 안전관리비는 A·B 두 줄이 나란히 오고 `note` 에 채택 표시가 붙는다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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_BillOfQuantities import bill_summary, build_bill
|
|
from B09_Estimation.B09_Estimation_Guards import DoubleCountError
|
|
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/base-data")
|
|
async def get_base_data_lists(project_id: UUID) -> JSONResponse:
|
|
"""**기초자료 네 표** — 노무비·재료비·경비 목록표 + 중기목록표 (사용자 확정 12번).
|
|
|
|
별표2 설계서 구성에 드는 표들이라 **없으면 설계서가 성립하지 않는다.** 서식은
|
|
실무 내역서(영월 기번6·봉화 기번41) 같은 이름 시트를 그대로 따랐다.
|
|
"""
|
|
from B09_Estimation.B09_Estimation_Lists import all_lists
|
|
|
|
try:
|
|
return JSONResponse(content={"status": "success", **all_lists(cached_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/price-sources")
|
|
async def get_price_sources(project_id: UUID) -> JSONResponse:
|
|
"""**자재단가대비표(A9) · 환율및기초자료(A10)** — 사용자 확정 12번의 남은 둘.
|
|
|
|
③(자재값 출처 둘 다)·⑮(유가 전국/지역)와 한 벌이라 **출처를 고르는 칸**과
|
|
**업체명·날짜·쪽수 자리**를 함께 낸다.
|
|
"""
|
|
from B09_Estimation.B09_Estimation_Lists_Sources import (
|
|
base_reference_data,
|
|
material_price_comparison,
|
|
)
|
|
|
|
try:
|
|
build = cached_build()
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"material_comparison": material_price_comparison(build),
|
|
"base_reference": base_reference_data(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)})
|
|
|
|
|
|
@router.get("/{project_id}/estimation/bill")
|
|
async def get_bill(project_id: UUID) -> JSONResponse:
|
|
"""④ 예산내역서 한 장 — B08 인계를 그대로 받아 계층을 세워 돌려준다.
|
|
|
|
⚠ **수량을 다시 세지 않는다.** B08 인계가 정본이고 여기서는 단가를 붙여 금액만
|
|
만든다(CLAUDE.md 5장 「같은 계산을 두 벌로 짜지 않는다」).
|
|
|
|
⚠ 단가가 없거나 밑수를 모르는 줄은 **0 으로 안 때우고** `missing` 으로 드러낸다 —
|
|
화면이 그 목록을 그대로 보인다.
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Router_Material import get_handoff
|
|
|
|
try:
|
|
response = await get_handoff(project_id)
|
|
payload = json.loads(bytes(response.body).decode("utf-8"))
|
|
except Exception:
|
|
logger.exception("B09 내역서 조회 실패(인계): project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=502,
|
|
content={"status": "error", "message": "B08 인계 자료를 받지 못했습니다."},
|
|
)
|
|
if "work_items" not in payload:
|
|
# B08 이 오류 응답을 준 경우 — 그 사유를 그대로 넘긴다(감추지 않는다).
|
|
return JSONResponse(status_code=502, content={"status": "error", **payload})
|
|
|
|
try:
|
|
result = build_bill(payload)
|
|
except DoubleCountError as error:
|
|
# 이중계상 감시에 걸린 경우 — 표를 그리지 않고 멈춘다.
|
|
logger.warning("B09 내역서 이중계상 감지: project_id=%s, %s", project_id, error)
|
|
return JSONResponse(status_code=409, 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": "예산내역서를 세우지 못했습니다."},
|
|
)
|
|
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"rows": [row.as_dict() for row in result.rows],
|
|
"excluded": [row.as_dict() for row in result.excluded],
|
|
"materials": [row.as_dict() for row in result.material_rows],
|
|
"summary": bill_summary(result),
|
|
"price_basis": result.price_basis.as_dict() if result.price_basis else {"entries": []},
|
|
}
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}/estimation/price-basis/{code}")
|
|
async def get_price_basis_detail(project_id: UUID, code: str) -> JSONResponse:
|
|
"""③ 단가산출서 한 장 — 그 단가가 무엇을 참조해 나왔는지."""
|
|
from B09_Estimation.B09_Estimation_PriceBasis import price_basis_detail
|
|
|
|
try:
|
|
body = price_basis_detail(code)
|
|
except Exception:
|
|
logger.exception("B09 단가산출서 조회 실패: project_id=%s, code=%s", project_id, code)
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "그 단가산출서를 찾지 못했습니다."},
|
|
)
|
|
return JSONResponse(content={"status": "success", **body})
|