"""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_Provenance import estimation_provenance 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, 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"]) def _with_provenance(body: dict[str, Any]) -> dict[str, Any]: """근거 사전을 응답에 얹는다 — **개발환경이 아니면 칸 자체를 안 만든다.** ⚠ 빈 dict 를 실으면 화면이 「사전이 있는데 비었다」로 읽어 빈 카드를 띄운다. 그래서 `None` 이면 **키를 넣지 않는다**(로직 보안의 문은 서버 쪽 하나뿐이다). """ provenance = estimation_provenance() if provenance is not None: body["provenance"] = provenance return body 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) #: 폐기물 분리발주 — 기본 아님(경비로 원가에 듦). 켜면 총원가 밖. waste_separate_order: bool = False 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, waste_separate_order=self.waste_separate_order, 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=_with_provenance({"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 = await _build_for(project_id) return JSONResponse( content=_with_provenance( { "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": "일위대가 목록을 못 만들었습니다."}, ) async def _project_root_of(project_id: UUID) -> str | None: """프로젝트 저장 폴더. 못 찾으면 `None` — 그때는 확정 기본값으로 돈다.""" from common_util.common_util_storage import resolve_stored_project_path from config.config_db import run_with_connection from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path try: stored = await run_with_connection(get_project_storage_relative_path, project_id) return str(resolve_stored_project_path(stored)) except Exception: logger.warning("B09 프로젝트 폴더를 못 찾았습니다 — 기본값으로 돕니다: %s", project_id) return None async def _build_for(project_id: UUID): """그 프로젝트가 **고른 값**으로 조립한 일위대가. ⚠ 범위 계수(작업효율)·장비 규격은 프로젝트마다 다를 수 있다(확정 ①). 전역 한 벌로 돌면 한 프로젝트에서 바꾼 값이 다른 프로젝트 금액까지 흔든다. """ from B09_Estimation.B09_Estimation_LaborSurcharge import parse_choices as parse_labor_surcharge from common_util.common_util_project_settings import estimation_settings root = await _project_root_of(project_id) settings = estimation_settings(root) if root else {} ranges = tuple( sorted((str(k), str(v)) for k, v in (settings.get("range_factor_choices") or {}).items()) ) machines = tuple( sorted((str(k), str(v)) for k, v in (settings.get("machine_choices") or {}).items()) ) # 공구손료·잡재료 — **비어 있는 것이 기본**이라 안 넣으면 줄이 안 선다(확정 5차 작은 것 1). # 유가 지역 — 안 고르면 전국평균(품셈 8-1-7 5호 「해당지역의 가격」). # 기계 수송비 — 거리·도로 구분이 있어야 선다(산림품셈 10-4 [주]). from B09_Estimation.B09_Estimation_Edits import edited_build, edits_key args = ( ranges, machines, str(settings.get("misc_material_percent") or ""), str(settings.get("fuel_region") or ""), str(settings.get("transport_distance_km") or ""), str(settings.get("transport_road") or ""), # 품 할인·할증(1-4) — **안 고르면 안 붙는다.** tuple(sorted(parse_labor_surcharge(settings.get("labor_surcharge")).items())), # 조종원 시간당 노임 자르는 자리 — 실무마다 다름(명세 7장 정정). 안 정하면 원 미만. str(settings.get("operator_wage_digits") or ""), ) # 사용자가 고친 값(PLAN 12장 2차) — 없으면 기본 조립 그 벌 그대로. return edited_build(args, edits_key(settings.get("edits"))) @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=_with_provenance( {"status": "success", **all_lists(await _build_for(project_id))} ) ) 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, ) from common_util.common_util_project_settings import estimation_settings try: build = await _build_for(project_id) root = await _project_root_of(project_id) settings = estimation_settings(root) if root else {} return JSONResponse( content=_with_provenance( { "status": "success", "material_comparison": material_price_comparison(build), "base_reference": base_reference_data( build, str(settings.get("fuel_region") or "") or None ), } ) ) except Exception: logger.exception("B09 단가 원천 표 실패: project_id=%s", project_id) return JSONResponse( status_code=500, content={"status": "error", "message": "단가 원천 표를 못 만들었습니다."}, ) @router.get("/{project_id}/estimation/basis-sheet") async def get_basis_sheet(project_id: UUID) -> JSONResponse: """**산출기초** — 줄에 달린 근거를 한 장으로 모은다(별표2 (5)(가) 열셋째). ⚠ 값을 다시 계산하지 않는다 — 일위대가·내역서가 낸 그대로를 모으기만 한다. """ from B09_Estimation.B09_Estimation_BasisSheet import basis_sheet from common_util.common_util_project_settings import estimation_settings try: build = await _build_for(project_id) root = await _project_root_of(project_id) settings = estimation_settings(root) if root else {} return JSONResponse( content=_with_provenance({"status": "success", **basis_sheet(build, settings)}) ) except Exception: logger.exception("B09 산출기초 실패: project_id=%s", project_id) return JSONResponse( status_code=500, content={"status": "error", "message": "산출기초를 못 만들었습니다."}, ) @router.get("/{project_id}/estimation/machine-expense") async def get_machine_expense(project_id: UUID) -> JSONResponse: """**각종 중기경비계산서** — 기종마다 한 장(별표2 (5)(가) 아홉째). 중기목록표가 「얼마」라면 이 장은 **「왜 그 값인가」**다 — 취득가·손료계수·연료·조종원을 그대로 보인다. """ from B09_Estimation.B09_Estimation_MachineExpenseSheet import machine_expense_report try: build = await _build_for(project_id) return JSONResponse(content={"status": "success", **machine_expense_report(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/design-doc-index") async def get_design_doc_index(project_id: UUID) -> JSONResponse: """**설계서 구성표** — 법이 정한 목차와 우리가 내는 것을 맞대 본다(별표2 (5)(가)). ⚠ 프로젝트마다 달라지는 값이 아니라 **우리가 무엇을 내는가**의 표다. 그래도 프로젝트 화면에서 보여야 설계서를 묶는 사람이 쓴다. """ from B09_Estimation.B09_Estimation_DesignDocIndex import design_doc_index try: return JSONResponse(content={"status": "success", **design_doc_index()}) 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=_with_provenance( {"status": "success", **detail_of(await _build_for(project_id), 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, structure_bill_prices try: # ⚠ **그 프로젝트가 고른 값**으로 조립한 단가표 — 구조물도 미리보기(`_build_for`)와 한 벌. # 기본값(`cached_build()`)을 쓰면 계수를 고른 프로젝트에서 화면과 내역이 갈림(브레인). build = await _build_for(project_id) 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: # 구조물도 호표 금액 — **이 내역과 같은 단가표**로(PLAN 6장 ②). structure_prices = await structure_bill_prices(project_id, build) except Exception: # 못 세면 호표 줄이 금액 없이 「못 받음」 사유로 섬 — 조용히 0 원 안 됨. logger.exception("B09 내역서 — 구조물도 호표 금액 실패: project_id=%s", project_id) structure_prices = {} try: result = build_bill(payload, build=build, structure_prices=structure_prices) 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": "예산내역서를 세우지 못했습니다."}, ) from B09_Estimation.B09_Estimation_Lists import bill_sheets return JSONResponse( content=_with_provenance( { "status": "success", # 집계표 넷·목록표 — 이 내역이 쓴 자원을 처음 쓰인 차례로 되모음(값은 새로 안 셈). **bill_sheets(result, build), "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": []} ), "unit_price_sheet": ( result.unit_price_sheet.as_dict() if result.unit_price_sheet 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, await _build_for(project_id)) 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})