eomsangdonandClaude Opus 5
50f3ab70d3
- ⚠ 앞 커밋의 「값 못 읽은 줄」 판정이 **머리 줄까지 잡고 있었음** — 「특별인부 | 벌목부 | 보통인부」처럼 자원 이름만 나열된 줄. 28 공종 중 대부분이 오탐이었음. **숫자가 있는데** 못 읽은 줄만 성분 빠짐으로 봄 (28 → 12) - 「0.2 × 30%」꼴을 0.06 으로 읽음. ⚠ 딱지의 30 % 를 **또 곱하지 않음** — 두 번 곱하면 0.018. 그 밖의 식(「5인/km」)은 기준이 달라 안 읽고 막은 채 둠 기초잡석 12-25 성립 (121,661.1원/㎥) 소할 할석공 0.06 + 부설다짐 보통인부 0.6 + 적사 굴착기 0.04535hr (Q = 3600 ÷ 22 × 0.7 × 0.55 × 1.0 × 0.35 = 22.05㎥/hr) 운반(덤프)은 품셈이 10장에서 따로 내므로 여기 안 붙음 ⇒ 옹벽 묶음 4조각이 모두 섬 — 타설 65,826.4 · 거푸집 81,321.9 · 철근 1,032,497.4 · 기초잡석 121,661.1 검증: pytest 174 통과(신규 4 — 오탐 짝 시험 포함: 머리 줄은 안 걸리고, 숫자 있는 미해독 줄은 걸림) Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
716 lines
30 KiB
Python
716 lines
30 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 hashlib
|
||
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+)?$")
|
||
|
||
#: 첫 칸이 **분류 딱지**이고 이름이 둘째 칸에 오는 표가 있다.
|
||
#: 예 — `['자재', '종 자', '', 'kg', '0.025']` · `['장비', '종자살포기', …]`.
|
||
#: 이 표를 첫 칸만 보고 읽으면 **자재·장비가 통째로 빠진다**(2026-09-07 실측 —
|
||
#: 씨앗뿜어붙이기에서 종자·비료·피복제·침식안정제·색소·장비 3종이 다 빠지고
|
||
#: 보통인부 한 줄만 남았다).
|
||
_GROUP_LABELS = ("자재", "장비", "인력", "노무", "재료", "기계")
|
||
|
||
#: 표 머리글·소계 행의 첫 칸에 오는 말. 자원이 아니므로 `unmatched` 로도 안 올린다.
|
||
#: 이것을 안 거르면 못 맞춘 목록이 머리글로 가득 차 **쓸 수 없는 목록**이 된다.
|
||
#: ⚠ **부분일치로 보면 안 된다.** 「계」를 부분일치로 잡으면 `건설기계운전사`·`비계공`·
|
||
#: `계장공` 이, 「작업」을 잡으면 `작업반장` 이, 「인력」을 잡으면 `인력운반공` 이
|
||
#: 통째로 사라진다(2026-09-07 실측 — 정상 자원 **70/745** 가 걸리고 있었음).
|
||
#: 그래서 **셀 전체가 그 말과 같을 때만** 머리글로 본다.
|
||
_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)
|
||
#: 이름 → 항목 색인. 자재까지 붙으면 7,700건이 넘어 매번 훑으면 느리다.
|
||
_index: dict[str, list[CatalogEntry]] | None = None
|
||
|
||
def by_name(self, name: str) -> list[CatalogEntry]:
|
||
if self._index is None:
|
||
index: dict[str, list[CatalogEntry]] = {}
|
||
for entry in self.entries:
|
||
index.setdefault(_normalize(entry.name), []).append(entry)
|
||
self._index = index
|
||
return self._index.get(_normalize(name), [])
|
||
|
||
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
|
||
# ⚠ **정확 일치만** — 부분일치는 정상 자원을 통째로 지운다(위 주석).
|
||
if text in _NON_RESOURCE_WORDS:
|
||
return True
|
||
# 머리글 조각이 이어 붙은 칸(「단위작업별」·「위치및면적」)도 머리글이다.
|
||
return _is_header_composite(text)
|
||
|
||
|
||
def _is_header_composite(text: str) -> bool:
|
||
"""머리글 낱말만으로 이루어진 칸인가 — 「단위작업별」·「위치및면적」 같은 것.
|
||
|
||
낱말을 차례로 벗겨 아무것도 안 남으면 머리글로 본다. 자원 이름은 낱말을 벗기면
|
||
반드시 무언가 남는다(`건설기계운전사` → `건설`·`운전사`).
|
||
"""
|
||
rest = text
|
||
for word in sorted(_NON_RESOURCE_WORDS, key=len, reverse=True):
|
||
rest = rest.replace(word, "")
|
||
rest = rest.replace("및", "").replace("별", "").strip()
|
||
return rest == ""
|
||
|
||
|
||
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_machine_catalog_entries(file_name: str = "mach_base_2026.json") -> list[CatalogEntry]:
|
||
"""기종 카탈로그 613건을 매칭용 항목으로 편다.
|
||
|
||
⚠ **규격이 매칭의 일부**다 — 「굴착기(무한궤도)」만 23 규격이라 이름만으로는
|
||
한 대가 안 정해진다(지시 4번). `CatalogEntry.spec` 에 규격을 실어 둔다.
|
||
"""
|
||
from B09_Estimation.B09_Estimation_MachineCost import load_machine_catalog
|
||
|
||
catalog = load_machine_catalog(file_name)
|
||
return [
|
||
CatalogEntry(code=m.machine_code, name=m.name, kind="machine", spec=m.specification)
|
||
for m in catalog.machines.values()
|
||
]
|
||
|
||
|
||
def load_material_catalog_entries(
|
||
file_name: str = "mat_price_public_2026-08-14.json",
|
||
) -> list[CatalogEntry]:
|
||
"""관급 자재 6,999건을 매칭용 항목으로 편다.
|
||
|
||
⚠ 같은 품명에 규격이 수백 개인 것이 있다(「연돌」 410 · 「육각볼트」 323).
|
||
**규격이 매칭의 일부**이므로 `spec` 을 반드시 싣는다.
|
||
"""
|
||
from B09_Estimation.B09_Estimation_MaterialCatalog import load_material_catalog
|
||
|
||
catalog = load_material_catalog(file_name)
|
||
return [
|
||
CatalogEntry(code=m.item_code, name=m.name, kind="material", spec=m.specification)
|
||
for m in catalog.items.values()
|
||
]
|
||
|
||
|
||
def load_combined_catalog() -> ResourceCatalog:
|
||
"""노임 + 기종 + **관급 자재** 를 한 벌로. 사급 자재는 아직 원천이 없다."""
|
||
labor = load_labor_catalog()
|
||
return ResourceCatalog(
|
||
entries=[
|
||
*labor.entries,
|
||
*load_machine_catalog_entries(),
|
||
*load_material_catalog_entries(),
|
||
],
|
||
aliases=labor.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
|
||
|
||
|
||
#: 기종 이름의 괄호 안은 **규격과 형식이 섞여** 있다 —
|
||
#: 「굴착기(무한궤도, 0.7㎥)」 는 이름 `굴착기(무한궤도)` + 규격 `0.7` 이다.
|
||
#: 형식(무한궤도·타이어)은 이름의 일부이고, 숫자가 든 조각만 규격이다.
|
||
_RE_MACHINE = re.compile(r"^(?P<base>[^()（）]+)[(（](?P<inner>[^)）]*)[)）]")
|
||
_RE_SIZE_TOKEN = re.compile(r"\d+(?:\.\d+)?")
|
||
|
||
|
||
def parse_machine_cell(cell: str) -> tuple[str, str]:
|
||
"""기종 셀을 카탈로그 이름과 규격으로 가른다.
|
||
|
||
「굴착기(무한궤도, 0.7㎥)」 → (`굴착기(무한궤도)`, `0.7`)
|
||
「굴착기 (무한궤도)」 → (`굴착기(무한궤도)`, ``) ← 규격은 옆 칸에 있다
|
||
괄호가 없으면 원문 그대로 돌려준다.
|
||
"""
|
||
text = _normalize(cell)
|
||
match = _RE_MACHINE.match(text)
|
||
if not match:
|
||
return text, ""
|
||
base = match.group("base")
|
||
parts = [p for p in re.split(r"[,·/]", match.group("inner")) if p]
|
||
form_parts = [p for p in parts if not _RE_SIZE_TOKEN.search(p)]
|
||
size_parts = [p for p in parts if _RE_SIZE_TOKEN.search(p)]
|
||
name = f"{base}({','.join(form_parts)})" if form_parts else base
|
||
spec = ""
|
||
if size_parts:
|
||
found = _RE_SIZE_TOKEN.search(size_parts[0])
|
||
spec = found.group(0) if found else ""
|
||
return name, spec
|
||
|
||
|
||
def spec_candidates(cells: list[str]) -> list[str]:
|
||
"""규격이 옆 칸에 있는 표가 많다 — 뒷 칸들에서 규격 후보를 모은다.
|
||
|
||
실측 배치: `['굴착기+부착용집게', '0.2㎥', 'hr', '2.71', …]` ·
|
||
`['굴착기 (무한궤도)', '굴착기(무한궤도,0.2㎥)', 'hr', '0.80', …]`
|
||
"""
|
||
found: list[str] = []
|
||
for cell in cells:
|
||
text = _normalize(cell)
|
||
if not text or len(text) > 30:
|
||
continue
|
||
_, spec = parse_machine_cell(text)
|
||
if spec:
|
||
found.append(spec)
|
||
continue
|
||
token = _RE_SIZE_TOKEN.fullmatch(text.rstrip("㎥㎡톤tonm³"))
|
||
if token:
|
||
found.append(token.group(0))
|
||
return found
|
||
|
||
|
||
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
|
||
|
||
|
||
#: 「0.2 × 30%」 꼴 — 값과 배분율이 **한 칸에** 적힌 표기(기초잡석 12-25).
|
||
#: ⚠ 이 값을 읽었으면 **딱지의 배분율을 또 곱하면 안 된다** — 같은 30 % 가 두 번 곱해진다.
|
||
_RE_RATIO_EXPRESSION = re.compile(r"^(\d+(?:\.\d+)?)\s*[×xX*]\s*(\d+(?:\.\d+)?)\s*%$")
|
||
|
||
|
||
def parse_amount_expression(cell: str) -> Decimal | None:
|
||
"""「0.2 × 30%」를 0.06 으로 읽는다. 그 밖의 식은 **읽지 않는다**.
|
||
|
||
식을 넓게 읽으려 들면 「5인/km」처럼 **기준이 다른 값**까지 삼킨다. 여기서 보는 것은
|
||
「값 × 비율%」 한 모양뿐이다.
|
||
"""
|
||
found = _RE_RATIO_EXPRESSION.match(_normalize(cell))
|
||
if found is None:
|
||
return None
|
||
return Decimal(found.group(1)) * Decimal(found.group(2)) / Decimal(100)
|
||
|
||
|
||
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
|
||
#: 규격 갈래 — 열이 자원인 표에서 행 이름(「무근구조물」). 없으면 빈 문자열.
|
||
#: **갈래마다 품이 다르므로 한 일위대가로 뭉치지 않는다.**
|
||
variant: str = ""
|
||
#: 분류 딱지가 달고 온 배분율 — 「인력(10%)」이면 `10`. 없으면 `None`.
|
||
#: ⚠ **이 값을 안 보면 단가가 조용히 틀린다** — 인력 몫 원단위를 전량에 곱하게 된다
|
||
#: (2026-09-08 실측: 측구터파기 39,575.6원/㎥ 이 인력 10 % 몫만이었다).
|
||
group_ratio_pct: Decimal | None = None
|
||
|
||
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,
|
||
"variant": self.variant,
|
||
"group_ratio_pct": None if self.group_ratio_pct is None else str(self.group_ratio_pct),
|
||
}
|
||
|
||
|
||
@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)
|
||
#: 자원은 알아봤는데 **값을 못 읽은** 줄이 있는 공종 — 그 단가는 「일부만 선 것」이다.
|
||
#: 기초잡석 12-25 가 `소할(30%) | 할석공(인) | 0.2 × 30%` 를 못 읽어 부설다짐만으로
|
||
#: 107,145 원이 서고 있었다(2026-09-08). **부분 성공이 가장 위험하다.**
|
||
partial_items: dict[str, str] = field(default_factory=dict)
|
||
|
||
|
||
def _resolve_cell(catalog: ResourceCatalog, name_cell: str, cells: list[str]):
|
||
"""셀 하나를 카탈로그 한 줄로 푼다 — 세 가지 모양을 차례로 시도한다.
|
||
|
||
① 셀 전체가 곧 이름 (「보통인부」)
|
||
② 기종 셀 (「굴착기(무한궤도, 0.7㎥)」 → 이름 + 규격)
|
||
③ 규격이 **옆 칸**에 있는 표 (「굴착기 (무한궤도)」 | 「0.7㎥」)
|
||
"""
|
||
machine_name, machine_spec = parse_machine_cell(name_cell)
|
||
plain_name, plain_spec = split_name_and_spec(name_cell)
|
||
|
||
for name, spec in ((machine_name, machine_spec), (plain_name, plain_spec)):
|
||
if not name:
|
||
continue
|
||
entry = catalog.resolve(name, spec)
|
||
if entry is not None:
|
||
return entry
|
||
|
||
# 이름은 맞는데 규격이 없어 못 고른 경우 — 옆 칸에서 규격을 찾는다.
|
||
for name in (machine_name, plain_name):
|
||
if len(catalog.by_name(name)) <= 1:
|
||
continue
|
||
for candidate in spec_candidates(cells[1:]):
|
||
entry = catalog.resolve(name, candidate)
|
||
if entry is not None:
|
||
return entry
|
||
return None
|
||
|
||
|
||
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 ""
|
||
|
||
# ⚠ **자원이 열 머리에 오는 표가 따로 있다** (2026-09-08 발견, 39 표).
|
||
# 「구 분 | 콘크리트공(인) | 보통인부(인)」처럼 **열이 자원**이고 행은 규격 갈래
|
||
# (무근·철근·소형구조물)다. 행을 자원으로 읽는 길로 보내면 통째로 못 맞춘다 —
|
||
# 콘크리트 타설(12-1)이 그래서 하나도 안 서고 있었다.
|
||
from B09_Estimation.B09_Estimation_ResourceAxis_Transposed import match_transposed_table
|
||
|
||
if match_transposed_table(node, table, catalog, result, basis_quantity, unit):
|
||
return
|
||
|
||
for index, row in enumerate(table.get("raw_row", [])):
|
||
cells = [str(c) for c in row]
|
||
if not cells:
|
||
continue
|
||
# 첫 칸이 분류 딱지(「자재」·「장비」)면 **이름은 둘째 칸**이다.
|
||
#
|
||
# ⚠ 딱지 목록만으로는 모자란다 — 첫 칸이 **공정 이름**인 표가 따로 있다
|
||
# (기초잡석 12-25: `소할(30%) | 할석공(인) | 0.2 × 30%`). 목록에 없는 말이라
|
||
# 통째로 못 맞추고 있었다. 그래서 **딱지 목록에 없더라도 첫 칸이 자원으로 안 풀리고
|
||
# 둘째 칸이 풀리면** 이름을 둘째 칸에서 읽는다 — 판정을 낱말이 아니라
|
||
# **풀리는지**로 한다. 배분율 꼬리표(「(30%)」)는 어느 쪽이든 첫 칸에서 읽는다.
|
||
name_cell = cells[0]
|
||
value_cells = cells[1:]
|
||
group_ratio = None
|
||
if len(cells) > 1 and (
|
||
_group_label_of(name_cell) is not None
|
||
or (
|
||
_resolve_cell(catalog, name_cell, cells) is None
|
||
and _resolve_cell(catalog, cells[1], cells[1:]) is not None
|
||
)
|
||
):
|
||
group_ratio = _group_ratio_of(name_cell)
|
||
name_cell = cells[1]
|
||
value_cells = cells[2:]
|
||
|
||
# 숫자 셀이 없는 행은 자원 줄이 아니다(제목·설명 행) — 목록에 안 올린다.
|
||
amount_cell = next(
|
||
(parse_amount(c) for c in value_cells if parse_amount(c) is not None), None
|
||
)
|
||
if amount_cell is None:
|
||
# 「0.2 × 30%」 꼴은 값과 배분율이 한 칸에 있다 — 읽었으면 딱지 배분율은 버린다.
|
||
expression = next(
|
||
(
|
||
parse_amount_expression(c)
|
||
for c in value_cells
|
||
if parse_amount_expression(c) is not None
|
||
),
|
||
None,
|
||
)
|
||
if expression is not None:
|
||
amount_cell = expression
|
||
group_ratio = None # ⚠ 이미 값 안에 들어 있다 — 또 곱하면 두 번이다
|
||
if amount_cell is None:
|
||
# ⚠ **이름은 자원인데 값을 못 읽은 줄**은 다르다 — 그 공종 단가는 성분이
|
||
# 빠진 채 서게 된다. 조용히 넘기지 않고 「일부만 섬」으로 표시한다.
|
||
# ⚠ **숫자가 아예 없는 줄은 머리 줄**이다 — 자원 이름만 나열된 줄
|
||
# (「특별인부 | 벌목부 | 보통인부」). 그것까지 「못 읽은 값」으로 세면
|
||
# 정상 공종이 무더기로 막힌다(2026-09-08: 28건 중 대부분이 이 오탐이었다).
|
||
# 숫자가 **있는데** 못 읽은 줄만 성분 빠짐으로 본다.
|
||
has_digit = any(ch.isdigit() for cell in value_cells for ch in cell)
|
||
if (
|
||
has_digit
|
||
and _resolve_cell(catalog, name_cell, [name_cell, *value_cells]) is not None
|
||
):
|
||
result.partial_items[node.get("work_item_code", "")] = (
|
||
f"{name_cell} 줄의 값을 못 읽었습니다"
|
||
)
|
||
result.unmatched.append(
|
||
UnmatchedRow(
|
||
work_item_code=node.get("work_item_code", ""),
|
||
pum_table_id=str(table.get("pum_table_id", "")),
|
||
cell=" | ".join(cells[:3]),
|
||
reason="자원은 알아봤으나 값을 못 읽었습니다 — 단가가 일부만 섭니다.",
|
||
)
|
||
)
|
||
continue
|
||
|
||
# ⚠ **카탈로그 조회를 먼저 한다.** 이름 필터를 앞에 두면 필터가 넓을 때
|
||
# 정상 자원이 조용히 사라진다(2026-09-07 실측 — 부분일치 필터가 매칭 14건을
|
||
# 지우고 있었음). 카탈로그에 있는 이름은 **정의상 자원**이다.
|
||
entry = _resolve_cell(catalog, name_cell, [name_cell, *value_cells])
|
||
if entry is None:
|
||
if is_non_resource_label(name_cell):
|
||
continue # 머리글·소계 — 못 맞춘 목록에도 안 올린다
|
||
name, spec = split_name_and_spec(name_cell)
|
||
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,
|
||
amount=amount,
|
||
amount_unit=unit,
|
||
raw_row_index=index,
|
||
group_ratio_pct=group_ratio,
|
||
)
|
||
)
|
||
|
||
|
||
#: 분류 딱지가 **비율을 달고 오는** 모양 — 「인력(10%)」·「장비(90%)」.
|
||
#: 2026-09-08 실측: 측구터파기(FP-09-12-01) 표가 이 모양이라 딱지 판정이 빗나가
|
||
#: **보통인부 0.23인이 통째로 빠지고 있었다**(그 공종의 자원 줄이 0 개였다).
|
||
#: 괄호 안이 **숫자·%·소수점뿐일 때만** 떼어 낸다 — 「보통인부(인)」 같은 단위 표기는
|
||
#: 떼면 안 되므로 넓게 잡지 않는다.
|
||
_RE_RATIO_SUFFIX = re.compile(r"[(（][\d.\s]*%?[)）]$")
|
||
|
||
|
||
def _group_label_of(cell: str) -> str | None:
|
||
"""첫 칸이 분류 딱지면 그 딱지를, 아니면 `None` 을 돌려준다.
|
||
|
||
⚠ 딱지 목록은 **정확 일치**를 유지한다(부분일치가 정상 자원을 지운 전례 —
|
||
`_NON_RESOURCE_WORDS` 주석). 비율 꼬리표만 떼고 다시 정확 일치로 본다.
|
||
"""
|
||
text = _normalize(cell)
|
||
if text in _GROUP_LABELS:
|
||
return text
|
||
stripped = _normalize(_RE_RATIO_SUFFIX.sub("", text))
|
||
return stripped if stripped in _GROUP_LABELS else None
|
||
|
||
|
||
def _group_ratio_of(cell: str) -> Decimal | None:
|
||
"""분류 딱지에 붙은 배분율. 「인력(10%)」 → `10`, 「자재」 → `None`."""
|
||
found = _RE_RATIO_SUFFIX.search(_normalize(cell))
|
||
if found is None:
|
||
return None
|
||
digits = found.group(0).strip("()（）%").strip()
|
||
return Decimal(digits) if digits else None
|
||
|
||
|
||
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 _master_file_fingerprint(master: dict[str, Any]) -> dict[str, str]:
|
||
"""공종 마스터 **파일 자체**의 지문. 낡은 파생물을 드러내는 유일한 근거다.
|
||
|
||
`dataset_version`(품셈 원판 지문)은 마스터가 다시 생성돼도 그대로라, 그것만
|
||
적어 두면 「내 자원 축이 옛 마스터에서 나왔다」는 사실이 안 보인다.
|
||
"""
|
||
effective_date = master.get("effective_date", "")
|
||
file_name = f"work_item_master_{effective_date}.json"
|
||
path = os.path.join(_project_root(), *_MASTER_SUBPATH, file_name)
|
||
try:
|
||
with open(path, "rb") as handle:
|
||
digest = hashlib.sha256(handle.read()).hexdigest()
|
||
except OSError:
|
||
return {"file": file_name, "sha256": ""}
|
||
return {"file": file_name, "sha256": digest}
|
||
|
||
|
||
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", {}),
|
||
# ⚠ 위 지문은 **품셈 원판**의 것이라 B08 이 마스터를 다시 생성해도 안 움직인다.
|
||
# 낡음을 실제로 드러내려면 **마스터 파일 자체의 지문**이 있어야 한다.
|
||
"source_master_file": _master_file_fingerprint(master),
|
||
"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}
|