Aislo/resources/master_data/scripts/master_keys.py

# -*- coding: utf-8 -*-
"""마스터 키 — 테이블ID(영문 대문자 2) + 6자리 일련번호 · 대장 `_키대장.json` (`_틀.md` 2장).

한 번 준 키는 안 바뀜 · 새 줄은 대장의 다음 번호. 원문이 준 번호·옛 열쇠는 줄의 「원문번호」.
"""

from __future__ import annotations

import json
import re
from decimal import Decimal
from pathlib import Path

MASTER = Path(__file__).resolve().parent.parent
BOOK = MASTER / "_키대장.json"
KEY = re.compile(r"[A-Z]{2}\d{6}")

# (그룹, 원문) → 테이블ID · 인력·기계는 원문 구분 없이 한 테이블
TABLES = {
    ("인력", None): "LB",
    ("재료", "시중물가"): "MT",
    ("재료", "나라장터자재"): "MN",
    ("재료", "오피넷유가"): "MO",
    ("재료", "품셈재료"): "MP",
    ("기계", None): "EQ",
    ("소요량", "산림품셈"): "QF",
    ("소요량", "건설품셈"): "QC",
    ("계수", "산림품셈"): "CF",
    ("계수", "건설품셈"): "CC",
    ("환율", "한국은행환율"): "FX",
    ("요율", "조달청제비율"): "RT",
    ("로직", "산림품셈"): "GF",
    ("로직", "건설품셈"): "GC",
    ("로직", "자체"): "GX",
}
GROUP_OF = {tid: group for (group, _), tid in TABLES.items()}
IDS = tuple(GROUP_OF)


def table_id(group: str, book: str | None) -> str:
    return TABLES.get((group, None)) or TABLES[(group, book)]


def id_of(ref: str) -> str:
    """키·「ID:원문번호」 의 테이블ID — 모르면 ""."""
    return ref[:2] if ref[:2] in GROUP_OF else ""


def load_book(path: Path = BOOK) -> dict:
    if path.is_file():
        return json.loads(path.read_text(encoding="utf-8"))
    return {"다음": {}, "키": {}}


def save_book(book: dict, path: Path = BOOK) -> None:
    """키 한 줄씩 — 머리 「다음」 뒤에 키 차례대로."""
    lines = ["{", '  "다음": ' + json.dumps(book["다음"], ensure_ascii=False) + ",", '  "키": {']
    items = sorted(book["키"].items())
    for i, (key, v) in enumerate(items):
        comma = "," if i < len(items) - 1 else ""
        lines.append(f"    {json.dumps(key)}: {json.dumps(v, ensure_ascii=False)}{comma}")
    lines += ["  }", "}"]
    path.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n")


def lookup(book: dict) -> dict:
    """(테이블ID, 원문번호) → 키 — 대장에서 한 번 세워 둠(`finish` 가 걷음)."""
    index = book.setdefault("_찾기", {})
    if not index:
        for key, v in book["키"].items():
            index[(key[:2], v["원문번호"])] = key
    return index


def issue(book: dict, tid: str, number: str, file: str) -> str:
    """(테이블ID, 원문번호) 의 키 — 대장에 있으면 그 키 · 없으면 다음 번호(원문번호 빈 새 줄 포함)."""
    index = lookup(book)
    key = index.get((tid, number)) if number else None
    if key:
        book["키"][key]["파일"] = file
        return key
    n = book["다음"].get(tid, 1)
    key = f"{tid}{n:06d}"
    book["다음"][tid] = n + 1
    book["키"][key] = {"원문번호": number, "파일": file}
    index[(tid, number)] = key
    return key


def finish(book: dict) -> dict:
    book.pop("_찾기", None)
    return book


def stamp(
    book: dict, tid: str, rows: list[dict], file: str, extra: dict | None = None
) -> list[dict]:
    """빌더가 만든 줄(「열쇠」 = 원문번호) → 「키」 · 「원문번호」 · extra 칸을 앞에 둔 줄."""
    out = []
    for row in rows:
        row = dict(row)
        number = str(row.pop("열쇠", row.get("원문번호")))
        row.pop("원문번호", None)
        key = row.pop("키", None) or issue(book, tid, number, file)
        out.append({"키": key, "원문번호": number, **(extra or {}), **row})
    return out


