Aislo/B08_Quantity/B08_Quantity_Build_WorkItemMaster.py

"""산림사업 표준품셈 476표 → 공종 마스터 정규화 (B08 일감 1번 · 공종 축).

무엇을 만드나
    `resources/data_work_item_master/work_item_master_<effective_date>.json` — 공종 계층 + 표 귀속.
    같은 폴더에 `form_undetermined_*.json`(형태 판정 실패분)과 `_manifest.json` 을 함께 낸다.

왜 이렇게 나누나 (PLAN 8-7 담당 경계)
    이 파일은 **공종 축만** 만든다. 자원 축(`resource_kind`·`resource_code`·`amount`)은
    단가표를 아는 B09 가 뒤 패스로 채운다. 그래서 각 표의 **원문 셀(`raw_row`)을 그대로 실어
    보낸다** — 버리면 B09 가 476표를 다시 열어야 한다.

공종의 정체 = 품셈의 **절 번호**다
    품셈은 「9-3-1. 인력」처럼 절 제목이 공종이고, 표의 행은 그 공종의 **조건별 변형**
    (토질·암종·규격)이다. 그래서 계층·정렬은 목차표(`F0001`)에서 나오고, 표는 그 절에 붙는다.

⚠ 최대 함정 — `pum_form` (PLAN 8-6)
    같은 숫자라도 뜻이 반대다.
      productivity(생산량형) : 「㎥/hr」·「㎥/1인/1일」 → 품 = 1 ÷ 값
      requirement(소요량형)  : 「100㎥당 x인」        → 품 = 값 ÷ 밑수
    태그가 없으면 뒤집힌 값이 조용히 들어간다. **판정 못 한 표는 빈칸으로 두지 않고
    `form_undetermined` 목록으로 뽑아** 사람이 보게 한다.

실행
    ./venv/Scripts/python.exe B08_Quantity/B08_Quantity_Build_WorkItemMaster.py
"""

from __future__ import annotations

import hashlib
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parent.parent
SOURCE = ROOT / "resources" / "data_cost_input_value" / "pum_forest_2026.json"
OUT_DIR = ROOT / "resources" / "data_work_item_master"

SCHEMA_VERSION = "1.0"
CODE_PREFIX = "FP"  # Forest Pumsem — 공종코드 접두. 품셈 절 번호를 그대로 싣는다.

# ── 형태 판정 ────────────────────────────────────────────────────────────
# 헤더·비고 문자열에서 찾는 표지. 앞의 것이 먼저 걸린다(생산량형이 더 좁은 표현이라 우선).
PRODUCTIVITY_MARKS = (
    "㎥/hr",
    "m3/hr",
    "㎡/hr",
    "본/hr",
    "/1인/1일",
    "/인/1일",
    "인/1일",
    "작업능력",
    "ha당 평균 작업량",
    "ha당 집재재적",
)
REQUIREMENT_MARKS = (
    "소요인력",
    "소요량",
    "당 주입량",
    "단위수량",
    "수 량",
    "수량",
    "인/km",
    "인/ha",
    "㏊당",
    "ha당",
    "당 소요",
)
# 값이 공종 품이 아니라 계산식 파라미터인 표. 공종으로 세우지 않는다.
COEFFICIENT_MARKS = ("손료계수", "시간당손료계수", "기계손료")
# 기계 시공능력 공식(Q = 3600·qo·K·f·E / Cm)의 파라미터 기호. 이 기호만 있는 표는 계수표다.
COEFFICIENT_ROW_KEYS = {"K", "f", "E", "Cm", "㎝(sec)", "q", "qo", "Q", "V", "L"}
# 품셈 1장(적용기준)·할증·공제·단위 표준 — 공종이 아니라 **기준표**다.
REFERENCE_MARKS = ("할증률", "할인", "공제율", "적재량", "단 위", "지위")
# 값 대신 「어디를 따르라」고만 적은 표. 품이 아니므로 공종으로 세우지 않는다.
REFERENCE_CELL_MARKS = ("별도계상", "구역화물", "적용한다", "따른다", "준용")
# 직종이 값의 주인인 표 = 소요량형. `보통인부(인)`·`콘크리트공(인)` 처럼 `(인)` 이 붙는다.
# ⚠ 원문에 `인 부` 처럼 낱말 사이 공백이 있어, 비교 전에 공백을 모두 지운다(`squeeze`).
OCCUPATION_RE = re.compile(
    r"\((?:인|조|인/일|인·일)\)|인부|기능공|운전사|특별인부|보통인부|콘크리트공|철근공|石工|석공|목공|용접공"
)
# 재료 소요량표의 값 단위. 직종이 없어도 이 단위가 값 열에 오면 소요량형이다.
MATERIAL_UNIT_MARKS = ("(kg)", "(㎏)", "(개)", "(매)", "(본)", "(ℓ)", "(L)", "(㎥)", "(㎡)", "(m)")


