"""B08 구조물 원단위·자재총괄 조회 라우터 (일감 6·7 · PLAN 8-2·8-6·8-7). 값은 어디서 오나 치수 정본은 **`structures.json` 하나**다(B05 가 주인). B08 은 자기 치수표를 들지 않고 그 제원을 읽어 전개할 뿐이다 — 도면은 H=1.5 인데 수량은 옛 치수로 도는 사고를 막는다. ⚠ `design_owner` 가 붙은 타입은 건너뛴다 측구가 그렇다 — 횡단 설계가 이미 터파기 단면적까지 셈하므로 구조물로 또 세면 **같은 것을 두 번 계상**한다(레지스트리 주석, 2026-09-07 조사). 건너뛴 것은 숨기지 않고 응답에 적는다. ⚠ 할증은 자재총괄 한 곳뿐이다 (㉠) 원단위표는 할증 **전** 값(`surcharge_applied: False`)으로 오고, 자재총괄이 한 번 붙인다. 응답에 두 깃발이 다 실리므로 화면·B09 가 어느 쪽 값인지 헷갈릴 일이 없다. """ from __future__ import annotations import json import logging from pathlib import Path from typing import Any from uuid import UUID from fastapi import APIRouter from fastapi.responses import JSONResponse from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path from B06_Section.B06_Section_Repository import get_cross_section_designs from B06_Section.B06_Section_Repository import get_workflow_route_context from B05_Profile.B05_Profile_Structures_Repository import load_structures from B05_Profile.B05_Profile_Structures_Schema import structure_type_map from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff, load_mapping, summarize from B08_Quantity.B08_Quantity_Engine_MaterialSummary import build_table as build_material_table from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table as build_unit_table from B08_Quantity.B08_Quantity_Engine_UnitQuantity import ( ground_types_from_designs, section_modes_from_designs, ) from common_util.common_util_project_settings import ( concrete_placing_method, quantity_settings, rock_classes, rock_method, ) from B08_Quantity.B08_Quantity_Engine_HaulInputs import haul_inputs from B08_Quantity.B08_Quantity_Engine_Preparation import frame_material_rows from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_structure_lengths import structure_lengths from config.config_db import run_with_connection logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"]) def _collect_structures( project_root: str, ) -> tuple[list[dict[str, Any]], dict[str, str], list[str]]: """전개 대상 구조물·타입명·건너뛴 사유를 함께 낸다.""" _revision, items = load_structures(project_root) types = structure_type_map() targets: list[dict[str, Any]] = [] names: dict[str, str] = {} skipped: list[str] = [] for item in items: payload = item.model_dump() type_id = str(payload.get("type_id") or "") definition = types.get(type_id) if definition is None: skipped.append(f"{type_id}: 레지스트리에 없는 타입") continue names[type_id] = definition.name if definition.design_owner: skipped.append( f"{definition.name}: {definition.design_owner} 가 이미 셈 — 중복 계상 방지" ) continue if definition.reference_only: skipped.append(f"{definition.name}: 전문 상세설계 대상 — 배치까지만") continue if definition.group == "B": # ⚠ B군(종단배수)은 **연장표**로 간다 — `common_util_structure_lengths` 가 # 겹친 구간을 합쳐 주기 때문이다. 구조물별로 세면 겹친 구간을 두 번 센다. # 여기서 빼지 않으면 **같은 시설이 두 줄로** 나간다. continue targets.append(payload) return targets, names, sorted(set(skipped)) async def _ground_types(project_id: UUID) -> dict[float, str]: """측점별 지반 갈래(`soil`·`ripping_rock`·`blasting_rock`). 구조물터파기(품셈 9-13)의 **토질 축**이 이 값으로 갈린다. 단면유형과 같은 자리에서 오므로 읽는 방식도 같다 — 못 읽으면 빈 표로 두고 판정이 「못 가름」이 되게 한다. """ try: context = await run_with_connection(get_workflow_route_context, project_id) route_id = int((context or {}).get("route_id") or 0) if not route_id: return {} designs = await run_with_connection(get_cross_section_designs, route_id) except Exception: logger.exception("B08 지반 갈래 조회 실패: project_id=%s", project_id) return {} return ground_types_from_designs(designs) async def _section_modes(project_id: UUID) -> dict[float, str]: """측점별 단면유형(`left_cut` 등). 구조물이 **성토면인가 절토면인가**를 가릴 때 쓴다. ⚠ 새 저장 키를 만들지 않는다 — 이미 저장되는 `design.section_mode` 를 읽기만 한다. 못 읽으면 빈 표로 두고, 판정이 「가를 근거 없음」이 되게 한다(성토로 눅이지 않음). """ try: context = await run_with_connection(get_workflow_route_context, project_id) route_id = int((context or {}).get("route_id") or 0) if not route_id: return {} designs = await run_with_connection(get_cross_section_designs, route_id) except Exception: logger.exception("B08 단면유형 조회 실패: project_id=%s", project_id) return {} return section_modes_from_designs(designs) @router.get("/{project_id}/quantity/material-summary") async def get_material_summary(project_id: UUID) -> JSONResponse: """구조물 원단위와 자재총괄을 **한 응답**으로 낸다. 자재총괄은 원단위의 `material` 성분만 모은 것이라 따로 부르면 같은 전개를 두 번 돈다. """ try: stored_path = await run_with_connection(get_project_storage_relative_path, project_id) project_root = resolve_stored_project_path(stored_path) except Exception: logger.exception("B08 자재총괄 조회 실패(경로): project_id=%s", project_id) return JSONResponse( status_code=404, content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."}, ) try: structures, names, skipped = _collect_structures(project_root) except Exception: logger.exception("B08 구조물 정본 읽기 실패: project_id=%s", project_id) return JSONResponse( status_code=500, content={"status": "error", "message": "구조물 정본을 읽지 못했습니다."}, ) settings = quantity_settings(project_root) unit_table = build_unit_table( structures, names, await _section_modes(project_id), await _ground_types(project_id), settings.get("rubble_base_thickness_m"), ) material_table = build_material_table( unit_table, supply_map=settings.get("material_supply") or {}, # 콘크리트 할증은 **레미콘일 때만** 붙는다 — 방식이 이름을 가른다(확정 3차 ⑥). concrete_placing_method=settings.get("concrete_placing_method"), ) # 묶음으로 서는 구조물의 조각을 화면에도 보인다 — 코드만으로는 사람이 검증 못 한다. handoff = build_handoff(unit_quantity_table=unit_table) composite = [ { "name": row["name"], "parts": row["composite_parts"], "not_ready": row.get("composite_not_ready"), } for row in handoff["work_items"] # 조각이 없어도(원단위 자체가 없어 못 세운 경우) 사유는 보여야 한다. if row.get("composite_parts") or row.get("composite_not_ready") ] return JSONResponse( content={ "unit_quantity": unit_table, "material": material_table, "composite": composite, "skipped_structures": skipped, "structure_count": len(structures), } ) async def project_haul_inputs(project_id: UUID) -> dict[str, Any]: """유토곡선(B06)이 받아야 할 **구조물 몫** — 채집석 공제 · 구조물 잔토. ⚠ **B06 이 이 함수를 부르면 된다.** 두 값 다 B08 전개에서 나오는 것이라 저쪽이 다시 세면 같은 계산이 두 벌이 된다(CLAUDE.md 5장). 값은 **양수 ㎥** 이고 빼고 더하는 것은 받는 쪽 몫이다. 못 읽으면 빈 값(`None`) — 0 으로 눅이지 않는다. """ try: stored_path = await run_with_connection(get_project_storage_relative_path, project_id) project_root = resolve_stored_project_path(stored_path) structures, names, _skipped = _collect_structures(project_root) unit_table = build_unit_table( structures, names, await _section_modes(project_id), await _ground_types(project_id), quantity_settings(project_root).get("rubble_base_thickness_m"), ) except Exception: logger.exception("B08 유토곡선 입력 조회 실패: project_id=%s", project_id) return haul_inputs(None) return haul_inputs(unit_table) @router.get("/{project_id}/quantity/haul-inputs") async def get_haul_inputs(project_id: UUID) -> JSONResponse: """같은 값을 화면·다른 창이 볼 수 있게 낸 자리. 계산은 위 함수 한 벌이다.""" return JSONResponse(content=await project_haul_inputs(project_id)) @router.get("/{project_id}/quantity/handoff") async def get_handoff(project_id: UUID) -> JSONResponse: """B09 로 넘길 두 벌 — 작업 공종 축과 자재 축 (일감 9). ⚠ **한 벌로 합치지 않는다.** 내역 줄은 작업 공종이고 자재는 자재다. 자재에 공종코드를 붙이면 자재가 내역 줄로 오해된다(8-2 이중계상 함정). ⚠ 토공·운반은 토적표 라우터가 이미 만드는 표를 그대로 받는다 — 여기서 다시 계산하지 않는다. 같은 값을 두 벌로 짜지 않는다는 규칙(CLAUDE.md 5장)이 여기에도 걸린다. """ try: stored_path = await run_with_connection(get_project_storage_relative_path, project_id) project_root = resolve_stored_project_path(stored_path) except Exception: logger.exception("B08 인계 조회 실패(경로): project_id=%s", project_id) return JSONResponse( status_code=404, content={"status": "error", "message": "프로젝트 저장 폴더를 찾지 못했습니다."}, ) structures, names, skipped = _collect_structures(project_root) settings = quantity_settings(project_root) unit_table = build_unit_table( structures, names, await _section_modes(project_id), await _ground_types(project_id), settings.get("rubble_base_thickness_m"), ) material_table = build_material_table( unit_table, supply_map=settings.get("material_supply") or {}, concrete_placing_method=settings.get("concrete_placing_method"), ) # 토공·운반 표는 토적표 라우터의 것을 그대로 쓴다 — 여기서 다시 만들지 않는다. earthwork = await _earthwork_tables(project_id) # 규준틀 재료 — **개소가 선 뒤에야 설 수 있어** 준비공 표를 받은 다음 자재 축에 얹는다. # ⚠ 값은 **제안값(실무 관측)**이고 산출 조건에서 고칠 수 있다 — 그 사실이 줄 사유에 적힌다. frame_rows = [ row for row in ((earthwork.get("preparation") or {}).get("rows") or []) if str(row.get("item") or "").endswith("규준틀") ] material_table = build_material_table( unit_table, supply_map=settings.get("material_supply") or {}, concrete_placing_method=settings.get("concrete_placing_method"), extra_materials=frame_material_rows(frame_rows, settings.get("frame_material") or {}), ) handoff = build_handoff( summary_table=earthwork.get("summary"), haul_table=earthwork.get("haul"), unit_quantity_table=unit_table, material_table=material_table, # 준비공·사방공 — 값이 서는 줄도, 못 내는 줄도 함께 넘긴다(빼면 빠진 줄이 안 보임). preparation_table=earthwork.get("preparation"), # B군 종단배수 — 겹침을 합친 연장. 그 규칙이 이미 그 함수에 있어 두 벌로 안 짠다. length_table=[row for row in structure_lengths(project_root) if row.get("group") == "B"], # 배수관 — 관 정본은 `pipe_points.json`, 연장은 측점 `design.pipe_length_m` 다. pipe_table=_pipe_table( project_root, earthwork.get("pipe_lengths") or [], earthwork.get("section_chainages") or [], ), ground_class_set=settings.get("rock_class_set"), ground_classes=rock_classes(settings), ground_methods={name: rock_method(settings, name) for name in rock_classes(settings)}, # 타설 방식 — 안 정했으면 기본값으로 서되 그 사실을 `placing_notes` 가 알린다. concrete_placing_method=concrete_placing_method(settings)[0], # 층따기 길이 — 면적 × 이 값으로 ㎥ 를 낸다(확정 2차 ①). 안 넣었으면 막히고 사유가 감. bench_cut_depth_m=settings.get("bench_cut_depth_m"), # 용수 유무 — 기본 「육상」은 **통상값**이다(확정 3차 ④). 사유·화면에 그 사실이 뜬다. structure_trench_water=settings.get("structure_trench_water"), ) handoff["summary"] = summarize(handoff) handoff["skipped_structures"] = skipped handoff["earthwork_available"] = bool(earthwork) return JSONResponse(content=handoff) def _pipe_table( project_root: str, pipe_lengths: list[dict[str, Any]], section_chainages: list[Any] | None = None, ) -> dict[str, Any]: """배수관 표 — 정본 셋을 읽어 잇는다. 못 읽으면 **빈 표**(줄이 안 서는 것이 정직하다). ⚠ 관 정본은 `structures.json` 이 아니라 `pipe_points.json` 이다 (레지스트리 `pipe` 타입이 `managed_by: pipe_points`). """ from B08_Quantity.B08_Quantity_Engine_Pipe import build_rows as build_pipe_rows from common_util.common_util_drainage_pipes import pipe_points_path_in path = pipe_points_path_in(Path(project_root)) if not path.is_file(): return {"rows": [], "notes": [], "pipe_count": 0, "ready_count": 0, "length_total_m": 0.0} try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): logger.exception("B08 관 지점 읽기 실패: %s", path) return { "rows": [], "notes": ["관 지점 파일을 읽지 못했습니다"], "pipe_count": 0, "ready_count": 0, "length_total_m": 0.0, } points = payload.get("points") or payload.get("items") or [] # 토적표 라우터가 실어 준 모양을 엔진이 읽는 모양으로 옮긴다. designs = [ {"chainage_m": row.get("chainage_m"), "design": {"pipe_length_m": row.get("pipe_length_m")}} for row in pipe_lengths ] return build_pipe_rows( points, designs, (load_mapping().pipe or {}), [float(x) for x in (section_chainages or []) if x is not None], ) async def _earthwork_tables(project_id: UUID) -> dict[str, Any]: """토적표 라우터가 만든 집계·운반 표를 얻는다. 노선이 없으면 빈 값.""" from B08_Quantity.B08_Quantity_Router_Earthwork import ( get_earthwork_table_for_current_route, ) try: response = await get_earthwork_table_for_current_route(project_id) except Exception: logger.exception("B08 인계 — 토적표 조회 실패: project_id=%s", project_id) return {} if response.status_code != 200: return {} import json as _json return _json.loads(bytes(response.body).decode("utf-8"))