"""B07 수량 산출 프로토타입 (처분 가능 — disposable). 사용자 평가용 시안이다. 이 파일 + `B07_DesignDetail_Router.py`의 include 2줄이 전부라 마음에 안 들면 그대로 지우면 된다. 정본을 읽기만 하고 어디에도 쓰지 않는다. 페이지: GET /api/projects/quantity-proto/page — 프로젝트 목록 GET /api/projects/quantity-proto/page/{id} — 수량 산출 보고서 구조물 출처(정본 3곳, CLAUDE.md 5장): ① pipe_points.json — 계곡 통과 시설(배수관·BOX암거·물넘이·세월교·독립 기슭막이) ② structures.json — 일반 구조물(C·D·E군) ③ 횡단 설계 patch(DB) — 기준벽 조작·다단 벽(extra_wall_counts·extra_spans) 원단위 출처: 실무 관측 **참조 전용** — 채용 미확정(개발 단계 사용자 협의 대상). · 울진1 수량집계 13종 (resources/knowledge/original/실무문서/_종합비교/04_임도구조물_원단위.md) · 울진소광 구조도 숨김탭 (resources/knowledge/original/실무문서/_원단위라이브러리_울진소광.md) """ import json import logging from collections import defaultdict from pathlib import Path from typing import Any from uuid import UUID from fastapi import APIRouter from fastapi.responses import HTMLResponse from B01_Dashboard.B01_Dashboard_Repository import list_all_projects from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path from B05_Profile.B05_Profile_Structures_Repository import load_structures from B06_Section.B06_Section_Repository import ( get_confirmed_route_context, get_cross_section_designs, get_longitudinal_section, ) from common_util.common_util_drainage_pipes import parse_pipe_points from common_util.common_util_storage import resolve_stored_project_path from config.config_db import get_db_pool logger = logging.getLogger(__name__) proto_router = APIRouter() # ---------------------------------------------------------------- 타입 이름표 _FACILITY_NAMES = { "pipe": "배수관", "box_culvert": "BOX암거", "ford_pavement": "물넘이포장", "ford_bridge": "세월교", "revetment": "기슭막이(독립)", } _REGISTRY_PATH = Path(__file__).parent.parent / "B05_Profile" / "B05_Profile_Structure_Types.json" _type_names: dict[str, str] | None = None def _type_name(type_id: str) -> str: global _type_names if _type_names is None: try: payload = json.loads(_REGISTRY_PATH.read_text(encoding="utf-8")) _type_names = {t["type_id"]: t["name"] for t in payload.get("types", [])} except (OSError, ValueError, KeyError): _type_names = {} return _type_names.get(type_id, type_id) # ---------------------------------------------------------------- 원단위 라이브러리 # ⚠ 참조 전용(과거 공사지 관측값). 프로그램 기본값 아님 — 채용은 사용자 협의로 확정. # items: {공종명(단위): 단위수량} _SRC_ULJIN1 = "울진1 수량집계" _SRC_SOGWANG = "울진소광 구조도 숨김탭" UNIT_RATES: dict[tuple[str, str], dict[str, Any]] = { ("basin", "Ø800"): { "label": "집수정 Ø800", "per": "개소", "source": _SRC_SOGWANG, "items": { "터파기(㎥)": 10.64, "되메우기(㎥)": 6.44, "잔토처리(㎥)": 4.20, "콘크리트(㎥)": 2.84, "이형철근 D13(kg)": 4.78, "면목 A25(m)": 12.67, "거푸집(㎡)": 21.28, }, }, ("erosion_check", "찰쌓기·평균"): { "label": "돌골막이(찰쌓기) — 임도 실측 6개소 평균", "per": "개소", "source": _SRC_SOGWANG, "note": "관측 높이 2.5~3.0m는 사방기술교본 '2m 이내' 초과 — 판정 없이 기록(지식DB 주석 그대로)", "items": { "돌쌓기(㎡)": 19.62, "돌붙임(㎡)": 2.63, "야면석(ton)": 19.58, "고임돌(㎥)": 3.34, "막자갈(㎥)": 10.54, "채움콘크리트(㎥)": 4.45, "물구멍(m)": 4.91, "바닥파기(㎥)": 8.61, "되메우기(㎥)": 5.87, "잔토처리(㎥)": 2.74, }, }, ("retaining_wall", "반중력식 H=2.0"): { "label": "반중력식옹벽 H=2.0", "per": "m", "source": _SRC_SOGWANG, "items": { "콘크리트(㎥)": 1.35, "버림콘크리트(㎥)": 0.15, "거푸집-유로폼(㎡)": 3.20, "거푸집-기초(㎡)": 0.60, "물빼기 Ø50(m)": 0.32, "이형철근 D13(kg)": 13.45, "이형철근 D16(kg)": 30.42, }, }, ("revet_wet", "H=1.5"): { "label": "돌기슭막이 찰쌓기 H=1.5", "per": "m", "source": _SRC_ULJIN1, "items": { "콘크리트(㎥)": 0.31, "모르터(㎥)": 0.014, "돌쌓기(㎡)": 1.57, "석재(ton)": 0.91, "터파기(㎥)": 1.55, "되메우기(㎥)": 0.30, "잔토처리(㎥)": 1.25, }, }, ("revet_wet", "H=2.0"): { "label": "돌기슭막이 찰쌓기 H=2.0", "per": "m", "source": _SRC_ULJIN1, "items": { "콘크리트(㎥)": 0.42, "모르터(㎥)": 0.019, "돌쌓기(㎡)": 2.09, "석재(ton)": 0.91, "터파기(㎥)": 2.06, "되메우기(㎥)": 0.40, "잔토처리(㎥)": 1.66, }, }, ("revet_wet", "H=2.5"): { "label": "돌기슭막이 찰쌓기 H=2.5", "per": "m", "source": _SRC_ULJIN1, "items": { "콘크리트(㎥)": 0.52, "모르터(㎥)": 0.023, "돌쌓기(㎡)": 2.61, "석재(ton)": 0.91, "터파기(㎥)": 2.58, "되메우기(㎥)": 0.50, "잔토처리(㎥)": 2.08, }, }, ("completion_sign", "화강암 300×200×600"): { "label": "임도준공판 (화강암 300×200×600)", "per": "개소", "source": _SRC_SOGWANG, "items": { "화강암(㎥)": 0.036, "기초콘크리트(㎥)": 0.014, "거푸집-유로폼(㎡)": 0.35, "터파기(㎥)": 0.105, "되메우기(㎥)": 0.08, "잔토처리(㎥)": 0.025, }, }, # 참고 표기용 — 관 부설 연장이 미산정이라 rollup에는 적용하지 않는다. ("pipe_ref", "Ø800"): { "label": "관공 Ø800 (참고 — 연장 미산정으로 미적용)", "per": "m", "source": _SRC_ULJIN1, "items": {"터파기(㎥)": 4.5, "되메우기(㎥)": 3.99, "잔토처리(㎥)": 2.93}, }, ("pipe_ref", "Ø1000"): { "label": "관공 Ø1000 (참고 — 연장 미산정으로 미적용)", "per": "m", "source": _SRC_ULJIN1, "items": {"터파기(㎥)": 5.6, "되메우기(㎥)": 4.81, "잔토처리(㎥)": 3.83}, }, } # ---------------------------------------------------------------- 데이터 수집 async def _collect(project_id: UUID) -> dict[str, Any]: """정본 3곳을 읽어 화면 조립에 필요한 원자료만 모은다 (쓰기 없음).""" pool = get_db_pool() async with pool.acquire() as connection: stored_path = await get_project_storage_relative_path(connection, project_id) route_context = await get_confirmed_route_context(connection, project_id) route_id = int(route_context["route_id"]) if route_context else None confirmed = False designs: list[dict[str, Any]] = [] if route_id is not None: longitudinal = await get_longitudinal_section(connection, project_id, route_id) confirmed = bool(longitudinal and longitudinal.get("status") == "CONFIRMED") rows = await get_cross_section_designs(connection, route_id) designs = [row for row in rows if isinstance(row.get("design"), dict)] root = Path(resolve_stored_project_path(stored_path)).resolve() pipe_file = root / "B04_PreProcess" / "drainage" / "edits" / "pipe_points.json" pipes: list[Any] = [] pipe_file_exists = pipe_file.is_file() if pipe_file_exists: try: document = json.loads(pipe_file.read_text(encoding="utf-8")) pipes = parse_pipe_points(document.get("points")) except (OSError, ValueError): logger.warning("수량 프로토: pipe_points.json 읽기 실패 — %s", pipe_file) revision, structures = load_structures(str(root)) return { "route_id": route_id, "confirmed": confirmed, "designs": designs, "pipes": pipes, "pipe_file_exists": pipe_file_exists, "structures": structures, "structures_revision": revision, } # ---------------------------------------------------------------- 집계·매칭 def _pipe_spec(point: Any) -> str: options = point.options or {} if point.facility == "pipe": kind = options.get("pipe_kind", "파형강관") dia = options.get("pipe_diameter_mm", 1000) return f"{kind} Ø{dia}" if point.facility == "box_culvert": return str(options.get("box_size", "규격 미지정")) return "—" def _span_text(point: Any) -> str: if point.start_m is not None and point.end_m is not None: return f"{point.start_m:.2f}~{point.end_m:.2f} (L={point.end_m - point.start_m:.2f}m)" return "미지정(폭 0)" def _structure_position(instance: Any) -> str: if instance.placement == "interval" and instance.start_m is not None: end = instance.end_m if instance.end_m is not None else instance.start_m return f"{instance.start_m:.2f}~{end:.2f} (L={end - instance.start_m:.2f}m)" if instance.chainage_m is not None: return f"{instance.chainage_m:.2f}m" return "—" def _wall_rows(designs: list[dict[str, Any]]) -> list[dict[str, Any]]: """횡단 patch에서 벽 조작·다단 벽 흔적이 있는 측점만 표로 만든다.""" rows: list[dict[str, Any]] = [] for row in designs: design = row["design"] adjust = design.get("revet_adjust") or {} counts = design.get("extra_wall_counts") or {} spans = design.get("extra_spans") or {} if not adjust and not any(counts.values() if isinstance(counts, dict) else []): continue walls = [] for key, patch in adjust.items() if isinstance(adjust, dict) else []: if not isinstance(patch, dict): continue height = patch.get("h") span = spans.get(key) if isinstance(spans, dict) else None length = span.get("length_m") if isinstance(span, dict) else None walls.append( { "key": key, "height": f"{height:.2f}m" if isinstance(height, (int, float)) else "자동", "material": patch.get("m") or "자동", "length": f"{length:.2f}m" if isinstance(length, (int, float)) else "—", } ) rows.append( { "chainage": float(row.get("chainage_m", 0)), "counts": counts if isinstance(counts, dict) else {}, "walls": walls, } ) rows.sort(key=lambda r: r["chainage"]) return rows def _rollup(data: dict[str, Any]) -> dict[str, Any]: """규격별 집계 → 원단위 매칭 → 공종별 합계. 매칭 불가는 사유와 함께 보류 목록으로.""" matched: list[dict[str, Any]] = [] # 산출근거 행 deferred: list[dict[str, Any]] = [] # 보류 행 totals: dict[str, float] = defaultdict(float) def apply(rate_key: tuple[str, str], quantity: float, positions: str) -> None: rate = UNIT_RATES[rate_key] lines = [] for item, unit_qty in rate["items"].items(): value = unit_qty * quantity totals[item] += value lines.append({"item": item, "unit_qty": unit_qty, "value": value}) matched.append( { "label": rate["label"], "per": rate["per"], "quantity": quantity, "source": rate["source"], "note": rate.get("note", ""), "positions": positions, "lines": lines, } ) # ① 배수 시설 — 관 본체는 연장 미산정으로 보류, 유입 집수정만 개소당 적용 basin_positions: list[str] = [] pipe_counts: dict[str, int] = defaultdict(int) for point in data["pipes"]: facility = _FACILITY_NAMES.get(point.facility, point.facility) options = point.options or {} if point.facility == "pipe": pipe_counts[_pipe_spec(point)] += 1 if options.get("inlet_type") == "집수정": basin_positions.append(f"{point.chainage_m:.2f}m") else: deferred.append( { "name": f"{facility} @{point.chainage_m:.2f}m", "reason": "원단위 미확보 (상세 치수·산식은 B06/구조도 확정 후)", } ) if basin_positions: apply(("basin", "Ø800"), len(basin_positions), " · ".join(basin_positions)) for spec, count in sorted(pipe_counts.items()): deferred.append( { "name": f"배수관 {spec} × {count}개소", "reason": "관 부설 연장 미산정 (B06 횡단 세트 연장 필요) — 관공 m당 원단위는 라이브러리 참고행", } ) # ② 일반 구조물 (structures.json) for instance in data["structures"]: name = _type_name(instance.type_id) position = _structure_position(instance) if instance.type_id == "erosion_check": apply(("erosion_check", "찰쌓기·평균"), 1, position) continue if instance.type_id == "completion_sign": apply(("completion_sign", "화강암 300×200×600"), 1, position) continue if instance.type_id == "retaining_wall" and instance.placement == "interval": length = (instance.end_m or 0) - (instance.start_m or 0) if length > 0: apply(("retaining_wall", "반중력식 H=2.0"), length, position) continue deferred.append( {"name": f"{name} @{position}", "reason": "원단위 미확보 또는 규격(치수) 미확정"} ) # ③ 횡단 벽 패치 — 높이·연장이 자동 계산 몫이라 전량 보류 (일람으로만 제시) wall_station_count = len(_wall_rows(data["designs"])) if wall_station_count: deferred.append( { "name": f"기준벽·다단 기슭막이 — 조작 측점 {wall_station_count}곳", "reason": "벽 높이(자동)·연장이 B06 엔진 산출 몫 — 규격 확정 후 돌기슭막이 H별 m당 원단위 적용 예정", } ) return {"matched": matched, "deferred": deferred, "totals": dict(totals)} # ---------------------------------------------------------------- HTML _CSS = """ body{font-family:'Malgun Gothic',sans-serif;margin:24px;background:#f5f6f8;color:#222} h1{font-size:20px} h2{font-size:16px;margin-top:28px;border-left:4px solid #4f8ef7;padding-left:8px} table{border-collapse:collapse;background:#fff;margin:8px 0;font-size:13px} th,td{border:1px solid #ccd;padding:4px 10px;text-align:left} th{background:#eef2fa} td.num{text-align:right;font-variant-numeric:tabular-nums} .banner{background:#fff3cd;border:1px solid #e0c060;padding:10px 14px;border-radius:6px;font-size:13px} .muted{color:#777;font-size:12px} .warn{color:#a55} .src{color:#476;font-size:12px} a{color:#3366cc} """ def _fmt(value: float) -> str: return f"{value:,.2f}" def _page(title: str, body: str) -> HTMLResponse: return HTMLResponse( "" f"{title}{body}" ) def _render_report(project: dict[str, Any] | None, project_id: UUID, data: dict[str, Any]) -> str: rollup = _rollup(data) name = (project or {}).get("name") or str(project_id) parts: list[str] = [f"

