feat(z01): 기초단가 다섯 고치기 — 노임·기계·자재·유가·요율을 kind 마다 한 표(자료 제 열쇠 @id · 판 기준일 칸) · 고친 값은 덮개 resources/data_master_override/{kind}.json 에만(원본은 읽기만) · 원본 바뀐 덮개는 덮개 값 유지 + source_changed · 주인 없음은 버리지 않고 모아 보기(서버 정렬 · 거름 · 쪽)와 한꺼번에 비우기 · 기계 시간당 단가는 계산값으로 잠그고 산출근거(@formula) · 노임·유가 덮개가 기계 계산에 닿음 · B09 hourly_cost_of 와 대조(브레인 계약 · 사용자 지시)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn

Z01_MasterData/Z01_MasterData_BasePrices.py

@@ -0,0 +1,347 @@
+"""Z01 기초단가 다섯 — 노임·기계·자재·유가·요율을 kind 마다 **한 표**로(2026-09-15 사용자 지시 · 브레인 계약).
+
+표 수준(한 번): columns · total · editable · locked{열: 왜 못 고치나} · formula{열: 식} · notice(요율만)
+줄 수준(해당할 때만): `@id`(자료 제 열쇠) · `@overrides`(고친 줄) · `@formula`(계산값 줄)
+`@id` — 줄 차례가 아니라 자료 열쇠(브레인 승인): 노임 `{dataset_id}/{occupation_code}` · 기계 machine_code ·
+  자재 item_code · 유가 `{scope}/{변수}[/{sido_code}]` · 요율 `{변수}[/{목록}/{구간칸=값;…}]`(구간 칸 이름 차례 · 값·식 칸 뺌)
+원본 파일 자리는 이름표 파일이 가리키는 곳(`Z01_MasterData_Tables`) — 원본은 읽기만, 고친 값은 덮개(`Z01_MasterData_Overrides`).
+"""
+
+from __future__ import annotations
+
+import json
+from decimal import Decimal
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+from Z01_MasterData import Z01_MasterData_BasePrices_Machine as machine
+from Z01_MasterData import Z01_MasterData_Overrides as overrides
+from Z01_MasterData import Z01_MasterData_Tables as tables
+
+KINDS = ("labor", "machine", "material", "oil", "rate")
+_KEY = "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음"
+_SOURCE = "공표 원문 칸 — 원본 갱신으로만 바뀜"
+_RATE_EDITABLE = (
+    "rate_percent",
+    "base_amount_krw",
+    "minimum_estimated_amount_krw",
+    "minimum_total_construction_amount_krw",
+    "rate_from_2033_percent",
+    "manager_thresholds.default_estimated_amount_krw",
+    "manager_thresholds.civil_main_work_estimated_amount_krw",
+)
+_RATE_NOTICE = "법이 정한 값 — 고치면 원본 갱신 때 구간 이름이 바뀌어 고친 값이 주인을 잃기 쉬움(주인 없음으로 남음)"
+
+
+def _source(file_id: str) -> Path:
+    labels = tables.load_labels()
+    entry = next((f for f in labels["files"] if f["file_id"] == file_id), None)
+    path = tables._file_path(entry) if entry else None
+    if path is None:
+        raise FileNotFoundError(f"이름표에 원본 자리가 없음: {file_id}")
+    return path
+
+
+@lru_cache(maxsize=16)
+def _read(path: str, _mtime_ns: int) -> dict[str, Any]:
+    return json.loads(Path(path).read_text(encoding="utf-8"))
+
+
+def _doc(file_id: str) -> tuple[dict[str, Any], str]:
+    path = _source(file_id)
+    return _read(str(path), path.stat().st_mtime_ns), path.name
+
+
+def _records(doc: dict[str, Any]) -> list[dict[str, Any]]:
+    return next(
+        v["records"] for v in doc["variables"].values() if isinstance(v.get("records"), list)
+    )
+
+
+def _labor() -> list[dict[str, Any]]:
+    rows = []
+    for file_id in ("labor_const", "labor_mfg"):
+        doc, name = _doc(file_id)
+        dataset = doc.get("dataset_id") or file_id
+        for r in _records(doc):
+            # 기준일 — 두 판(건설 · 제조) 기준일이 달라 한 표에 섞임 → 칸으로 드러냄
+            row = {
+                "@id": f"{dataset}/{r['occupation_code']}",
+                "@source": name,
+                "dataset_id": dataset,
+                "effective_date": doc.get("effective_date"),
+                **r,
+            }
+            row.setdefault("daily_wage_krw", None)  # 미공표 직종 — 사람이 넣을 수 있음
+            rows.append(row)
+    return rows
+
+
+def _material() -> list[dict[str, Any]]:
+    doc, name = _doc("mat_price_public")
+    date = doc.get("effective_date")
+    return [
+        {"@id": str(r["item_code"]), "@source": name, "effective_date": date, **r}
+        for r in _records(doc)
+    ]
+
+
+def _oil() -> list[dict[str, Any]]:
+    rows = []
+    for file_id in ("oil", "oil_regional"):
+        doc, name = _doc(file_id)
+        for variable, v in doc["variables"].items():
+            head = {k: x for k, x in v.items() if k not in ("records", "value")}
+            common = {
+                "scope": v.get("scope"),
+                "fuel": variable,
+                "effective_date": doc.get("effective_date"),  # 전국 · 지역 판 기준일이 다름
+                **head,
+            }
+            if "records" not in v:
+                rows.append(
+                    {
+                        "@id": f"{v.get('scope')}/{variable}",
+                        "@source": name,
+                        **common,
+                        "value": v.get("value"),
+                    }
+                )
+            for r in v.get("records") or []:
+                rows.append(
+                    {
+                        "@id": f"{v.get('scope')}/{variable}/{r['sido_code']}",
+                        "@source": name,
+                        **common,
+                        **r,
+                    }
+                )
+    return rows
+
+
+def _flat(d: dict[str, Any], prefix: str = "") -> dict[str, Any]:
+    out: dict[str, Any] = {}
+    for k, v in d.items():
+        if isinstance(v, dict):
+            out.update(_flat(v, f"{prefix}{k}."))
+        elif not isinstance(v, list):
+            out[f"{prefix}{k}"] = v
+    return out
+
+
+def _rate() -> list[dict[str, Any]]:
+    doc, name = _doc("rates")
+    rows = []
+    for variable, v in doc["variables"].items():
+        rows.append(
+            {
+                "@id": variable,
+                "@source": name,
+                "variable": variable,
+                "part": "",
+                "effective_date": doc.get("effective_date"),
+                **_flat(v),
+            }
+        )
+        for part, items in v.items():
+            if not (isinstance(items, list) and items and all(isinstance(x, dict) for x in items)):
+                continue
+            for item in items:
+                bracket = ";".join(
+                    f"{k}={item[k]}"
+                    for k in sorted(item)
+                    if k not in _RATE_EDITABLE
+                    and k != "formula"
+                    and not isinstance(item[k], (dict, list))
+                )
+                rows.append(
+                    {
+                        "@id": f"{variable}/{part}/{bracket}",
+                        "@source": name,
+                        "variable": variable,
+                        "part": part,
+                        "effective_date": doc.get("effective_date"),
+                        "base": v.get("base"),
+                        **item,
+                    }
+                )
+    return rows
+
+
+_BUILDERS = {"labor": _labor, "material": _material, "oil": _oil, "rate": _rate}
+
+
+def base_rows(kind: str) -> list[dict[str, Any]]:
+    """덮개 얹기 전 원본 줄 — 기계는 입력 칸까지(계산 칸은 덮개를 얹은 뒤 셈)."""
+    if kind == "machine":
+        doc, name = _doc("mach_base")
+        return machine.base_rows(name, doc.get("effective_date"))
+    return _BUILDERS[kind]()
+
+
+def _machine_inputs() -> machine.Inputs:
+    """기계가 끌어 쓰는 밑값 — **덮개를 얹은** 노임(건설)·전국평균 유가 + 그 판들의 기준일."""
+    labor = [r for r in overrides.overlay("labor", _labor()) if r["dataset_id"] == "labor_const"]
+    oil = [
+        r for r in overrides.overlay("oil", _oil()) if r["@id"] == f"national_average/{r['fuel']}"
+    ]
+    return machine.Inputs(
+        wages={
+            str(r["occupation_code"]): None
+            if r["daily_wage_krw"] is None
+            else Decimal(str(r["daily_wage_krw"]))
+            for r in labor
+        },
+        wage_date=labor[0]["effective_date"] if labor else None,
+        oil={r["fuel"]: Decimal(str(r["value"])) for r in oil},
+        oil_dates={r["fuel"]: r.get("date") or r["effective_date"] for r in oil},
+        operating_date=_doc("pum_const")[0].get("effective_date"),
+    )
+
+
+def rows(kind: str) -> list[dict[str, Any]]:
+    """덮개를 얹은 줄 — 기계는 그 위에서 시간당 단가를 다시 셈."""
+    out = overrides.overlay(kind, base_rows(kind))
+    if kind == "machine":
+        inputs = _machine_inputs()
+        out = [machine.compute(r, inputs) for r in out]
+    return out
+
+
+def _columns(kind: str, all_rows: list[dict[str, Any]]) -> list[str]:
+    keys: dict[str, None] = {}
+    for r in all_rows:
+        keys.update(dict.fromkeys(k for k in r if not k.startswith("@")))
+    return list(keys)
+
+
+def spec(kind: str, columns: list[str]) -> dict[str, Any]:
+    """표 수준 — editable · locked · formula (· notice)."""
+    if kind == "machine":
+        return machine.spec(columns)
+    editable = {
+        "labor": ["daily_wage_krw"],
+        "material": ["price_krw"],
+        "oil": ["value"],
+        "rate": [c for c in _RATE_EDITABLE if c in columns],
+    }[kind]
+    keys = {"labor": {"dataset_id", "occupation_code"}, "material": {"item_code"}}.get(kind, set())
+    if kind == "oil":
+        keys = {"scope", "fuel", "sido_code"}
+    if kind == "rate":
+        locked = {
+            c: "규칙·구간 칸(로직) — 원본 갱신으로만 바뀜" for c in columns if c not in editable
+        }
+    else:
+        locked = {c: _KEY if c in keys else _SOURCE for c in columns if c not in editable}
+    out: dict[str, Any] = {"editable": editable, "locked": locked, "formula": {}}
+    if kind == "rate":
+        out["notice"] = _RATE_NOTICE
+    return out
+
+
+def row_label(kind: str, row: dict[str, Any] | None) -> str:
+    if row is None:
+        return ""
+    if kind == "labor":
+        return str(row.get("occupation_name") or "")
+    if kind == "machine":
+        return f"{row.get('machine_name') or ''} {row.get('specification') or ''}".strip()
+    if kind == "material":
+        return str(row.get("specification") or row.get("classification_name") or "")
+    if kind == "oil":
+        return f"{row.get('source_product_name') or row.get('fuel')} {row.get('sido_name') or '전국평균'}"
+    named = (tables.load_labels().get("value_overrides") or {}).get(
+        f"rates::variables/{row['variable']}"
+    ) or {}
+    bracket = row["@id"].split("/", 2)[2].replace(";", " · ") if row["part"] else ""
+    return f"{named.get('name_ko') or row['variable']} {bracket}".strip()
+
+
+def column_meta(kind: str, key: str) -> dict[str, Any]:
+    """열 이름 — 이름표 `synthetic_tables['@base_prices/{kind}']` → `columns[열key]` → 영문 key."""
+    labels = tables.load_labels()
+    own = (labels.get("synthetic_tables") or {}).get(f"@base_prices/{kind}") or {}
+    named = next((c for c in own.get("columns") or [] if c.get("key") == key), None)
+    named = named or labels["columns"].get(key) or {}
+    return {
+        "key": key,
+        "label": named.get("name_ko") or key,
+        "unit": named.get("unit") or "",
+        "hidden": named.get("visible") is False,
+    }
+
+
+def public(row: dict[str, Any]) -> dict[str, Any]:
+    """내보낼 줄 — 안쪽 칸(`@source`)은 뺌."""
+    return {k: v for k, v in row.items() if k != "@source"}
+
+
+def table(
+    kind: str, page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE, q: str = ""
+) -> dict[str, Any]:
+    all_rows = rows(kind)
+    columns = _columns(kind, all_rows)
+    needle = q.strip().lower()
+    hits = (
+        [
+            r
+            for r in all_rows
+            if needle in json.dumps(public(r), ensure_ascii=False, default=str).lower()
+        ]
+        if needle
+        else all_rows
+    )
+    size = max(1, min(size, tables.MAX_PAGE_SIZE))
+    start = (max(page, 1) - 1) * size
+    return {
+        "columns": [column_meta(kind, c) for c in columns],
+        "rows": [public(r) for r in hits[start : start + size]],
+        "total": len(hits),
+        **spec(kind, columns),
+    }
+
+
+def edit(kind: str, row_id: str, values: dict[str, Any], by: Any) -> tuple[int, dict[str, Any]]:
+    """고치기 — (상태 코드, 고친 뒤 그 줄 또는 막은 까닭)."""
+    base = next((r for r in base_rows(kind) if r["@id"] == row_id), None)
+    if base is None:
+        return 404, {"message": f"없는 줄: {row_id}"}
+    info = spec(kind, _columns(kind, rows(kind)))
+    for column, value in values.items():
+        if column not in info["editable"] or column not in base:
+            reason = info["locked"].get(column) or "이 줄에 없는 칸"
+            return 400, {"message": f"못 고치는 칸 {column} — {reason}"}
+        problem = overrides.check_value(value)
+        if problem:
+            return 400, {"message": f"{column}: {problem}"}
+    overrides.write(kind, base, values, by)
+    return 200, public(next(r for r in rows(kind) if r["@id"] == row_id))
+
+
+def override_items(
+    kind: str = "", state: str = "", page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE
+) -> dict[str, Any]:
+    """모아 보기 — 서버가 정렬(원본 바뀜 → 주인 없음 → 고쳐짐) · 거름(kind·state) · 쪽 · kind 별 state 수."""
+    out = []
+    counts: dict[str, dict[str, int]] = {}
+    for each in KINDS:  # 수는 늘 다섯 다 셈 — 거름은 줄에만
+        base = base_rows(each)
+        by_id = {r["@id"]: r for r in base}
+        for item in overrides.items(each, base):
+            counts.setdefault(each, {}).setdefault(item["state"], 0)
+            counts[each][item["state"]] += 1
+            if (kind and each != kind) or (state and item["state"] != state):
+                continue
+            item["row_label"] = row_label(each, by_id.get(item["row_id"]))
+            item["column_label"] = column_meta(each, item["column"])["label"]
+            out.append(item)
+    out.sort(key=lambda i: overrides.STATE_ORDER.index(i["state"]))
+    size = max(1, min(size, tables.MAX_PAGE_SIZE))
+    start = (max(page, 1) - 1) * size
+    return {"items": out[start : start + size], "total": len(out), "counts": counts}
+
+
+def clear_overrides(kind: str, state: str) -> int:
+    return overrides.clear(kind, state, base_rows(kind))

