feat(B08): B군 종단배수를 연장표로 받음 — 겹친 구간을 두 번 안 셈

랩탑 제안(두 창 합의). common_util_structure_lengths 가 이미 규칙 셋을 갖고
있는데 B08 이 structures.json 을 직접 읽어 세면 그것을 두 벌로 짜게 됨.

그 함수가 갖고 있는 것
  ① 겹친 구간 합치기 (같은 시설을 겹쳐 놓으면 단순 합은 두 번 셈)
  ② 측구 제외 (design_owner — 횡단이 이미 ditch_area_m2 로 셈)
  ③ 관 소관 제외 (managed_by)

⚠ B군만 그 함수로 감 — C군(돌쌓기·옹벽)은 그대로 구조물별 줄
  그 함수는 종류별로 뭉쳐 내는데, C군은 측점·규격이 줄마다 다르고 자재가
  줄마다 나옴. 실무 내역도 B군은 「산마루측구 40m」 한 줄, C군은 구조물별 줄

실측: 겹친 산마루측구 둘(80~120 · 100~140) → 80m 이 아니라 60m
      비고 「입력 구간 합 80m 에서 겹친 20m 를 뺀 값」

⚠ 만들다 낸 것 하나 — 겹침 설명을 blocked_reason 에 넣었더니 B09 가
「막힌 줄」로 읽어 금액을 안 붙였음. 비고(spec_class_basis)로 옮김.
막힌 사유는 진짜 막힐 때만(공종 못 이음 · 연장 0).

시험 다섯 — 종류별 한 줄 · 겹침이 합쳐짐 · ⚠ 겹침 설명은 막힘이 아님 ·
⚠ 안 겹치면 원합을 안 실음(반영률로 오해됨) · 공종 못 이으면 사유와 함께 막힘

시험: 709 passed · 24 skipped (B05 코리도 1건 기존 깨짐, 무관).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-08 10:22:52 +09:00
co-authored by Claude Opus 5
parent cbf991d5be
commit d3a763d9c9
2 changed files with 79 additions and 0 deletions
@@ -918,6 +918,74 @@ def _pipe_rows(pipe_table: dict[str, Any]) -> list[dict[str, Any]]:
return rows
def _length_rows(
length_table: list[dict[str, Any]], mapping: WorkItemMapping
) -> list[dict[str, Any]]:
"""B군 종단배수 — **종류별 한 줄**로 낸다(연장이 곧 수량).
⚠ **왜 구조물별로 안 내나** — `common_util_structure_lengths` 가 **겹친 구간을 합쳐**
준다. 같은 시설을 겹쳐 놓으면 구조물별로 세는 순간 그 구간을 **두 번** 센다.
그 규칙(겹침 합치기 · 측구 제외 · 관 소관 제외)이 이미 그 함수에 있으므로
**두 벌로 짜지 않는다**(2026-09-08 랩탑 창 제안, 두 창 합의).
⚠ **C군(돌쌓기·옹벽 등)은 여기로 오지 않는다** — 그 함수는 종류별로 뭉쳐 내는데,
C군은 **측점·규격이 줄마다 달라** 구조물별로 서야 하고 자재도 줄마다 나온다.
실무 내역도 B군은 「산마루측구 40m」 한 줄, C군은 구조물별 줄이다.
⚠ 겹침이 있으면(`length_m != raw_length_m`) **숨기지 않고 비고에 적는다.**
"""
rows: list[dict[str, Any]] = []
for entry in length_table or []:
type_id = str(entry.get("type_id") or "")
found = mapping.for_structure(type_id)
code = (found or {}).get("work_item_code")
length = float(entry.get("length_m") or 0.0)
raw = float(entry.get("raw_length_m") or length)
note = ""
if abs(raw - length) > 1e-9:
note = f"입력 구간 합 {raw:g}m 에서 겹친 {raw - length:g}m 를 뺀 값"
# ⚠ 겹침 설명은 **비고**이지 막힌 사유가 아니다 — `blocked_reason` 에 넣으면
# 받는 쪽이 「막힌 줄」로 읽어 금액을 안 붙인다(2026-09-08 실측에서 그랬다).
reason = ""
if code is None:
reason = f"{entry.get('name') or type_id} — 품셈 공종을 아직 못 이었습니다"
elif length <= 0:
reason = f"{entry.get('name') or type_id} — 연장이 0 이라 값이 서지 않습니다"
rows.append(
{
"work_item_code": code,
"name": str(entry.get("name") or type_id),
"spec": f"{entry.get('count')}개소",
"unit": "m",
"quantity": length,
"quantity_gross": raw if note else None,
"application_ratio_pct": None,
"application_ratio_breakdown": None,
"quantity_breakdown": None,
"ground_class": None,
"haul_distance_m": None,
"haul_equipment": None,
"station_from": None,
"station_to": None,
"excavation_method": None,
"spec_detail": f"{entry.get('count')}개소",
"composite_parts": None,
"structure_kind": None,
"blocked_kind": None if (code and length > 0) else BLOCKED_FORMULA_MISSING,
"blocked_reason": reason,
"variant_axis": None,
"variant_value": None,
"spec_class": None,
"spec_class_basis": note,
"composite_not_ready": None,
"in_bill": bool(code and length > 0),
"in_bill_reason": "" if (code and length > 0) else reason,
"origin": ORIGIN_STRUCTURE,
}
)
return rows
def _material_rows(material_table: dict[str, Any]) -> list[dict[str, Any]]:
"""자재 줄 — **공종코드를 붙이지 않는다.** 자재 축은 B09 카탈로그가 잇는다(8-7)."""
rows: list[dict[str, Any]] = []
@@ -946,6 +1014,7 @@ def build_handoff(
unit_quantity_table: dict[str, Any] | None = None,
material_table: dict[str, Any] | None = None,
preparation_table: dict[str, Any] | None = None,
length_table: list[dict[str, Any]] | None = None,
pipe_table: dict[str, Any] | None = None,
mapping: WorkItemMapping | None = None,
ground_class_set: str | None = None,
@@ -973,6 +1042,8 @@ def build_handoff(
unmatched.extend(misses)
# 준비공·사방공 — **못 내는 줄도 사유와 함께** 보낸다(빼면 빠진 줄이 안 보인다).
# B군 종단배수 — 겹침을 합친 연장으로 종류별 한 줄(위 `_length_rows` 주석).
work_items.extend(_length_rows(length_table or [], table))
work_items.extend(_preparation_rows(preparation_table or {}))
# 배수관 — 정본 셋(관 지점·측점 연장·매핑)을 이은 결과. 못 서는 줄도 사유와 함께 감.
work_items.extend(_pipe_rows(pipe_table or {}))
@@ -37,6 +37,7 @@ from common_util.common_util_project_settings import (
rock_method,
)
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__)
@@ -68,6 +69,11 @@ def _collect_structures(
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))
@@ -163,6 +169,8 @@ async def get_handoff(project_id: UUID) -> JSONResponse:
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,