Aislo/resources/master_data/scripts/_build_pum.py

# -*- coding: utf-8 -*-
"""원천 벌 뽑기 — 품셈 md → `pum_forest_2026.json` · `pum_const_2026.json` (2026-09-18 브레인).

왜 — 원문 md 가 고쳐지면(줄 보탬 · 표 머리 복원) 원천 벌의 **줄 번호가 밀려** 공종 마스터가 남의 줄을
읽었음(산림 밑수 238 → 105). 머리글로 다시 맞추니 12장에 같은 머리글이 많아 **이웃 표를 집었고**
직접공사비가 조용히 움직였음. 원천 벌을 뽑던 추출기가 저장소에 없었음.

무엇을 하나
  - 표 = md 의 `|` 표(머리 줄 + `|---|` 줄 + 몸 줄). 칸 글자를 **그대로** 옮김 —
    ⚠ 2단 머리를 펼치지 않음(윗머리 「시 간 당(10-7)」 에 밑수·절 번호가 들어 있음).
  - ⚠ **줄 번호로 옛 표를 찾지 않음.** 표 번호(`F0001`·`C0001`)와 절(`section`)은 옛 벌에서 이어받되,
    짝은 이렇게 지음 — ① 글자(머리·몸)가 같은 표를 **차례를 지키며** 맞댐(앞뒤가 뒤바뀌지 않으니
    머리글이 같은 이웃 표로 건너뛸 수 없음) ② 맞은 표 사이 빈 자리의 표 수가 같으면 **같은 차례끼리**
    (글자만 고쳐진 표) ③ 수가 다르면 그 안에서 머리가 같은 표끼리 차례대로.
  - 새 표 = 가장 큰 번호 다음 번호 · 절은 표 위로 올라가며 처음 만나는 `###` 제목이나
    「숫자-숫자 글」 줄(옛 벌의 절 잡는 법 — 산림 476표가 이 규칙과 같음). ⚠ 알림에 뜸 — 사람이 볼 것.
  - 사라진 표 = 번호를 버리고 알림에 적음(다른 표에 번호를 물려주지 않음).
  - 줄 번호·sha256 은 새 md 에서. 표가 한 칸도 안 바뀌면 파일을 그대로 둠(`generated_at` 도).

돌리기: `./venv/Scripts/python.exe resources/data_cost_input_value/_build_pum.py [--check]`
  `--check` — 쓰지 않고 달라질 것만 알림 · 달라지면 끝 코드 1.
금액 불변은 `resources/tester/test_work_item_key_gate.py` 원가계산서 시험(직접공사비 5,983,724)이 잼 —
다시 뽑은 뒤 공종 마스터를 다시 짓고 그 시험을 돌릴 것.
"""

from __future__ import annotations

import difflib
import hashlib
import json
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Callable

ROOT = Path(__file__).resolve().parents[2]
CATALOGS = {
    "pum_forest": ROOT / "resources/data_cost_input_value/pum_forest_2026.json",
    "pum_const": ROOT / "resources/data_cost_input_value/pum_const_2026.json",
}
PREFIX = {"pum_forest": "F", "pum_const": "C"}

