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.
     형식이 둘 이상이면 고르지 않는다(공압식·전기식은 다른 장비 · 판정 Ⓒ).
  ③ **범위 별칭** — 별칭표 한 벌(`common_util_aliases`)의 `resource` 줄을 공종 범위 안에서만 쓴다.
     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,
)
from common_util.common_util_aliases import AliasError, in_scope, load_aliases, parse_aliases

_EXT_SUBPATH = ("resources", "data_resource_catalog")
EXT_CATALOG_FILE = "resource_catalog_ext_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 parse_scoped_aliases(rows: list[dict[str, Any]]) -> list[dict[str, str]]:
    """자원 축 별칭만 — 검사(scope 없음 · 겹친 범위의 두 코드는 오류)는 별칭표 한 벌이 한다."""
    try:
        return [row for row in parse_aliases(rows) if row["axis"] == "resource"]
    except AliasError as error:
        raise ResourceAxisError(str(error)) from error


def load_scoped_aliases() -> list[dict[str, str]]:
    """자원 축 별칭 — 정본은 `resources/data_aliases/`(B08·B09 한 벌)."""
    return load_aliases("resource")


#: 옆 칸 규격 — 「0.6㎥」·「10.3㎥/min」. 값「0.09」·식「(2.45+3.05)/2/10㎡」·단위「hr」 는 아님.
_RE_SIDE_SPEC = re.compile(r"^\d+(?:\.\d+)?(?:㎥|m3|㎥/min|㎥/분|ton|톤|㎾|kW|㎜|mm|㎝|cm|인치)$")


def _writes_spec(side_cells: list[str] | tuple[str, ...]) -> bool:
    """옆 칸이 규격을 적었나 — 이름이 같아도 별칭이 그 규격을 덮으면 안 됨."""
    return any(_RE_SIDE_SPEC.match(_normalize(cell)) for cell in side_cells)


def scoped_alias_entry(
    catalog: ResourceCatalog,
    name_cell: str,
    work_item_code: str,
    side_cells: list[str] | tuple[str, ...] = (),
) -> CatalogEntry | None:
    """범위 별칭이 가리키는 **항목 그 자체**(코드로 곧장) — 없으면 `None`.

    ⚠ 이름만 돌려주면 규격이 조인 키인 기계(소형브레이커 공압식 넷)는 다시 「규격 미정」 으로
      흐려짐 — 별칭이 코드를 적었으면 그 코드가 곧 답(2026-09-14 · 9-19-3 소형브레이커).
    ⚠ 별칭은 **빈 곳을 채움** — 옆 칸이 규격을 적었으면 안 덮음(같은 9-19-1 성토면 「굴착기 |
      0.6㎥」 가 절토면 [주]① 0.7 로 덮이던 자리 · 2026-09-14 판정 Ⓐ).
    """
    if _writes_spec(side_cells):
        return None
    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
        return next((entry for entry in catalog.entries if entry.code == row["to"]), None)
    return None


def apply_scoped_alias(
    catalog: ResourceCatalog,
    name_cell: str,
    work_item_code: str,
    side_cells: list[str] | tuple[str, ...] = (),
) -> str:
    """범위 안이면 카탈로그 쪽 **이름**으로 바꾼다. 범위 밖이거나 대상 코드가 없으면 원문 그대로."""
    if _writes_spec(side_cells):
        return name_cell  # 이름만 갈면 옆 칸 규격으로 대상 계열(무한궤도)이 조용히 골라짐
    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 "카탈로그에 없는 이름 (기계·자재 카탈로그 미확보 포함)"