revert(Z01): 롤백된 마스터 DB·화면 작업이 dev 로 되들어온 것 걷어냄

- 사용자 「완전히 되돌려」 뒤 dev 를 거쳐 다시 들어온 1-2·1-3·1-4 코드 26 경로를 9629430a 모습으로(고친 17 되돌림 · 새로 생긴 9 지움)
- 대상: Z01_MasterData 17 · main.py · ui_template_locale_b3.ts · resources/tester 의 Z01 시험 7
- 품셈 md·도구는 안 건드림 · 이력 재작성 없음(되돌림 커밋)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0145cr1DMC5Ft8fouQSXtp69
This commit is contained in:
2026-09-19 01:36:21 +09:00
co-authored by Claude Opus 5
parent 6b417d3b79
commit f2fe4f4214
26 changed files with 1012 additions and 3598 deletions
+111 -158
View File
@@ -1,20 +1,16 @@
"""Z01 기초단가 다섯 — 노임·기계·자재·유가·요율을 kind 마다 **한 표**로(2026-09-15 사용자 지시 · 브레인 계약).
표 수준(한 번): columns · total · editable · locked{열: 왜 못 고치나} · formula{열: 식} · notice(요율만)
줄 수준: `@id`(자료 제 열쇠) · `changed_columns`(초기값과 다른 칸) · `is_added`(관리자 추가 줄) · `@formula`(계산값 줄)
줄 수준(해당할 때만): `@id`(자료 제 열쇠) · `@overrides`(고친 줄) · `@formula`(계산값 줄)
`@id` — 줄 차례가 아니라 자료 열쇠(브레인 승인): 노임 `{dataset_id}/{occupation_code}` · 기계 machine_code ·
자재 item_code · 유가 `{scope}/{변수}[/{sido_code}]`(원본 값 그대로 · 열 이름은 이름표 것) · 요율 `{변수}[/{목록}/{구간칸=값;…}]`(구간 칸 이름 차례 · 값·식 칸 뺌)
정본 = DB `master_base_price`(2026-09-18 PLAN 1-0 · `Z01_MasterData_Repository_BasePrices`) — 파일 원본(`base_rows`)은
초기값 1회 주입(`Z01_MasterData_Seed`)의 재료 · 공종 축 둘만 아직 파일 읽기.
원본 파일 자리는 이름표 파일이 가리키는 곳(`Z01_MasterData_Tables`) — 원본은 읽기만, 고친 값은 덮개(`Z01_MasterData_Overrides`).
"""
from __future__ import annotations
import asyncio
import json
import math
import re
import uuid
from datetime import date
from decimal import Decimal
from functools import lru_cache
@@ -23,7 +19,7 @@ from typing import Any
from Z01_MasterData import Z01_MasterData_BasePrices_Machine as machine
from Z01_MasterData import Z01_MasterData_BasePrices_Tables as base_tables
from Z01_MasterData import Z01_MasterData_Repository_BasePrices as repo
from Z01_MasterData import Z01_MasterData_Overrides as overrides
from Z01_MasterData import Z01_MasterData_Tables as tables
from Z01_MasterData import Z01_MasterData_WorkItems as work_items
@@ -32,7 +28,7 @@ KINDS = ("labor", "machine", "material", "oil", "rate", *base_tables.SPEC, *work
SOURCES_DIR = (
tables.RESOURCES / "data_master_sources"
) # 출처표(데스크탑 서브) — 어디서 받고 최신이 무엇인지
_KEY = "자료 열쇠 — 바꾸면 다른 표가 이 줄을 못 찾음"
_KEY = "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음"
_SOURCE = "공표 원문 칸 — 원본 갱신으로만 바뀜"
_RATE_EDITABLE = (
"rate_percent",
@@ -43,14 +39,14 @@ _RATE_EDITABLE = (
"manager_thresholds.default_estimated_amount_krw",
"manager_thresholds.civil_main_work_estimated_amount_krw",
)
#: ⚠ **지금 상태를 먼저** — 고친 값을 읽는 곳이 Z01 밖에 0(2026-09-17 서브 훑기 · 브레인 「없으면 없다고」).
#: ⚠ **지금 상태를 먼저** — 덮개를 읽는 곳이 Z01 밖에 0(2026-09-17 서브 훑기 · 브레인 「없으면 없다고」).
#: 뒤 문장은 프로젝트 복사(요구 14)가 붙으면 참이 되는 약속이라 지우지 않고 「그 길이 붙으면」으로 둠.
_SCOPE_NOTICE = (
"⚠ **아직 어디에도 안 쓰임** — 고친 값은 마스터에만 쌓임 · 프로젝트로 옮기는 길(프로젝트 복사)이 아직 없음."
" 그 길이 붙으면: 여기서 고친 값은 **새로 만드는 프로젝트부터** 쓰임 — 이미 만든 프로젝트 금액은 안 바뀜"
"(프로젝트는 만들 때 뜬 사본을 봄)"
)
_RATE_NOTICE = "법이 정한 값 — 고시가 바뀔 때만 고칠 것(구간 칸은 규칙이라 못 고침)"
_RATE_NOTICE = "법이 정한 값 — 고치면 원본 갱신 때 구간 이름이 바뀌어 고친 값이 주인을 잃기 쉬움(주인 없음으로 남음)"
def _source(file_id: str) -> Path:
@@ -281,7 +277,7 @@ def merge_same_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
def base_rows(kind: str) -> list[dict[str, Any]]:
"""파일 원본 줄 — 초기값 주입의 재료 · 기계는 입력 칸까지(계산 칸은 읽을 때 셈)."""
"""덮개 얹기 전 원본 줄 — 기계는 입력 칸까지(계산 칸은 덮개를 얹은 뒤 셈)."""
if kind == "machine":
doc, name = _doc("mach_base")
return machine.base_rows(name, edition_of(doc))
@@ -294,67 +290,33 @@ def base_rows(kind: str) -> list[dict[str, Any]]:
return _BUILDERS[kind]()
def _machine_inputs(labor: list[dict[str, Any]], oil: list[dict[str, Any]]) -> machine.Inputs:
"""기계가 끌어 쓰는 밑값 — DB 의 노임(건설)·전국평균 유가 + 그 판들의 기준일(빈 값 줄은 뺌)."""
labor = [r for r in labor if r.get("source") == "labor_const" and r.get("occupation_code")]
def _machine_inputs() -> machine.Inputs:
"""기계가 끌어 쓰는 밑값 — **덮개를 얹은** 노임(건설)·전국평균 유가 + 그 판들의 기준일."""
labor = [r for r in overrides.overlay("labor", _labor()) if r["source"] == "labor_const"]
oil = [
r
for r in oil
if r["@id"] == f"national_average/{r.get('fuel')}" and r.get("price_krw_per_l") is not None
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.get("daily_wage_krw") is None
if r["daily_wage_krw"] is None
else Decimal(str(r["daily_wage_krw"]))
for r in labor
},
wage_date=labor[0].get("effective_date") if labor else None,
wage_date=labor[0]["effective_date"] if labor else None,
oil={r["fuel"]: Decimal(str(r["price_krw_per_l"])) for r in oil},
oil_dates={r["fuel"]: r.get("date") or r.get("effective_date") 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]),
)
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 _row(record: dict[str, Any]) -> dict[str, Any]:
"""DB 한 줄 → 표 줄. `@data` = 저장된 값 그대로(쓸 때 여기에 고친 칸만 얹음 · 기계 계산 칸이 섞이지 않게)."""
data, seed = record["data"], record["seed"]
changed = []
if seed is not None:
changed = [
c
for c in dict.fromkeys([*seed, *data])
if not c.startswith("@") and not same(data.get(c), seed.get(c))
]
row = {"@id": record["row_key"], **data}
row.update(
{"@id": record["row_key"], "@data": data, "@added": seed is None, "@changed": changed}
)
return row
async def _stored(kind: str) -> list[dict[str, Any]]:
return [_row(r) for r in await repo.load(kind)]
async def rows(kind: str) -> list[dict[str, Any]]:
"""DB 정본 줄(공종 축 둘만 파일) — 기계는 DB 노임·유가로 시간당 단가를 읽을 때마다 셈(식은 코드 · 브레인)."""
if kind in work_items.KINDS:
return work_items.base_rows(kind)
if kind != "machine":
return await _stored(kind)
out, labor, oil = await asyncio.gather(_stored("machine"), _stored("labor"), _stored("oil"))
inputs = _machine_inputs(labor, oil)
return [machine.compute(r, inputs) for r in out]
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]:
@@ -484,26 +446,45 @@ def group_label(group: str) -> str:
return named.get("name_ko") or group
async def kinds() -> list[dict[str, Any]]:
def kinds() -> list[dict[str, Any]]:
"""화면이 상자를 세울 목록 — **서버가 냄**(화면이 제 코드에 들면 새 kind 가 조용히 안 뜸).
갈래 · 갈래 한글 이름 · kind · 한글 이름(이름표) · 줄 수(DB · 공종 축은 파일).
갈래 · 갈래 한글 이름 · kind · 한글 이름(이름표) · 줄 수.
"""
counted = await repo.counts()
return [
{
"group": kind_group(kind),
"group_label": group_label(kind_group(kind)),
"kind": kind,
"label": kind_label(kind),
"rows": len(work_items.base_rows(kind))
if kind in work_items.KINDS
else counted.get(kind, 0),
"rows": len(rows(kind)),
}
for kind in KINDS
]
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 in work_items.KINDS:
return str(row.get("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('region_name') or '전국평균'}"
if kind in base_tables.SPEC: # 기초데이터 아홉 — 축 칸을 이어 사람이 읽을 한 줄로
return " · ".join(str(row[c]) for c in row.get("@axis", ()) if row.get(c) is not None)
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]:
"""열 이름 — 이름표 `merged_tables[kind]` → `columns[열key]` → 영문 key(이름은 이름표에만 · 브레인 ②)."""
labels = tables.load_labels()
@@ -520,12 +501,27 @@ def column_meta(kind: str, key: str) -> dict[str, Any]:
}
_INNER = ("@source", "@axis", "@data", "@added", "@changed")
#: 뜻 칸에서 뺄 자리 — 판·공표 날짜는 갱신 때마다 바뀌어 「뜻 바뀜」과 구별이 안 됨
_NOT_MEANING = ("effective_date", "date", "notice_datetime")
def meaning_of(kind: str, row: dict[str, Any], editable: list[str]) -> dict[str, Any]:
"""그 줄의 **값이 아닌 글자 칸**(이름·규격·단위·인도조건 …) — 이름표가 숨김 아니라 한 열만.
갱신으로 같은 코드에 다른 기준이 붙으면(노임 국가유산직종 개편 · 자재 KS 개정) 값만으로는 못 잡음.
"""
out = {}
for key, value in row.items():
if key.startswith("@") or key in editable or key in _NOT_MEANING or key.endswith("_date"):
continue
if isinstance(value, str) and value.strip() and not column_meta(kind, key)["hidden"]:
out[key] = value
return out
def public(row: dict[str, Any]) -> dict[str, Any]:
"""내보낼 줄 — 안쪽 칸(`@source`·`@axis`·`@data`…)은 뺌."""
return {k: v for k, v in row.items() if k not in _INNER}
"""내보낼 줄 — 안쪽 칸(`@source`·`@axis`)은 뺌."""
return {k: v for k, v in row.items() if k not in ("@source", "@axis")}
#: 찾기만 같게 보는 글자 — 원문이 물결표·가운뎃점을 여러 벌로 씀(∼ 26 · ~ 8 · ․ · ㆍ) · 자간 공백(「굴 삭 기」).
@@ -542,7 +538,6 @@ def fold(text: str) -> str:
def table(
kind: str,
all_rows: list[dict[str, Any]],
page: int = 1,
size: int = tables.DEFAULT_PAGE_SIZE,
q: str = "",
@@ -550,7 +545,7 @@ def table(
desc: bool = False,
filters: dict[str, str] | None = None,
) -> dict[str, Any]:
"""한 표 한 쪽 — `all_rows` 는 `rows(kind)` 가 준 전체(DB 읽기와 셈을 떼어 둠)."""
all_rows = rows(kind)
columns = _columns(kind, all_rows)
needle = fold(q)
hits = (
@@ -574,14 +569,7 @@ def table(
"columns": [column_meta(kind, c) for c in columns],
# 못 고치는 칸(계산값)도 **세우기는 됨** — 화면은 이 목록으로 눌릴 제목을 가림
"sortable": list(columns),
"rows": [
{
**public(r),
"changed_columns": r.get("@changed", []),
"is_added": r.get("@added", False),
}
for r in hits[start : start + size]
],
"rows": [public(r) for r in hits[start : start + size]],
"total": len(hits),
**spec(kind, columns),
"notice": notice(kind),
@@ -590,84 +578,49 @@ def table(
}
class SaveError(ValueError):
"""[저장] 을 막는 까닭 — 하나라도 있으면 아무것도 안 씀(전부 아니면 전무)."""
def __init__(self, status: int, message: str) -> None:
super().__init__(message)
self.status = status
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 _axis_of(kind: str) -> dict[str, tuple[str, ...]]:
"""줄마다 축 칸 — 파일 원본에서(표 모양이라 값 고치기로 안 바뀜 · 주입 data 에 `@axis` 가 없어도 섬)."""
if kind not in base_tables.SPEC:
return {}
return {r["@id"]: tuple(r.get("@axis") or ()) for r in base_rows(kind)}
def _cell_problem(
row: dict[str, Any], column: str, value: Any, info: dict[str, Any], axis: dict, fillable: set
) -> str | None:
"""한 칸 고치기를 막는 까닭 — 추가 줄은 저장 칸이면 글자 칸도(이름·규격을 사람이 넣은 줄)."""
if row["@added"]:
if column not in fillable:
return f"못 고치는 칸 {column} — {info['locked'].get(column) or '이 표에 없는 칸'}"
elif column in axis.get(row["@id"], ()):
return f"못 고치는 칸 {column} — {_KEY}"
elif column not in info["editable"] or column not in row:
return f"못 고치는 칸 {column} — {info['locked'].get(column) or '이 줄에 없는 칸'}"
if column in info["editable"]:
problem = check_value(value)
return f"{column}: {problem}" if problem else None
if isinstance(value, (dict, list)):
return f"{column}: 글자나 숫자만 넣을 수 있음"
return None
def plan_save(
kind: str,
shown: list[dict[str, Any]],
edits: list[dict[str, Any]],
added: list[dict[str, Any]],
deleted: list[str],
) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], list[str]]:
"""[저장] 한 번 — 다 맞춰 본 뒤 쓸 것만: (고친 줄 data · 새 줄 data · 지울 줄). 막히면 SaveError.
`shown` = `rows(kind)`(기계는 계산 칸까지 붙은 것) — 쓰는 값은 그 줄 `@data` 에 고친 칸만 얹음.
새 줄 열쇠 = `added/{난수}`(브레인 계약에 열쇠 규칙이 없어 서버가 줌 · 응답으로 알림).
"""
by_key = {r["@id"]: r for r in shown}
info = spec(kind, _columns(kind, shown))
axis = _axis_of(kind)
fillable = {c for r in shown for c in r["@data"] if not c.startswith("@")} # 계산 칸 제외
changed: dict[str, dict[str, Any]] = {}
for edit in edits:
row = by_key.get(edit["row_key"])
if row is None:
raise SaveError(404, f"없는 줄: {edit['row_key']}")
problem = _cell_problem(row, edit["column"], edit["value"], info, axis, fillable)
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 in (base.get("@axis") or ()):
return 400, {"message": f"못 고치는 칸 {column} — {_KEY}"}
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:
raise SaveError(400, problem)
changed.setdefault(row["@id"], dict(row["@data"]))[edit["column"]] = edit["value"]
new: dict[str, dict[str, Any]] = {}
for data in added:
for column, value in data.items():
problem = _cell_problem({"@added": True}, column, value, info, axis, fillable)
if problem:
raise SaveError(400, problem)
new[f"added/{uuid.uuid4().hex[:12]}"] = dict(data)
missing = [key for key in deleted if key not in by_key]
if missing:
raise SaveError(404, f"없는 줄: {', '.join(missing)}")
return changed, new, list(dict.fromkeys(deleted))
return 400, {"message": f"{column}: {problem}"}
overrides.write(kind, base, values, by, meaning_of(kind, base, info["editable"]))
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"]))
meta = column_meta(each, item["column"])
item["column_label"] = meta["label"]
item["unit"] = meta["unit"]
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))