# ── 빌더 잇기 — 빌더는 옛 모양(「열쇠」 · 원문별 파일 이름)으로 쓰고 adopt 가 새 틀로 받음 ──
LABOR = {
    "건설노임": "건설업",
    "제조노임": "제조업",
    "엔지니어링노임": "엔지니어링",
    "측량노임": "측량",
    "건설사업관리노임": "건설사업관리",
    "SW노임": "SW",
    "산림노임": "산림",
    "준용대상": "준용대상",
}
MERGED = {
    "인력": ("인력.json", "조사", LABOR),
    "기계": ("기계.json", "세부분류", {"건설품셈": "건설품셈", "산림품셈": "산림품셈"}),
}
DIVISIONS = ("공통", "토목", "건축", "기계설비", "유지관리")
OLD_CHAPTER = re.compile(r"^(소요량|계수|로직)_(산림품셈|건설품셈)_(\D*)(\d+)장\.json$")
CHAPTER = re.compile(r"^(소요량|계수|로직)_(산림품셈|건설품셈|자체)_(\d\d)장_.+\.json$")
_BODY = MASTER.parents[1] / "resources/knowledge/original/원가계산"
_OLDREF = re.compile(r"(인력|재료|기계|소요량|계수|로직|환율|요율):([^:\s·]+):(.+?)(?=\s·\s|$)")
_STR = re.compile(r'"(?:[^"\\]|\\.)*"')


def title(book: str, division: str, number: int) -> str:
    """원문 장 제목 — 본문 폴더 이름 그대로(「제07장_지붕_및_홈통공사」 → 「지붕_및_홈통공사」)."""
    base = _BODY / "산림_표준품셈/본문"
    if book == "건설품셈":
        index = DIVISIONS.index(division) + 1
        base = _BODY / "건설공사_표준품셈/본문" / f"{index:02d}_{division}부문"
    (hit,) = base.glob(f"제{number:02d}장_*")
    return hit.name.split("_", 1)[1]


def chapter_file(group: str, book: str, division: str, number: int) -> str:
    return f"{group}_{book}_{number:02d}장_{title(book, division, number)}.json"


def _items(data: dict) -> str:
    return "표" if data.get("그룹") in ("소요량", "계수") else "줄"


def _read(path: Path, exact: bool = True) -> dict:
    text = path.read_text(encoding="utf-8")
    return json.loads(text, parse_float=Decimal, parse_int=Decimal) if exact else json.loads(text)


def old_view(name: str, folder: Path = MASTER) -> dict:
    """옛 파일 이름으로 읽기(빌더용 · 수는 float) — 줄의 「열쇠」 = 원문번호 · 키·갈래 칸 뺌.
    준용·통합·연결 같은 참조는 키 그대로."""
    m = re.match(r"^(인력|기계)_(.+)\.json$", name)
    if m:
        file, slot, labels = MERGED[m.group(1)]
        data = _read(folder / file, exact=False)
        label = labels[m.group(2)]
        head = {"그룹": m.group(1), **data[slot][label]}
        rows = [r for r in data["줄"] if r.get(slot) == label]
    else:
        c = OLD_CHAPTER.match(name)
        path = folder / (chapter_file(c[1], c[2], c[3], int(c[4])) if c else name)
        data = _read(path, exact=False)
        head, rows, slot = (
            {k: v for k, v in data.items() if k != _items(data)},
            data[_items(data)],
            "",
        )
    drop = ("키", "원문번호", slot)
    out = [{"열쇠": r["원문번호"], **{k: v for k, v in r.items() if k not in drop}} for r in rows]
    return {**head, _items(head): out}


def _old_ref(book: dict, text: str) -> str:
    """옛 참조 `그룹:원문:열쇠` → 키(대장에 없으면 그대로 — check_master 가 잡음)."""
    group, src, number = (text.split(":", 2) + ["", ""])[:3]
    try:
        tid = table_id(group, src)
    except KeyError:
        return text
    return lookup(book).get((tid, number), text)


