Aislo/B09_Estimation/B09_Estimation_ResourceAxis_Join.py

"""B09 원가계산 — 자원 축 **조인 키 규칙** (`_ResourceAxis` 보조 · 2026-09-13 축 C 1장).

세 가지를 한 자리에 둔다.
  ① **규격이 조인 키인 항목** — `resources/data_resource_catalog/` 의 `AR-` 자원 목록
     (기존 카탈로그에 없는 자원). 이름이 맞아도 **규격이 같아야** 고르고,
     **후보가 하나여도 자동 채택하지 않는다**(명세 §11 · 판정 Ⓑ).
     품셈 칸이 규격을 안 주면 「규격 미정 — 후보 N」으로 드러낸다.
  ② **형식이 하나뿐인 계열 이름** — 「공기압축기(3.5㎥/min)」 → 「공기압축기(이동식)」 3.5.
     형식이 둘 이상이면 고르지 않는다(공압식·전기식은 다른 장비 · 판정 Ⓒ).
  ③ **범위 별칭** `{from, to, scope, pum_edition}` — 공종 코드 범위 안에서만 바꾼다(명세 5장).
     scope 없는 줄 · 겹친 범위에서 두 코드로 가는 이름은 **읽을 때 오류**로 세운다.
     ⚠ 대체(대응이 아예 없어 갈음)는 별칭이 아니다 — 여기 넣지 않는다.

⚠ 단가는 다루지 않는다 — 코드·이름·규격만 선다(7장 단가 보류).
"""

from __future__ import annotations

import json
import os
import re
from typing import Any

from B09_Estimation.B09_Estimation_ResourceAxis import (
    RANGE_DASHES,
    CatalogEntry,
    ResourceAxisError,
    ResourceCatalog,
    _normalize,
    _project_root,
)
from B09_Estimation.B09_Estimation_ResourceAxis_Sources import (
    parse_amount,
    parse_machine_cell,
    split_name_and_spec,
)

_EXT_SUBPATH = ("resources", "data_resource_catalog")
EXT_CATALOG_FILE = "resource_catalog_ext_2026-01-01.json"
ALIASES_FILE = "aliases_2026-01-01.json"

#: 명세 §2 ③ — `AR-<M|L|X>-<8자리 소문자 16진>`. 종류 글자는 PriceKind 글자 그대로.
_RE_AR_CODE = re.compile(r"^AR-([MLX])-[0-9a-f]{8}$")
_KIND_LETTER = {"material": "M", "labor": "L", "machine": "X"}

#: 규격 키 — 공백을 지우고 물결표만 한 종류로 모은다(`normalize_variant_key` 와 같은 두 규칙).
_TILDES = "".join(ch for ch in RANGE_DASHES if ch not in "-–‐")
#: 옆 칸에서 글자 규격을 찾을 때 **단위 칸**은 건너뛴다.
_UNIT_WORDS = frozenset(
    {"kg", "㎏", "g", "t", "ton", "톤", "㎥", "m3", "㎡", "m2", "m", "㎝", "㎜", "mm", "개", "본",
     "매", "EA", "ea", "개소", "인", "hr", "h", "시간", "시간(h)", "대", "회", "L", "ℓ", "%", "식"}
)  # fmt: skip
#: 「0.016/2.5/2」 — 나눗셈으로 적힌 **값 칸**. 여기서 멈춘다.
_RE_QUOTIENT = re.compile(r"^\d+(?:\.\d+)?(?:/\d+(?:\.\d+)?)+$")
#: 규격 첫 수 — 「5,500ℓ」 처럼 천 단위 쉼표가 든 것도 한 수로 읽는다.
_RE_NUMBER_TOKEN = re.compile(r"\d{1,3}(?:,\d{3})+(?:\.\d+)?|\d+(?:\.\d+)?")


def _read_optional(file_name: str) -> dict[str, Any]:
    """파일이 없으면 빈 벌 — 목록이 없던 때와 같이 돈다."""
    path = os.path.join(_project_root(), *_EXT_SUBPATH, file_name)
    if not os.path.isfile(path):
        return {}
    with open(path, encoding="utf-8") as handle:
        return json.load(handle)


