Aislo/resources/tester/test_work_item_master_source_match.py

"""원문 md ↔ 공종 마스터 맞대기 — 2026-09-17 브레인 일감(랩탑 메인).

안티그래비티가 손으로 하던 대조를 시험으로 옮김. 원문은
`(산림청고시 제2025-82호) 산림사업 표준품셈.md` 한 벌, 마스터는 `pum_forest_2026.json`(표 원본)과
`work_item_master_2026-01-01.json`(공종 축).

- **어긋남이 0 이어야 하는 것** — 표 수·차례 · 셀 값(글자까지). 하나라도 다르면 빨강.
- **알려진 어긋남이 있는 것** — 표 자리(부록 표가 공종에 붙음) · 밑수 · [주] · 표 성격. 지금 어긋난 것은
  `fixtures/work_item_master_source_known.json` 에 적어 두고, **그 목록 밖**에서 새로 어긋나면 빨강.
  ⚠ 목록에 있는데 **고쳐졌거나 모습이 바뀐 것도 빨강** — 목록이 낡지 않게.
  고친 뒤 목록 지우기: `python resources/tester/test_work_item_master_source_match.py --prune`
  (지우기만 함 · 새 어긋남은 안 보탬 — 새 것은 원인을 보고 사람이 적을 것).

⚠ **빌더 코드를 안 씀** — 원문 읽기를 여기서 따로 함. 같은 코드로 재면 같은 틀림을 못 잡음.
"""

from __future__ import annotations

import json
import re
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
MD = (
    ROOT
    / "resources/knowledge/original/행정규칙/임도 품셈 적용기준 (현 산림사업 표준품셈)/첨부"
    / "(산림청고시 제2025-82호) 산림사업 표준품셈.md"
)
PUM = ROOT / "resources/data_cost_input_value/pum_forest_2026.json"
MASTER = ROOT / "resources/data_work_item_master/work_item_master_2026-01-01.json"
KNOWN = Path(__file__).resolve().parent / "fixtures/work_item_master_source_known.json"

#: 목차 표 — 공종 줄이 아니라 목차를 세우는 데 쓰임.
TOC_TABLE = "F0001"
#: 이 줄 뒤는 부록(할인·할증 공종표 · 적용 공종 · 단가산출서 예시) — 공종 품이 아니라 `orphan_tables` 몫.
APPENDIX = "〔부록 1〕"
#: 알려진 목록 갈래. `form_blind`·`cost_rule` 은 어긋남이 아니라 **시험이 못 가리는 자리**를 적어 둔 것.
KIND_NAMES = {
    "place": "표 자리",
    "basis": "밑수",
    "notes": "[주]",
    "form": "표 성격",
    "form_blind": "표 성격 — 품 표지를 못 알아봄",
    "cost_rule": "비목 구성표 — 사람이 봐야 함",
}

# ── 원문 읽기 ────────────────────────────────────────────────────────────
_SEP = re.compile(r"\s*\|(\s*:?-+:?\s*\|)+\s*")


def _cells(line: str) -> list[str]:
    s = line.strip()
    s = s[1:-1] if s.endswith("|") else s[1:]
    return [c.strip() for c in s.split("|")]


def md_tables(lines: list[str]) -> list[dict]:
    """`|` 표 = 머리 줄 + `|---|` 줄 + 몸 줄. 나온 차례대로 F0001… (줄 번호는 1부터)."""
    tables: list[dict] = []
    i = 0
    while i < len(lines):
        if (
            lines[i].lstrip().startswith("|")
            and i + 1 < len(lines)
            and _SEP.fullmatch(lines[i + 1])
        ):
            j = i + 2
            while j < len(lines) and lines[j].lstrip().startswith("|"):
                j += 1
            tables.append(
                {
                    "id": f"F{len(tables) + 1:04d}",
                    "line": i + 1,
                    "end": j,  # 표 다음 줄의 0부터 번호
                    "headers": _cells(lines[i]),
                    "rows": [_cells(x) for x in lines[i + 2 : j]],
                }
            )
            i = j
        else:
            i += 1
    return tables


