refactor(B08,B09): 폴더째 old_code 로 옮기고 빈 화면 둘만 남김 (PLAN 7-3)
B08_Quantity 86 · B09_Estimation 111 파일을 old_code/ 로 옮김(지우지 않음). 화면은 메뉴·주소·단계 막대만 남은 빈 틀 둘 — main.py 라우터 13 개는 끊음. B07 이 빌려 쓰던 비탈 길이·면적은 필요한 함수만 B07_DesignDetail_Engine_SlopeGeometry 로 옮겨 적음(면적 적분·노면 면적·측점 묶음은 안 옮김) · 시험 하나를 새로 둠. B07 구조물도 조립(Cad_StandardSheet)은 2026-09-13 에 이미 도면 목록에서 빠져 부르는 곳이 없어 old_code 로 같이 보냄 — 구조물 그림은 되살리지 않음. B06 구조물 몫 조회는 빈 값으로 두어 화면이 그대로 서게 함. B08·B09 를 부르던 시험 25 개도 old_code/resources/tester 로 옮김. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PAdA5ThqmtVSsbzjusk1cJ
This commit is contained in:
@@ -0,0 +1,275 @@
|
||||
"""B09 원가계산서 · 제비율 요율표 탭 API (PLAN 12장 · 랩탑 메인).
|
||||
|
||||
⚠ 옛 `/estimation/cost`(수기 입력)는 그대로 둔다 — 새 탭은 **프로젝트 값**으로 선다:
|
||||
직접비 = 예산내역서(`/estimation/bill`) 합계 · 기준 입력 = `estimation.cost_sheet` 저장값 ·
|
||||
폐기물처리비 = B08 「임목폐기물 처리」 톤 × 수동 처리단가(분리발주 여부도 B08 산출 조건).
|
||||
⚠ 화면이 값을 만들지 않는다 — 서식 줄·금액·산식은 전부 여기서 낸다.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from B09_Estimation.B09_Estimation_CostSheet import (
|
||||
AMOUNT_FIELDS,
|
||||
CHOICE_FIELDS,
|
||||
FIELD_HINTS,
|
||||
SETTINGS_KEY,
|
||||
WORK_TYPE_FIELDS,
|
||||
clean_settings,
|
||||
cost_input,
|
||||
field_options,
|
||||
sheet_rows,
|
||||
status_line,
|
||||
)
|
||||
from B09_Estimation.B09_Estimation_CostSheet_Forms import FORM_LABELS, form_rows
|
||||
from B09_Estimation.B09_Estimation_Engine_Cost import calculate_cost
|
||||
from B09_Estimation.B09_Estimation_RateOverride import apply_overrides, clean_overrides
|
||||
from B09_Estimation.B09_Estimation_Rates import RateLookupError, load_rate_dataset
|
||||
from B09_Estimation.B09_Estimation_RateTable import rate_sections
|
||||
from B09_Estimation.B09_Estimation_Rounding import OutputPlace, round_at
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api/projects", tags=["B09 Estimation — Cost Sheet"])
|
||||
|
||||
TREE_WASTE_ITEM = "임목폐기물 처리"
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
def _waste(bill: dict[str, Any], quantity: dict[str, Any]) -> tuple[Decimal, bool, str]:
|
||||
"""(금액, 분리발주, 사유) — 톤은 내역서 제외 줄에서, 단가는 B08 수동 입력에서."""
|
||||
tons = next(
|
||||
(
|
||||
Decimal(str(row.get("quantity") or 0))
|
||||
for row in bill.get("excluded") or []
|
||||
if row.get("name") == TREE_WASTE_ITEM
|
||||
),
|
||||
Decimal(0),
|
||||
)
|
||||
price = quantity.get("tree_waste_unit_price_krw_per_ton")
|
||||
separate = bool(quantity.get("waste_separate_order"))
|
||||
if tons <= 0:
|
||||
return Decimal(0), separate, "임목폐기물 톤이 아직 안 섬(수량산출 산출 조건)"
|
||||
if not price:
|
||||
return Decimal(0), separate, f"임목폐기물 {tons:,.2f}톤 — 처리단가 없음(수동 입력 전)"
|
||||
amount = Decimal(math.floor(tons * Decimal(str(price))))
|
||||
return amount, separate, f"임목폐기물 {tons:,.2f}톤 × {price:,.0f}원 — ⚠ 수동 단가(미확정)"
|
||||
|
||||
|
||||
def cost_from_bill(root: str, bill: dict[str, Any], direct: dict[str, Decimal]):
|
||||
"""(엔진 입력, 원가계산 결과, 폐기물 사유) — 직접비만 갈아 끼우면 계약·기성도 같은 길로 섬.
|
||||
|
||||
기준 입력·요율 덮어쓰기·폐기물 톤은 프로젝트 저장값 그대로. `RateLookupError` 는 부른 쪽이 받음.
|
||||
"""
|
||||
from common_util.common_util_project_settings import estimation_settings, quantity_settings
|
||||
|
||||
stored = dict(estimation_settings(root).get(SETTINGS_KEY) or {})
|
||||
waste, separate, waste_note = _waste(bill, quantity_settings(root))
|
||||
overrides = tuple(estimation_settings(root).get(OVERRIDES_KEY) or ())
|
||||
data = cost_input(direct, stored, waste, separate, overrides)
|
||||
return data, calculate_cost(data), waste_note
|
||||
|
||||
|
||||
@router.get("/{project_id}/estimation/cost-sheet")
|
||||
async def get_cost_sheet(project_id: UUID, form: str = "general") -> JSONResponse:
|
||||
"""원가계산서 한 장 — 서식 줄 · 기준 입력 선택지 · 상태줄.
|
||||
|
||||
`form` — 일반형식만 금액을 냄. 수공·실적일반·실적수공은 **서식 차례만**(실적형 임도 미적용 ·
|
||||
수공 세부 경비 법정 요율 없음 줄은 막고 사유 · 0 원으로 안 채움).
|
||||
"""
|
||||
from B09_Estimation.B09_Estimation_Router import 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={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
||||
)
|
||||
response = await get_bill(project_id)
|
||||
bill = json.loads(bytes(response.body).decode("utf-8"))
|
||||
if response.status_code != 200 or "summary" not in bill:
|
||||
# 내역서가 안 서면 원가계산서도 못 섬 — 사유를 그대로 넘긴다.
|
||||
return JSONResponse(status_code=response.status_code or 502, content=bill)
|
||||
summary = bill["summary"]
|
||||
place = OutputPlace.RESOURCE_SUMMARY
|
||||
direct = {
|
||||
part: round_at(Decimal(str(summary.get(f"direct_{part}_krw") or 0)), place)
|
||||
for part in ("material", "labor", "expense")
|
||||
}
|
||||
stored = dict(estimation_settings(root).get(SETTINGS_KEY) or {})
|
||||
overrides = tuple(estimation_settings(root).get(OVERRIDES_KEY) or ())
|
||||
try:
|
||||
data, result, waste_note = cost_from_bill(root, bill, direct)
|
||||
dataset = load_rate_dataset(data.rate_file_name)
|
||||
except RateLookupError as error:
|
||||
return JSONResponse(status_code=422, content={"status": "error", "message": str(error)})
|
||||
return JSONResponse(
|
||||
content={
|
||||
"status": "success",
|
||||
"form": "일반",
|
||||
"rows": sheet_rows(result, data),
|
||||
"status_line": status_line(result, data),
|
||||
"settings": {
|
||||
**{key: str(getattr(data, key)) for key, _ in (*WORK_TYPE_FIELDS, *CHOICE_FIELDS)},
|
||||
**{key: str(getattr(data, key)) for key, _ in AMOUNT_FIELDS},
|
||||
"duration_days": data.duration_days,
|
||||
},
|
||||
"stored": stored,
|
||||
"fields": {
|
||||
"work_types": [{"key": k, "label": label} for k, label in WORK_TYPE_FIELDS],
|
||||
"choices": [{"key": k, "label": label} for k, label in CHOICE_FIELDS],
|
||||
"hints": FIELD_HINTS,
|
||||
"amounts": [{"key": k, "label": label} for k, label in AMOUNT_FIELDS],
|
||||
"options": field_options(dataset),
|
||||
},
|
||||
"waste_note": waste_note,
|
||||
"bill_missing_count": len(summary.get("missing") or []),
|
||||
"bill_unconfirmed_count": int(summary.get("unconfirmed_count") or 0),
|
||||
"rate_version": result.rate_version,
|
||||
"notes": result.notes,
|
||||
"rate_override_count": len(overrides),
|
||||
"form_key": "general",
|
||||
"form_labels": FORM_LABELS,
|
||||
**(_other_form(form) if form != "general" else {}),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.put("/{project_id}/estimation/cost-sheet")
|
||||
async def put_cost_sheet(project_id: UUID, body: dict[str, Any]) -> JSONResponse:
|
||||
"""기준 입력 저장 — `estimation.cost_sheet` 한 칸만 통째로 갈아 끼움."""
|
||||
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={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
||||
)
|
||||
cleaned = clean_settings(body, load_rate_dataset())
|
||||
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", "stored": cleaned})
|
||||
|
||||
|
||||
#: 프로젝트 요율 덮어쓰기 저장 자리 — `estimation` 구획 안 한 칸(마스터 요율 파일은 안 건드림).
|
||||
OVERRIDES_KEY = "rate_overrides"
|
||||
|
||||
|
||||
def _other_form(form: str) -> dict[str, Any]:
|
||||
"""나머지 세 형식 — 금액 없는 서식 줄로 갈아 끼움(기준 입력판 자료는 그대로)."""
|
||||
if form not in FORM_LABELS:
|
||||
return {"form_error": f"모르는 형식입니다: {form}"}
|
||||
return {
|
||||
"form_key": form,
|
||||
"form": FORM_LABELS[form],
|
||||
"rows": form_rows(form),
|
||||
"status_line": {},
|
||||
"form_note": (
|
||||
"금액 미산출 — 이 형식의 요율표(수공 세부 경비·표준시장단가 제비율)가 없어"
|
||||
" 서식 차례만 보임"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@router.get("/{project_id}/estimation/rate-table")
|
||||
async def get_rate_table(project_id: UUID) -> JSONResponse:
|
||||
"""제비율 요율표 — 마스터 요율에 **이 프로젝트의 덮어쓰기**를 얹어 DFM 첫 탭 차례로.
|
||||
|
||||
칸마다 기본값(마스터)·지금 값·주소를 실어 화면이 갈라 보이고 고칠 수 있게 한다.
|
||||
"""
|
||||
from common_util.common_util_project_settings import estimation_settings
|
||||
|
||||
root = await _root(project_id)
|
||||
master = load_rate_dataset()
|
||||
overrides = list((estimation_settings(root) if root else {}).get(OVERRIDES_KEY) or [])
|
||||
try:
|
||||
current = apply_overrides(master, overrides)
|
||||
except RateLookupError as error:
|
||||
# 판이 바뀌어 주소가 안 맞는 덮어쓰기 — 조용히 버리지 않고 알림(마스터 값으로 보임).
|
||||
return JSONResponse(
|
||||
content={
|
||||
"status": "success",
|
||||
"rate_version": master.version_stamp,
|
||||
"sections": rate_sections(master),
|
||||
"overrides": overrides,
|
||||
"override_error": str(error),
|
||||
}
|
||||
)
|
||||
return JSONResponse(
|
||||
content={
|
||||
"status": "success",
|
||||
"rate_version": master.version_stamp,
|
||||
"sections": rate_sections(current, master),
|
||||
"overrides": overrides,
|
||||
"override_error": "",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.put("/{project_id}/estimation/rate-table")
|
||||
async def put_rate_table(project_id: UUID, body: dict[str, Any]) -> JSONResponse:
|
||||
"""덮어쓰기 저장 — 목록을 **통째로** 갈아 끼움(빈 목록 = 전부 기본값으로 되돌림).
|
||||
|
||||
⚠ 한 줄이라도 사유가 비거나 주소가 틀리면 **아무것도 안 저장**하고 까닭을 돌려줌.
|
||||
"""
|
||||
from datetime import datetime
|
||||
|
||||
from common_util.common_util_project_settings import estimation_settings, save_section
|
||||
|
||||
root = await _root(project_id)
|
||||
if root is None:
|
||||
return JSONResponse(
|
||||
status_code=404,
|
||||
content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
||||
)
|
||||
kept, errors = clean_overrides(body.get("overrides"), load_rate_dataset())
|
||||
if errors:
|
||||
return JSONResponse(
|
||||
status_code=400,
|
||||
content={"status": "error", "message": " · ".join(errors), "errors": errors},
|
||||
)
|
||||
previous = {
|
||||
(o.get("variable"), o.get("table"), str(sorted((o.get("match") or {}).items()))): o
|
||||
for o in estimation_settings(root).get(OVERRIDES_KEY) or []
|
||||
}
|
||||
now = datetime.now().isoformat(timespec="seconds")
|
||||
for entry in kept:
|
||||
# 값·사유가 그대로면 입력 시각을 지킴 — 「언제 고쳤나」가 저장할 때마다 바뀌지 않게.
|
||||
before = previous.get(
|
||||
(entry["variable"], entry["table"], str(sorted(entry["match"].items())))
|
||||
)
|
||||
same = before and (before.get("rate_percent"), before.get("reason")) == (
|
||||
entry["rate_percent"],
|
||||
entry["reason"],
|
||||
)
|
||||
entry["entered_at"] = before.get("entered_at") if same else now
|
||||
try:
|
||||
save_section(root, "estimation", {OVERRIDES_KEY: kept}, replace_keys=(OVERRIDES_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", "overrides": kept})
|
||||
Reference in New Issue
Block a user