def parse_ext_entries(rows: list[dict[str, Any]]) -> list[CatalogEntry]:
    """`AR-` 자원 줄 → 조인 키 항목. 코드 모양·종류 글자·겹침을 **읽을 때** 막는다."""
    entries: list[CatalogEntry] = []
    for row in rows:
        code, kind = str(row.get("code") or ""), str(row.get("kind") or "")
        found = _RE_AR_CODE.match(code)
        if found is None or found.group(1) != _KIND_LETTER.get(kind):
            raise ResourceAxisError(f"자원 코드 모양이 규약과 다릅니다: {code!r} ({kind})")
        if not str(row.get("name") or "").strip() or not str(row.get("unit") or "").strip():
            raise ResourceAxisError(f"자원 줄에 이름·단위가 있어야 합니다: {code}")
        entries.append(
            CatalogEntry(
                code=code,
                name=str(row["name"]),
                kind=kind,
                spec=str(row.get("spec") or ""),
                strict_spec=True,
            )
        )
    if len({entry.code for entry in entries}) != len(entries):
        raise ResourceAxisError("자원 코드가 겹칩니다 — 난수 8자리를 다시 뽑을 것")
    return entries


def load_ext_entries(file_name: str = EXT_CATALOG_FILE) -> list[CatalogEntry]:
    """기존 카탈로그에 없는 자원(`AR-`)."""
    return parse_ext_entries(list(_read_optional(file_name).get("entries") or []))


def in_scope(work_item_code: str, scope: str) -> bool:
    """`FP-*` 범위 — 그 코드 자신이거나 그 아래. ⚠ `AX`·`AR` 에는 계층이 없어 쓰지 않는다."""
    return work_item_code == scope or work_item_code.startswith(scope + "-")


def parse_scoped_aliases(rows: list[dict[str, Any]]) -> list[dict[str, str]]:
    """자원 축 별칭만 골라 검사한다. **scope 없는 줄 · 겹친 범위의 두 코드는 오류.**"""
    aliases: list[dict[str, str]] = []
    for row in rows:
        if row.get("axis") != "resource":
            continue
        picked = {
            key: str(row.get(key) or "").strip() for key in ("from", "to", "scope", "pum_edition")
        }
        if not all(picked.values()):
            raise ResourceAxisError(
                f"별칭 줄에 from·to·scope·pum_edition 이 다 있어야 합니다: {row}"
            )
        aliases.append(picked)
    for index, left in enumerate(aliases):
        for right in aliases[index + 1 :]:
            if (
                _normalize(left["from"]) == _normalize(right["from"])
                and left["to"] != right["to"]
                and left["pum_edition"] == right["pum_edition"]
                and (
                    in_scope(left["scope"], right["scope"])
                    or in_scope(right["scope"], left["scope"])
                )
            ):
                raise ResourceAxisError(
                    f"같은 이름 「{left['from']}」이 겹친 범위에서 두 코드로 갑니다: "
                    f"{left['to']} · {right['to']}"
                )
    return aliases


def load_scoped_aliases(file_name: str = ALIASES_FILE) -> list[dict[str, str]]:
    return parse_scoped_aliases(list(_read_optional(file_name).get("aliases") or []))


def apply_scoped_alias(catalog: ResourceCatalog, name_cell: str, work_item_code: str) -> str:
    """범위 안이면 카탈로그 쪽 **이름**으로 바꾼다. 범위 밖이거나 대상 코드가 없으면 원문 그대로."""
    wanted = _normalize(name_cell)
    for row in catalog.scoped_aliases:
        if _normalize(row["from"]) != wanted or not in_scope(work_item_code, row["scope"]):
            continue
        target = next((entry for entry in catalog.entries if entry.code == row["to"]), None)
        if target is not None:
            return target.name
    return name_cell


def spec_key(text: str) -> str:
    tight = _normalize(text)
    return "".join("~" if ch in _TILDES else ch for ch in tight)