#: 밑수가 될 수 있는 세는 단위. 「회」「%」「분」은 밑수가 아님(2026-09-17 데스크탑 서브 원문 확인).
_UNIT = "㎥|m3|㎡|m2|㏊|ha|㎞|km|m|매|본|개소|개|인|공㎥|공|kg|㎏|톤|ton|주|식|대|일|기|조|롤|RM|시간|hr"
_SAME = {"㏊": "ha", "㎞": "km", "m3": "㎥", "m2": "㎡", "㎏": "kg", "ton": "톤"}
_ONE = re.compile(rf"^([\d.]*)\s*({_UNIT})$")
_PAREN_END = re.compile(r"[(（]([^()（）]*)[)）]+\s*$")
_PAREN_ANY = re.compile(r"[(（]([^()（）]*)[)）]")


def parse_basis(text: str) -> tuple[float, str] | str | None:
    """괄호 속 글 하나 → `(수, 단위)` · 분모가 둘이면 `"둘"` · 밑수 글이 아니면 `None`.

    - `인/100㎡` · `단위 : 인/ha` → 분모가 밑수 · `100본당` · `일 당` → 「당」 앞이 밑수
    - `ha당 소요량` · `1km 소요량기준` → 「소요량」 앞이 밑수
    - ⚠ 「인」「공」은 숫자가 붙을 때만(`(단위 : 인)` 은 품의 단위 — 2026-09-09 원문 대조)
    """
    t = re.sub(r"^\s*단\s*위\s*[:：]?\s*", "", text).strip()
    t = re.sub(r"(\d),(\d{3})", r"\1\2", t)
    t = re.sub(r"\s*당?\s*소요량\s*(기준)?$", "당", t)
    if "/" in t:
        denominators = t.split("/")[1:]
        den = re.sub(r"\s*당$", "", denominators[-1]).strip()
        if len(denominators) > 1 or re.search(r"[,，·\s]", den):
            return "둘"
    elif t.endswith("당"):
        den = t[:-1].strip()
    else:
        return None
    m = _ONE.match(den)
    if m is None:
        return None
    raw, unit = m.group(1), m.group(2)
    if unit in ("인", "공") and not raw:
        return None
    try:
        return (float(raw) if raw else 1.0), _SAME.get(unit, unit)
    except ValueError:
        return None


def md_basis(lines: list[str], table: dict) -> tuple[tuple[float, str] | None, str]:
    """`(밑수, 근거 글)`. 표 바로 위 본문 → 표머리 차례(본문이 정본). 못 찾으면 `(None, "")`.

    본문은 가까운 줄부터 올라가며 줄 끝 괄호를 봄 — 앞 표·`[주]` 를 만나면 멈추고 제목 줄은 보고 멈춤.
    표머리는 칸 속 괄호와 칸 전체를 봄 — 서로 다른 밑수가 둘 이상이면 못 정함.
    """
    for k in range(table["line"] - 2, -1, -1):
        s = lines[k].strip()
        if s.startswith("|") or s.startswith("[주"):
            break
        if (m := _PAREN_END.search(s)) and isinstance(hit := parse_basis(m.group(1)), tuple):
            return hit, s
        if s.startswith("#"):
            break
    found: dict[tuple[float, str], str] = {}
    for head in table["headers"]:
        for text in [m.group(1) for m in _PAREN_ANY.finditer(head)] + [head]:
            if isinstance(hit := parse_basis(text), tuple):
                found.setdefault(hit, head)
    if len(found) == 1:
        ((hit, head),) = found.items()
        return hit, f"표머리 {head}"
    return None, ""


_NOTE_CONT = re.compile(r"^(-\s*[①-⑳]|-\s*-|[※*·])")


