Aislo/resources/master_data/scripts/_migrate_base.py

# -*- coding: utf-8 -*-
"""1_기초 옛 파일 7 → 새 틀(`_틀.md` 2~4장) 요소 파일 6 옮김 · 대조.

돌리기: `./venv/Scripts/python.exe resources/master_data/scripts/_migrate_base.py`
옛 파일은 `old/` 에서 읽기만 함. 새 파일은 `resources/master_data/` 첫 층에 씀. 값은 바꾸지 않음.
`dump_master` 는 `_build_mat_price.py` 도 씀.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

MD = Path(__file__).resolve().parents[1]
OLD = MD / "old"

UNIT = {
    "KRW/day": "원/일",
    "KRW/L": "원/L",
    "KRW/USD": "원/USD",
    "KRW/JPY100": "원/JPY100",
    "KRW/EUR": "원/EUR",
    "KRW/GBP": "원/GBP",
    "KRW/CNY": "원/CNY",
}
STATUS = {"published": "공표", "not_published": "미공표"}


def dump_master(path: Path, head: dict, rows: list[dict]) -> None:
    """머리는 들여쓰기 2칸 · 줄은 한 줄씩(길면 펼침) · UTF-8 · LF."""
    out = ["{"]
    for key, val in head.items():
        text = json.dumps(val, ensure_ascii=False, indent=2).replace("\n", "\n  ")
        out.append(f"  {json.dumps(key, ensure_ascii=False)}: {text},")
    out.append('  "줄": [')
    for i, row in enumerate(rows):
        line = json.dumps(row, ensure_ascii=False, separators=(", ", ": "))
        if len(line) <= 450:
            body = "{ " + line[1:-1] + " }"
        else:
            body = json.dumps(row, ensure_ascii=False, indent=2).replace("\n", "\n    ")
        out.append("    " + body + ("," if i < len(rows) - 1 else ""))
    out += ["  ]", "}"]
    path.write_bytes(("\n".join(out) + "\n").encode("utf-8"))


def load(name: str) -> dict:
    return json.loads((OLD / name).read_text(encoding="utf-8"))


def origin(sources: list[dict]) -> list[dict]:
    return [{"경로": s["path"], "해시": s["sha256"], "역할": s["role"]} for s in sources]


def pick(rec: dict, allowed: set[str]) -> None:
    extra = set(rec) - allowed
    assert not extra, f"모르는 칸 {extra}"


# ── 인력 ────────────────────────────────────────────────────────────────
LABOR_EXTRA = {  # 옛 칸 → 새 칸
    "hours_per_day": "일시간",
    "reliability": "신뢰도",
    "industry": "업종",
    "variation_coefficient": "변동계수",
    "source_flag": "원문기호",
}
LABOR_KEYS = {
    "occupation_code",
    "occupation_name",
    "unit",
    "daily_wage_krw",
    "status",
    *LABOR_EXTRA,
}


def labor_rows(recs: list[dict], origin_name: str) -> list[dict]:
    rows = []
    for r in recs:
        pick(r, LABOR_KEYS)
        row = {
            "열쇠": r["occupation_code"],
            "이름": r["occupation_name"],
            "규격": "",
            "단위": UNIT[r["unit"]],
            "값": r.get("daily_wage_krw"),
            "출처": f"{origin_name} {r['occupation_code']}",
        }
        for old, new in LABOR_EXTRA.items():
            if old in r:
                row[new] = r[old]
        row["상태"] = STATUS[r["status"]]
        rows.append(row)
    return rows


def labor_back(rows: list[dict]) -> list[dict]:
    inv_unit = {v: k for k, v in UNIT.items()}
    inv_status = {v: k for k, v in STATUS.items()}
    out = []
    for w in rows:
        r = {
            "occupation_code": w["열쇠"],
            "occupation_name": w["이름"],
            "unit": inv_unit[w["단위"]],
        }
        for old, new in LABOR_EXTRA.items():
            if new in w:
                r[old] = w[new]
        if w["값"] is not None:
            r["daily_wage_krw"] = w["값"]
        r["status"] = inv_status[w["상태"]]
        out.append(r)
    return out


# ── 재료 ────────────────────────────────────────────────────────────────
MAT_EXTRA = {
    "classification_code": "분류코드",
    "notice_datetime": "공시일시",
    "notice_number": "공시번호",
    "business_division_code": "사업구분코드",
    "business_division_name": "사업구분",
    "vat_basis": "부가세",
    "price_type": "가격종류",
    "delivery_condition": "인도조건",
    "field": "분야",
    "price_adopted_reason": "비고",
}
MAT_KEYS = {"item_code", "classification_name", "specification", "unit", "price_krw", *MAT_EXTRA}


def mat_rows(recs: list[dict]) -> list[dict]:
    rows = []
    for r in recs:
        pick(r, MAT_KEYS)
        row = {
            "열쇠": r["item_code"],
            "이름": r["classification_name"],
            "규격": r["specification"],
            "단위": r["unit"],
            "값": r["price_krw"],
            "출처": f"나라장터자재 {r['item_code']}",
        }
        for old, new in MAT_EXTRA.items():
            if old in r:
                row[new] = r[old]
        rows.append(row)
    return rows


def mat_back(rows: list[dict]) -> list[dict]:
    inv = {v: k for k, v in MAT_EXTRA.items()}
    out = []
    for w in rows:
        r = {
            "item_code": w["열쇠"],
            "classification_name": w["이름"],
            "specification": w["규격"],
            "unit": w["단위"],
            "price_krw": w["값"],
        }
        for new, old in inv.items():
            if new in w:
                r[old] = w[new]
        out.append(r)
    return out


# ── 유가 · 환율 ─────────────────────────────────────────────────────────
FUEL = {"oil_gasoline": "휘발유", "oil_diesel": "경유"}
FX_NAME = {
    "fx_usd": "USD",
    "fx_jpy100": "JPY100",
    "fx_eur": "EUR",
    "fx_gbp": "GBP",
    "fx_cny": "CNY",
}


def oil_rows(avg: dict, reg: dict) -> list[dict]:
    rows = []
    for var, v in avg["variables"].items():
        rows.append(
            {
                "열쇠": f"{FUEL[var]}:전국평균",
                "이름": v["source_product_name"],
                "규격": "",
                "단위": UNIT[v["unit"]],
                "값": v["value"],
                "출처": f"오피넷유가 {v['source_product_code']}",
                "구분": "전국평균",
                "지역": "전국평균",
                "기준일": v["date"],
                "상품코드": v["source_product_code"],
            }
        )
    for var, v in reg["variables"].items():
        for r in v["records"]:
            rows.append(
                {
                    "열쇠": f"{FUEL[var]}:{r['sido_name']}",
                    "이름": v["source_product_name"],
                    "규격": "",
                    "단위": UNIT[v["unit"]],
                    "값": r["value"],
                    "출처": f"오피넷유가 {v['source_product_code']}",
                    "구분": "시도별",
                    "지역": r["sido_name"],
                    "지역코드": r["sido_code"],
                    "기준일": v["date"],
                    "상품코드": v["source_product_code"],
                }
            )
    return rows


def fx_rows(fx: dict) -> list[dict]:
    return [
        {
            "열쇠": v["source_item_code"],
            "이름": FX_NAME[var],
            "규격": "",
            "단위": UNIT[v["unit"]],
            "값": v["value"],
            "출처": f"한국은행환율 {v['source_item_code']}",
            "기준일": v["date"],
        }
        for var, v in fx["variables"].items()
    ]


# ── 요율 ────────────────────────────────────────────────────────────────
RATE_NAME = {
    "rate_sanjae": "산재보험료",
    "rate_goyong": "고용보험료",
    "rate_health": "건강보험료",
    "rate_pension": "국민연금보험료",
    "rate_care": "노인장기요양보험료",
    "rate_safety_pct": "산업안전보건관리비",
    "rate_safety_base": "산업안전보건관리비 기초액",
    "rate_indirect_labor": "간접노무비",
    "rate_other_expense": "기타경비",
    "rate_environment": "환경보전비",
    "rate_retirement_mutual_aid": "퇴직공제부금비",
    "rate_wage_claim_contribution": "임금채권보장분담금",
    "rate_asbestos_contribution": "석면피해구제분담금",
    "rate_subcontract_payment_guarantee": "하도급대금 지급보증 수수료",
    "rate_equipment_payment_guarantee": "건설기계대여대금 지급보증 수수료",
    "rate_performance_guarantee_fee": "공사이행보증 수수료",
    "rate_overhead": "일반관리비",
    "rate_profit": "이윤",
    "rate_vat": "부가가치세",
}


def rate_rows(rates: dict) -> list[dict]:
    assert set(rates["variables"]) == set(RATE_NAME)
    return [
        {"열쇠": k, "이름": RATE_NAME[k], "값": v, "출처": "조달청제비율"}
        for k, v in rates["variables"].items()
    ]


# ── 옮기기 ──────────────────────────────────────────────────────────────
F_LC = "1_기초_노임_건설업_대한건설협회_2026-01-01.json"
F_LM = "1_기초_노임_제조업_중소기업중앙회_2026-07-01.json"
F_OR = "1_기초_유가_시도별_오피넷_2026-09-09.json"
F_OA = "1_기초_유가_전국평균_오피넷_2026-08-14.json"
F_MT = "1_기초_자재_시설공통_나라장터_2026-08-14.json"
F_RT = "1_기초_제비율_토목_조달청_2026-04-13.json"
F_FX = "1_기초_환율_일별_한국은행_2026-08-14.json"

NEW = {
    "lc": "인력_건설노임.json",
    "lm": "인력_제조노임.json",
    "mt": "재료_나라장터자재.json",
    "oil": "재료_오피넷유가.json",
    "fx": "환율_한국은행환율.json",
    "rt": "요율_조달청제비율.json",
}


def migrate() -> None:
    lc, lm, mt = load(F_LC), load(F_LM), load(F_MT)
    orr, oa, rt, fx = load(F_OR), load(F_OA), load(F_RT), load(F_FX)

    lcv = lc["variables"]["labor_rate"]
    dump_master(
        MD / NEW["lc"],
        {
            "그룹": "인력",
            "원문": "건설노임",
            "판": lc["effective_date"],
            "공표일": lc["publication_date"],
            "원천": origin(lc["sources"]),
            "비고": [lcv["reliability_note"]],
        },
        labor_rows(lcv["records"], "건설노임"),
    )
    lmv = lm["variables"]["labor_mfg"]
    dump_master(
        MD / NEW["lm"],
        {
            "그룹": "인력",
            "원문": "제조노임",
            "판": lm["effective_date"],
            "공표일": lm["publication_date"],
            "조사월": lm["survey_month"],
            "원천": origin(lm["sources"]),
            "비고": [lmv["industry_note"]],
        },
        labor_rows(lmv["records"], "제조노임"),
    )
    dump_master(
        MD / NEW["mt"],
        {
            "그룹": "재료",
            "원문": "나라장터자재",
            "판": mt["effective_date"],
            "원천": origin(mt["sources"]),
            "선정규칙": mt["selection_policy"],
            "원천행수": mt["source_row_count"],
            "제외묶음": mt["excluded_named_groups"],
            "0원미회수": {
                "비고": mt["zero_price_unrecovered"]["note"],
                "줄": mt["zero_price_unrecovered"]["items"],
            },
        },
        mat_rows(mt["variables"]["mat_price"]["records"]),
    )
    dump_master(
        MD / NEW["oil"],
        {
            "그룹": "재료",
            "원문": "오피넷유가",
            "판": orr["effective_date"],
            "원천": origin(oa["sources"]) + origin(orr["sources"]),
            "비고": [orr["note"]],
        },
        oil_rows(oa, orr),
    )
    dump_master(
        MD / NEW["fx"],
        {
            "그룹": "환율",
            "원문": "한국은행환율",
            "판": fx["effective_date"],
            "원천": origin(fx["sources"]),
        },
        fx_rows(fx),
    )
    dump_master(
        MD / NEW["rt"],
        {
            "그룹": "요율",
            "원문": "조달청제비율",
            "판": rt["effective_date"],
            "원천": origin(rt["sources"]),
            "처리규칙": rt["processing_rules"],
        },
        rate_rows(rt),
    )


# ── 대조 ────────────────────────────────────────────────────────────────
def new(key: str) -> dict:
    return json.loads((MD / NEW[key]).read_text(encoding="utf-8"))


def num_sum(vals) -> float:
    return round(sum(v for v in vals if isinstance(v, (int, float))), 6)


def leaves(o):
    if isinstance(o, dict):
        for v in o.values():
            yield from leaves(v)
    elif isinstance(o, list):
        for v in o:
            yield from leaves(v)
    elif isinstance(o, (int, float)) and not isinstance(o, bool):
        yield o


def check() -> bool:
    res = []  # (이름, 옛 줄, 새 줄, 옛 합, 새 합, 같음)

    def add(name, old_rows, new_rows, old_sum, new_sum, same):
        res.append((name, len(old_rows), len(new_rows), old_sum, new_sum, same))

    for key, f, v, fn in (
        ("lc", F_LC, "labor_rate", "건설노임"),
        ("lm", F_LM, "labor_mfg", "제조노임"),
    ):
        old = load(f)["variables"][v]["records"]
        nw = new(key)["줄"]
        add(
            fn,
            old,
            nw,
            num_sum(r.get("daily_wage_krw") for r in old),
            num_sum(r["값"] for r in nw),
            labor_back(nw) == old,
        )
    old = load(F_MT)["variables"]["mat_price"]["records"]
    nw = new("mt")["줄"]
    add(
        "나라장터자재",
        old,
        nw,
        num_sum(r["price_krw"] for r in old),
        num_sum(r["값"] for r in nw),
        mat_back(nw) == old,
    )
    oa, orr = load(F_OA), load(F_OR)
    nw = new("oil")["줄"]
    old_vals = [v["value"] for v in oa["variables"].values()] + [
        r["value"] for v in orr["variables"].values() for r in v["records"]
    ]
    n_old = len(oa["variables"]) + sum(len(v["records"]) for v in orr["variables"].values())
    keys_ok = len({r["열쇠"] for r in nw}) == len(nw)
    same = keys_ok and [r["값"] for r in nw] == old_vals and nw == oil_rows(oa, orr)
    res.append(
        ("오피넷유가", n_old, len(nw), num_sum(old_vals), num_sum(r["값"] for r in nw), same)
    )
    fx = load(F_FX)
    nw = new("fx")["줄"]
    old_vals = [v["value"] for v in fx["variables"].values()]
    same = [r["값"] for r in nw] == old_vals and nw == fx_rows(fx)
    res.append(
        (
            "한국은행환율",
            len(old_vals),
            len(nw),
            num_sum(old_vals),
            num_sum(r["값"] for r in nw),
            same,
        )
    )
    rt = load(F_RT)
    nw = new("rt")
    same = {r["열쇠"]: r["값"] for r in nw["줄"]} == rt["variables"] and nw["처리규칙"] == rt[
        "processing_rules"
    ]
    old_sum = num_sum(leaves(rt["variables"]))
    res.append(
        (
            "조달청제비율",
            len(rt["variables"]),
            len(nw["줄"]),
            old_sum,
            num_sum(leaves([r["값"] for r in nw["줄"]])),
            same,
        )
    )

    print(f"{'파일':<12}{'옛 줄':>7}{'새 줄':>7}{'옛 값합':>16}{'새 값합':>16}  일치")
    ok = True
    for name, a, b, sa, sb, same in res:
        good = a == b and sa == sb and same
        ok &= good
        print(f"{name:<12}{a:>7}{b:>7}{sa:>16}{sb:>16}  {'예' if good else '아니오'}")
    return ok


if __name__ == "__main__":
    sys.stdout.reconfigure(encoding="utf-8")
    migrate()
    sys.exit(0 if check() else 1)