def _convert(book: dict, row: dict, names: dict) -> dict:
    """줄 속 옛 참조 — 준용·통합 · 후보 글(키 옆 이름) · 품셈재료 연결 · 시중물가 조달 나라장터 연결."""
    for col in ("준용", "통합"):
        value = row.get(col)
        if isinstance(value, str) and ":" in value:
            row[col] = _old_ref(book, value)
    if isinstance(row.get("후보"), str):

        def one(m):
            key = _old_ref(book, m.group(0))
            return f"{key} {names.get(key, m.group(3))}" if KEY.fullmatch(key) else key

        row["후보"] = _OLDREF.sub(one, row["후보"])
    link = row.get("연결")
    if isinstance(link, str) and not KEY.fullmatch(link):
        row["연결"] = lookup(book).get(("MT", link), link)
    value = row.get("값")
    supply = value.get("조달") if isinstance(value, dict) else None
    if isinstance(supply, dict) and str(supply.get("출처") or "").startswith("나라장터:"):
        code = supply["출처"].split(":", 1)[1]
        if not KEY.fullmatch(code):  # 묶음은 새로 만듦 — 옛 줄과 나눠 쓰지 않게
            source = "나라장터:" + lookup(book).get(("MN", code), code)
            row["값"] = {**value, "조달": {**supply, "출처": source}}
    return row


def _store():
    import sys

    sys.path.insert(0, str(MASTER.parents[1]))
    from M01_MasterData import M01_MasterData_Store as st

    return st


def dump_rows(head: dict, rows: list[dict]) -> str:
    """요소 파일 — 머리는 들여쓰기 2칸 · 줄은 한 줄씩(길면 펼침)."""
    st = _store()
    out = ["{"] + [f"  {st._atom(k)}: {st.dump(v, 2)}," for k, v in head.items()]
    out.append('  "줄": [')
    for i, row in enumerate(rows):
        line = st._flat(row)
        body = line if len(line) <= 450 else st.dump(row, 4, expand=True)
        out.append("    " + body + ("," if i < len(rows) - 1 else ""))
    return "\n".join(out + ["  ]", "}"]) + "\n"


def keep_text(text: str, old: dict, new: dict) -> str | None:
    """글 모양 그대로 — 「열쇠」 칸 · 바뀐 글 · 머리 부문·차례만 갈음 · 읽어 맞대어 다르면 None."""
    rows_old, rows_new = old[_items(old)], new[_items(new)]
    spots = list(re.finditer(r'"열쇠": ("(?:[^"\\]|\\.)*"|[-\d.]+)', text))
    if len(spots) != len(rows_new):
        return None
    parts, at = [], 0
    for m, row in zip(spots, rows_new):
        number = json.dumps(row["원문번호"], ensure_ascii=False)
        parts += [text[at : m.start()], f'"키": "{row["키"]}", "원문번호": {number}']
        at = m.end()
    text = "".join(parts) + text[at:]
    swap = {}
    for a_row, b_row in zip(rows_old, rows_new):
        for k in b_row:
            if k in a_row and a_row[k] != b_row[k]:
                for a, b in zip(_strings(a_row[k]), _strings(b_row[k])):
                    if a != b:
                        swap[json.dumps(a, ensure_ascii=False)] = json.dumps(b, ensure_ascii=False)
    text = _STR.sub(lambda m: swap.get(m.group(0), m.group(0)), text)
    add = "".join(
        f"  {json.dumps(k, ensure_ascii=False)}: {_store()._atom(new[k])},\n"
        for k in ("부문", "차례")
        if k in new and k not in old
    )
    if add:
        text = re.sub(r'(\n  "원문": "[^"]*",\n)', lambda m: m.group(1) + add, text, count=1)
    same = json.loads(text, parse_float=Decimal, parse_int=Decimal) == new
    return text if same else None