def md_notes(lines: list[str], table: dict) -> list[str]:
    """표 바로 아래 `[주]` 덩어리. 첫 글줄이 `[주` 가 아니면 없음.

    이어지는 줄 = `- ②` 꼴 · `- -` 꼴 · `※`·`*`·`·` 로 시작하는 줄. 빈 줄은 건너뜀. 그 밖의 줄에서 끝.
    ⚠ `- 2. 소제목` 같은 본문 목록은 [주]가 아님.
    ⚠ [주] 사이에 낀 `> 【산출 예시】` 덩어리는 건너뛰고 이어 봄 — 그 뒤에 `- ⑥ … 100% 가산` 이 옴(4-3).
    """
    out: list[str] = []
    for k in range(table["end"], len(lines)):
        s = lines[k].strip()
        if not s or (out and s.startswith(">")):
            continue
        if not out and s.startswith("[주"):
            out.append(s)
        elif out and _NOTE_CONT.match(s):
            out.append(s)
        else:
            break
    return out


_WORKER = re.compile(
    r"인\s*부|기능공|운전|목\s*공|석\s*공|철근공|콘크리트공|기\s*사|벌목부|작업반장|기술자|"
    r"용접공|미장공|방수공|배관공|비계공"
)
#: 품 낱말 — 공백을 지운 칸에서 찾음(`주  재  료` → `주재료`).
_WORK_MARK = re.compile(
    r"\(인\)|소요인력|소요량|주입량|작업능력|작업량|수집량|집재량|공정량|시공량|살포량|처리량|"
    r"인력구분|인력구성|적용인부|인원\(명\)|^주재료$|인/|/인|\(시간\)|\(hr\)|[가-힣]\((kg|㎏|g|㎥|ℓ)\)"
)
_QTY_UNIT = {"인", "매", "개", "본", "kg", "㎏", "㎥", "㎡", "m", "ℓ", "대", "톤", "hr", "시간"}
#: 숫자 칸 — `0.16` · 범위 `3.3∼5.9` · 단위 붙음 `1.5인`·`10ℓ`·`5인/km`·`2.18Roll` ·
#: 한 칸에 여러 값 `0.04 0.06 0.10`.
_NUMBER = re.compile(
    r"^\d[\d,]*(\.\d+)?(\s*[～∼~-]\s*\d[\d,]*(\.\d+)?)?"
    r"(\s*(인|ℓ|시간|㎥|㎡|개|본|매|kg|m|%|Roll)(/[a-z㎞㏊]+)?)?(\s+[-\d,.]+)*$"
)


def md_work_mark(table: dict) -> str | None:
    """품 표 표지 칸 — 있으면 그 칸 글, 없으면 `None`. 표에 숫자 칸이 있을 때만 봄.

    ① **짧은 칸(20자 이하)** 에 직종 이름·품 낱말(`소요인력`·`인력구분`·`(인)`·`시멘트(kg)` …)
    ② 한 줄에 수량 단위 칸(`인`·`매`·`㎥` …)과 숫자 칸이 함께 있음
    ③ 단위만 늘어선 줄(`| | ㎥ | ㎥ | ㎥ |`) — 값은 아래 줄에 있음(7-1-1 수확)
    긴 설명 글 속 「보통인부」는 이름이 아님(손료계수표 적용기준 칸).
    ⚠ 못 알아보는 품 표는 알려진 목록 `form_blind` 에 까닭과 함께 둠 — 그 표는 성격이 뒤집혀도 못 잡음.
    """
    rows = [table["headers"], *table["rows"]]
    cells = [c for row in rows for c in row]
    if not any(_NUMBER.match(c) for c in cells):
        return None
    for c in cells:
        if len(c) <= 20 and (_WORKER.search(c) or _WORK_MARK.search(c.replace(" ", ""))):
            return c
    for row in rows:
        unit = next((c for c in row if c in _QTY_UNIT), None)
        if unit and any(_NUMBER.match(c) for c in row):
            return unit
        filled = {c for c in row[1:] if c}
        if len(filled) == 1 and filled <= _QTY_UNIT:
            return next(iter(filled))
    return None


#: 비목 이름 — 이것만으로 첫 칸이 채워진 표가 비목 구성표(`재료비 | 설치비 재료비의 5%`).
_COST_ITEMS = {"재료비", "설치비", "노무비", "경비", "제작비", "운송비", "자재비", "주재료비"}


