feat(B08): 토적표 엔진·API — 평균단면적법 체적화

PLAN 8-4b 열 명세대로. B06 이 이미 낸 측점별 단면적을 다시 재지 않고
체적화만 함 — 새 수량을 낳지 않으므로 캐시·조작 경로 불필요(CLAUDE.md 5장).

열 구성 (실무 토적표 = 오솔길 1.BOM 36열과 1:1)
  측점·거리·절토[토사·암 각 단면적/입적/보정량]·측구터파기[토사·암 각 3칸]
  ·보정량계·성토[단면적/입적]·유용토·차인토량·누가토량.
  사면 계열(층따기·면고르기·법면보호공·지장목제거)은 사면길이가 아직 없어 일감 3.

계산 규칙
  체적 = (앞 단면적 + 현 단면적)/2 x 거리. 첫 측점은 앞이 없어 체적 없음.
  보정량 = 체적 x 다짐 환산계수 — 정의처는 EARTHWORK_CONVERSION_FACTORS 한 곳,
  여기서 값을 다시 적지 않음(토사 0.90 / 리핑암 1.15 / 발파암 1.30).
  값을 자르지 않음 — 품셈 1-2-2 는 표기 규칙이고 절사는 화면 몫(PLAN 8-16).
  원가 쪽(줄마다 원 단위 절사)과 규칙이 반대라 섞지 말 것.

TODO(미결 · PLAN 8-4b) — 설계가 측구를 토사·암으로 안 나눠 줌(ditch_area_m2 한 값).
  잠정으로 그 측점 절토 토사:암 면적비로 안분함. 측구는 절토부에 파므로 같은
  지반을 만난다는 것이 근거. 설계가 측구 지반을 따로 내면 _split_ditch 만 교체.

API — GET /api/projects/{id}/quantity/{route_id}/earthwork-table,
  경로 생략형은 워크플로 최신 노선으로. main.py 는 자기 두 줄만 추가.

검증 — tmp/tests/test_b08_earthwork_table.py 14건 통과.
  거창 실무 BOM 실측값 재현(단면적 1.89 → 체적 9.45 → 보정 8.505,
  다음 측점 18.10 → 16.29, 측구 0.18㎡ → 10m당 1.80 → 1.62).
  공용 브라우저에서 실 API 호출로 route 150·측점 65곳 확인 — 측점 20 에서
  (0.2723+1.9615)/2x20 = 22.338, x0.9 = 20.1042, 암 25.427 x1.15 = 29.24105,
  측구 안분 0.0784+0.1016 = 0.18 로 전건 일치.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

B08_Quantity/B08_Quantity_Engine_EarthworkTable.py

