eomsangdonandClaude Opus 5
036da3f9f9
777줄이던 _Handoff_Rows 를 565 + 255 로 가름. 줄의 모양은 안 바뀜 — 갈라낸 넷을 본체에서 다시 내보내 부르는 쪽 import 도 그대로임. 상태 낱말은 _Preparation_Status 한 벌에서 읽음. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
256 lines
12 KiB
Python
256 lines
12 KiB
Python
"""인계 줄 — **준비공·배수관·연장·자재** 네 갈래 (`Engine_Handoff_Rows` 에서 갈라냄).
|
|
|
|
⚠ **왜 갈랐나** — 700줄 제한(2026-09-09). **줄의 모양은 하나도 안 바뀐다.**
|
|
⚠ 줄 빌더가 여럿이라 칸을 하나 늘리면 **여기와 저기를 함께** 고쳐야 한다 — 계약 시험
|
|
(`tmp/tests/test_b08_handoff_contract.py`)이 「모든 줄이 같은 칸을 갖는가」로 그것을 지킨다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from B08_Quantity.B08_Quantity_Engine_Handoff_Mapping import (
|
|
BLOCKED_FORMULA_MISSING,
|
|
BLOCKED_INPUT_MISSING,
|
|
BLOCKED_UNIT_DATA_MISSING,
|
|
ORIGIN_PIPE,
|
|
ORIGIN_PREPARATION,
|
|
ORIGIN_STRUCTURE,
|
|
WorkItemMapping,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Engine_Preparation_Status import (
|
|
STATUS_COUNTED_ELSEWHERE as PREP_COUNTED_ELSEWHERE,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Engine_Preparation_Status import (
|
|
STATUS_NOT_APPLICABLE as PREP_NOT_APPLICABLE,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Engine_Preparation_Status import (
|
|
STATUS_PENDING as PREP_PENDING,
|
|
)
|
|
from B08_Quantity.B08_Quantity_Engine_Preparation_Status import (
|
|
STATUS_READY as PREP_READY,
|
|
)
|
|
|
|
|
|
def _preparation_rows(preparation_table: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""준비공·사방공 줄 — **값이 서는 줄도, 못 서는 줄도** 함께 보낸다.
|
|
|
|
⚠ **줄을 빼면 「빠졌다는 사실조차 안 보인다」** (2026-09-08 보조 창 제보).
|
|
받는 쪽 화면에서 「내역서에 원래 없는 것」과 「우리가 아직 못 내는 것」이 구별되지 않는다.
|
|
그래서 못 내는 줄도 `in_bill: False` + `blocked_reason` 으로 실어 보낸다 —
|
|
**금액은 안 붙되 「무엇이 채워지면 풀리는지」가 함께 간다.**
|
|
|
|
⚠ 이 표가 통째로 안 가고 있었다 — 표토제거(값 있음·`FP-09-15`)·규준틀(개소·`FP-11-02`)이
|
|
화면에는 서는데 인계에는 없었다. 「사유를 실어 달라」는 요청을 보다 드러났다.
|
|
"""
|
|
rows: list[dict[str, Any]] = []
|
|
for row in preparation_table.get("rows") or []:
|
|
status = str(row.get("status") or "")
|
|
amount = row.get("amount")
|
|
ready = status == PREP_READY and amount is not None
|
|
rows.append(
|
|
{
|
|
"work_item_code": row.get("work_item_code"),
|
|
"name": str(row.get("item") or ""),
|
|
"spec": str(row.get("group") or ""),
|
|
"unit": str(row.get("unit") or ""),
|
|
"quantity": float(amount or 0.0),
|
|
"quantity_gross": 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,
|
|
# ⚠ **값이 서는 줄에도 근거를 싣는다**(2026-09-09) — 종전에는 대분류 이름만
|
|
# 실어, 값이 서는 순간 **왜 그 값인지가 사라졌다**(제근이 면적 축으로 확정돼
|
|
# 값이 서자 「교차 참조: 건설품셈 3-9-2」가 화면에서 사라진 자리).
|
|
# 못 서는 줄은 종전대로 `blocked_reason` 이 따로 든다.
|
|
"spec_detail": " · ".join(
|
|
part
|
|
for part in (str(row.get("group") or ""), str(row.get("reason") or ""))
|
|
if part
|
|
),
|
|
"composite_parts": None,
|
|
"structure_kind": None,
|
|
# ⚠ 못 서는 까닭을 그대로 넘긴다 — 받는 쪽이 「만들어야 할 것」 목록에 얹는다.
|
|
"blocked_kind": None if ready else _prep_blocked_kind(status),
|
|
"blocked_reason": "" if ready else str(row.get("reason") or status),
|
|
# ⚠ 준비공 줄도 갈래를 실어 보낸다(2026-09-09) — 종전에는 늘 `None` 이라
|
|
# 표토 운반처럼 **부모 공종코드**로 가는 줄이 B09 에서 「후보 N건」에 머물렀다.
|
|
"variant_axis": row.get("variant_axis"),
|
|
"variant_value": row.get("variant_value"),
|
|
"secondary_axes": None,
|
|
"spec_class": None,
|
|
"spec_class_basis": str(row.get("reason") or ""),
|
|
"composite_not_ready": None,
|
|
# 값이 없는 줄은 **내역에 세우지 않는다** — 0 원 줄을 만들면 더 나쁘다.
|
|
"in_bill": ready,
|
|
"in_bill_reason": "" if ready else str(row.get("reason") or status),
|
|
"origin": ORIGIN_PREPARATION,
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def _prep_blocked_kind(status: str) -> str | None:
|
|
"""준비공 줄의 상태를 **막힌 갈래 셋** 중 하나로 옮긴다. 모르는 상태는 `None`."""
|
|
if status == PREP_PENDING:
|
|
return BLOCKED_INPUT_MISSING
|
|
if status == PREP_COUNTED_ELSEWHERE:
|
|
# 다른 표에서 이미 선 줄 — 막힌 것이 아니라 **여기서 세면 안 되는** 줄이다.
|
|
return None
|
|
if status == PREP_NOT_APPLICABLE:
|
|
return None
|
|
return BLOCKED_UNIT_DATA_MISSING
|
|
|
|
|
|
def _pipe_rows(pipe_table: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""배수관 줄 — 값이 서는 줄도, 못 서는 줄도 함께 보낸다(준비공과 같은 규칙).
|
|
|
|
⚠ 터파기·되메우기를 붙이지 않는다 — 관 부설과 굴착이 각각 오면 **같은 굴착을 두 번** 센다
|
|
(B09 ㉡ 가드와 같은 자리).
|
|
"""
|
|
rows: list[dict[str, Any]] = []
|
|
for row in pipe_table.get("rows") or []:
|
|
ready = bool(row.get("in_bill"))
|
|
rows.append(
|
|
{
|
|
"work_item_code": row.get("work_item_code"),
|
|
"name": f"배수관({row.get('kind')})",
|
|
"spec": f"Ø{row.get('variant_value')}" if row.get("variant_value") else "",
|
|
"unit": str(row.get("unit") or "m"),
|
|
"quantity": float(row.get("quantity") or 0.0),
|
|
"quantity_gross": None,
|
|
"application_ratio_pct": None,
|
|
"application_ratio_breakdown": None,
|
|
"quantity_breakdown": None,
|
|
"ground_class": None,
|
|
"haul_distance_m": None,
|
|
"haul_equipment": None,
|
|
"station_from": row.get("chainage_m"),
|
|
"station_to": row.get("chainage_m"),
|
|
"excavation_method": None,
|
|
"spec_detail": f"Ø{row.get('variant_value')}" if row.get("variant_value") else "",
|
|
"composite_parts": None,
|
|
"structure_kind": None,
|
|
"blocked_kind": row.get("blocked_kind"),
|
|
"blocked_reason": str(row.get("blocked_reason") or ""),
|
|
# 갈래는 **저장 원본값**만 — 「…㎜ 이하」 구간 나누기는 원문을 읽는 쪽 몫.
|
|
"variant_axis": row.get("variant_axis"),
|
|
"variant_value": row.get("variant_value"),
|
|
"secondary_axes": None,
|
|
"spec_class": None,
|
|
"spec_class_basis": str(row.get("blocked_reason") or ""),
|
|
"composite_not_ready": None,
|
|
"in_bill": ready,
|
|
"in_bill_reason": "" if ready else str(row.get("blocked_reason") or ""),
|
|
"origin": ORIGIN_PIPE,
|
|
}
|
|
)
|
|
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)
|
|
# 구간 목록 — **겹침을 지운 뒤**의 것이라 그 합이 곧 `length_m` 이다
|
|
# (80~120 과 100~140 은 80~140 한 줄로 합쳐져 온다, 2026-09-08 랩탑 창).
|
|
# ⚠ 표기(`NO.4+0.0`)는 만들지 않는다 — 측점 간격을 아는 화면 몫이다.
|
|
spans = [
|
|
span
|
|
for span in (entry.get("spans") or [])
|
|
if span.get("start_m") is not None and span.get("end_m") is not None
|
|
]
|
|
span_note = " · ".join(f"{s['start_m']:g}~{s['end_m']:g}m" for s in spans)
|
|
note = f"구간 {span_note}" if span_note else ""
|
|
if abs(raw - length) > 1e-9:
|
|
겹침 = f"입력 구간 합 {raw:g}m 에서 겹친 {raw - length:g}m 를 뺀 값"
|
|
note = f"{note} · {겹침}" if note else 겹침
|
|
# ⚠ 겹침 설명은 **비고**이지 막힌 사유가 아니다 — `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": spans[0]["start_m"] if spans else None,
|
|
"station_to": spans[-1]["end_m"] if spans else 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,
|
|
"secondary_axes": 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]] = []
|
|
for row in material_table.get("rows") or []:
|
|
rows.append(
|
|
{
|
|
"material_name": row.get("name"),
|
|
"spec": row.get("spec") or "",
|
|
"unit": row.get("unit"),
|
|
"net_amount": row.get("net_amount"),
|
|
"total_amount": row.get("total_amount"),
|
|
"surcharge_pct": row.get("surcharge_pct"),
|
|
"surcharge_note": row.get("note") or "",
|
|
"supply_type": row.get("supply"),
|
|
"install_by": row.get("install_by"),
|
|
"source_structure": row.get("sources") or [],
|
|
}
|
|
)
|
|
return rows
|