def md_cost_rows(table: dict) -> list[list[str]] | None:
    """비목 구성표면 그 줄들(뒤 빈 칸 뗌), 아니면 `None`.

    ⚠ 이 꼴은 **모양은 잡히나 성격은 원문으로 못 가림** — 「설치비 = 재료비의 5%」 가 품(자원 줄)인지
    참조(비율 지시)인지는 설계 방식의 결정이라 사람이 봐야 함(알려진 목록 `cost_rule`).
    """
    rows = [r for r in table["rows"] if r and r[0].strip()]
    if not rows or any(r[0].replace(" ", "") not in _COST_ITEMS for r in rows):
        return None
    out = []
    for r in rows:
        cut = list(r)
        while cut and not cut[-1]:
            cut.pop()
        out.append(cut)
    return out


# ── 맞대기 ──────────────────────────────────────────────────────────────
def _norm(text: str) -> str:
    return " ".join(str(text).split())


@pytest.fixture(scope="module")
def src() -> dict:
    return load_sources()


def load_sources() -> dict:
    lines = MD.read_text(encoding="utf-8").splitlines()
    pum = json.loads(PUM.read_text(encoding="utf-8"))["variables"]["pum"]["tables"]
    master = json.loads(MASTER.read_text(encoding="utf-8"))
    entries = [(w, t) for w in master["work_items"] for t in w.get("tables", [])]
    return {
        "lines": lines,
        "tables": md_tables(lines),
        "pum": pum,
        "master": master,
        "entries": entries,
        "by_id": {t["pum_table_id"]: (w, t) for w, t in entries},
    }


def current_mismatches(src: dict) -> dict[str, dict]:
    """표 자리·밑수·[주]·표 성격 어긋남 — 알려진 목록과 같은 꼴. 밑수·[주]·성격은 공종에 붙은 표만 봄."""
    lines = src["lines"]
    out: dict[str, dict] = {kind: {} for kind in KIND_NAMES}
    appendix = next(i for i, x in enumerate(lines, 1) if x.strip().startswith(APPENDIX))
    orphans = {o["pum_table_id"] for o in src["master"]["orphan_tables"]}
    for table in src["tables"]:
        in_appendix = table["line"] > appendix
        if table["id"] in orphans and not in_appendix:
            out["place"][table["id"]] = {
                "section": None,
                "md": "본문 표",
                "master": "공종에 안 붙음",
            }
        if table["id"] not in src["by_id"]:
            continue
        if in_appendix:
            out["place"][table["id"]] = {
                "section": src["by_id"][table["id"]][0]["number"],
                "md": "부록 표",
                "master": "공종에 붙음",
            }
        work, entry = src["by_id"][table["id"]]
        section = work["number"]

        md_hit, evidence = md_basis(lines, table)
        md_value = list(md_hit) if md_hit else None
        unit = entry["basis_unit"]
        master_value = [entry["basis_quantity"], _SAME.get(unit, unit)] if unit else None
        if md_value != master_value:
            out["basis"][table["id"]] = {
                "section": section,
                "md": md_value,
                "md_evidence": evidence,
                "master": master_value,
            }

        md_note = md_notes(lines, table)
        if md_note != entry["notes"]:
            out["notes"][table["id"]] = {
                "section": section,
                "md_only": [x for x in md_note if x not in entry["notes"]],
                "master_only": [x for x in entry["notes"] if x not in md_note],
            }

        form = entry["pum_form"]
        if cost_rows := md_cost_rows(table):
            out["cost_rule"][table["id"]] = {
                "section": section,
                "master": form,
                "md_rows": cost_rows,
            }
            continue  # 성격은 사람 몫 — 품 표지로 재지 않음
        mark = md_work_mark(table)
        if form in ("reference", "coefficient") and mark:
            out["form"][table["id"]] = {"section": section, "master": form, "md_mark": mark}
        elif form in ("requirement", "productivity") and not mark:
            out["form_blind"][table["id"]] = {"section": section, "master": form}
    return out