수량 산출 프로토타입 — {name}

"] parts.append( "" ) # 데이터 원천 상태 parts.append("

데이터 원천

") parts.append( f"" ) parts.append( f"" ) patch_state = ( f"확정 노선 route_id={data['route_id']} — 설계 저장 측점 {len(data['designs'])}곳" if data["route_id"] is not None else "확정 노선 없음 — 횡단 벽 패치 제외" ) if data["route_id"] is not None and not data["confirmed"]: patch_state += " (B06 종단 미확정)" parts.append(f"
정본상태
pipe_points.json (배수·독립 기슭막이){'있음 — ' + str(len(data['pipes'])) + '건' if data['pipe_file_exists'] else '없음'}
structures.json (일반 구조물, rev {data['structures_revision']}){len(data['structures'])}건
횡단 설계 patch (DB){patch_state}
") # ① 배수 시설 일람 parts.append("

① 배수 시설 일람 (pipe_points.json)

") if data["pipes"]: parts.append( "" ) for point in sorted(data["pipes"], key=lambda p: p.chainage_m): options = point.options or {} inlet = options.get("inlet_type", "기슭막이(기본)") if point.facility == "pipe" else "—" parts.append( f"" f"" ) parts.append("
측점시설규격부속 구간(전~후)유입구출처
{point.chainage_m:.2f}m{_FACILITY_NAMES.get(point.facility, point.facility)}{_pipe_spec(point)}{_span_text(point)}{inlet}{point.source}
") else: parts.append("