Z01_MasterData/Z01_MasterData_BasePrices_Machine.py

@@ -0,0 +1,213 @@
+"""Z01 기초단가 — 기계 한 표: 기종 한 줄 + 시간당 단가(계산값) · 산출근거(2026-09-15 브레인 계약).
+
+기계단가는 취득가 613 · 손료계수 387 · 운전경비(연료·잡재료·조종원 — 건설품셈 8-4) · 노임 · 유가가 계산으로 물림.
+  고치는 칸 = 취득가 · 손료계수(이 표) — 노임·유가는 그 표에서 고치고 여기는 **덮개 얹은 값**을 끌어 씀
+  계산 칸 = 손료 · 연료 · 조종원 · 시간당 단가 — editable 에서 빼고 locked 에 까닭
+⚠ 셈은 B09 함수를 그대로 부름(`load_machine_catalog` · `load_operating_records` · `hourly_machine_cost`) — 식을 두 벌로 안 둠.
+  `hourly_cost_of`(B09) 와 같은 값인지 `test_z01_base_prices.py` 가 대조. 반만 선 값(연료·조종원 빠짐)은 시간당 단가를 안 냄.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from decimal import Decimal
+from functools import lru_cache
+from typing import Any
+
+from B09_Estimation.B09_Estimation_MachineCost import (
+    OPERATOR_WAGE_FORMULA,
+    hourly_machine_cost,
+    load_machine_catalog,
+)
+from B09_Estimation.B09_Estimation_MachineOperating import FUEL_VARIABLES, load_operating_records
+from Z01_MasterData.Z01_MasterData_Tables import RESOURCES
+
+#: B09 셈이 여는 원본 셋 — 바뀌면 캐시를 새로 읽음
+_INPUT_FILES = ("mach_base_2026.json", "pum_const_2026.json", "labor_const_2026-01-01.json")
+EDITABLE = ["price_thousand_krw", "loss_coefficient_per_hour"]
+COMPUTED = ("loss_krw_per_hour", "fuel_krw_per_hour", "operator_krw_per_hour", "hourly_krw")
+#: 조종원 식 글자 — B09 식 문자열(좌→우 차례)을 사람이 읽게만 바꿈
+_WAGE_TEXT = OPERATOR_WAGE_FORMULA.replace("*", " × ").replace("/", " ÷ ")
+_FORMULA = {
+    "loss_krw_per_hour": "손료 = 취득가(천원) × 1,000 × 시간당 손료계수 (건설품셈 8-1-6)",
+    "fuel_krw_per_hour": "연료 = 주연료(ℓ/hr) × (1 + 잡재료%) × 전국평균 유가 (건설품셈 8-4 · 8-1-7)",
+    "operator_krw_per_hour": (
+        f"조종원 = 일 노임 × 인원 × {_WAGE_TEXT}"
+        " · 원 미만 절사 (건설품셈 8-1-2 · 건협 임금적용요령 4-나)"
+    ),
+    "hourly_krw": "시간당 사용료 = 손료 + 연료 + 조종원",
+}
+_LOCKED_COMPUTED = "계산값 — 밑값(취득가·손료계수는 이 표 · 노임·유가는 그 표)을 고치면 따라 바뀜"
+_LOCKED_OPERATING = "건설품셈 8-4 운전경비표(로직)에서 옴 — 기초단가가 아님"
+_LOCKED = {
+    "machine_code": "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음",
+    "machine_name": "공표 원문 칸 — 원본 갱신으로만 바뀜",
+    "specification": "공표 원문 칸 — 원본 갱신으로만 바뀜",
+    "economic_life_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜",
+    "annual_standard_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜",
+    "fuel_kind": _LOCKED_OPERATING,
+    "fuel_liters_per_hour": _LOCKED_OPERATING,
+    "misc_material_percent": _LOCKED_OPERATING,
+    "operator_occupation_code": _LOCKED_OPERATING,
+    "operator_person_days": _LOCKED_OPERATING,
+    "fuel_price_krw_per_l": "유가 표에서 고침 — 여기는 덮개 얹은 전국평균을 끌어 씀",
+    "operator_daily_wage_krw": "노임 표에서 고침 — 여기는 덮개 얹은 건설 노임을 끌어 씀",
+    "hourly_note": "계산이 왜 안 섰는지 — 밑값이 비면 섬",
+    "effective_date": "원본 판 기준일 — 원본 갱신으로만 바뀜",
+    "operating_effective_date": "운전경비 원단위가 나온 건설품셈 판 기준일",
+    "fuel_price_date": "끌어 쓴 유가 판 기준일 — 유가 표 판",
+    "operator_wage_date": "끌어 쓴 노임 판 기준일 — 노임 표 판",
+    **dict.fromkeys(COMPUTED, _LOCKED_COMPUTED),
+}
+
+
+def _num(value: Decimal | int | None) -> int | float | None:
+    if value is None:
+        return None
+    value = Decimal(value)
+    return int(value) if value == value.to_integral_value() else float(value)
+
+
+def _fmt(value: Decimal | int | float) -> str:
+    text = f"{Decimal(str(value)):,.4f}".rstrip("0").rstrip(".")
+    return text
+
+
+@dataclass(frozen=True)
+class Inputs:
+    """기계 셈이 다른 표에서 끌어 오는 값 — 덮개를 얹은 뒤 · 판 기준일과 함께."""
+
+    wages: dict[str, Decimal | None]
+    wage_date: str | None
+    oil: dict[str, Decimal]
+    oil_dates: dict[str, str | None]
+    operating_date: str | None
+
+
+@lru_cache(maxsize=2)
+def _cached(_mtimes: tuple[int, ...]):
+    return load_machine_catalog(), {r.machine_code: r for r in load_operating_records().records}
+
+
+def _inputs():
+    folder = RESOURCES / "data_cost_input_value"
+    return _cached(tuple((folder / name).stat().st_mtime_ns for name in _INPUT_FILES))
+
+
+def base_rows(source_name: str, effective_date: str | None) -> list[dict[str, Any]]:
+    """기종 한 줄 — 원본 칸(카탈로그가 되살린 이름·규격 · 손료계수 포함)과 운전경비 원단위."""
+    catalog, records = _inputs()
+    rows = []
+    for code, m in catalog.machines.items():
+        record = records.get(code)
+        rows.append(
+            {
+                "@id": code,
+                "@source": source_name,
+                "machine_code": code,
+                "effective_date": effective_date,
+                "machine_name": m.name,
+                "specification": m.specification,
+                "price_thousand_krw": _num(m.price_thousand_krw),
+                "loss_coefficient_per_hour": _num(m.loss_coefficient_per_hour),
+                "economic_life_hours": m.economic_life_hours,
+                "annual_standard_hours": m.annual_standard_hours,
+                "fuel_kind": record.fuel_kind if record else None,
+                "fuel_liters_per_hour": _num(record.fuel_liters_per_hour) if record else None,
+                "misc_material_percent": _num(record.misc_material_percent) if record else None,
+                "operator_occupation_code": record.operator_occupation_code if record else None,
+                "operator_person_days": _num(record.operator_person_days) if record else None,
+            }
+        )
+    return rows
+
+
+def compute(row: dict[str, Any], inputs: Inputs) -> dict[str, Any]:
+    """덮개 얹은 줄 → 시간당 단가 칸을 붙인 새 줄. B09 `hourly_cost_of` 와 같은 길."""
+    catalog, records = _inputs()
+    code = row["@id"]
+    coef = row["loss_coefficient_per_hour"]
+    record = records.get(code)
+    out = {
+        **row,
+        "operating_effective_date": inputs.operating_date if record is not None else None,
+        "fuel_price_krw_per_l": None,
+        "fuel_price_date": None,
+        "operator_daily_wage_krw": None,
+        "operator_wage_date": None,
+        **dict.fromkeys(COMPUTED),
+    }
+    if coef is None:
+        out["hourly_note"] = "손료계수 없음 — 시간당 단가 안 섬"
+        return out
+    spec = replace(
+        catalog.machines[code],
+        price_thousand_krw=Decimal(str(row["price_thousand_krw"])),
+        loss_coefficient_per_hour=Decimal(str(coef)),
+    )
+    liters = fuel_price = wage = None
+    if record is not None:
+        liters = record.fuel_liters_per_hour
+        if liters is not None:
+            variable = FUEL_VARIABLES.get(record.fuel_kind, "")
+            fuel_price = inputs.oil.get(variable)
+            out["fuel_price_date"] = inputs.oil_dates.get(variable)
+            if record.misc_material_percent is not None:
+                liters = liters * (Decimal(1) + record.misc_material_percent / Decimal(100))
+        wage = inputs.wages.get(record.operator_occupation_code)
+        out["operator_daily_wage_krw"] = _num(wage)
+        out["operator_wage_date"] = inputs.wage_date
+        if wage is not None and record.operator_person_days is not None:
+            wage = wage * record.operator_person_days  # 인원 칸이 비면 B09 처럼 일 노임 그대로
+    cost = hourly_machine_cost(
+        spec,
+        fuel_liters_per_hour=liters if fuel_price is not None else None,
+        fuel_price_per_liter=fuel_price,
+        operator_daily_wage=wage,
+    )
+    money = cost.money
+    out["fuel_price_krw_per_l"] = _num(fuel_price)
+    out["loss_krw_per_hour"] = _num(money.expense)
+    formula = {
+        "loss_krw_per_hour": (
+            f"{_fmt(money.expense)} = {_fmt(spec.price_thousand_krw)}천원 × 1,000 × {spec.loss_coefficient_per_hour}"
+        )
+    }
+    if fuel_price is not None:
+        out["fuel_krw_per_hour"] = _num(money.material)
+        misc = record.misc_material_percent if record else None
+        formula["fuel_krw_per_hour"] = (
+            f"{_fmt(money.material)} = {_fmt(record.fuel_liters_per_hour)}ℓ"
+            + (f" × (1 + {_fmt(misc)}%)" if misc is not None else "")
+            + f" × {_fmt(fuel_price)}원/ℓ"
+        )
+    if wage is not None:
+        out["operator_krw_per_hour"] = _num(money.labor)
+        persons = record.operator_person_days
+        formula["operator_krw_per_hour"] = (
+            f"{_fmt(money.labor)} = {_fmt(out['operator_daily_wage_krw'])}원"
+            + (f" × {_fmt(persons)}인" if persons is not None else "")
+            + f" × {_WAGE_TEXT} (원 미만 절사)"
+        )
+    if cost.gaps:
+        out["hourly_note"] = " · ".join(cost.gaps)
+    else:
+        out["hourly_krw"] = _num(money.total)
+        out["hourly_note"] = None
+        formula["hourly_krw"] = (
+            f"{_fmt(money.total)} = {_fmt(money.expense)} + {_fmt(money.material)} + {_fmt(money.labor)}"
+        )
+    out["@formula"] = formula
+    return out
+
+
+def spec(columns: list[str]) -> dict[str, Any]:
+    return {
+        "editable": list(EDITABLE),
+        "locked": {
+            c: _LOCKED.get(c, "공표 원문 칸 — 원본 갱신으로만 바뀜")
+            for c in columns
+            if c not in EDITABLE
+        },
+        "formula": dict(_FORMULA),
+    }

Z01_MasterData/Z01_MasterData_Overrides.py

@@ -0,0 +1,166 @@
+"""Z01 기초단가 덮개 — 사람이 고친 값만 담는 층(2026-09-15 브레인 틀 · 사용자 지시 「기초단가를 고칠 수 있게」).
+
+  원본  resources/data_*/…                              읽기만 — 주기 갱신은 이 파일을 통째로 갈아끼움
+  덮개  resources/data_master_override/{kind}.json      고친 것만 — 주기가 달라 kind 마다 한 파일
+  읽기  원본 줄 위에 덮개를 얹음 · 줄 수준 `@overrides[열] = {original, value, state, current}`
+
+state 셋(판정은 **여기 한 곳** — 화면이 따로 재지 않음):
+  edited          고쳐짐 — 원본 그대로(덮개를 쓸 때 본 원본 = 지금 원본)
+  source_changed  고쳐짐 — 원본이 바뀜 · ⚠ 덮개 값을 그대로 씀(자동으로 원본을 쓰면 금액이 말없이 바뀜 — 사람이 정함)
+  orphan          주인 없음 — 갱신으로 그 줄·칸이 사라짐 · 버리지 않고 모아 보기에 남김
+되돌리기(「새 원본 값으로 받기」 도 같음) = 값 null → 그 칸을 덮개에서 뺌.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import threading
+from datetime import datetime
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+from common_util.common_util_json import atomic_write_json
+
+OVERRIDE_DIR = Path(__file__).resolve().parent.parent / "resources" / "data_master_override"
+EDITED, SOURCE_CHANGED, ORPHAN = "edited", "source_changed", "orphan"
+STATE_ORDER = (SOURCE_CHANGED, ORPHAN, EDITED)  # 모아 보기 차례 — 확인할 것이 맨 위
+_LOCK = threading.Lock()  # ponytail: 한 프로세스 잠금 — 서버를 여럿 띄우면 파일 잠금이 필요
+
+
+def _path(kind: str) -> Path:
+    return OVERRIDE_DIR / f"{kind}.json"
+
+
+def load(kind: str) -> dict[str, dict[str, dict[str, Any]]]:
+    """`@id` → 열 → {value, original, source, at, by}. 파일이 없으면 빈 덮개."""
+    path = _path(kind)
+    if not path.is_file():
+        return {}
+    return json.loads(path.read_text(encoding="utf-8")).get("overrides") or {}
+
+
+def same(a: Any, b: Any) -> bool:
+    """값 비교 — 215907 과 215907.0 은 같음 · None 은 None 과만 같음."""
+    if a is None or b is None:
+        return a is None and b is None
+    if isinstance(a, (int, float)) and isinstance(b, (int, float)):
+        return Decimal(str(a)) == Decimal(str(b))
+    return a == b
+
+
+def check_value(value: Any) -> str | None:
+    """넣을 수 있는 값인가 — 막는 까닭(없으면 None). null 은 되돌리기라 통과."""
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return "숫자만 넣을 수 있음"
+    if not math.isfinite(value) or value < 0:
+        return "0 이상 유한한 수만 넣을 수 있음"
+    return None
+
+
+def mark(original: Any, value: Any, current: Any) -> dict[str, Any]:
+    state = EDITED if same(original, current) else SOURCE_CHANGED
+    return {"original": original, "value": value, "state": state, "current": current}
+
+
+def overlay(kind: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """원본 줄(바꾸지 않음) 위에 덮개를 얹은 **새 줄들**."""
+    store = load(kind)
+    out = []
+    for row in rows:
+        entries = store.get(row["@id"])
+        if not entries:
+            out.append(row)
+            continue
+        row = dict(row)
+        marks = {}
+        for column, entry in entries.items():
+            if column not in row:
+                continue  # 주인 없는 칸 — 모아 보기에만
+            marks[column] = mark(entry.get("original"), entry.get("value"), row[column])
+            row[column] = entry.get("value")
+        if marks:
+            row["@overrides"] = marks
+        out.append(row)
+    return out
+
+
+def write(kind: str, base_row: dict[str, Any], values: dict[str, Any], by: Any) -> None:
+    """고친 값을 덮개에 씀 — null 이거나 지금 원본과 같으면 그 칸을 뺌. 원본은 안 건드림."""
+    row_id = base_row["@id"]
+    with _LOCK:
+        store = load(kind)
+        entries = dict(store.get(row_id) or {})
+        for column, value in values.items():
+            current = base_row.get(column)
+            if value is None or same(value, current):
+                entries.pop(column, None)
+                continue
+            entries[column] = {
+                "value": value,
+                "original": current,  # 이 값을 보고 고침 — 갱신 뒤 원본과 달라지면 source_changed
+                "source": base_row.get("@source", ""),
+                "at": datetime.now().astimezone().isoformat(timespec="seconds"),
+                "by": by,
+            }
+        if entries:
+            store[row_id] = entries
+        else:
+            store.pop(row_id, None)
+        _save(kind, store)
+
+
+def _save(kind: str, store: dict[str, dict[str, dict[str, Any]]]) -> None:
+    atomic_write_json(
+        _path(kind),
+        {
+            "schema_version": "1.0",
+            "kind": kind,
+            "note": "사람이 고친 기초단가만 — 원본 파일은 그대로 · Z01 마스터 데이터 화면이 씀",
+            "overrides": store,
+        },
+    )
+
+
+def clear(kind: str, state: str, base_rows: list[dict[str, Any]]) -> int:
+    """한 kind 의 그 state 덮개를 한꺼번에 뺌 — 품목 퇴출로 여럿이 주인을 잃는 갱신 뒤(브레인 ②). 뺀 수."""
+    with _LOCK:
+        doomed = {(i["row_id"], i["column"]) for i in items(kind, base_rows) if i["state"] == state}
+        if not doomed:
+            return 0
+        store = load(kind)
+        for row_id, column in doomed:
+            store[row_id].pop(column, None)
+            if not store[row_id]:
+                del store[row_id]
+        _save(kind, store)
+        return len(doomed)
+
+
+def items(kind: str, base_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """모아 보기 한 kind 몫 — 주인 없는 것까지 · 차례는 부르는 쪽이 STATE_ORDER 로."""
+    by_id = {r["@id"]: r for r in base_rows}
+    out = []
+    for row_id, entries in load(kind).items():
+        row = by_id.get(row_id)
+        for column, entry in entries.items():
+            if row is None or column not in row:
+                current, state = None, ORPHAN
+            else:
+                current = row[column]
+                state = mark(entry.get("original"), entry.get("value"), current)["state"]
+            out.append(
+                {
+                    "kind": kind,
+                    "row_id": row_id,
+                    "column": column,
+                    "original": entry.get("original"),
+                    "current": current,
+                    "value": entry.get("value"),
+                    "state": state,
+                }
+            )
+    return out

Z01_MasterData/Z01_MasterData_Router.py

@@ -1,19 +1,32 @@
-"""Z01 마스터 데이터 라우터 — 읽기 전용(고치기는 다음 차례 · 2026-09-15 브레인).
+"""Z01 마스터 데이터 라우터 — 읽기 + 기초단가 고치기(2026-09-15 브레인).
 
   GET /api/master-data/tree                                  갈래 → 파일 → 표
   GET /api/master-data/rows?file=&table=&page=&size=&q=      표 줄(쪽 나누기 · 검색)
+  GET /api/master-data/base-prices/{kind}?page=&size=&q=     기초단가 한 표(labor|machine|material|oil|rate)
+  PUT /api/master-data/base-prices/{kind}/{row_id}           {values:{열:값}} → 덮개에만 씀 · null = 되돌리기
+  GET /api/master-data/overrides                             고친 것·원본 바뀐 것·주인 없는 것(서버 정렬)
 ⚠ 권한은 등록하는 쪽(`main.py` · 랩탑 서브)이 `dependencies=[verify_session, require_system_admin]` 로 붙임.
 """
 
 from __future__ import annotations
 
-from fastapi import APIRouter, HTTPException
+from typing import Any, Literal
 
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from common_util.common_util_auth import verify_session
+from Z01_MasterData import Z01_MasterData_BasePrices as base_prices
 from Z01_MasterData import Z01_MasterData_Tables as tables
 
 router = APIRouter(prefix="/api/master-data", tags=["Z01 MasterData"])
 
 
+class BasePriceEdit(BaseModel):
+    values: dict[str, Any]
+
+
 @router.get("/tree")
 def get_tree() -> dict:
     return tables.tree()
@@ -27,3 +40,44 @@ def get_rows(
     if result is None:
         raise HTTPException(status_code=404, detail="없는 파일이나 표입니다.")
     return result
+
+
+def _kind(kind: str) -> str:
+    if kind not in base_prices.KINDS:
+        raise HTTPException(status_code=404, detail=f"없는 기초단가: {kind}")
+    return kind
+
+
+@router.get("/base-prices/{kind}")
+def get_base_prices(
+    kind: str, page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE, q: str = ""
+) -> dict:
+    return base_prices.table(_kind(kind), page=page, size=size, q=q)
+
+
+@router.put("/base-prices/{kind}/{row_id:path}")
+def put_base_price(
+    kind: str, row_id: str, body: BasePriceEdit, session: dict = Depends(verify_session)
+) -> JSONResponse:
+    status, payload = base_prices.edit(_kind(kind), row_id, body.values, session.get("user_id"))
+    if status != 200:
+        raise HTTPException(status_code=status, detail=payload["message"])
+    return JSONResponse(content=payload)
+
+
+@router.get("/overrides")
+def get_overrides(
+    kind: str = "", state: str = "", page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE
+) -> dict:
+    return base_prices.override_items(kind and _kind(kind), state, page=page, size=size)
+
+
+class OverrideClear(BaseModel):
+    kind: str
+    state: Literal["edited", "source_changed", "orphan"]
+
+
+@router.post("/overrides/clear")
+def clear_overrides(body: OverrideClear, _session: dict = Depends(verify_session)) -> dict:
+    """한 kind 의 그 state 덮개를 한꺼번에 뺌 — kind·state 둘 다 있어야 함(통째로 비우기 막음)."""
+    return {"removed": base_prices.clear_overrides(_kind(body.kind), body.state)}