def sha256_of(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def norm(text: Any) -> str:
    """공백을 하나로 줄인 문자열. `None` 은 빈 문자열."""
    return " ".join(str(text).split()) if text is not None else ""


def parse_toc(rows: list[list[str]]) -> list[dict[str, Any]]:
    """목차표(F0001) → 계층 목록. 장(제n장)·절(n-n)·항(n-n-n)·목(n-n-n-n)."""
    nodes: list[dict[str, Any]] = []
    order = 0
    for raw in rows:
        cells = [norm(c) for c in raw]
        cells = [c for c in cells if c]
        if not cells:
            continue
        key = cells[0]
        name = cells[1] if len(cells) > 1 else ""
        if m := re.fullmatch(r"제(\d+)장", key):
            number = m.group(1)
        elif re.fullmatch(r"\d+(?:-\d+){1,3}", key):
            number = key
        else:
            continue  # 부록 등 — 공종 계층이 아니다.
        order += 256  # STmate 의 SORTCODE 관례(256 간격) — 중간 삽입 여유.
        parts = number.split("-")
        nodes.append(
            {
                "work_item_code": f"{CODE_PREFIX}-" + "-".join(p.zfill(2) for p in parts),
                "number": number,
                "name": name,
                "level": len(parts),
                "parent_code": (
                    f"{CODE_PREFIX}-" + "-".join(p.zfill(2) for p in parts[:-1])
                    if len(parts) > 1
                    else None
                ),
                "sort_order": order,
                "tables": [],
            }
        )
    return nodes


def section_number(section: str) -> str | None:
    """`"9-3-1. 인력"` → `"9-3-1"`. 번호가 없으면 `None`."""
    m = re.match(r"^\s*(\d+(?:-\d+){0,3})[.\s]", section + " ")
    return m.group(1) if m else None


def detect_form(table: dict[str, Any], chapter: str | None) -> tuple[str, str]:
    """`(pum_form, 근거 문구)`. 판정 못 하면 `("undetermined", 이유)`.

    순서가 뜻을 가진다 — 좁은 표지부터 본다. 계수·기준표를 먼저 걸러 내야
    「작업능력」 같은 흔한 낱말이 기준표를 공종으로 오인하지 않는다.
    """
    hay = " ".join(norm(h) for h in table.get("headers", []))
    keys = [norm(r[0]) for r in table.get("rows", []) if r]
    head_rows = [norm(c) for r in table.get("rows", [])[:3] for c in r]
    first_rows = " ".join(head_rows)
    both = f"{hay} || {first_rows}"
    squeezed = both.replace(" ", "")  # `인 부` → `인부`. 원문 자간 공백을 지운 뒤 비교한다.

    for mark in COEFFICIENT_MARKS:
        if mark in hay:
            return "coefficient", f"헤더 '{mark}'"
    if keys and set(keys) <= COEFFICIENT_ROW_KEYS:
        return "coefficient", f"행 키가 시공능력 공식 기호뿐 {sorted(set(keys))}"
    if keys and all(re.fullmatch(r"[a-zA-Zqf][₀-₉0-9]?", k) for k in keys):
        return "coefficient", f"행 키가 기호뿐 {sorted(set(keys))}"

    # 1장은 적용기준 장 자체다 — 공종이 아니라 기준표로 둔다(PLAN 8-14 「목록은 규정에서」).
    if chapter == "1":
        return "reference", "품셈 제1장(적용기준)"
    for mark in REFERENCE_MARKS:
        if mark in hay:
            return "reference", f"헤더 '{mark}'"
    for mark in REFERENCE_CELL_MARKS:
        if mark in squeezed:
            return "reference", f"'{mark}' — 값이 아니라 참조 지시"

    # ⚠ 생산량형을 **직종보다 먼저** 본다. 「작업능력(㎥/hr)」 표의 비고란에 흔히
    # 「보통인부 1인/일」이 붙어 있어, 직종을 먼저 보면 생산량형이 소요량형으로 뒤집힌다.
    # 그 뒤집힘이 곧 PLAN 8-6 이 경고한 「값이 조용히 반대로 들어가는」 사고다.
    for mark in PRODUCTIVITY_MARKS:
        if mark in hay:
            return "productivity", f"헤더 '{mark}'"

    # 직종이 값의 주인이면 소요량형이다 — 「보통인부(인) 0.16」 은 ㎥당 품이다.
    if OCCUPATION_RE.search(squeezed):
        return "requirement", "직종 표기((인)·인부·공)"
    for mark in MATERIAL_UNIT_MARKS:
        if mark in hay:
            return "requirement", f"값 단위 '{mark}'"

    for mark in PRODUCTIVITY_MARKS:
        if mark in both:
            return "productivity", f"본문 '{mark}'"
    for mark in REQUIREMENT_MARKS:
        if mark in both:
            return "requirement", f"'{mark}'"
    return "undetermined", "헤더·첫 행에 단위·밑수·직종 표지 없음"


BASIS_RE = re.compile(r"(\d[\d,.]*)\s*(㎥|m3|㎡|m2|㏊|ha|km|㎞|m|인|본|개|kg|㎏|톤|ton)\s*당")


def detect_basis(table: dict[str, Any]) -> tuple[float | None, str | None]:
    """「100㎥당」 같은 밑수. 없으면 `(None, None)` — 단위당 1 로 단정하지 않는다."""
    hay = " ".join(norm(h) for h in table.get("headers", []))
    hay += " " + " ".join(norm(c) for r in table.get("rows", [])[:2] for c in r)
    if m := BASIS_RE.search(hay):
        try:
            return float(m.group(1).replace(",", "")), m.group(2)
        except ValueError:
            return None, m.group(2)
    return None, None


def variant_axis(table: dict[str, Any]) -> list[str]:
    """행이 갈리는 축 — 표의 첫 열 값들(토질·암종·규격). 값 열은 뺀다."""
    seen: list[str] = []
    for row in table.get("rows", []):
        key = norm(row[0]) if row else ""
        if key and key not in seen:
            seen.append(key)
    return seen[:24]


def build() -> dict[str, Any]:
    data = json.loads(SOURCE.read_text(encoding="utf-8"))
    tables = data["variables"]["pum"]["tables"]
    toc_table = next(t for t in tables if t["table_id"] == "F0001")
    nodes = parse_toc(toc_table["rows"])
    by_number = {n["number"]: n for n in nodes}

    attached = 0
    orphans: list[dict[str, Any]] = []
    undetermined: list[dict[str, Any]] = []

    for table in tables:
        if table["table_id"] == "F0001":
            continue  # 목차 자신은 공종이 아니다.
        section = norm(table.get("section"))
        number = section_number(section)
        chapter = number.split("-")[0] if number else None
        form, why = detect_form(table, chapter)
        basis_qty, basis_unit = detect_basis(table)
        entry = {
            "pum_table_id": table["table_id"],
            "section": section,
            "source_line": table.get("line"),
            "pum_form": form,
            "form_basis": why,
            "basis_quantity": basis_qty,
            "basis_unit": basis_unit,
            "variant_key": variant_axis(table),
            "condition_note": [norm(h) for h in table.get("headers", []) if norm(h)],
            "raw_row": table.get("rows", []),  # 원문 셀 — B09 자원 축이 읽는다.
        }
        if form == "undetermined":
            undetermined.append(
                {
                    "pum_table_id": table["table_id"],
                    "section": section,
                    "headers": entry["condition_note"],
                    "first_rows": table.get("rows", [])[:2],
                    "reason": why,
                }
            )
        node = by_number.get(number) if number else None
        if node is None:
            orphans.append({"pum_table_id": table["table_id"], "section": section})
            continue
        node["tables"].append(entry)
        attached += 1

    src_meta = {
        "dataset_id": data["dataset_id"],
        "effective_date": data["effective_date"],
        "sha256": sha256_of(SOURCE),
        "file": SOURCE.name,
    }
    return {
        "schema_version": SCHEMA_VERSION,
        "dataset_id": "work_item_master_forest",
        "effective_date": data["effective_date"],
        "generated_at": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"),
        "dataset_version": src_meta,
        "policy": {
            "axis": "work_item_only",
            "resource_axis_owner": "B09",
            "no_invented_values": True,
            "raw_row_preserved": True,
        },
        "stats": {
            "toc_nodes": len(nodes),
            "tables_total": len(tables) - 1,
            "tables_attached": attached,
            "tables_orphan": len(orphans),
            "form_undetermined": len(undetermined),
        },
        "orphan_tables": orphans,
        "work_items": nodes,
    }, undetermined


def main() -> None:
    master, undetermined = build()
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    date = master["effective_date"]
    master_path = OUT_DIR / f"work_item_master_{date}.json"
    undet_path = OUT_DIR / f"form_undetermined_{date}.json"

    master_path.write_text(json.dumps(master, ensure_ascii=False, indent=1), encoding="utf-8")
    undet_path.write_text(
        json.dumps(
            {
                "schema_version": SCHEMA_VERSION,
                "dataset_id": "work_item_master_form_undetermined",
                "effective_date": date,
                "note": "형태를 못 정한 표. 사람이 보고 productivity/requirement/coefficient 로 확정할 것.",
                "items": undetermined,
            },
            ensure_ascii=False,
            indent=1,
        ),
        encoding="utf-8",
    )

    manifest = {
        "schema_version": SCHEMA_VERSION,
        "dataset_id": "data_work_item_master_manifest",
        "generated_at": master["generated_at"],
        "built_by": "B08_Quantity/B08_Quantity_Build_WorkItemMaster.py",
        "source": master["dataset_version"],
        "files": [
            {
                "file": p.name,
                "sha256": sha256_of(p),
                "size_bytes": p.stat().st_size,
            }
            for p in (master_path, undet_path)
        ],
    }
    (OUT_DIR / "_manifest.json").write_text(
        json.dumps(manifest, ensure_ascii=False, indent=1), encoding="utf-8"
    )

    s = master["stats"]
    print(f"목차 계층      {s['toc_nodes']}")
    print(
        f"표 귀속        {s['tables_attached']} / {s['tables_total']}  (미귀속 {s['tables_orphan']})"
    )
    print(f"형태 미판정    {s['form_undetermined']}")
    print(f"산출           {master_path.relative_to(ROOT)}")


if __name__ == "__main__":
    main()