배치된 배수 시설이 없다.

") # ② 일반 구조물 일람 parts.append("

② 일반 구조물 일람 (structures.json)

") if data["structures"]: parts.append( "" ) for instance in data["structures"]: option_text = ", ".join(f"{k}={v}" for k, v in (instance.options or {}).items()) or "—" parts.append( f"" f"" ) parts.append("
타입위치옵션배치상태
{_type_name(instance.type_id)}{_structure_position(instance)}{option_text}{instance.placement_source}{instance.status}
") else: parts.append("

배치된 일반 구조물이 없다.

") # ③ 횡단 벽 패치 일람 parts.append("

③ 기준벽·다단 기슭막이 (횡단 설계 patch)

") wall_rows = _wall_rows(data["designs"]) if wall_rows: parts.append( "" ) for row in wall_rows: counts = row["counts"] count_text = f"{counts.get('outlet', 0)} / {counts.get('basin', 0)}" if not row["walls"]: parts.append( f"" ) for index, wall in enumerate(row["walls"]): lead = ( ( f"" f"" ) if index == 0 else "" ) parts.append( f"{lead}" f"" ) parts.append("
측점다단(유출/계류)벽 키높이재질단별 구간
{row['chainage']:.2f}m{count_text}벽 조작값 없음(단 수만 지정)
{row['chainage']:.2f}m{count_text}
{wall['key']}{wall['height']}{wall['material']}{wall['length']}
") parts.append( "

