Aislo/B09_Estimation/B09_Estimation_ResourceAxis.py

"""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+)?$")

#: 첫 칸이 **분류 딱지**이고 이름이 둘째 칸에 오는 표가 있다.
#: 예 — `['자재', '종      자', '', '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


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 _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 ""

    for index, row in enumerate(table.get("raw_row", [])):
        cells = [str(c) for c in row]
        if not cells:
            continue
        # 첫 칸이 분류 딱지(「자재」·「장비」)면 **이름은 둘째 칸**이다.
        name_cell = cells[0]
        value_cells = cells[1:]
        if _normalize(name_cell) in _GROUP_LABELS and len(cells) > 1:
            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:
            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,
            )
        )


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}