eomsangdonandClaude Opus 5
89effeca07
별표2 (5)(가) 열셋째. 근거는 줄마다 이미 있었고 묶는 자리만 없었음. 원가계산에 「산출기초」 탭으로 세움. 넷으로 접음 — ① 어느 판으로 계산했나(데이터 기준일·지문) ② 무엇을 골랐나(고른 값만) ③ 공종마다 무엇을 근거로 했나(줄 문구 그대로) ④ 못 채운 자리(0 으로 안 때운 자리). - 판 목록은 장부(_manifest.json)를 그대로 읽음 — 목록을 따로 적으면 한쪽만 고쳐짐. - ⚠ 여기서 값을 다시 계산하지 않음 — 금액 칸이 아예 없음(두 벌 방지, 시험으로 못 박음). ⇒ 「설계서 구성」의 산출기초가 반쪽 → 있음. 법이 정한 13 중 9 가 서고, 우리가 더 낼 것은 공사설명서 하나(서식·설계하중 표기는 사용자에게 받아야 함). 곁들여: 자재총괄의 「미분류」를 「미정(발주기관 결정)」으로 고침 — 발주기관이 정할 자리인데 우리가 못 만든 것처럼 읽히던 문구(V-14). 값·판정은 그대로. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
784 lines
36 KiB
Python
784 lines
36 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 = await _build_for(project_id)
|
|
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": "일위대가 목록을 못 만들었습니다."},
|
|
)
|
|
|
|
|
|
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 [주]).
|
|
return cached_build(
|
|
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())),
|
|
)
|
|
|
|
|
|
@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(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={
|
|
"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={"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/factors")
|
|
async def get_factor_choices(project_id: UUID) -> JSONResponse:
|
|
"""**산출 조건** — 품셈이 범위로 준 계수와 장비 규격 (사용자 확정 ① 딸림 지시).
|
|
|
|
「값을 코드에 박고 끝내지 말 것 · 화면에 칸으로 세우고 근거를 보이고 바꿀 수 있게」
|
|
라는 지시대로, **지금 값 · 고를 수 있는 것 · 왜 그 값인지**를 함께 낸다.
|
|
"""
|
|
from B09_Estimation.B09_Estimation_FactorChoices import (
|
|
BASIS_NOTES,
|
|
DEFAULT_CHOICE,
|
|
MACHINE_CHOICES,
|
|
MACHINE_OPTION_CODES,
|
|
machine_choices,
|
|
scan_range_factors,
|
|
)
|
|
from B09_Estimation.B09_Estimation_MachineCost import load_machine_catalog
|
|
from B09_Estimation.B09_Estimation_UnitPrice import load_work_item_master
|
|
from common_util.common_util_project_settings import estimation_settings
|
|
|
|
try:
|
|
root = await _project_root_of(project_id)
|
|
settings = estimation_settings(root) if root else {}
|
|
stored = settings.get("range_factor_choices") or {}
|
|
|
|
ranges = []
|
|
for item in scan_range_factors(load_work_item_master()):
|
|
choice = str(stored.get(item.key) or DEFAULT_CHOICE)
|
|
ranges.append(
|
|
{
|
|
"key": item.key,
|
|
"work_item_code": item.work_item_code,
|
|
"work_item_name": item.work_item_name,
|
|
"factor": item.factor,
|
|
"raw_cell": item.raw_cell,
|
|
"chosen": choice,
|
|
"value": str(item.value_of(choice)),
|
|
"is_default": choice == DEFAULT_CHOICE,
|
|
"options": item.options(),
|
|
"basis": BASIS_NOTES.get(item.key, []),
|
|
}
|
|
)
|
|
|
|
catalog = load_machine_catalog()
|
|
picked = machine_choices(settings)
|
|
machines = []
|
|
for code, entry in MACHINE_CHOICES.items():
|
|
options = []
|
|
for machine_code in MACHINE_OPTION_CODES:
|
|
machine = catalog.machines.get(machine_code)
|
|
if machine is None:
|
|
continue
|
|
options.append(
|
|
{
|
|
"key": machine_code,
|
|
"label": f"{machine.name} {machine.specification}".strip(),
|
|
}
|
|
)
|
|
machines.append(
|
|
{
|
|
"work_item_code": code,
|
|
"work_item_name": entry["work_item_name"],
|
|
"chosen": picked.get(code, entry["default_code"]),
|
|
"default": entry["default_code"],
|
|
"is_default": picked.get(code) == entry["default_code"],
|
|
"source": entry["source"],
|
|
"options": options,
|
|
"basis": entry["basis"],
|
|
}
|
|
)
|
|
|
|
from B09_Estimation.B09_Estimation_PriceBook import PriceKind
|
|
from B09_Estimation.B09_Estimation_UnitPrice import (
|
|
MISC_MATERIAL_MAX_PERCENT,
|
|
MISC_MATERIAL_MIN_PERCENT,
|
|
)
|
|
|
|
from B09_Estimation.B09_Estimation_Transport import ASSUMPTION_TEXT as TRANSPORT_ASSUMPTION
|
|
from B09_Estimation.B09_Estimation_Transport import BASIS_TEXT as TRANSPORT_BASIS
|
|
from B09_Estimation.B09_Estimation_Transport import ROAD_CLASSES as TRANSPORT_ROADS
|
|
from B09_Estimation.B09_Estimation_Transport import TRANSPORT_VARIANTS
|
|
from B09_Estimation.B09_Estimation_Transport import WORK_ITEM_CODE as TRANSPORT_CODE
|
|
|
|
# 「넣을 데가 있는가」 — 주재료비가 선 일위대가가 몇인지 세어 그대로 알린다.
|
|
# 지금은 사급 자재 단가가 미결(확정 5차 큰 것 8)이라 0 이 정상이다.
|
|
from B09_Estimation.B09_Estimation_LaborSurcharge import (
|
|
COMBINE_NOTE as LABOR_SURCHARGE_COMBINE,
|
|
)
|
|
from B09_Estimation.B09_Estimation_LaborSurcharge import SEAT_NOTE as LABOR_SURCHARGE_SEAT
|
|
from B09_Estimation.B09_Estimation_LaborSurcharge import load_series, parse_choices
|
|
from B09_Estimation.B09_Estimation_LaborSurcharge import total_percent
|
|
|
|
LABOR_SURCHARGE_SCOPE = (
|
|
"⚠ 26계열의 [주] 는 대개 조림·숲가꾸기·방제 작업을 지목합니다 — 임도 토공에 붙이라는"
|
|
" 지시가 원문에 없으므로, 각 계열의 [주] 를 보고 그 작업일 때만 고르십시오."
|
|
)
|
|
labor_surcharge_series = load_series()
|
|
labor_surcharge_chosen = parse_choices(settings.get("labor_surcharge"))
|
|
labor_surcharge_total, labor_surcharge_reasons = total_percent(labor_surcharge_chosen)
|
|
|
|
prices = await _build_for(project_id)
|
|
book = prices.book
|
|
transport_notes = list(prices.transport_notes)
|
|
transport_prices = {}
|
|
for variant in TRANSPORT_VARIANTS:
|
|
code = f"B-{TRANSPORT_CODE}#{variant['key']}"
|
|
if code in book.titles:
|
|
transport_prices[variant["key"]] = f"{book.resolve(code).total:,.0f}"
|
|
with_material = sum(
|
|
1
|
|
for unit_code, unit_title in book.titles.items()
|
|
if unit_title.kind is PriceKind.UNIT_PRICE and book.material_base(unit_code) > 0
|
|
)
|
|
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"ranges": ranges,
|
|
"machines": machines,
|
|
"misc_material": {
|
|
"percent": str(settings.get("misc_material_percent") or ""),
|
|
"min": str(MISC_MATERIAL_MIN_PERCENT),
|
|
"max": str(MISC_MATERIAL_MAX_PERCENT),
|
|
"basis": [
|
|
"산림품셈 1-2-6 — 「각 항목에 명시되어 있지 않는 잡재료 및 소모재료 등을"
|
|
" 계상하고자 할 때에는 주재료비(재료비의 할증수량 제외)의 2～5%까지"
|
|
" 별도 계상하되 산정 근거를 명시하여야 한다」",
|
|
"⚠ 비워 두면 안 붙습니다 — 지금은 안 붙고 있는 상태입니다"
|
|
" (사용자 확정 2026-09-09 「지금은 안 넣되 숫자 넣으면 되게 열어 둘 것」).",
|
|
],
|
|
"base_items": with_material,
|
|
"base_note": (
|
|
""
|
|
if with_material
|
|
else "⚠ 지금은 일위대가에 주재료비가 선 공종이 하나도 없습니다"
|
|
" — 자재는 자재대 표에서 따로 금액이 섭니다. 값을 넣어도 붙을 밑수가"
|
|
" 없으므로, 사급 자재 단가가 서는 날 이 칸이 함께 살아납니다."
|
|
),
|
|
},
|
|
"transport": {
|
|
"distance_km": str(settings.get("transport_distance_km") or ""),
|
|
"road": str(settings.get("transport_road") or ""),
|
|
"roads": [
|
|
{"key": row["key"], "label": row["label"]} for row in TRANSPORT_ROADS
|
|
],
|
|
"variants": [
|
|
{
|
|
"key": variant["key"],
|
|
"label": variant["label"],
|
|
"unit_price_krw": transport_prices.get(variant["key"], ""),
|
|
}
|
|
for variant in TRANSPORT_VARIANTS
|
|
],
|
|
"basis": [TRANSPORT_BASIS, TRANSPORT_ASSUMPTION],
|
|
"notes": transport_notes,
|
|
},
|
|
"labor_surcharge": {
|
|
"chosen": labor_surcharge_chosen,
|
|
"total_percent": f"{labor_surcharge_total:g}",
|
|
"reasons": labor_surcharge_reasons,
|
|
"series": [
|
|
{
|
|
"key": item["key"],
|
|
"title": item["title"],
|
|
"section": item["section"],
|
|
"source_note": item["source_note"],
|
|
"options": [
|
|
{
|
|
"key": option["key"],
|
|
"label": f"{option['label']} · {option['percent']:g}%",
|
|
}
|
|
for option in item["options"]
|
|
],
|
|
}
|
|
for item in labor_surcharge_series
|
|
],
|
|
"basis": [LABOR_SURCHARGE_SEAT, LABOR_SURCHARGE_COMBINE, LABOR_SURCHARGE_SCOPE],
|
|
},
|
|
"notes": [
|
|
"고를 수 있는 것은 원문에 적힌 값뿐입니다 — 그 밖의 수는 만들지 않습니다.",
|
|
"바꾸면 그 공종 단가가 바로 달라집니다. [저장]한 값은 이 프로젝트에만 걸립니다.",
|
|
],
|
|
}
|
|
)
|
|
except Exception:
|
|
logger.exception("B09 산출 조건 조회 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "산출 조건을 못 불러왔습니다."},
|
|
)
|
|
|
|
|
|
class FactorChoiceBody(BaseModel):
|
|
"""고른 값 — 안 보낸 칸은 그대로 둔다."""
|
|
|
|
range_factor_choices: dict[str, str] | None = None
|
|
machine_choices: dict[str, str] | None = None
|
|
#: 공구손료·잡재료 비율 — **빈 문자열이면 안 붙는다**(칸을 도로 비우는 길).
|
|
misc_material_percent: str | None = None
|
|
#: 유가 지역(시도코드) — **빈 문자열이면 전국평균**으로 돌아간다.
|
|
fuel_region: str | None = None
|
|
#: 기계 수송 거리(㎞)·도로 구분 — **비면 수송비 줄이 안 선다.**
|
|
transport_distance_km: str | None = None
|
|
transport_road: str | None = None
|
|
#: 품 할인·할증(1-4) — 계열코드 → 고른 행. **빈 값이면 그 계열을 끄는 것.**
|
|
labor_surcharge: dict[str, str] | None = None
|
|
|
|
|
|
@router.put("/{project_id}/estimation/factors")
|
|
async def put_factor_choices(project_id: UUID, body: FactorChoiceBody) -> JSONResponse:
|
|
"""산출 조건을 이 프로젝트에 저장한다. **다른 구획은 손대지 않는다.**"""
|
|
from B09_Estimation.B09_Estimation_FactorChoices import CHOICE_KEYS, MACHINE_OPTION_CODES
|
|
from common_util.common_util_project_settings import save_section
|
|
|
|
from common_util.common_util_project_settings import estimation_settings
|
|
|
|
root = await _project_root_of(project_id)
|
|
if root is None:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
|
)
|
|
values: dict[str, Any] = {}
|
|
if body.range_factor_choices is not None:
|
|
# ⚠ 모르는 값은 안 받는다 — 원문에 없는 수가 설정으로 들어오면 그것이 임의 수치다.
|
|
values["range_factor_choices"] = {
|
|
str(key): str(value)
|
|
for key, value in body.range_factor_choices.items()
|
|
if str(value) in CHOICE_KEYS
|
|
}
|
|
if body.machine_choices is not None:
|
|
values["machine_choices"] = {
|
|
str(key): str(value)
|
|
for key, value in body.machine_choices.items()
|
|
if str(value) in MACHINE_OPTION_CODES
|
|
}
|
|
if body.misc_material_percent is not None:
|
|
from B09_Estimation.B09_Estimation_UnitPrice import parse_misc_material_percent
|
|
|
|
try:
|
|
percent = parse_misc_material_percent(body.misc_material_percent)
|
|
except ValueError as exc:
|
|
# ⚠ 조용히 깎아 넣지 않는다 — 범위 밖 값을 상한으로 접으면 사용자가 넣은 값과
|
|
# 금액이 어긋난 채로 선다.
|
|
return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
|
|
values["misc_material_percent"] = "" if percent is None else str(percent)
|
|
if body.fuel_region is not None:
|
|
from B09_Estimation.B09_Estimation_MachineOperating import load_regional_fuel_table
|
|
|
|
region = str(body.fuel_region).strip()
|
|
table, _ = load_regional_fuel_table()
|
|
if region and region not in table:
|
|
# ⚠ 판에 없는 지역을 받아 두면 조용히 전국평균으로 서고 사용자는 지역값인 줄 안다.
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={
|
|
"status": "error",
|
|
"message": f"유가 판에 없는 지역입니다: {region}",
|
|
},
|
|
)
|
|
values["fuel_region"] = region
|
|
if body.transport_distance_km is not None:
|
|
from B09_Estimation.B09_Estimation_Transport import parse_distance_km
|
|
|
|
try:
|
|
distance = parse_distance_km(body.transport_distance_km)
|
|
except ValueError as exc:
|
|
return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
|
|
values["transport_distance_km"] = "" if distance is None else str(distance)
|
|
if body.transport_road is not None:
|
|
from B09_Estimation.B09_Estimation_Transport import road_class
|
|
|
|
road = str(body.transport_road).strip()
|
|
if road and road_class(road) is None:
|
|
# ⚠ 원문 표에 없는 도로 구분을 받아 두면 속도를 못 골라 조용히 안 선다.
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"status": "error", "message": f"원문에 없는 도로 구분입니다: {road}"},
|
|
)
|
|
values["transport_road"] = road
|
|
if body.labor_surcharge is not None:
|
|
from B09_Estimation.B09_Estimation_LaborSurcharge import parse_choices
|
|
|
|
# ⚠ **원문에 있는 선택지만** 받는다 — 없는 율이 설정으로 들어오면 그것이 임의 수치다.
|
|
stored = dict(estimation_settings(root).get("labor_surcharge") or {})
|
|
for series_key, option_key in body.labor_surcharge.items():
|
|
if str(option_key).strip():
|
|
stored[str(series_key)] = str(option_key)
|
|
else:
|
|
stored.pop(str(series_key), None) # 빈 값 = 그 계열 끄기
|
|
values["labor_surcharge"] = parse_choices(stored)
|
|
try:
|
|
save_section(root, "estimation", values, replace_keys=tuple(values))
|
|
return JSONResponse(content={"status": "success", **values})
|
|
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(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
|
|
|
|
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})
|