feat(z01): 기초단가 열 이름을 이름표에서만 읽음 — 합친 표 열 key 를 이름표 merged_tables 에 맞춤(노임 source · 유가 source·region_code·region_name·price_krw_per_l · 기계 fuel_type·fuel_rate_l_per_hour·operator_person_per_day·hourly_loss_krw·hourly_fuel_krw·hourly_operator_krw·hourly_total_krw) · 표 수준 알림을 서버가 한 벌로(새로 만드는 프로젝트부터 쓰임 · 요율은 법정값 경고까지) · 출처 받는 자리 빈 곳은 「없음」(브레인 ①②③)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn
This commit is contained in:
2026-09-16 08:50:45 +09:00
co-authored by Claude Opus 5
parent 909b055aa9
commit 05bab3b61f
3 changed files with 100 additions and 62 deletions
+36 -21
View File
@@ -3,7 +3,7 @@
표 수준(한 번): columns · total · editable · locked{열: 왜 못 고치나} · formula{열: 식} · notice(요율만)
줄 수준(해당할 때만): `@id`(자료 제 열쇠) · `@overrides`(고친 줄) · `@formula`(계산값 줄)
`@id` — 줄 차례가 아니라 자료 열쇠(브레인 승인): 노임 `{dataset_id}/{occupation_code}` · 기계 machine_code ·
자재 item_code · 유가 `{scope}/{변수}[/{sido_code}]` · 요율 `{변수}[/{목록}/{구간칸=값;…}]`(구간 칸 이름 차례 · 값·식 칸 뺌)
자재 item_code · 유가 `{scope}/{변수}[/{sido_code}]`(원본 값 그대로 · 열 이름은 이름표 것) · 요율 `{변수}[/{목록}/{구간칸=값;…}]`(구간 칸 이름 차례 · 값·식 칸 뺌)
원본 파일 자리는 이름표 파일이 가리키는 곳(`Z01_MasterData_Tables`) — 원본은 읽기만, 고친 값은 덮개(`Z01_MasterData_Overrides`).
"""
@@ -35,6 +35,10 @@ _RATE_EDITABLE = (
"manager_thresholds.default_estimated_amount_krw",
"manager_thresholds.civil_main_work_estimated_amount_krw",
)
_SCOPE_NOTICE = (
"여기서 고친 값은 **새로 만드는 프로젝트부터** 쓰임 — 이미 만든 프로젝트 금액은 안 바뀜"
"(프로젝트는 만들 때 뜬 사본을 봄)"
)
_RATE_NOTICE = "법이 정한 값 — 고치면 원본 갱신 때 구간 이름이 바뀌어 고친 값이 주인을 잃기 쉬움(주인 없음으로 남음)"
@@ -98,7 +102,7 @@ def sources(kind: str) -> list[dict[str, Any]]:
"source_id": entry.get("source_id"),
"name": entry.get("name_ko") or entry.get("source_id"),
"publisher": entry.get("publisher") or "",
"where": entry.get("where_to_get") or "",
"where": entry.get("where_to_get") or "없음", # 안 지어냄(브레인)
"cycle": entry.get("cycle") or "",
"latest_published": latest.isoformat() if latest else None,
"our_edition": ours.isoformat() if ours else None,
@@ -125,7 +129,7 @@ def _labor() -> list[dict[str, Any]]:
row = {
"@id": f"{dataset}/{r['occupation_code']}",
"@source": name,
"dataset_id": dataset,
"source": dataset,
"effective_date": edition_of(doc),
**r,
}
@@ -148,9 +152,9 @@ def _oil() -> list[dict[str, Any]]:
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")}
head = {k: x for k, x in v.items() if k not in ("records", "value", "scope")}
common = {
"scope": v.get("scope"),
"source": v.get("scope"),
"fuel": variable,
"effective_date": edition_of(doc), # 전국 · 지역 판 기준일이 다름
**head,
@@ -161,7 +165,7 @@ def _oil() -> list[dict[str, Any]]:
"@id": f"{v.get('scope')}/{variable}",
"@source": name,
**common,
"value": v.get("value"),
"price_krw_per_l": v.get("value"),
}
)
for r in v.get("records") or []:
@@ -170,7 +174,9 @@ def _oil() -> list[dict[str, Any]]:
"@id": f"{v.get('scope')}/{variable}/{r['sido_code']}",
"@source": name,
**common,
**r,
"region_code": r["sido_code"],
"region_name": r["sido_name"],
"price_krw_per_l": r["value"],
}
)
return rows
@@ -238,7 +244,7 @@ def base_rows(kind: str) -> list[dict[str, Any]]:
def _machine_inputs() -> machine.Inputs:
"""기계가 끌어 쓰는 밑값 — **덮개를 얹은** 노임(건설)·전국평균 유가 + 그 판들의 기준일."""
labor = [r for r in overrides.overlay("labor", _labor()) if r["dataset_id"] == "labor_const"]
labor = [r for r in overrides.overlay("labor", _labor()) if r["source"] == "labor_const"]
oil = [
r for r in overrides.overlay("oil", _oil()) if r["@id"] == f"national_average/{r['fuel']}"
]
@@ -250,7 +256,7 @@ def _machine_inputs() -> machine.Inputs:
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={r["fuel"]: Decimal(str(r["price_krw_per_l"])) for r in oil},
oil_dates={r["fuel"]: r.get("date") or r["effective_date"] for r in oil},
operating_date=edition_of(_doc("pum_const")[0]),
)
@@ -272,29 +278,35 @@ def _columns(kind: str, all_rows: list[dict[str, Any]]) -> list[str]:
return list(keys)
def notice(kind: str) -> list[str]:
"""표 수준 알림 — 화면이 따로 가지면 문구가 두 벌이 됨(브레인 ③)."""
return [_SCOPE_NOTICE] + ([_RATE_NOTICE] if kind == "rate" else [])
def spec(kind: str, columns: list[str]) -> dict[str, Any]:
"""표 수준 — editable · locked · formula (· notice)."""
"""표 수준 — editable · locked · formula."""
if kind == "machine":
return machine.spec(columns)
editable = {
"labor": ["daily_wage_krw"],
"material": ["price_krw"],
"oil": ["value"],
"oil": ["price_krw_per_l"],
"rate": [c for c in _RATE_EDITABLE if c in columns],
}[kind]
keys = {"labor": {"dataset_id", "occupation_code"}, "material": {"item_code"}}.get(kind, set())
keys = {"labor": {"source", "occupation_code"}, "material": {"item_code"}}.get(kind, set())
if kind == "oil":
keys = {"scope", "fuel", "sido_code"}
keys = {"source", "fuel", "region_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
return {
"editable": editable,
"locked": locked,
"formula": {},
}
def row_label(kind: str, row: dict[str, Any] | None) -> str:
@@ -307,7 +319,7 @@ def row_label(kind: str, row: dict[str, Any] | None) -> str:
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 '전국평균'}"
return f"{row.get('source_product_name') or row.get('fuel')} {row.get('region_name') or '전국평균'}"
named = (tables.load_labels().get("value_overrides") or {}).get(
f"rates::variables/{row['variable']}"
) or {}
@@ -316,10 +328,12 @@ def row_label(kind: str, row: dict[str, Any] | None) -> str:
def column_meta(kind: str, key: str) -> dict[str, Any]:
"""열 이름 — 이름표 `synthetic_tables['@base_prices/{kind}']` → `columns[열key]` → 영문 key."""
"""열 이름 — 이름표 `merged_tables[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)
merged = next(
(t for t in labels.get("merged_tables") or [] if t.get("key") in (kind, f"{kind}s")), {}
)
named = next((c for c in merged.get("columns") or [] if c.get("key") == key), None)
named = named or labels["columns"].get(key) or {}
return {
"key": key,
@@ -356,6 +370,7 @@ def table(
"rows": [public(r) for r in hits[start : start + size]],
"total": len(hits),
**spec(kind, columns),
"notice": notice(kind),
"source": sources(kind),
}
@@ -25,17 +25,17 @@ 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")
COMPUTED = ("hourly_loss_krw", "hourly_fuel_krw", "hourly_operator_krw", "hourly_total_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": (
"hourly_loss_krw": "손료 = 취득가(천원) × 1,000 × 시간당 손료계수 (건설품셈 8-1-6)",
"hourly_fuel_krw": "연료 = 주연료(ℓ/hr) × (1 + 잡재료%) × 전국평균 유가 (건설품셈 8-4 · 8-1-7)",
"hourly_operator_krw": (
f"조종원 = 일 노임 × 인원 × {_WAGE_TEXT}"
" · 원 미만 절사 (건설품셈 8-1-2 · 건협 임금적용요령 4-나)"
),
"hourly_krw": "시간당 사용료 = 손료 + 연료 + 조종원",
"hourly_total_krw": "시간당 사용료 = 손료 + 연료 + 조종원",
}
_LOCKED_COMPUTED = "계산값 — 밑값(취득가·손료계수는 이 표 · 노임·유가는 그 표)을 고치면 따라 바뀜"
_LOCKED_OPERATING = "건설품셈 8-4 운전경비표(로직)에서 옴 — 기초단가가 아님"
@@ -45,11 +45,11 @@ _LOCKED = {
"specification": "공표 원문 칸 — 원본 갱신으로만 바뀜",
"economic_life_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜",
"annual_standard_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜",
"fuel_kind": _LOCKED_OPERATING,
"fuel_liters_per_hour": _LOCKED_OPERATING,
"fuel_type": _LOCKED_OPERATING,
"fuel_rate_l_per_hour": _LOCKED_OPERATING,
"misc_material_percent": _LOCKED_OPERATING,
"operator_occupation_code": _LOCKED_OPERATING,
"operator_person_days": _LOCKED_OPERATING,
"operator_person_per_day": _LOCKED_OPERATING,
"fuel_price_krw_per_l": "유가 표에서 고침 — 여기는 덮개 얹은 전국평균을 끌어 씀",
"operator_daily_wage_krw": "노임 표에서 고침 — 여기는 덮개 얹은 건설 노임을 끌어 씀",
"hourly_note": "계산이 왜 안 섰는지 — 밑값이 비면 섬",
@@ -112,11 +112,11 @@ def base_rows(source_name: str, effective_date: str | None) -> list[dict[str, An
"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,
"fuel_type": record.fuel_kind if record else None,
"fuel_rate_l_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,
"operator_person_per_day": _num(record.operator_person_days) if record else None,
}
)
return rows
@@ -167,24 +167,24 @@ def compute(row: dict[str, Any], inputs: Inputs) -> dict[str, Any]:
)
money = cost.money
out["fuel_price_krw_per_l"] = _num(fuel_price)
out["loss_krw_per_hour"] = _num(money.expense)
out["hourly_loss_krw"] = _num(money.expense)
formula = {
"loss_krw_per_hour": (
"hourly_loss_krw": (
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)
out["hourly_fuel_krw"] = _num(money.material)
misc = record.misc_material_percent if record else None
formula["fuel_krw_per_hour"] = (
formula["hourly_fuel_krw"] = (
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)
out["hourly_operator_krw"] = _num(money.labor)
persons = record.operator_person_days
formula["operator_krw_per_hour"] = (
formula["hourly_operator_krw"] = (
f"{_fmt(money.labor)} = {_fmt(out['operator_daily_wage_krw'])}"
+ (f" × {_fmt(persons)}" if persons is not None else "")
+ f" × {_WAGE_TEXT} (원 미만 절사)"
@@ -192,9 +192,9 @@ def compute(row: dict[str, Any], inputs: Inputs) -> dict[str, Any]:
if cost.gaps:
out["hourly_note"] = " · ".join(cost.gaps)
else:
out["hourly_krw"] = _num(money.total)
out["hourly_total_krw"] = _num(money.total)
out["hourly_note"] = None
formula["hourly_krw"] = (
formula["hourly_total_krw"] = (
f"{_fmt(money.total)} = {_fmt(money.expense)} + {_fmt(money.material)} + {_fmt(money.labor)}"
)
out["@formula"] = formula
+45 -22
View File
@@ -75,7 +75,7 @@ def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> No
expected = {
"labor": (261, ["daily_wage_krw"]),
"material": (6999, ["price_krw"]),
"oil": (36, ["value"]),
"oil": (36, ["price_krw_per_l"]),
"machine": (613, ["price_thousand_krw", "loss_coefficient_per_hour"]),
}
for kind, (total, editable) in expected.items():
@@ -91,8 +91,8 @@ def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> No
assert set(table["locked"]) <= columns and not set(table["locked"]) & set(editable), kind
assert _row(client, "labor", "labor_const/1001")["daily_wage_krw"] == 215907
assert _row(client, "labor", "labor_mfg/1")["occupation_name"] == "CAD설계사(기계)"
assert _row(client, "oil", "national_average/oil_diesel")["value"] == 1846.39
assert _row(client, "oil", "regional/oil_diesel/01")["sido_name"] == "서울"
assert _row(client, "oil", "national_average/oil_diesel")["price_krw_per_l"] == 1846.39
assert _row(client, "oil", "regional/oil_diesel/01")["region_name"] == "서울"
rate = _get(client, "rate", size=500)
assert rate["total"] == 218 and "rate_percent" in rate["editable"]
by_id = {r["@id"]: r for r in rate["rows"]}
@@ -107,16 +107,16 @@ def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> No
def test_기계는_시간당_단가를_계산값으로_잠그고_산출근거를_실음(client: TestClient) -> None:
table = _get(client, "machine", q="0101-0007", size=5)
assert "hourly_krw" in table["locked"] and "hourly_krw" in table["formula"]
assert "hourly_total_krw" in table["locked"] and "hourly_total_krw" in table["formula"]
row = next(r for r in table["rows"] if r["@id"] == "0101-0007")
parts = row["loss_krw_per_hour"] + row["fuel_krw_per_hour"] + row["operator_krw_per_hour"]
assert row["hourly_krw"] == pytest.approx(parts)
parts = row["hourly_loss_krw"] + row["hourly_fuel_krw"] + row["hourly_operator_krw"]
assert row["hourly_total_krw"] == pytest.approx(parts)
assert (
"hourly_krw" in row["@formula"]
and f"{int(row['hourly_krw']):,}" in row["@formula"]["hourly_krw"]
"hourly_total_krw" in row["@formula"]
and f"{int(row['hourly_total_krw']):,}" in row["@formula"]["hourly_total_krw"]
)
res = _put(client, "machine", "0101-0007", {"hourly_krw": 1})
assert res.status_code == 400 and table["locked"]["hourly_krw"] in res.text
res = _put(client, "machine", "0101-0007", {"hourly_total_krw": 1})
assert res.status_code == 400 and table["locked"]["hourly_total_krw"] in res.text
def test_기계_계산값은_B09_시간당_사용료와_같음(client: TestClient) -> None:
@@ -135,12 +135,12 @@ def test_기계_계산값은_B09_시간당_사용료와_같음(client: TestClien
try:
cost = hourly_cost_of(code)
except MachineCostError: # 손료계수 없는 기종 — B09 도 못 셈
assert rows[code]["hourly_krw"] is None and rows[code]["hourly_note"], code
assert rows[code]["hourly_total_krw"] is None and rows[code]["hourly_note"], code
continue
if cost.gaps:
assert rows[code]["hourly_krw"] is None, code # 반만 선 값은 안 냄
assert rows[code]["hourly_total_krw"] is None, code # 반만 선 값은 안 냄
continue
assert rows[code]["hourly_krw"] == pytest.approx(float(cost.money.total)), code
assert rows[code]["hourly_total_krw"] == pytest.approx(float(cost.money.total)), code
checked += 1
assert checked >= 5
@@ -162,7 +162,10 @@ def test_고친_값은_덮개에만_쓰고_원본은_안_건드림(client: TestC
assert (
_row(client, "labor", "labor_const/1001")["@overrides"]["daily_wage_krw"]["value"] == 230000
)
assert _put(client, "oil", "national_average/oil_diesel", {"value": 1900}).status_code == 200
assert (
_put(client, "oil", "national_average/oil_diesel", {"price_krw_per_l": 1900}).status_code
== 200
)
assert sorted(p.name for p in store.iterdir()) == [
"labor.json",
"oil.json",
@@ -182,12 +185,12 @@ def test_밑값을_고치면_기계_계산값이_따라_바뀜(client: TestClien
operator = base["operator_occupation_code"]
wage = _row(client, "labor", f"labor_const/{operator}")["daily_wage_krw"]
_put(client, "labor", f"labor_const/{operator}", {"daily_wage_krw": wage + 80000})
_put(client, "oil", "national_average/oil_diesel", {"value": 2000})
_put(client, "oil", "national_average/oil_diesel", {"price_krw_per_l": 2000})
_put(client, "machine", "0101-0007", {"price_thousand_krw": base["price_thousand_krw"] * 2})
after = _row(client, "machine", "0101-0007")
assert after["operator_krw_per_hour"] > base["operator_krw_per_hour"]
assert after["fuel_krw_per_hour"] > base["fuel_krw_per_hour"]
assert after["loss_krw_per_hour"] == pytest.approx(base["loss_krw_per_hour"] * 2)
assert after["hourly_operator_krw"] > base["hourly_operator_krw"]
assert after["hourly_fuel_krw"] > base["hourly_fuel_krw"]
assert after["hourly_loss_krw"] == pytest.approx(base["hourly_loss_krw"] * 2)
price = base["price_thousand_krw"]
assert after["@overrides"] == {
"price_thousand_krw": {
@@ -313,8 +316,11 @@ def test_빈_칸에_새로_넣은_값도_덮개_원본에_값이_생기면_원
def test_요율_표는_법정값_경고를_실음(client: TestClient) -> None:
rate = _get(client, "rate", size=1)
assert "갱신" in rate["notice"] and "주인" in rate["notice"]
assert "notice" not in _get(client, "material", size=1)
assert any("갱신" in line and "주인" in line for line in rate["notice"])
for kind in ("labor", "machine", "material", "oil", "rate"): # 브레인 ③ — 문구는 서버가 한 벌로
notice = _get(client, kind, size=1)["notice"]
assert any("새로 만드는 프로젝트" in line and "이미 만든" in line for line in notice), kind
assert len(rate["notice"]) == 2 and len(_get(client, "material", size=1)["notice"]) == 1
def test_막는_자리(client: TestClient) -> None:
@@ -353,8 +359,8 @@ def test_합친_표는_줄마다_판_기준일을_실음(client: TestClient) ->
m = _row(client, "machine", "0101-0007")
assert (m["effective_date"], m["operating_effective_date"]) == ("2026-01-01", "2026-01-01")
assert (m["fuel_price_date"], m["operator_wage_date"]) == ("2026-08-14", "2026-01-01")
no_record = next(r for r in _get(client, "machine", size=500)["rows"] if r["fuel_kind"] is None)
assert no_record["operating_effective_date"] is None and no_record["hourly_krw"] is None
no_record = next(r for r in _get(client, "machine", size=500)["rows"] if r["fuel_type"] is None)
assert no_record["operating_effective_date"] is None and no_record["hourly_total_krw"] is None
def test_주인_없는_덮개는_모아서_한꺼번에_다룸(client: TestClient, store: Path) -> None:
@@ -473,3 +479,20 @@ def test_모아_보기_항목에_단위(client: TestClient) -> None:
_put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000})
(item,) = client.get("/api/master-data/overrides").json()["items"]
assert item["unit"] == "" and item["column_label"] == "일 노임"
def test_열_이름은_이름표에서만_옴(client: TestClient) -> None:
"""브레인 ② — 이름·판정은 한 곳에만. 합친 표 열 이름은 이름표 `merged_tables` 가 정본."""
labels = json.loads(
(ROOT / "resources/data_master_labels/labels_2026-01-01.json").read_text(encoding="utf-8")
)
merged = {t["key"]: t for t in labels["merged_tables"]}
for kind in ("labor", "material", "oil"):
named = {c["key"]: c for c in merged[kind]["columns"]}
for column in _get(client, kind, size=1)["columns"]:
if column["key"] in named:
assert column["label"] == named[column["key"]]["name_ko"], (kind, column["key"])
assert column["unit"] == named[column["key"]]["unit"]
assert not [c for c in _get(client, kind, size=1)["columns"] if c["label"] == c["key"]], (
kind
)