벽 좌우·상하·높이·형태는 횡단 설계 patch, 다단 단별 구간값은 extra_spans 정본 (CLAUDE.md 5장).

" ) else: parts.append("

벽 조작·다단 지정 측점이 없다.

") # ④ 산출근거 (원단위 × 수량) parts.append("

④ 산출근거 — 원단위 × 수량

") if rollup["matched"]: for entry in rollup["matched"]: parts.append( f"

{entry['label']} × {_fmt(entry['quantity'])}{entry['per']}" f" [출처: {entry['source']} — 참조 전용]
" f"위치: {entry['positions']}" + (f"
{entry['note']}" if entry["note"] else "") + "

" ) parts.append( "" ) for line in entry["lines"]: parts.append( f"" f"" ) parts.append("
공종단위수량수량산출값
{line['item']}{line['unit_qty']}{_fmt(entry['quantity'])}{_fmt(line['value'])}
") else: parts.append("

원단위 매칭 가능한 구조물이 없다.

") # ⑤ 공종별 합계 parts.append("

⑤ 공종별 수량 합계

") if rollup["totals"]: parts.append("") for item, value in sorted(rollup["totals"].items()): parts.append(f"") parts.append("
공종합계
{item}{_fmt(value)}
") else: parts.append("

합산할 항목이 없다.

