eomsangdonandClaude Opus 5
6e7570eba6
메인 창의 공종 축(`resources/data_work_item_master/`)은 **읽기만** 하고, 결과는 **내 소관 경로** `resources/data_cost_resource_axis/` 에 냄 — 메인이 품셈을 다시 돌리면 그 폴더가 덮이므로 안에 섞으면 사라짐. - ⚠ **`pum_form` 을 먼저 봄** — `productivity` = 1 ÷ 값 · `requirement` = 값 ÷ 기준수량. **뒤집으면 20배 틀림.** 형태를 모르는 표(`undetermined`)는 **값을 안 씀**, 한쪽으로 단정하지 않고 멈춤. 테스트가 뒤집힘을 자릿수로 잡음. - `coefficient`(19)·`reference`(94)는 공종이 아니므로 **일위대가 항목으로 안 세움**. - **규격이 없으면 매칭 성공으로 안 침** — 「굴착기」와 「굴착기 0.7㎥」는 단가가 다름. 이름이 여럿이면 규격 없이는 못 고르고 `unmatched` 로 감. - **못 맞춘 것은 빈칸이 아니라 목록** — 다만 표 머리글·소계·기호(`f`·`E`)·규격 구간 (`10∼12`·`50이상`)·절번호는 자원이 아니므로 먼저 걷어냄. 안 걸렀을 때 1,550건이던 목록이 **274건의 실제 공백**으로 줄어 쓸 수 있는 목록이 됨. 산출 — 매칭 **86행**(노무, 7직종: 보통인부 59·특별인부 19·조경공 4 등) · 못 맞춘 **274건**(상위: 치즐소모량·자재·굴착기(무한궤도) 등). ⚠ **노임 카탈로그만 있어 노무만 맞음** — 기계·자재 카탈로그가 아직 없어 그 계열은 전부 못 맞춘 목록으로 감. 사유 문구에 그 사실을 적어 「이름이 틀림」과 구분함. 자체검증 — 신규 12건 포함 `pytest tmp/tests/ -q` **97 passed** · ruff 통과 · 409줄. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
410 lines
15 KiB
Python
410 lines
15 KiB
Python
"""B09 원가계산 — 자원 축 매칭 (PLAN 8-6 · 9-3).
|
||
|
||
메인 창이 낸 **공종 축**(`resources/data_work_item_master/`)을 **읽기만** 하고, 그 위에
|
||
**자원 축**(`resource_kind`·`resource_code`·`resource_spec`·`amount`·`amount_unit`)을
|
||
붙여 **별도 파일**로 낸다. 원본은 고치지 않는다 — 메인이 품셈을 다시 돌리면 덮이므로
|
||
그 안에 섞으면 사라진다.
|
||
|
||
지켜야 할 것
|
||
1. **`pum_form` 을 먼저 본다.** `productivity`(생산량형) = **1 ÷ 값**,
|
||
`requirement`(소요량형) = **값 ÷ basis_quantity**. ⚠ **뒤집으면 20배 틀린다.**
|
||
직종 이름부터 보면 「작업능력(㎥/hr)」 표의 비고란 「보통인부 1인/일」에 끌려
|
||
생산량형을 소요량형으로 읽는다(메인이 실제로 한 번 뒤집혔다가 잡은 자리).
|
||
2. **`coefficient` · `reference` 는 공종이 아니다.** 일위대가 항목으로 세우지 않는다.
|
||
`undetermined` 는 **값을 쓰지 않는다.**
|
||
3. **규격(`resource_spec`)이 없으면 매칭 성공으로 치지 않는다.** 「굴착기」와
|
||
「굴착기 0.7㎥」는 단가가 다르다 — 이름만 맞추면 조용히 틀린 단가가 붙는다.
|
||
4. **못 맞춘 것은 빈칸이 아니라 `unmatched` 목록**으로 낸다.
|
||
5. **자재는 할증 전 값**이다 (PLAN 8-7 ㉠). 할증은 자재총괄 한 곳뿐이다.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from decimal import Decimal, InvalidOperation
|
||
from typing import Any
|
||
|
||
#: 자원 축을 붙일 수 있는 표 형태. 나머지는 값을 쓰지 않는다.
|
||
USABLE_FORMS = frozenset({"productivity", "requirement"})
|
||
#: 공종이 아닌 표 — 일위대가 항목으로 세우지 않는다.
|
||
NON_WORK_ITEM_FORMS = frozenset({"coefficient", "reference"})
|
||
#: 형태 판정이 안 된 표 — 값을 쓰지 않는다.
|
||
UNUSABLE_FORMS = frozenset({"undetermined"})
|
||
|
||
_MASTER_SUBPATH = ("resources", "data_work_item_master")
|
||
_CATALOG_SUBPATH = ("resources", "data_cost_input_value")
|
||
|
||
#: 규격이 이름 안에 붙어 있는 흔한 모양 — 「굴착기(0.7㎥)」·「덤프트럭 15톤」.
|
||
_RE_SPEC = re.compile(r"[(（]([^)）]+)[)）]|(\d+(?:\.\d+)?\s*(?:톤|ton|㎥|m3|㎡|㎜|mm|HP|kW))")
|
||
_RE_NUMBER = re.compile(r"^-?\d+(?:,\d{3})*(?:\.\d+)?$")
|
||
|
||
#: 표 머리글·소계 행의 첫 칸에 흔히 오는 말. 자원이 아니므로 `unmatched` 로도 안 올린다.
|
||
#: 이것을 안 거르면 못 맞춘 목록이 머리글로 가득 차 **쓸 수 없는 목록**이 된다.
|
||
_NON_RESOURCE_WORDS = (
|
||
"구분",
|
||
"합계",
|
||
"소계",
|
||
"계",
|
||
"단위",
|
||
"비고",
|
||
"규격",
|
||
"명칭",
|
||
"품명",
|
||
"종류",
|
||
"항목",
|
||
"적용",
|
||
"기준",
|
||
"산출",
|
||
"비율",
|
||
"할증",
|
||
"할인",
|
||
"직접노무비",
|
||
"재료비",
|
||
"경비",
|
||
"위치",
|
||
"면적",
|
||
"수량",
|
||
"공종",
|
||
"작업",
|
||
"내역",
|
||
"총계",
|
||
"인력",
|
||
"장비",
|
||
"기계",
|
||
)
|
||
|
||
|
||
class ResourceAxisError(ValueError):
|
||
"""자원 축을 붙일 수 없는 경우. 조용히 넘기지 않는다."""
|
||
|
||
|
||
def _project_root() -> str:
|
||
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
|
||
def _read_json(*parts: str) -> dict[str, Any]:
|
||
with open(os.path.join(_project_root(), *parts), encoding="utf-8") as handle:
|
||
return json.load(handle)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CatalogEntry:
|
||
"""단가 카탈로그 한 줄 — 매칭 대상."""
|
||
|
||
code: str
|
||
name: str
|
||
kind: str
|
||
spec: str = ""
|
||
|
||
|
||
@dataclass
|
||
class ResourceCatalog:
|
||
"""이름 → 코드. **같은 이름에 규격이 여럿이면 규격 없이는 못 고른다.**"""
|
||
|
||
entries: list[CatalogEntry] = field(default_factory=list)
|
||
aliases: dict[str, str] = field(default_factory=dict)
|
||
|
||
def by_name(self, name: str) -> list[CatalogEntry]:
|
||
cleaned = _normalize(name)
|
||
return [e for e in self.entries if _normalize(e.name) == cleaned]
|
||
|
||
def resolve(self, name: str, spec: str) -> CatalogEntry | None:
|
||
"""이름(+규격)으로 한 줄을 고른다. 못 고르면 None — 0 으로 안 때운다."""
|
||
found = self.by_name(name)
|
||
if not found:
|
||
return None
|
||
if len(found) == 1:
|
||
return found[0]
|
||
# 이름이 여럿이면 규격이 있어야 고를 수 있다.
|
||
if not spec:
|
||
return None
|
||
narrowed = [e for e in found if _normalize(e.spec) == _normalize(spec)]
|
||
return narrowed[0] if len(narrowed) == 1 else None
|
||
|
||
|
||
#: 첫 칸이 자원 이름이 **아닌** 표가 많다 — 규격 구간표(「10∼12」), 기호표(「f」·「E」),
|
||
#: 치수표 등. 그런 셀을 못 맞춘 목록에 넣으면 목록이 못 쓰게 되므로 먼저 거른다.
|
||
_RE_RANGE_CELL = re.compile(r"^\d+(?:\.\d+)?\s*[∼~〜\-–]\s*\d+(?:\.\d+)?$")
|
||
_RE_HANGUL = re.compile(r"[가-힣]")
|
||
|
||
|
||
def is_non_resource_label(cell: str) -> bool:
|
||
"""표 머리글·소계 행이거나, 애초에 자원 이름이 올 자리가 아닌 셀인가.
|
||
|
||
못 맞춘 목록에 이런 것이 섞이면 목록 자체가 못 쓰게 된다. 여기서 먼저 걷어낸다.
|
||
"""
|
||
text = _normalize(cell)
|
||
if not text:
|
||
return True
|
||
if text.startswith(("※", "<", "(", "-", "ㆍ", "·")):
|
||
return True
|
||
if _RE_RANGE_CELL.match(text): # 규격 구간표의 첫 칸
|
||
return True
|
||
# 자원 이름은 숫자로 시작하지 않는다 — 「50이상」·「100m이하」·「2.집재」는 구간·절번호다.
|
||
if text[0].isdigit():
|
||
return True
|
||
# 자원 이름은 한글 두 자 이상이다. 기호(`f`·`E`)·숫자·단위만 있는 칸은 자원이 아니다.
|
||
if len(_RE_HANGUL.findall(text)) < 2:
|
||
return True
|
||
return any(word in text for word in _NON_RESOURCE_WORDS)
|
||
|
||
|
||
def _normalize(text: str) -> str:
|
||
"""표 셀의 공백·개행 흔들림을 지운다. 「경 암」·「연 암」 같은 것."""
|
||
return re.sub(r"\s+", "", str(text or "")).strip()
|
||
|
||
|
||
def load_labor_catalog(file_name: str = "labor_const_2026-01-01.json") -> ResourceCatalog:
|
||
"""노임 카탈로그 132직종 + `aliases`. 코드는 `occupation_code`."""
|
||
payload = _read_json(*_CATALOG_SUBPATH, file_name)
|
||
variables = payload["variables"]
|
||
records = variables["labor_rate"]["records"]
|
||
entries = [
|
||
CatalogEntry(code=str(r["occupation_code"]), name=r["occupation_name"], kind="labor")
|
||
for r in records
|
||
]
|
||
return ResourceCatalog(entries=entries, aliases=dict(variables.get("aliases", {})))
|
||
|
||
|
||
def load_work_item_master(
|
||
file_name: str = "work_item_master_2026-01-01.json",
|
||
) -> dict[str, Any]:
|
||
"""메인 창 산출물 — **읽기 전용**."""
|
||
return _read_json(*_MASTER_SUBPATH, file_name)
|
||
|
||
|
||
def split_name_and_spec(cell: str) -> tuple[str, str]:
|
||
"""셀 문자열에서 이름과 규격을 가른다.
|
||
|
||
「유압식백호우 (무한궤도,0.7㎥)」 → (`유압식백호우`, `무한궤도,0.7㎥`)
|
||
「덤프트럭 15톤」 → (`덤프트럭`, `15톤`)
|
||
"""
|
||
text = str(cell or "").strip()
|
||
match = _RE_SPEC.search(text)
|
||
if not match:
|
||
return text, ""
|
||
spec = (match.group(1) or match.group(2) or "").strip()
|
||
name = (text[: match.start()] + text[match.end() :]).strip(" ,()（）")
|
||
return name, spec
|
||
|
||
|
||
def parse_amount(cell: str) -> Decimal | None:
|
||
"""숫자 셀만 값으로 본다. 숫자가 아니면 None — 억지로 읽지 않는다."""
|
||
text = _normalize(cell)
|
||
if not _RE_NUMBER.match(text):
|
||
return None
|
||
try:
|
||
return Decimal(text.replace(",", ""))
|
||
except InvalidOperation: # pragma: no cover - 정규식이 먼저 거른다
|
||
return None
|
||
|
||
|
||
def convert_amount(raw: Decimal, *, pum_form: str, basis_quantity: Decimal | None) -> Decimal:
|
||
"""표 형태에 맞춰 소요량으로 환산한다.
|
||
|
||
⚠ **여기가 20배 틀리는 자리다.**
|
||
- `productivity`(생산량형, 예 「㎥/1인/1일」) → **1 ÷ 값**
|
||
- `requirement`(소요량형, 예 「100㎥당 인부 x인」) → **값 ÷ basis_quantity**
|
||
"""
|
||
if pum_form == "productivity":
|
||
if raw == 0:
|
||
raise ResourceAxisError("생산량이 0 이라 소요량으로 뒤집을 수 없습니다")
|
||
return Decimal(1) / raw
|
||
if pum_form == "requirement":
|
||
divisor = basis_quantity if basis_quantity else Decimal(1)
|
||
if divisor == 0:
|
||
raise ResourceAxisError("기준 수량이 0 입니다")
|
||
return raw / divisor
|
||
raise ResourceAxisError(f"자원 축을 붙일 수 없는 표 형태입니다: {pum_form}")
|
||
|
||
|
||
@dataclass
|
||
class ResourceRow:
|
||
"""자원 축 한 줄 — 공종(표) 하나에 붙는 자원 하나."""
|
||
|
||
work_item_code: str
|
||
pum_table_id: str
|
||
pum_form: str
|
||
resource_kind: str
|
||
resource_code: str
|
||
resource_name: str
|
||
resource_spec: str
|
||
amount: Decimal
|
||
amount_unit: str
|
||
raw_row_index: int
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"work_item_code": self.work_item_code,
|
||
"pum_table_id": self.pum_table_id,
|
||
"pum_form": self.pum_form,
|
||
"resource_kind": self.resource_kind,
|
||
"resource_code": self.resource_code,
|
||
"resource_name": self.resource_name,
|
||
"resource_spec": self.resource_spec,
|
||
"amount": str(self.amount),
|
||
"amount_unit": self.amount_unit,
|
||
"raw_row_index": self.raw_row_index,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class UnmatchedRow:
|
||
"""못 맞춘 것 — **빈칸으로 두지 않고 여기 모은다**."""
|
||
|
||
work_item_code: str
|
||
pum_table_id: str
|
||
cell: str
|
||
reason: str
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"work_item_code": self.work_item_code,
|
||
"pum_table_id": self.pum_table_id,
|
||
"cell": self.cell,
|
||
"reason": self.reason,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class AxisResult:
|
||
rows: list[ResourceRow] = field(default_factory=list)
|
||
unmatched: list[UnmatchedRow] = field(default_factory=list)
|
||
skipped_forms: dict[str, int] = field(default_factory=dict)
|
||
|
||
|
||
def match_table(
|
||
node: dict[str, Any],
|
||
table: dict[str, Any],
|
||
catalog: ResourceCatalog,
|
||
result: AxisResult,
|
||
) -> None:
|
||
"""표 하나에 자원 축을 붙인다. 값이 안 서면 `unmatched` 로 보낸다."""
|
||
form = table.get("pum_form", "")
|
||
if form in NON_WORK_ITEM_FORMS or form in UNUSABLE_FORMS or form not in USABLE_FORMS:
|
||
result.skipped_forms[form] = result.skipped_forms.get(form, 0) + 1
|
||
return
|
||
|
||
basis = table.get("basis_quantity")
|
||
basis_quantity = None if basis is None else Decimal(str(basis))
|
||
unit = table.get("basis_unit") or ""
|
||
|
||
for index, row in enumerate(table.get("raw_row", [])):
|
||
cells = [str(c) for c in row]
|
||
if not cells:
|
||
continue
|
||
name_cell = cells[0]
|
||
if is_non_resource_label(name_cell):
|
||
continue
|
||
|
||
# 숫자 셀이 없는 행은 자원 줄이 아니다(제목·설명 행) — 목록에 안 올린다.
|
||
amount_cell = next(
|
||
(parse_amount(c) for c in cells[1:] if parse_amount(c) is not None), None
|
||
)
|
||
if amount_cell is None:
|
||
continue
|
||
|
||
name, spec = split_name_and_spec(name_cell)
|
||
entry = catalog.resolve(name, spec)
|
||
if entry is None:
|
||
found = catalog.by_name(name)
|
||
if len(found) > 1:
|
||
reason = "규격이 없어 같은 이름 여럿 중 고를 수 없음"
|
||
else:
|
||
# 지금 가진 카탈로그는 노임뿐이다. 기계·자재는 카탈로그 자체가 없어
|
||
# 못 맞추는 것이므로 사유를 갈라 적는다 — 「이름이 틀림」과 다르다.
|
||
reason = "카탈로그에 없는 이름 (기계·자재 카탈로그 미확보 포함)"
|
||
result.unmatched.append(
|
||
UnmatchedRow(node["work_item_code"], table["pum_table_id"], name_cell, reason)
|
||
)
|
||
continue
|
||
|
||
try:
|
||
amount = convert_amount(amount_cell, pum_form=form, basis_quantity=basis_quantity)
|
||
except ResourceAxisError as error:
|
||
result.unmatched.append(
|
||
UnmatchedRow(node["work_item_code"], table["pum_table_id"], name_cell, str(error))
|
||
)
|
||
continue
|
||
|
||
result.rows.append(
|
||
ResourceRow(
|
||
work_item_code=node["work_item_code"],
|
||
pum_table_id=table["pum_table_id"],
|
||
pum_form=form,
|
||
resource_kind=entry.kind,
|
||
resource_code=entry.code,
|
||
resource_name=entry.name,
|
||
resource_spec=entry.spec or spec,
|
||
amount=amount,
|
||
amount_unit=unit,
|
||
raw_row_index=index,
|
||
)
|
||
)
|
||
|
||
|
||
def build_resource_axis(master: dict[str, Any], catalog: ResourceCatalog) -> AxisResult:
|
||
"""공종 축 전체를 훑어 자원 축을 만든다."""
|
||
result = AxisResult()
|
||
for node in master.get("work_items", []):
|
||
for table in node.get("tables", []):
|
||
match_table(node, table, catalog, result)
|
||
return result
|
||
|
||
|
||
#: 자원 축 산출물이 나가는 자리 — **메인의 `data_work_item_master/` 안에 넣지 않는다.**
|
||
#: 메인이 품셈을 다시 돌리면 그 폴더가 덮이므로 섞으면 사라진다.
|
||
OUTPUT_SUBPATH = ("resources", "data_cost_resource_axis")
|
||
|
||
|
||
def write_resource_axis(
|
||
result: AxisResult,
|
||
master: dict[str, Any],
|
||
*,
|
||
output_dir: str | None = None,
|
||
) -> dict[str, str]:
|
||
"""자원 축과 못 맞춘 목록을 파일로 낸다. 만든 파일 경로를 돌려준다."""
|
||
directory = output_dir or os.path.join(_project_root(), *OUTPUT_SUBPATH)
|
||
os.makedirs(directory, exist_ok=True)
|
||
effective_date = master.get("effective_date", "")
|
||
|
||
axis_path = os.path.join(directory, f"resource_axis_{effective_date}.json")
|
||
unmatched_path = os.path.join(directory, f"unmatched_{effective_date}.json")
|
||
|
||
axis_payload = {
|
||
"schema_version": "1.0",
|
||
"dataset_id": "resource_axis_forest",
|
||
"effective_date": effective_date,
|
||
# 어느 공종 축 판에 붙인 것인지 — 세 쪽을 그대로 옮겨 적는다(PLAN 9-2).
|
||
"source_dataset_version": master.get("dataset_version", {}),
|
||
"policy": {
|
||
"axis": "resource_only",
|
||
"work_item_axis_owner": "B08",
|
||
"material_amounts_are_before_surcharge": True,
|
||
},
|
||
"stats": {
|
||
"rows": len(result.rows),
|
||
"unmatched": len(result.unmatched),
|
||
"skipped_forms": result.skipped_forms,
|
||
},
|
||
"rows": [r.as_dict() for r in result.rows],
|
||
}
|
||
unmatched_payload = {
|
||
"schema_version": "1.0",
|
||
"effective_date": effective_date,
|
||
"note": (
|
||
"못 맞춘 자원 이름. 빈칸으로 두지 않고 여기 모은다. "
|
||
"기계·자재 카탈로그가 아직 없어 그 계열은 전부 여기로 온다."
|
||
),
|
||
"rows": [u.as_dict() for u in result.unmatched],
|
||
}
|
||
|
||
for path, payload in ((axis_path, axis_payload), (unmatched_path, unmatched_payload)):
|
||
with open(path, "w", encoding="utf-8", newline="\n") as handle:
|
||
json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
|
||
|
||
return {"resource_axis": axis_path, "unmatched": unmatched_path}
|