def load_known() -> dict[str, dict]:
    """알려진 어긋남 목록. **「할 일 아님」 칸(`_할일_아님`)도 함께 읽는다.**

    ⚠ 왜 합쳐 읽나 — ㉮(원문에 없음·거절 자리·판정으로 닫힘)는 **바퀴마다 다시 세어지지 않게**
    갈래에서 빼 별도 칸에 둔다(2026-09-18 브레인). 그렇다고 비교에서까지 빠지면 그 줄이 다시
    「새로 어긋남」으로 빨개진다 — 그래서 **세는 곳에서만 빼고 비교에는 넣는다.**
    """
    data = json.loads(KNOWN.read_text(encoding="utf-8"))
    for kind, rows in ((data.get("_할일_아님") or {}).get("줄") or {}).items():
        data.setdefault(kind, {}).update(rows)
    return data


#: 비교에서 뺄 칸 — **사람이 적은 메모**다. 자료가 아니라 설명이라 어긋남 판정에 안 쓴다.
#: `class`·`class_why` 는 2026-09-18 에 붙인 갈래(㉮ 영영 아님 · ㉯ 남의 몫 · ㉰ 사용자 판단).
_MEMO_FIELDS = ("why", "class", "class_why", "was", "owner", "owner_ask", "갈래")


def _without_why(entry: dict) -> dict:
    return {k: v for k, v in entry.items() if k not in _MEMO_FIELDS}


def test_할일_아님_칸이_갈려_있다() -> None:
    """㉮ — **영영 할 일이 아닌 것**은 갈래에서 빼 별도 칸에 둔다(2026-09-18 브레인).

    ⚠ 왜 못박나 — 섞여 있으면 바퀴마다 「어긋남」으로 다시 세어져 **없는 할 일이 생긴다.**
    ⚠ 지우지는 않는다 — 나중에 「이거 왜 안 했지」 할 때 사유를 보아야 한다.
    """
    block = json.loads(KNOWN.read_text(encoding="utf-8")).get("_할일_아님")
    assert block, "「할 일 아님」 칸이 없다"
    rows = block["줄"]
    total = sum(len(v) for v in rows.values())
    assert total == 34
    assert block["갈래별"] == {"거절 자리": 11, "그 밖": 18, "부록 예시": 1, "원문 부재": 4}
    for kind, got in rows.items():
        for tid, v in got.items():
            assert str(v.get("class", "")).startswith("㉮"), f"{kind}/{tid}"
            assert v.get("class_why"), f"{kind}/{tid} 사유 없음"  # 왜 할 일이 아닌지
            assert v.get("갈래") in ("원문 부재", "부록 예시", "거절 자리", "그 밖"), (
                f"{kind}/{tid}"
            )
    # 세는 칸에서는 빠지고 비교에는 들어간다 — 그래서 빨개지지 않는다
    merged = load_known()
    for kind, got in rows.items():
        assert set(got) <= set(merged[kind])


# ── 시험 ────────────────────────────────────────────────────────────────
def test_표_수와_자리(src: dict) -> None:
    """원문 표가 원본(`pum_forest`)에 같은 차례·같은 줄로 있고, 마스터에 빠짐없이 한 번씩 있음."""
    md_ids = [(t["id"], t["line"]) for t in src["tables"]]
    pum_ids = [(t["table_id"], t["line"]) for t in src["pum"]]
    assert len(md_ids) == len(pum_ids), f"표 수 — 원문 {len(md_ids)} · 원본 {len(pum_ids)}"
    moved = [(a, b) for a, b in zip(md_ids, pum_ids) if a != b]
    assert not moved, f"표 자리 어긋남 {len(moved)}건 (원문, 원본) — 앞 5: {moved[:5]}"

    attached = [t["pum_table_id"] for _, t in src["entries"]]
    doubled = sorted({x for x in attached if attached.count(x) > 1})
    assert not doubled, f"두 공종에 붙은 표: {doubled}"
    orphans = [o["pum_table_id"] for o in src["master"]["orphan_tables"]]
    all_ids = {i for i, _ in md_ids}
    missing = sorted(all_ids - set(attached) - set(orphans) - {TOC_TABLE})
    assert not missing, f"마스터에 없는 원문 표: {missing}"
    extra = sorted((set(attached) | set(orphans)) - all_ids)
    assert not extra, f"원문에 없는 마스터 표: {extra}"