def same_spec(catalog_spec: str, text: str) -> bool:
    """규격이 같은가. 카탈로그가 **수만** 적은 규격(「3.5」)은 품셈 첫 수(「3.5㎥/min」)로 본다."""
    left, right = spec_key(catalog_spec), spec_key(text)
    if not left or not right:
        return False
    if left == right:
        return True
    number = parse_amount(left)
    if number is None:
        return False
    found = _RE_NUMBER_TOKEN.search(right)
    return found is not None and parse_amount(found.group(0)) == number


def text_spec_candidates(cells: list[str]) -> list[str]:
    """옆 칸의 **글자 규격** — 「복합비료」·「∅200mm」·「직경4～6㎝, 길이120㎝ 기준」.

    값 칸(수·나눗셈)을 만나면 멈춘다 — 그 뒤는 소요량·비고라 규격이 아니다. 단위 칸은 건너뛴다.
    """
    found: list[str] = []
    for cell in cells:
        text = _normalize(cell)
        if not text:
            continue
        if parse_amount(text) is not None or _RE_QUOTIENT.match(text):
            break
        if text in _UNIT_WORDS or len(text) > 40:
            continue
        found.append(str(cell).strip())
    return found


def pick_by_spec(found: list[CatalogEntry], spec: str) -> CatalogEntry | None:
    """조인 키 항목 고르기 — 규격이 같은 것이 **정확히 한 건**일 때만."""
    hits = [entry for entry in found if same_spec(entry.spec, spec)]
    return hits[0] if len(hits) == 1 else None


def _family_base(name: str) -> str:
    return _normalize(name).split("(")[0].split("（")[0]


def family_members(catalog: ResourceCatalog, name_cell: str) -> list[CatalogEntry]:
    """괄호 앞 이름이 같은 카탈로그 항목(「공기압축기」 → 「공기압축기(이동식)」 여섯 규격)."""
    if catalog._family is None:
        index: dict[str, list[CatalogEntry]] = {}
        for entry in catalog.entries:
            if "(" in entry.name or "（" in entry.name:
                index.setdefault(_family_base(entry.name), []).append(entry)
        catalog._family = index
    name, _ = parse_machine_cell(name_cell)
    return catalog._family.get(_family_base(name), [])


def resolve_family(
    catalog: ResourceCatalog, name_cell: str, cells: list[str]
) -> CatalogEntry | None:
    """카탈로그에 그 이름이 없고 **형식이 하나뿐인 계열**일 때만 규격으로 고른다."""
    name, spec = parse_machine_cell(name_cell)
    if catalog.by_name(name):
        return None  # 이름이 그대로 있으면 계열로 넓히지 않는다
    members = family_members(catalog, name_cell)
    if len({_normalize(entry.name) for entry in members}) != 1:
        return None
    specs = [text for text in (spec, *text_spec_candidates(cells[1:])) if text]
    hits = [entry for entry in members if any(same_spec(entry.spec, text) for text in specs)]
    return hits[0] if len(hits) == 1 else None


def unmatched_reason(catalog: ResourceCatalog, name_cell: str) -> str:
    """못 맞춘 까닭 — 「규격 미정」·「같은 이름 여럿」·「카탈로그에 없는 이름」을 가른다."""
    name, _ = split_name_and_spec(name_cell)
    found = catalog.by_name(name)
    if any(entry.strict_spec for entry in found):
        pool = found
    else:
        pool = [] if found else family_members(catalog, name_cell)
    if pool:
        labels = sorted(
            {
                f"{entry.name} {entry.spec}".strip()[:30]
                for entry in pool
                if entry.spec or not entry.strict_spec
            }
        )
        if not labels:
            codes = ", ".join(sorted(entry.code for entry in pool)[:2])
            return f"규격 미정 — 규격 목록 미확보(코드 {codes} 만 섬) · 설계·라이브러리가 고름"
        shown = " / ".join(labels[:3]) + (" …" if len(labels) > 3 else "")
        return f"규격 미정 — 후보 {len(labels)}건 ({shown}) · 설계·라이브러리가 고름"
    if len(found) > 1:
        return "규격이 없어 같은 이름 여럿 중 고를 수 없음"
    # 기계·자재는 카탈로그가 없어 못 맞추는 것이라 사유를 갈라 적는다 — 「이름이 틀림」과 다르다.
    return "카탈로그에 없는 이름 (기계·자재 카탈로그 미확보 포함)"