def _strings(v):
    if isinstance(v, str):
        yield v
    elif isinstance(v, dict):
        for x in v.values():
            yield from _strings(x)
    elif isinstance(v, list):
        for x in v:
            yield from _strings(x)


def order_chapters(folder: Path = MASTER) -> None:
    """장 파일 머리 「차례」 — 그룹마다 산림 1~14장 → 건설 공통 · 토목 · 건축 · 기계설비 · 유지관리."""
    found = {}
    for path in folder.glob("*.json"):
        m = CHAPTER.match(path.name)
        if m:
            data = _read(path)
            division = data.get("부문") or ""
            rank = (m[2] != "산림품셈", DIVISIONS.index(division) if division else -1, int(m[3]))
            found.setdefault(m[1], []).append((rank, path.name, data.get("차례")))
    for items in found.values():
        for i, (_, name, had) in enumerate(sorted(items), start=1):
            if had != i:
                path = folder / name
                text = re.sub(
                    r'"차례": \d+', f'"차례": {i}', path.read_text(encoding="utf-8"), count=1
                )
                path.write_text(text, encoding="utf-8", newline="\n")


def adopt(folder: Path = MASTER) -> list[str]:
    """빌더가 옛 모양으로 쓴 파일 → 새 틀(키 · 원문번호 · 한 테이블 · 장 파일 이름) · 옛 파일은 지움.
    키는 대장의 (테이블ID, 원문번호) 키 · 없으면 다음 번호. 로직은 키로 적어 넣으므로 받지 않음."""
    book = load_book(folder / BOOK.name)
    names = {}
    if (folder / "인력.json").is_file():
        names = {r["키"]: r.get("이름") for r in _read(folder / "인력.json")["줄"]}
    done = []
    for path in sorted(folder.glob("*.json")):
        if path.name.startswith("_"):
            continue
        data = _read(path)
        rows = data.get(_items(data)) or []
        if not rows or "열쇠" not in rows[0]:
            continue
        group, src = data["그룹"], data["원문"]
        if group == "로직":
            raise SystemExit(f"{path.name} · 로직은 키로 적음 — 옛 모양은 받지 않음")
        c = OLD_CHAPTER.match(path.name)
        target = chapter_file(c[1], c[2], c[3], int(c[4])) if c else path.name
        target = MERGED[group][0] if group in MERGED else target
        if c and c[3]:  # 건설품셈 원문번호 앞에 부문(「기계설비 7-1-1 …」)
            head = c[3] + " "
            rows = [
                r if str(r["열쇠"]).startswith(head) else {**r, "열쇠": head + r["열쇠"]}
                for r in rows
            ]
            data[_items(data)] = rows
        stamped = [
            _convert(book, r, names) for r in stamp(book, table_id(group, src), rows, target)
        ]
        if group in MERGED:
            file, slot, labels = MERGED[group]
            label = labels[src]
            whole = _read(folder / file) if (folder / file).is_file() else {"그룹": group}
            heads = whole.setdefault(slot, {})
            heads[label] = {k: v for k, v in data.items() if k not in ("그룹", "줄")}
            stamped = [
                {"키": r["키"], "원문번호": r["원문번호"], slot: label, **r} for r in stamped
            ]
            keep = [r for r in whole.get("줄", []) if r.get(slot) != label]
            rows_all = sorted(keep + stamped, key=lambda r: r["키"])
            text = dump_rows({k: v for k, v in whole.items() if k != "줄"}, rows_all)
        else:
            head = {"그룹": group, "원문": src}
            if c and c[3]:
                head["부문"] = c[3]
            if c:
                head["차례"] = Decimal(0)  # order_chapters 가 채움
            rest = {k: v for k, v in data.items() if k not in head}
            new = {**head, **rest, _items(data): stamped}
            old_text = path.read_text(encoding="utf-8")
            text = keep_text(old_text, data, new) or _store().dump(new)
        (folder / target).write_text(text, encoding="utf-8", newline="\n")
        if target != path.name:
            path.unlink()
        done.append(target)
    save_book(finish(book), folder / BOOK.name)
    order_chapters(folder)
    return done