") # ⑥ 산출 보류 parts.append("

⑥ 산출 보류 (사유)

") if rollup["deferred"]: parts.append("") for row in rollup["deferred"]: parts.append(f"") parts.append("
대상사유
{row['name']}{row['reason']}
") else: parts.append("

보류 항목 없음.

") parts.append( "

근거 문서: resources/knowledge/original/실무문서/_원단위라이브러리_울진소광.md · " "_종합비교/04_임도구조물_원단위.md · technical_info/01_임도/04_수량분석정보/ (수량산출_일반·구조물_수량·배수공_수량)

" ) return "".join(parts) # ---------------------------------------------------------------- 라우트 @proto_router.get("/quantity-proto/page", response_class=HTMLResponse) async def quantity_proto_index() -> HTMLResponse: """프로젝트 선택 페이지.""" projects = await list_all_projects() rows = "".join( f"{p.get('name') or p['id']}" f"{p.get('region') or '—'}{p.get('status') or '—'}{p.get('updated_at') or ''}" for p in projects ) body = ( "

수량 산출 프로토타입 — 프로젝트 선택

" "" f"{rows}
프로젝트지역상태수정
" if projects else "

수량 산출 프로토타입

프로젝트가 없다.

" ) return _page("수량 산출 프로토타입", body) @proto_router.get("/quantity-proto/page/{project_id}", response_class=HTMLResponse) async def quantity_proto_report(project_id: UUID) -> HTMLResponse: """프로젝트 1건의 수량 산출 보고서.""" try: data = await _collect(project_id) except FileNotFoundError as error: return _page( "수량 산출 프로토타입", f"

수량 산출 프로토타입

{error}

" ) projects = await list_all_projects() project = next((p for p in projects if str(p.get("id")) == str(project_id)), None) body = _render_report(project, project_id, data) body += "

← 프로젝트 선택으로

" return _page("수량 산출 프로토타입", body)