eomsangdonandClaude Opus 5
0bb14ce2b3
화면: 복제로 만든 개인 항목 선택 → 동료 목록 받음 → 확인창 「받는 쪽 목록에 받음으로 뜸 · 그 사람 것은 안 덮음」 · 공용 storage 라 취소(동료 폴더 안 생김 확인) · 개인 항목 지워 되돌림 · 쓰기는 임시 storage 시험 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VBGFXB9AbJBwXP19z75Qq
701 lines
30 KiB
Python
701 lines
30 KiB
Python
"""B08 구조물도(표준도) 라우터 — 장 목록 조회와 **제원 입력**.
|
|
|
|
⚠ **2026-09-13 B07 에서 이관**(PLAN 3장 · 사용자 확정 ③ 「구조물도는 B08 로」).
|
|
옛 자리 `B07_DesignDetail_Router_Standard.py` · 옛 주소 `/standard-sheets`.
|
|
|
|
표준도는 도면이자 **입력 화면**이다(PLAN 4-5b). `phase: "detail"` 칸(돌 종류·조달·뒷길이·
|
|
전면 기울기)을 그리는 화면이 없어(2026-09-09 실측) 그 자리를 여기가 맡는다.
|
|
**장 하나 = 제원 조합 하나**라 한 번 고치면 그 조합의 개소 전부에 걸린다.
|
|
|
|
⚠ 값을 여기서 셈하지 않는다 — 정본(`structures.json`)에 적기만 하고 표·그림은 다음 조회에서
|
|
그 정본으로 다시 선다. 표준도가 두 번째 정본이 되면 안 된다(CLAUDE.md 5장).
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from common_util.common_util_auth import verify_session
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
from config.config_db import run_with_connection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"])
|
|
|
|
|
|
async def _project_root(project_id: UUID) -> str | None:
|
|
"""프로젝트 저장 폴더(절대경로). 못 찾으면 `None`."""
|
|
try:
|
|
stored_path = await run_with_connection(get_project_storage_relative_path, project_id)
|
|
return str(Path(resolve_stored_project_path(stored_path)).resolve())
|
|
except Exception:
|
|
logger.exception("B08 구조물도 — 저장 폴더 조회 실패: project_id=%s", project_id)
|
|
return None
|
|
|
|
|
|
def project_structure_sheets(
|
|
project_root: str,
|
|
section_modes: dict[float, str] | None,
|
|
ground_types: dict[float, str] | None = None,
|
|
designs: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""구조물도 장 목록 — **원단위 탭(`material-summary`)과 같은 입력**으로 전개해 접는다.
|
|
|
|
⚠ 기초잡석 두께를 안 넘기면 산출 조건에서 바꿔도 **구조물도만 옛 두께(0.2)** 로 선다
|
|
(2026-09-13 이관 때 잡음 — 옛 B07 창구가 두께·지반 갈래를 안 넘겼음).
|
|
⚠ 늦게 부른다(함수 안 import) — B08 은 B05 를 부르고 B05 는 다시 B07 을 부를 수 있어
|
|
모듈 맨 위에서 부르면 맞물린다.
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureFigure import attach_figures
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import project_templates
|
|
from B08_Quantity.B08_Quantity_Engine_StructureSheet import build_standard_sheets
|
|
from B08_Quantity.B08_Quantity_Engine_StructureTemplate import apply_templates
|
|
from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table as build_unit_table
|
|
from B08_Quantity.B08_Quantity_Router_Material import _collect_structures
|
|
from common_util.common_util_project_settings import quantity_settings
|
|
|
|
# 설계를 넘겨야 다단 추가 기슭막이 장이 원단위 탭과 같이 섬(④).
|
|
structures, names, skipped = _collect_structures(project_root, designs)
|
|
settings = quantity_settings(project_root)
|
|
templates = project_templates(project_root)
|
|
unit_table = build_unit_table(
|
|
structures,
|
|
names,
|
|
section_modes,
|
|
ground_types,
|
|
settings.get("rubble_base_thickness_m"),
|
|
structure_formulas=settings.get("structure_formula_overrides"),
|
|
structure_templates=templates,
|
|
)
|
|
payload = build_standard_sheets(unit_table, section_modes)
|
|
# 양식이 있는 종류는 줄마다 식·설명·반올림·갈 곳을 실음(PLAN 3장 ④ · 명세 13장).
|
|
apply_templates(payload, settings, templates)
|
|
# 상단 그림(읽기 전용 SVG 자료) — 못 그리는 장은 까닭(PLAN 12장).
|
|
attach_figures(payload, settings.get("rubble_base_thickness_m"))
|
|
# 왜 안 실렸는지 — 「구조물이 없다」와 「걸러졌다」를 화면이 가릴 수 있어야 한다.
|
|
payload["skipped_structures"] = skipped
|
|
return payload
|
|
|
|
|
|
async def _sheets_of(project_id: UUID, project_root: str) -> dict[str, Any]:
|
|
"""조회·저장이 **같은 장 목록**을 보게 하는 한 문.
|
|
|
|
단면유형·지반 갈래는 원단위 탭 창구(`B08_Quantity_Router_Material`)를 그대로 씀.
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Router_Material import _designs, _ground_types, _section_modes
|
|
|
|
modes = await _section_modes(project_id)
|
|
ground = await _ground_types(project_id)
|
|
designs = await _designs(project_id)
|
|
return await asyncio.to_thread(project_structure_sheets, project_root, modes, ground, designs)
|
|
|
|
|
|
def _not_found() -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."},
|
|
)
|
|
|
|
|
|
class StandardSheetSpecRequest(BaseModel):
|
|
"""표준도 장 하나의 제원. **빈 값(null)은 「정한 적 없음」**이라 그 칸을 지운다."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
sheet_key: str
|
|
base_revision: int = Field(ge=0)
|
|
stone_kind: str | None = None
|
|
stone_supply: str | None = None
|
|
back_len_cm: int | str | None = None
|
|
face_slope_ratio: float | str | None = None
|
|
foundation: str | None = None
|
|
stone_coeff_basis: str | None = None
|
|
fill_concrete_mpa: str | None = None
|
|
thickness_top_m: float | str | None = None
|
|
thickness_bottom_m: float | str | None = None
|
|
blinding_concrete: str | None = None
|
|
|
|
|
|
class RoundingEdit(BaseModel):
|
|
"""줄 반올림 — 갈래는 명세 13장 일곱 가지만. 자리수가 음수면 10의 자리 쪽."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
mode: Literal["none", "floor", "trunc", "round", "ceil_away", "ceil", "round_half_even"]
|
|
digits: int = Field(default=0, ge=-6, le=10)
|
|
|
|
|
|
class FormulaEdit(BaseModel):
|
|
"""고친 식 한 줄 — 빈 식(null·"")은 **양식 식으로 되돌림**."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
seq: int = Field(ge=1)
|
|
formula: str | None = Field(default=None, max_length=2000)
|
|
#: 빠지면(null) 양식 반올림 — 식과 같이 **그 줄의 뜻한 상태 전부**를 보냄.
|
|
rounding: RoundingEdit | None = None
|
|
|
|
|
|
class StructureSheetFormulaRequest(BaseModel):
|
|
"""구조물도 장 하나에서 고친 식들(PLAN 3장 ⑤) — 저장은 그 장의 **양식 + 프로젝트** 단위."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
sheet_key: str
|
|
rows: list[FormulaEdit] = Field(max_length=200)
|
|
|
|
|
|
@router.put("/{project_id}/quantity/structure-sheets/formulas")
|
|
async def put_structure_sheet_formulas(
|
|
project_id: UUID, payload: StructureSheetFormulaRequest
|
|
) -> JSONResponse:
|
|
"""장 하나의 고친 식을 **산출 조건에 저장**하고 **서버가 같은 풀이기로 다시 푼 값**을 돌려줌.
|
|
|
|
⚠ 브라우저가 푼 값을 받아 적지 않음 — 식만 받고 값은 Node 로 다시 냄(판정 Ⓐ · 데이터 3층).
|
|
⚠ 개인 라이브러리에 저장하지 않음 — 그것은 [내 라이브러리에 저장] 단추 몫(명세 13장 Ⓒ, 4장).
|
|
⚠ 식이 풀리지 않아도 **막지 않고 저장**하되 줄마다 오류를 돌려줌 — 어디가 틀렸는지 보임.
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import project_templates
|
|
from B08_Quantity.B08_Quantity_Engine_StructureTemplate import (
|
|
OVERRIDES_KEY,
|
|
save_sheet_overrides,
|
|
template_of,
|
|
)
|
|
from common_util.common_util_project_settings import quantity_settings, save_section
|
|
|
|
project_root = await _project_root(project_id)
|
|
if project_root is None:
|
|
return _not_found()
|
|
sheets = (await _sheets_of(project_id, project_root)).get("sheets") or []
|
|
picked = next((s for s in sheets if s.get("key") == payload.sheet_key), None)
|
|
template = (
|
|
template_of(str(picked.get("type_id") or ""), project_templates(project_root))
|
|
if picked
|
|
else None
|
|
)
|
|
if picked is None or template is None:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "양식이 있는 구조물도 장을 찾지 못했습니다."},
|
|
)
|
|
try:
|
|
current = quantity_settings(project_root).get(OVERRIDES_KEY) or {}
|
|
# 장 이름이 아니라 **양식**에 묶음 — 제원을 고쳐도 식이 따라가고 같은 양식의 장이 모두 받음.
|
|
merged, changed = save_sheet_overrides(
|
|
current, str(picked["type_id"]), template, [row.model_dump() for row in payload.rows]
|
|
)
|
|
except ValueError as exc:
|
|
return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
|
|
# ⚠ 통째로 갈아 끼움 — 병합이면 되돌린(지운) 식이 남음(`save_section` replace_keys).
|
|
await asyncio.to_thread(
|
|
save_section,
|
|
project_root,
|
|
"quantity",
|
|
{OVERRIDES_KEY: merged},
|
|
replace_keys=[OVERRIDES_KEY],
|
|
)
|
|
|
|
sheets = (await _sheets_of(project_id, project_root)).get("sheets") or []
|
|
fresh = next((s for s in sheets if s.get("key") == payload.sheet_key), None) or {}
|
|
rows = fresh.get("rows") or []
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"changed": changed,
|
|
"rows": rows,
|
|
# 막지 않고 알림 — 어느 줄이 왜 안 섰는지.
|
|
"errors": [f"{row['name']}: {row['error']}" for row in rows if row.get("error")],
|
|
}
|
|
)
|
|
|
|
|
|
@router.put("/{project_id}/quantity/structure-sheets/spec")
|
|
async def put_structure_sheet_spec(
|
|
project_id: UUID, payload: StandardSheetSpecRequest
|
|
) -> JSONResponse:
|
|
"""장 하나의 제원을 고쳐 **그 조합의 구조물 전부**에 반영한다.
|
|
|
|
⚠ 값을 여기서 셈하지 않는다 — 정본(`structures.json`)에 적기만 하고, 표·그림은 다음
|
|
조회에서 그 정본으로 다시 선다. 표준도가 두 번째 정본이 되면 안 된다(CLAUDE.md 5장).
|
|
"""
|
|
from B05_Profile.B05_Profile_Structures_Repository import load_structures, save_structures
|
|
from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
|
|
from B08_Quantity.B08_Quantity_Engine_StructureSheet_Edit import (
|
|
apply_spec,
|
|
clean_spec,
|
|
drop_unregistered,
|
|
)
|
|
|
|
project_root = await _project_root(project_id)
|
|
if project_root is None:
|
|
return _not_found()
|
|
|
|
try:
|
|
sheets = (await _sheets_of(project_id, project_root)).get("sheets") or []
|
|
except Exception:
|
|
logger.exception("B08 구조물도 제원 저장 실패(장 목록): project_id=%s", project_id)
|
|
sheets = []
|
|
picked = next((s for s in sheets if s.get("key") == payload.sheet_key), None)
|
|
if picked is None:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "그 표준도 장을 찾지 못했습니다."},
|
|
)
|
|
member_ids = {
|
|
str(m.get("structure_id")) for m in picked.get("members") or [] if m.get("structure_id")
|
|
}
|
|
|
|
spec, notes = clean_spec(payload.model_dump(exclude={"sheet_key", "base_revision"}))
|
|
# ⚠ 등록부에 없는 칸은 저장소가 거절한다 — 한 칸 때문에 **전부** 못 저장되지 않게 거른다.
|
|
type_id = str(picked.get("type_id") or "")
|
|
definition = structure_type_map().get(type_id)
|
|
allowed = {field.key for field in definition.options} if definition else set()
|
|
spec, missing = drop_unregistered(type_id, spec, allowed)
|
|
notes.extend(missing)
|
|
try:
|
|
revision, stored = await asyncio.to_thread(load_structures, project_root)
|
|
updated, changed = apply_spec(stored, member_ids, spec)
|
|
outside = member_ids - {str(item.structure_id) for item in stored}
|
|
if outside:
|
|
# 관 지점 시설(A1)은 `pipe_points.json` 이 정본 — 조용히 안 먹히지 않게 알림.
|
|
notes.append(
|
|
f"관 지점 시설 {len(outside)}개소(배수관 기슭막이 등)는 이 화면에서 제원을 못 적음"
|
|
" — B05 배수 시설·구조물 집계표에서 고칠 것"
|
|
)
|
|
new_revision = await asyncio.to_thread(
|
|
save_structures, project_root, updated, base_revision=payload.base_revision
|
|
)
|
|
except Exception as exc:
|
|
logger.exception("B08 구조물도 제원 저장 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=409,
|
|
content={"status": "error", "message": f"제원을 저장하지 못했습니다 — {exc}"},
|
|
)
|
|
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"project_id": str(project_id),
|
|
"revision": new_revision,
|
|
"previous_revision": revision,
|
|
"changed": changed,
|
|
# 범위 밖 값·표에 없는 규격은 **막지 않고 알린다**(실무에 1:0.7 이 실재).
|
|
"notes": notes,
|
|
}
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}/quantity/structure-sheets/library")
|
|
async def get_structure_library(
|
|
project_id: UUID, type_id: str, session: dict[str, Any] = Depends(verify_session)
|
|
) -> JSONResponse:
|
|
"""가져오기 고르개 — **로그인한 사람**의 개인·회사 단과 프로그램 기본에서 그 종류의 항목.
|
|
|
|
⛔ 3단을 읽는 곳은 여기와 가져오기뿐 — 표를 그릴 때는 안 읽음(PLAN 4장 · 판정 Ⓑ).
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import (
|
|
list_items,
|
|
project_templates,
|
|
tier_dirs,
|
|
)
|
|
|
|
project_root = await _project_root(project_id)
|
|
if project_root is None:
|
|
return _not_found()
|
|
dirs = tier_dirs(session.get("company_id"), session.get("user_id"))
|
|
current = project_templates(project_root).get(type_id) or {}
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"items": await asyncio.to_thread(list_items, dirs, type_id),
|
|
"current": {"code": current.get("code"), "imported_from": current.get("imported_from")},
|
|
"can_publish": {tier: _publish_dir(session, tier) is not None for tier in PUBLISH},
|
|
}
|
|
)
|
|
|
|
|
|
class LibraryImportRequest(BaseModel):
|
|
"""가져올 항목 하나 — 단과 코드로 고름(이름은 겹칠 수 있음)."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type_id: str = Field(min_length=1, max_length=100)
|
|
tier: Literal["personal", "received", "company", "program"]
|
|
code: str = Field(pattern=r"^AX-ST-[0-9a-f]{8}$")
|
|
|
|
|
|
@router.put("/{project_id}/quantity/structure-sheets/library/import")
|
|
async def put_structure_library_import(
|
|
project_id: UUID,
|
|
payload: LibraryImportRequest,
|
|
session: dict[str, Any] = Depends(verify_session),
|
|
) -> JSONResponse:
|
|
"""고른 항목을 **프로젝트 작업본에 박고** 그 종류의 고친 식·줄 조합·수동 단가를 비움.
|
|
|
|
⚠ 비우는 까닭 — 셋 다 줄 차례에 묶였는데 새 양식과 차례가 안 맞을 수 있음. 묻는 것은 화면 몫.
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import (
|
|
find_item,
|
|
import_item,
|
|
tier_dirs,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Engine_StructureTemplate import OVERRIDES_KEY
|
|
from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import MANUAL_KEY, ROWS_KEY
|
|
from common_util.common_util_project_settings import quantity_settings, save_section
|
|
|
|
project_root = await _project_root(project_id)
|
|
if project_root is None:
|
|
return _not_found()
|
|
dirs = tier_dirs(session.get("company_id"), session.get("user_id"))
|
|
item = await asyncio.to_thread(find_item, dirs, payload.tier, payload.code)
|
|
if item is None or item.get("type_id") != payload.type_id:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "가져올 양식 항목을 찾지 못했습니다."},
|
|
)
|
|
await asyncio.to_thread(import_item, project_root, item, payload.tier)
|
|
settings = quantity_settings(project_root)
|
|
current = {key: settings.get(key) or {} for key in (OVERRIDES_KEY, ROWS_KEY, MANUAL_KEY)}
|
|
touched = [key for key, value in current.items() if payload.type_id in value]
|
|
if touched:
|
|
rest = {
|
|
key: {k: v for k, v in current[key].items() if k != payload.type_id} for key in touched
|
|
}
|
|
await asyncio.to_thread(save_section, project_root, "quantity", rest, replace_keys=touched)
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"code": payload.code,
|
|
"imported_from": payload.tier,
|
|
"cleared_formulas": len(current[OVERRIDES_KEY].get(payload.type_id) or {}),
|
|
"cleared_unit_price_rows": payload.type_id in current[ROWS_KEY],
|
|
"cleared_manual_prices": len(current[MANUAL_KEY].get(payload.type_id) or {}),
|
|
}
|
|
)
|
|
|
|
|
|
def _personal_dir(session: dict[str, Any]) -> Path | None:
|
|
"""로그인한 사람의 개인 단. 회사·사람이 없으면(시스템 관리자 등) `None`."""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import tier_dirs
|
|
|
|
return tier_dirs(session.get("company_id"), session.get("user_id")).get("personal")
|
|
|
|
|
|
def _no_personal() -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=403,
|
|
content={"status": "error", "message": "개인 라이브러리는 회사에 속한 사용자만 씁니다."},
|
|
)
|
|
|
|
|
|
#: 발행 단 — 회사는 마스터·시스템 관리자, 프로그램 기본은 시스템 관리자(2026-09-14 브레인 승인).
|
|
PUBLISH = ("company", "program")
|
|
|
|
|
|
def _publish_dir(session: dict[str, Any], tier: str) -> Path | None:
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import program_library_dir, tier_dirs
|
|
|
|
admin = session.get("role") == "SYSTEM_ADMIN"
|
|
if tier == "program":
|
|
return program_library_dir() if admin else None
|
|
if tier == "company" and (admin or session.get("is_master")):
|
|
return tier_dirs(session.get("company_id"), None).get("company")
|
|
return _personal_dir(session) if tier == "personal" else None
|
|
|
|
|
|
class LibrarySaveRequest(BaseModel):
|
|
"""[내 라이브러리에 저장]·[발행] — 어느 장의 양식을 어느 단에 쓸지(같은 모양)."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
sheet_key: str
|
|
tier: Literal["personal", "company", "program"] = "personal"
|
|
|
|
|
|
@router.put("/{project_id}/quantity/structure-sheets/library/personal")
|
|
@router.put("/{project_id}/quantity/structure-sheets/library/publish")
|
|
async def put_structure_library_personal(
|
|
project_id: UUID,
|
|
payload: LibrarySaveRequest,
|
|
session: dict[str, Any] = Depends(verify_session),
|
|
) -> JSONResponse:
|
|
"""그 장의 양식 + 이 프로젝트에서 고친 식을 **로그인한 사람 개인 단**에 한 벌로 씀.
|
|
|
|
⚠ 프로젝트 작업본은 안 바꿈 — 반대 방향(작업본 → 개인 단)이라(판정 Ⓑ).
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import (
|
|
project_templates,
|
|
save_personal,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Engine_StructureTemplate import OVERRIDES_KEY, template_of
|
|
from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import ROWS_KEY
|
|
from common_util.common_util_project_settings import quantity_settings
|
|
|
|
folder = _publish_dir(session, payload.tier)
|
|
if folder is None:
|
|
if payload.tier == "personal":
|
|
return _no_personal()
|
|
return JSONResponse(
|
|
status_code=403, content={"status": "error", "message": "발행 권한이 없습니다."}
|
|
)
|
|
project_root = await _project_root(project_id)
|
|
if project_root is None:
|
|
return _not_found()
|
|
sheets = (await _sheets_of(project_id, project_root)).get("sheets") or []
|
|
picked = next((s for s in sheets if s.get("key") == payload.sheet_key), None)
|
|
type_id = str((picked or {}).get("type_id") or "")
|
|
template = template_of(type_id, project_templates(project_root)) if picked else None
|
|
if template is None:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "양식이 있는 구조물도 장을 찾지 못했습니다."},
|
|
)
|
|
settings = quantity_settings(project_root)
|
|
overrides = (settings.get(OVERRIDES_KEY) or {}).get(type_id)
|
|
# ⛔ 수동 단가(`MANUAL_KEY`)는 안 넘김 — 프로젝트의 값(브레인 판정).
|
|
rows = (settings.get(ROWS_KEY) or {}).get(type_id)
|
|
code = await asyncio.to_thread(save_personal, folder, template, overrides, rows, payload.tier)
|
|
return JSONResponse(content={"status": "success", "code": code, "edited": len(overrides or {})})
|
|
|
|
|
|
@router.delete("/{project_id}/quantity/structure-sheets/library/personal")
|
|
async def delete_structure_library_personal(
|
|
project_id: UUID, type_id: str, session: dict[str, Any] = Depends(verify_session)
|
|
) -> JSONResponse:
|
|
"""[내 것 지우기] — 개인 단의 그 종류만 지움. ⛔ 프로젝트에 박힌 것은 안 바꿈(판정 Ⓑ)."""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import delete_personal
|
|
|
|
folder = _personal_dir(session)
|
|
if folder is None:
|
|
return _no_personal()
|
|
deleted = await asyncio.to_thread(delete_personal, folder, type_id)
|
|
return JSONResponse(content={"status": "success", "deleted": deleted})
|
|
|
|
|
|
async def _price_build(project_id: UUID) -> Any:
|
|
"""그 프로젝트가 고른 값으로 조립한 B09 단가표 — B09 창구와 **같은 한 벌**(읽기만)."""
|
|
from B09_Estimation.B09_Estimation_Router import _build_for
|
|
|
|
return await _build_for(project_id)
|
|
|
|
|
|
async def _sheet_template(
|
|
project_id: UUID, project_root: str, sheet_key: str
|
|
) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
|
|
"""장 하나와 그 양식(프로젝트에 박힌 것 → 프로그램 기본)."""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import project_templates
|
|
from B08_Quantity.B08_Quantity_Engine_StructureTemplate import template_of
|
|
|
|
sheets = (await _sheets_of(project_id, project_root)).get("sheets") or []
|
|
picked = next((s for s in sheets if s.get("key") == sheet_key), None)
|
|
if picked is None:
|
|
return None, None
|
|
return picked, template_of(str(picked.get("type_id") or ""), project_templates(project_root))
|
|
|
|
|
|
def _no_sheet() -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "그 구조물도 장을 찾지 못했습니다."},
|
|
)
|
|
|
|
|
|
async def _price_build_or_error(project_id: UUID) -> tuple[Any, JSONResponse | None]:
|
|
try:
|
|
return await _price_build(project_id), None
|
|
except Exception:
|
|
logger.exception("B08 구조물도 일위대가 — 단가표 조립 실패: project_id=%s", project_id)
|
|
return None, JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "단가표를 조립하지 못했습니다."},
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}/quantity/structure-sheets/unit-price")
|
|
async def get_structure_unit_price(project_id: UUID, sheet_key: str) -> JSONResponse:
|
|
"""장 하나의 하단 **일위대가 표**(미리보기) — 줄 조합은 양식(+고친 것), 단가는 B09 단가표.
|
|
|
|
⚠ 장 조회와 창구를 나눔 — 단가표 첫 조립이 십여 초라 장 조회를 늦추지 않게.
|
|
⚠ 값을 여기서 정본으로 적지 않음 — 보이기만(내역 금액은 B09 가 셈).
|
|
⚠ `editor` — 화면이 고칠 줄 조합·양식 줄·수동 단가. 값(금액)은 늘 이 서버 계산만 씀.
|
|
"""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import available_templates
|
|
from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import (
|
|
MANUAL_KEY,
|
|
ROWS_KEY,
|
|
unit_price_table,
|
|
with_rows,
|
|
)
|
|
from B09_Estimation.B09_Estimation_UnitPrice import find_variant_code
|
|
from common_util.common_util_project_settings import quantity_settings
|
|
|
|
project_root = await _project_root(project_id)
|
|
if project_root is None:
|
|
return _not_found()
|
|
picked, template = await _sheet_template(project_id, project_root, sheet_key)
|
|
if picked is None:
|
|
return _no_sheet()
|
|
if not template:
|
|
return JSONResponse(content={"status": "success", "unit_price": None})
|
|
build, error = await _price_build_or_error(project_id)
|
|
if error:
|
|
return error
|
|
settings = quantity_settings(project_root)
|
|
row_edits = settings.get(ROWS_KEY) or {}
|
|
type_id = str(template.get("type_id") or "")
|
|
manual = (settings.get(MANUAL_KEY) or {}).get(type_id) or {}
|
|
effective = with_rows(template, row_edits.get(type_id))
|
|
# 하위 양식도 그 종류에서 고친 줄 조합으로 풂 — 수동 단가는 맨 윗 표만.
|
|
library = [
|
|
with_rows(t, row_edits.get(t.get("type_id"))) for t in available_templates(project_root)
|
|
]
|
|
table = await asyncio.to_thread(
|
|
unit_price_table,
|
|
effective,
|
|
picked,
|
|
build.book,
|
|
lambda code, value: find_variant_code(code, value, build),
|
|
library,
|
|
manual,
|
|
)
|
|
editor = {
|
|
"rows": effective["unit_price"]["rows"],
|
|
"default_rows": (template.get("unit_price") or {}).get("rows") or [],
|
|
"edited": type_id in row_edits,
|
|
"manual_prices": manual,
|
|
}
|
|
return JSONResponse(content={"status": "success", "unit_price": table, "editor": editor})
|
|
|
|
|
|
class UnitPriceRowEdit(BaseModel):
|
|
"""일위대가 줄 하나 — 수량은 원단위 줄(`from_row`) 또는 박힌 값, 코드는 고르개가 채움."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
seq: int = Field(ge=1)
|
|
name: str | None = Field(default=None, max_length=200)
|
|
spec: str | None = Field(default=None, max_length=200)
|
|
unit: str | None = Field(default=None, max_length=20)
|
|
from_row: int | None = Field(default=None, ge=1)
|
|
quantity: float | None = None
|
|
ref_code: str | None = Field(default=None, max_length=100)
|
|
work_item_code: str | None = Field(default=None, max_length=100)
|
|
variant_from: str | None = Field(default=None, max_length=50)
|
|
sub_vars: dict[str, float | str] | None = None
|
|
|
|
|
|
class ManualPrice(BaseModel):
|
|
"""수동 단가 — 단위당 3분할과 출처. 넣은 날짜는 서버가 붙임."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
material: float = Field(default=0, ge=0)
|
|
labor: float = Field(default=0, ge=0)
|
|
expense: float = Field(default=0, ge=0)
|
|
source: str = Field(default="", max_length=200)
|
|
|
|
|
|
class UnitPriceEditRequest(BaseModel):
|
|
"""한 화면에서 고친 일위대가 — **저장 자리는 둘**(줄 조합 = 양식+프로젝트 · 단가 = 프로젝트)."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
sheet_key: str
|
|
rows: list[UnitPriceRowEdit] = Field(max_length=100)
|
|
manual_prices: dict[str, ManualPrice] = Field(default_factory=dict)
|
|
|
|
|
|
@router.put("/{project_id}/quantity/structure-sheets/unit-price")
|
|
async def put_structure_unit_price(project_id: UUID, payload: UnitPriceEditRequest) -> JSONResponse:
|
|
"""고친 줄 조합·수동 단가를 산출 조건에 저장. 금액은 화면이 다시 조회해 서버 계산으로 받음."""
|
|
from datetime import date
|
|
|
|
from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import (
|
|
MANUAL_KEY,
|
|
ROWS_KEY,
|
|
save_manual,
|
|
save_rows,
|
|
)
|
|
from common_util.common_util_project_settings import quantity_settings, save_section
|
|
|
|
rows = [row.model_dump(exclude_none=True) for row in payload.rows]
|
|
if len({row["seq"] for row in rows}) != len(rows):
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content={"status": "error", "message": "일위대가 줄 차례가 겹칩니다."},
|
|
)
|
|
project_root = await _project_root(project_id)
|
|
if project_root is None:
|
|
return _not_found()
|
|
picked, template = await _sheet_template(project_id, project_root, payload.sheet_key)
|
|
if picked is None or template is None:
|
|
return _no_sheet()
|
|
type_id = str(template.get("type_id") or "")
|
|
settings = quantity_settings(project_root)
|
|
merged_rows, changed = save_rows(settings.get(ROWS_KEY) or {}, type_id, template, rows)
|
|
prices = {seq: price.model_dump() for seq, price in payload.manual_prices.items()}
|
|
merged_manual = save_manual(
|
|
settings.get(MANUAL_KEY) or {}, type_id, rows, prices, date.today().isoformat()
|
|
)
|
|
await asyncio.to_thread(
|
|
save_section,
|
|
project_root,
|
|
"quantity",
|
|
{ROWS_KEY: merged_rows, MANUAL_KEY: merged_manual},
|
|
replace_keys=[ROWS_KEY, MANUAL_KEY],
|
|
)
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"changed_rows": changed,
|
|
"edited": type_id in merged_rows,
|
|
"manual_prices": len(merged_manual.get(type_id) or {}),
|
|
}
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}/quantity/structure-sheets/price-search")
|
|
async def get_structure_price_search(
|
|
project_id: UUID, q: str, kind: Literal["work", "resource"] = "work"
|
|
) -> JSONResponse:
|
|
"""고르개 — 이 프로젝트 단가표에서 품셈(일위대가)·자원(자재·노임·중기)을 낱말로 찾음."""
|
|
from B08_Quantity.B08_Quantity_Engine_StructureUnitPrice import search_titles
|
|
|
|
build, error = await _price_build_or_error(project_id)
|
|
if error:
|
|
return error
|
|
items = await asyncio.to_thread(search_titles, build.book, q[:100], kind)
|
|
return JSONResponse(content={"status": "success", "items": items})
|
|
|
|
|
|
@router.get("/{project_id}/quantity/structure-sheets")
|
|
async def get_structure_sheets(project_id: UUID) -> JSONResponse:
|
|
"""구조물도(표준도) **장 목록 + 원단위 수량표**.
|
|
|
|
⚠ 수량을 여기서 새로 셈하지 않는다 — B08 원단위 전개를 그대로 받아 **제원 조합으로 묶고
|
|
단위당으로 접기만** 한다(계산 자리는 한 곳, CLAUDE.md 5장).
|
|
"""
|
|
project_root = await _project_root(project_id)
|
|
if project_root is None:
|
|
return _not_found()
|
|
|
|
try:
|
|
payload = await _sheets_of(project_id, project_root)
|
|
except Exception:
|
|
logger.exception("B08 구조물도 전개 실패: project_id=%s", project_id)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"status": "error", "message": "구조물 원단위를 전개하지 못했습니다."},
|
|
)
|
|
|
|
payload["status"] = "success"
|
|
payload["project_id"] = str(project_id)
|
|
return JSONResponse(content=payload)
|