def test_셀_값이_글자까지_같음(src: dict) -> None:
    """표머리·줄 칸을 글자까지 맞댐 — 원문 ↔ 원본(`headers`·`rows`) ↔ 마스터(`condition_note`·`raw_row`)."""
    diffs: list[str] = []

    def compare(tid: str, line: int, what: str, md_rows: list, other: list) -> None:
        if len(md_rows) != len(other):
            diffs.append(f"{tid} L{line} {what} 줄 수 — 원문 {len(md_rows)} · {len(other)}")
        for r, (a, b) in enumerate(zip(md_rows, other)):
            for c in range(max(len(a), len(b))):
                va = a[c] if c < len(a) else "(없음)"
                vb = b[c] if c < len(b) else "(없음)"
                if va != vb:
                    diffs.append(
                        f"{tid} L{line + 2 + r} {what} {r + 1}행 {c + 1}열 — 원문 {va!r} · {vb!r}"
                    )

    pum = {t["table_id"]: t for t in src["pum"]}
    for table in src["tables"]:
        tid, line = table["id"], table["line"]
        p = pum.get(tid)
        if p is None:
            continue  # 표 수 시험이 잡음
        if table["headers"] != p["headers"]:
            diffs.append(f"{tid} L{line} 원본 표머리 — 원문 {table['headers']} · {p['headers']}")
        compare(tid, line, "원본", table["rows"], p["rows"])
        if tid in src["by_id"]:
            _, entry = src["by_id"][tid]
            if entry["source_line"] != line:
                diffs.append(
                    f"{tid} 마스터 줄 번호 — 원문 L{line} · 마스터 L{entry['source_line']}"
                )
            heads = [_norm(h) for h in table["headers"] if _norm(h)]
            if heads != entry["condition_note"]:
                diffs.append(
                    f"{tid} L{line} 마스터 표머리 — 원문 {heads} · {entry['condition_note']}"
                )
            compare(tid, line, "마스터", table["rows"], entry["raw_row"])
    assert not diffs, f"셀 어긋남 {len(diffs)}건 — 앞 30:\n" + "\n".join(diffs[:30])


@pytest.mark.parametrize("kind", list(KIND_NAMES))
def test_알려진_어긋남_밖은_없음(src: dict, kind: str) -> None:
    """알려진 목록과 똑같아야 함 — 새로 어긋남 · 모습 바뀜 · 고쳐짐 셋 다 빨강.

    목록 칸 `why` 는 사람이 적은 까닭 — 맞대기에서 뺌.
    """
    now = current_mismatches(src)[kind]
    known = {k: _without_why(v) for k, v in load_known()[kind].items()}
    lines: list[str] = []
    for tid in sorted(set(now) - set(known)):
        lines.append(f"  새로 어긋남 {tid}: {json.dumps(now[tid], ensure_ascii=False)}")
    for tid in sorted(set(now) & set(known)):
        if now[tid] != known[tid]:
            lines.append(
                f"  모습 바뀜 {tid}: 목록 {json.dumps(known[tid], ensure_ascii=False)}"
                f"\n           지금 {json.dumps(now[tid], ensure_ascii=False)}"
            )
    for tid in sorted(set(known) - set(now)):
        lines.append(f"  고쳐짐 {tid} — 목록에서 지울 것(--prune)")
    assert not lines, f"{KIND_NAMES[kind]} {len(lines)}건 — 원문 ↔ 마스터:\n" + "\n".join(lines)


if __name__ == "__main__" and "--prune" in sys.argv:
    # 고쳐졌거나 모습이 바뀐 줄만 지움 — 새 어긋남은 안 보탬(보태면 빨강이 조용히 묻힘).
    current = current_mismatches(load_sources())
    data = load_known()
    for kind in KIND_NAMES:
        before = len(data[kind])
        data[kind] = {
            k: v for k, v in data[kind].items() if current[kind].get(k) == _without_why(v)
        }
        print(f"{KIND_NAMES[kind]}: {before} → {len(data[kind])}")
    KNOWN.write_text(json.dumps(data, ensure_ascii=False, indent=1) + "\n", encoding="utf-8")