_SEP = re.compile(r"\s*\|(\s*:?-+:?\s*\|)+\s*")
#: 옛 벌의 절 줄 — `12-2 표면 마무리를 따른다.` 처럼 본문 인용도 절로 잡혀 있음(마스터 빌더가 바로잡음).
_SECTION_LINE = re.compile(r"(\d+-\d+(?:-\d+)*\.?\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]:
    """md `|` 표 목록 — `line` 은 머리 줄 번호(1부터)."""
    tables, 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(
                {
                    "line": i + 1,
                    "headers": _cells(lines[i]),
                    "rows": [_cells(x) for x in lines[i + 2 : j]],
                }
            )
            i = j
        else:
            i += 1
    return tables


def section_above(lines: list[str], line: int) -> str | None:
    """새 표의 절 — 위로 올라가며 처음 만나는 `###` 제목 또는 「숫자-숫자 글」 줄(표 줄은 건너뜀)."""
    for k in range(line - 2, -1, -1):
        s = lines[k]
        if s.startswith("### "):
            return s[4:].strip()
        if s.lstrip().startswith("|"):
            continue
        if m := _SECTION_LINE.search(s):
            return m.group(1).strip()
    return None


def _sig(table: dict) -> str:
    return json.dumps([table["headers"], table["rows"]], ensure_ascii=False)


def align(old: list[dict], new: list[dict]) -> list[int | None]:
    """`new[j]` 의 짝 `old` 차례(없으면 None). 차례를 지킴 — 짝끼리 앞뒤가 뒤바뀌지 않음."""
    pair: list[int | None] = [None] * len(new)

    def pair_blocks(a: list[str], b: list[str], i0: int, j0: int, inner: Callable | None) -> None:
        matcher = difflib.SequenceMatcher(None, a, b, autojunk=False)
        for tag, i1, i2, j1, j2 in matcher.get_opcodes():
            if tag == "equal" or (tag == "replace" and i2 - i1 == j2 - j1):
                for k in range(j2 - j1):
                    pair[j0 + j1 + k] = i0 + i1 + k
            elif tag == "replace" and inner:
                inner(i0 + i1, i0 + i2, j0 + j1, j0 + j2)

    def by_headers(i1: int, i2: int, j1: int, j2: int) -> None:
        heads = lambda ts: [json.dumps(t["headers"], ensure_ascii=False) for t in ts]  # noqa: E731
        matcher = difflib.SequenceMatcher(
            None, heads(old[i1:i2]), heads(new[j1:j2]), autojunk=False
        )
        for tag, a1, a2, b1, b2 in matcher.get_opcodes():
            if tag == "equal":
                for k in range(b2 - b1):
                    pair[j1 + b1 + k] = i1 + a1 + k

    pair_blocks([_sig(t) for t in old], [_sig(t) for t in new], 0, 0, by_headers)
    return pair


def extract(old_doc: dict, read: Callable[[str], bytes], dataset: str) -> tuple[dict, dict]:
    """옛 벌 + 원문 → (새 벌, 알림). `read(path)` 는 원문 파일 바이트."""
    old_tables = old_doc["variables"]["pum"]["tables"]
    per_file = "source_file" in old_tables[0]
    sources = [s for s in old_doc["sources"] if s.get("role") == "primary"]
    numbers = [int(t["table_id"][1:]) for t in old_tables]
    next_number = max(numbers) + 1
    report = {"same": 0, "moved": 0, "changed": [], "new": [], "removed": []}
    tables, new_sources = [], []
    for src in old_doc["sources"]:
        data = read(src["path"])
        new_sources.append({**src, "sha256": hashlib.sha256(data).hexdigest()})
    for src in sources:
        lines = read(src["path"]).decode("utf-8").splitlines()
        old = [t for t in old_tables if not per_file or t["source_file"] == src["path"]]
        new = md_tables(lines)
        for j, k in enumerate(align(old, new)):
            t = new[j]
            if k is None:
                table_id = f"{PREFIX[dataset]}{next_number:04d}"
                next_number += 1
                section = section_above(lines, t["line"])
                report["new"].append({"table_id": table_id, "line": t["line"], "section": section})
            else:
                table_id, section = old[k]["table_id"], old[k]["section"]
                if _sig(old[k]) != _sig(t):
                    report["changed"].append({"table_id": table_id, "line": t["line"]})
                elif old[k]["line"] != t["line"]:
                    report["moved"] += 1
                else:
                    report["same"] += 1
            entry = {"table_id": table_id, "section": section, "line": t["line"], "headers": t["headers"], "rows": t["rows"]}  # fmt: skip
            if per_file:
                entry["source_file"] = src["path"]
            tables.append(entry)
        kept = {old[k]["table_id"] for k in align(old, new) if k is not None}
        report["removed"] += [{"table_id": t["table_id"], "section": t["section"]} for t in old if t["table_id"] not in kept]  # fmt: skip
    doc = {**old_doc, "sources": new_sources}
    doc["variables"] = {
        **old_doc["variables"],
        "pum": {**old_doc["variables"]["pum"], "tables": tables},
    }
    if doc != old_doc:
        doc["generated_at"] = datetime.now(timezone(timedelta(hours=9))).isoformat(
            timespec="seconds"
        )
    return doc, report


def _read(path: str) -> bytes:
    return (ROOT / path).read_bytes()


def main(check: bool) -> int:
    dirty = False
    for dataset, path in CATALOGS.items():
        old_doc = json.loads(path.read_text(encoding="utf-8"))
        doc, report = extract(old_doc, _read, dataset)
        changed = doc != old_doc
        dirty |= changed
        print(
            f"── {dataset}: 표 {len(doc['variables']['pum']['tables'])} · 그대로 {report['same']} · "
            f"줄만 밀림 {report['moved']} · 글자 바뀜 {len(report['changed'])} · 새 표 {len(report['new'])} · "
            f"사라진 표 {len(report['removed'])}" + ("" if changed else " — 바뀐 것 없음")
        )
        for key in ("changed", "new", "removed"):
            for item in report[key]:
                print(f"   {key}: {json.dumps(item, ensure_ascii=False)}")
        if changed and not check:
            path.write_text(json.dumps(doc, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return 1 if (check and dirty) else 0


if __name__ == "__main__":
    sys.exit(main("--check" in sys.argv))