@@ -0,0 +1,209 @@
+"""토적표 — 측점별 단면적을 평균단면적법으로 체적화한다 (B08 일감 2 · PLAN 8-4b).
+
+무엇을 만드나
+    실무 토적표의 열 구성 그대로다. 거창 실무 워크북 `토적표` 시트와 오솔길 `1.BOM` 36열이
+    서로 1:1 로 맞물리는 것을 확인해 열 이름을 그대로 옮겼다(PLAN 8-4b).
+
+    측점 · 거리 · 절토[토사·암 각 (단면적·입적·보정량)] · 측구터파기[토사·암 각 3칸]
+    · 보정량계 · 성토[단면적·입적] · 유용토 · 차인토량 · 누가토량
+
+    사면 계열(층따기·면고르기·법면보호공·지장목제거)은 사면길이가 아직 없어 일감 3에서 붙인다.
+
+평균단면적법 (신규 문서 5장 「다. 공사수량의 산출」)
+    체적 = (앞 측점 단면적 + 현 측점 단면적) ÷ 2 × 두 측점 사이 거리.
+    첫 측점은 앞이 없으므로 체적이 없다(거창 실무 토적표도 첫 행 체적이 비어 있다).
+
+보정량 = 체적 × 토량환산계수(다짐)
+    절취한 흙이 다져지면 줄거나 부푼다. 성토에 쓸 수 있는 양으로 환산한 것이 보정량이다.
+    계수의 유일한 정의처는 `config.config_system_design.EARTHWORK_CONVERSION_FACTORS` 이며
+    여기서 값을 다시 적지 않는다.
+
+⚠ 숫자는 자르지 않는다 (PLAN 8-16)
+    품셈 1-2-2 의 소수 자리는 **표기 규칙**이다. 계산은 전정밀로 두고 화면·출력에서만
+    반올림한다. 원가 쪽(줄마다 원 단위 절사)과 규칙이 반대이므로 섞지 말 것.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Iterable
+
+from config.config_system_design import EARTHWORK_CONVERSION_FACTORS
+
+# 절토 암을 어느 환산계수로 볼지 — 측점의 `cut_rock_kind` 를 그대로 쓴다.
+# 값이 없으면 리핑암으로 본다(발파암보다 보수적으로 적은 쪽).
+_DEFAULT_ROCK_KIND = "ripping_rock"
+
+
+def _factor(kind: str) -> float:
+    """지반유형 → 다짐 환산계수. 모르는 유형이면 토사로 본다."""
+    entry = EARTHWORK_CONVERSION_FACTORS.get(kind) or EARTHWORK_CONVERSION_FACTORS["soil"]
+    return float(entry["compacted"])
+
+
+@dataclass(slots=True)
+class StationArea:
+    """토적표 한 줄이 필요로 하는 측점 값. B06 설계 결과에서 그대로 옮겨 담는다."""
+
+    chainage_m: float
+    cut_soil_area_m2: float = 0.0
+    cut_rock_area_m2: float = 0.0
+    fill_area_m2: float = 0.0
+    ditch_area_m2: float = 0.0
+    cut_rock_kind: str | None = None
+
+    @classmethod
+    def from_design(cls, chainage_m: float, design: dict[str, Any]) -> "StationArea":
+        def num(key: str) -> float:
+            value = design.get(key)
+            return float(value) if isinstance(value, (int, float)) else 0.0
+
+        return cls(
+            chainage_m=float(chainage_m),
+            cut_soil_area_m2=num("cut_soil_area_m2"),
+            cut_rock_area_m2=num("cut_rock_area_m2"),
+            fill_area_m2=num("fill_area_m2"),
+            ditch_area_m2=num("ditch_area_m2"),
+            cut_rock_kind=design.get("cut_rock_kind") or None,
+        )
+
+
+@dataclass(slots=True)
+class EarthworkRow:
+    """토적표 한 줄. 열 이름은 실무 토적표(PLAN 8-4b)를 따른다."""
+
+    chainage_m: float
+    distance_m: float = 0.0
+    cut_soil_area_m2: float = 0.0
+    cut_soil_volume_m3: float = 0.0
+    cut_soil_adjusted_m3: float = 0.0
+    cut_rock_area_m2: float = 0.0
+    cut_rock_volume_m3: float = 0.0
+    cut_rock_adjusted_m3: float = 0.0
+    ditch_soil_area_m2: float = 0.0
+    ditch_soil_volume_m3: float = 0.0
+    ditch_soil_adjusted_m3: float = 0.0
+    ditch_rock_area_m2: float = 0.0
+    ditch_rock_volume_m3: float = 0.0
+    ditch_rock_adjusted_m3: float = 0.0
+    adjusted_total_m3: float = 0.0
+    fill_area_m2: float = 0.0
+    fill_volume_m3: float = 0.0
+    diverted_m3: float = 0.0
+    balance_m3: float = 0.0
+    cumulative_m3: float = 0.0
+    notes: list[str] = field(default_factory=list)
+
+
+def _split_ditch(area: StationArea) -> tuple[float, float]:
+    """측구터파기 단면적을 토사·암으로 가른다.
+
+    ⚠ TODO(미결 · PLAN 8-4b) — 설계가 측구를 토사·암으로 나눠 주지 않는다(`ditch_area_m2`
+    한 값뿐). 실무 토적표는 둘로 갈라 적으므로, **그 측점의 절토 토사:암 면적비로 안분**한다.
+    측구는 절토부에 파므로 같은 지반을 만난다는 것이 근거다. 설계가 측구 지반을 따로 내주게
+    되면 이 함수만 갈아끼운다.
+    """
+    ditch = area.ditch_area_m2
+    if ditch <= 0:
+        return 0.0, 0.0
+    soil, rock = area.cut_soil_area_m2, area.cut_rock_area_m2
+    total = soil + rock
+    if total <= 0:
+        return ditch, 0.0  # 절토가 없으면 토사로 본다.
+    return ditch * soil / total, ditch * rock / total
+
+
+def build_rows(stations: Iterable[StationArea]) -> list[EarthworkRow]:
+    """측점 목록 → 토적표 줄 목록. 측점은 이정 순으로 정렬해 받는다."""
+    ordered = sorted(stations, key=lambda s: s.chainage_m)
+    rows: list[EarthworkRow] = []
+    previous: StationArea | None = None
+    previous_ditch: tuple[float, float] = (0.0, 0.0)
+    cumulative = 0.0
+
+    for station in ordered:
+        ditch_soil, ditch_rock = _split_ditch(station)
+        soil_factor = _factor("soil")
+        rock_factor = _factor(station.cut_rock_kind or _DEFAULT_ROCK_KIND)
+        row = EarthworkRow(
+            chainage_m=station.chainage_m,
+            cut_soil_area_m2=station.cut_soil_area_m2,
+            cut_rock_area_m2=station.cut_rock_area_m2,
+            ditch_soil_area_m2=ditch_soil,
+            ditch_rock_area_m2=ditch_rock,
+            fill_area_m2=station.fill_area_m2,
+        )
+        if previous is not None:
+            distance = station.chainage_m - previous.chainage_m
+            row.distance_m = distance
+
+            def mean_volume(before: float, now: float) -> float:
+                return (before + now) / 2.0 * distance
+
+            row.cut_soil_volume_m3 = mean_volume(
+                previous.cut_soil_area_m2, station.cut_soil_area_m2
+            )
+            row.cut_rock_volume_m3 = mean_volume(
+                previous.cut_rock_area_m2, station.cut_rock_area_m2
+            )
+            row.ditch_soil_volume_m3 = mean_volume(previous_ditch[0], ditch_soil)
+            row.ditch_rock_volume_m3 = mean_volume(previous_ditch[1], ditch_rock)
+            row.fill_volume_m3 = mean_volume(previous.fill_area_m2, station.fill_area_m2)
+
+            row.cut_soil_adjusted_m3 = row.cut_soil_volume_m3 * soil_factor
+            row.cut_rock_adjusted_m3 = row.cut_rock_volume_m3 * rock_factor
+            row.ditch_soil_adjusted_m3 = row.ditch_soil_volume_m3 * soil_factor
+            row.ditch_rock_adjusted_m3 = row.ditch_rock_volume_m3 * rock_factor
+
+        row.adjusted_total_m3 = (
+            row.cut_soil_adjusted_m3
+            + row.cut_rock_adjusted_m3
+            + row.ditch_soil_adjusted_m3
+            + row.ditch_rock_adjusted_m3
+        )
+        # 유용토 = 그 측점에서 절취분과 성토분이 서로 만나는 몫.
+        row.diverted_m3 = min(row.adjusted_total_m3, row.fill_volume_m3)
+        row.balance_m3 = row.adjusted_total_m3 - row.fill_volume_m3
+        cumulative += row.balance_m3
+        row.cumulative_m3 = cumulative
+
+        rows.append(row)
+        previous = station
+        previous_ditch = (ditch_soil, ditch_rock)
+    return rows
+
+
+def totals(rows: list[EarthworkRow]) -> dict[str, float]:
+    """합계 행. 단면적은 합이 뜻이 없어 싣지 않는다(실무 토적표도 비워 둔다)."""
+    keys = (
+        "distance_m",
+        "cut_soil_volume_m3",
+        "cut_soil_adjusted_m3",
+        "cut_rock_volume_m3",
+        "cut_rock_adjusted_m3",
+        "ditch_soil_volume_m3",
+        "ditch_soil_adjusted_m3",
+        "ditch_rock_volume_m3",
+        "ditch_rock_adjusted_m3",
+        "adjusted_total_m3",
+        "fill_volume_m3",
+        "diverted_m3",
+        "balance_m3",
+    )
+    return {key: sum(getattr(row, key) for row in rows) for key in keys}
+
+
+def build_table(stations: Iterable[StationArea]) -> dict[str, Any]:
+    """화면·API 가 그대로 쓰는 모양. 값은 자르지 않는다(PLAN 8-16)."""
+    rows = build_rows(stations)
+    return {
+        "method": "average_end_area",
+        "conversion_factors": EARTHWORK_CONVERSION_FACTORS,
+        "rows": [row.__dict__ if not hasattr(row, "__slots__") else _as_dict(row) for row in rows],
+        "totals": totals(rows),
+        "station_count": len(rows),
+    }
+
+
+def _as_dict(row: EarthworkRow) -> dict[str, Any]:
+    return {name: getattr(row, name) for name in EarthworkRow.__slots__}

B08_Quantity/B08_Quantity_Router_Earthwork.py

@@ -0,0 +1,64 @@
+"""B08 토적표 조회 라우터 (일감 2 · PLAN 8-4b).
+
+값은 어디서 오나
+    측점별 단면적은 **B06 이 이미 낸 정본**이다(`cross_sections.data.design` 의
+    `cut_soil_area_m2`·`cut_rock_area_m2`·`fill_area_m2`·`ditch_area_m2`).
+    B08 은 그것을 다시 재지 않고 **평균단면적법으로 체적화만** 한다.
+
+계산 자리 (CLAUDE.md 5장)
+    초기값은 서버가 한 번 계산해 영구저장한다. 여기서는 저장된 단면적을 읽어 표를 만든다 —
+    새 수량을 낳지 않으므로 캐시·조작 경로가 따로 필요 없다.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+from B06_Section.B06_Section_Repository import (
+    get_cross_section_designs,
+    get_workflow_route_context,
+)
+from B08_Quantity.B08_Quantity_Engine_EarthworkTable import StationArea, build_table
+from config.config_db import run_with_connection
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"])
+
+
+def _stations(designs: list[dict[str, Any]]) -> list[StationArea]:
+    return [
+        StationArea.from_design(item["chainage_m"], item.get("design") or {}) for item in designs
+    ]
+
+
+@router.get("/{project_id}/quantity/{route_id}/earthwork-table")
+async def get_earthwork_table(project_id: UUID, route_id: int) -> JSONResponse:
+    """토적표 — 측점별 단면적을 평균단면적법으로 체적화한 표."""
+    try:
+        designs = await run_with_connection(get_cross_section_designs, route_id)
+    except Exception:
+        logger.exception("B08 토적표 조회 실패: project_id=%s route_id=%s", project_id, route_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "토적표를 만들지 못했습니다."},
+        )
+    table = build_table(_stations(designs))
+    table["route_id"] = route_id
+    return JSONResponse(content=table)
+
+
+@router.get("/{project_id}/quantity/earthwork-table")
+async def get_earthwork_table_for_current_route(project_id: UUID) -> JSONResponse:
+    """경로를 안 주면 워크플로가 보고 있는 경로로 낸다 — 화면이 route_id 를 모를 때 쓴다."""
+    context = await run_with_connection(get_workflow_route_context, project_id)
+    if not context or not context.get("route_id"):
+        return JSONResponse(
+            status_code=404,
+            content={"status": "error", "message": "이 프로젝트에 확정된 노선이 없습니다."},
+        )
+    return await get_earthwork_table(project_id, int(context["route_id"]))

main.py

@@ -59,6 +59,7 @@ from B06_Section.B06_Section_Router_HaulPlan import (
 from B07_DesignDetail.B07_DesignDetail_Router import router as b07_design_router
 from B07_DesignDetail.B07_DesignDetail_Router_Frame import router as b07_frame_router
 from B08_Quantity.B08_Quantity_Router import router as b08_quantity_router
+from B08_Quantity.B08_Quantity_Router_Earthwork import router as b08_earthwork_router
 from common_util.common_util_audit import note_api_call, record_call_burst
 from common_util.common_util_auth import (
     require_company,
@@ -535,6 +536,7 @@ app.include_router(b06_section_haul_plan_router, dependencies=protected_with_com
 app.include_router(b07_design_router, dependencies=protected_with_company)
 app.include_router(b07_frame_router, dependencies=protected_with_company)
 app.include_router(b08_quantity_router, dependencies=protected_with_company)
+app.include_router(b08_earthwork_router, dependencies=protected_with_company)
 
 
 # ─────────────────────────────────────────────────────────────────────────