refactor(M01): Z01_MasterData 폴더째 old_code 로 옮기고 등록·문구 끊음 (PLAN 7-2)
라우터·라우트·문구에서 Z01 을 끊음 — M01 만 남김. 시험 여섯도 old_code/resources/tester 로 같이 옮김. Z01 이 읽던 옛 자료 셋은 디스크에 이미 없음 · master_data/old 는 B08·B09 도 읽어 남김. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PAdA5ThqmtVSsbzjusk1cJ
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,626 @@
|
||||
"""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
|
||||
import re
|
||||
from datetime import date
|
||||
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_BasePrices_Tables as base_tables
|
||||
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
|
||||
|
||||
#: 기초단가 다섯 + 기초데이터 열 + 공종 축 둘(산림·건설) — 이름은 자료 이름 그대로
|
||||
KINDS = ("labor", "machine", "material", "oil", "rate", *base_tables.SPEC, *work_items.KINDS)
|
||||
SOURCES_DIR = (
|
||||
tables.RESOURCES / "data_master_sources"
|
||||
) # 출처표(데스크탑 서브) — 어디서 받고 최신이 무엇인지
|
||||
_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",
|
||||
)
|
||||
#: ⚠ **지금 상태를 먼저** — 덮개를 읽는 곳이 Z01 밖에 0(2026-09-17 서브 훑기 · 브레인 「없으면 없다고」).
|
||||
#: 뒤 문장은 프로젝트 복사(요구 14)가 붙으면 참이 되는 약속이라 지우지 않고 「그 길이 붙으면」으로 둠.
|
||||
_SCOPE_NOTICE = (
|
||||
"⚠ **아직 어디에도 안 쓰임** — 고친 값은 마스터에만 쌓임 · 프로젝트로 옮기는 길(프로젝트 복사)이 아직 없음."
|
||||
" 그 길이 붙으면: 여기서 고친 값은 **새로 만드는 프로젝트부터** 쓰임 — 이미 만든 프로젝트 금액은 안 바뀜"
|
||||
"(프로젝트는 만들 때 뜬 사본을 봄)"
|
||||
)
|
||||
_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 not None:
|
||||
return path
|
||||
found = sorted(tables.RESOURCES.glob(f"data_*/{file_id}*.json"))
|
||||
if not found:
|
||||
raise FileNotFoundError(f"원본 자리를 못 찾음: {file_id}")
|
||||
return found[-1]
|
||||
|
||||
|
||||
@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 edition_of(doc: dict[str, Any]) -> str | None:
|
||||
"""판 기준일 — 자료마다 자리가 달라 **여기 한 곳**에서 뽑음(`effective_date` · 파생본은 `derived_from.effective_date`)."""
|
||||
derived = doc.get("derived_from")
|
||||
return doc.get("effective_date") or (
|
||||
derived.get("effective_date") if isinstance(derived, dict) else None
|
||||
)
|
||||
|
||||
|
||||
def _date(text: Any) -> date | None:
|
||||
try:
|
||||
return date.fromisoformat(str(text))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def sources(kind: str) -> list[dict[str, Any]]:
|
||||
"""표 수준 출처 — kind 하나에 판이 둘인 곳(노임·유가·기계)이 있어 목록.
|
||||
|
||||
`our_edition` 은 출처표 글자가 아니라 **실제 파일**에서 읽음(글자가 굳으면 또 뒤처짐) ·
|
||||
`outdated` 판정은 서버만(최신 공표일 > 우리 판 · 둘 중 하나라도 모르면 None).
|
||||
"""
|
||||
found = sorted(SOURCES_DIR.glob("sources_*.json")) if SOURCES_DIR.is_dir() else []
|
||||
if not found:
|
||||
return []
|
||||
listing = _read(str(found[-1]), found[-1].stat().st_mtime_ns)
|
||||
out = []
|
||||
for entry in listing.get("sources") or []:
|
||||
if entry.get("kind") != kind:
|
||||
continue
|
||||
editions = []
|
||||
for name in entry.get("files") or []:
|
||||
path = (tables.ROOT / name).resolve()
|
||||
if path.is_file() and path.is_relative_to(tables.RESOURCES):
|
||||
editions.append(_date(edition_of(_read(str(path), path.stat().st_mtime_ns))))
|
||||
ours = min((e for e in editions if e), default=None)
|
||||
latest = _date(entry.get("latest_published"))
|
||||
out.append(
|
||||
{
|
||||
"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 "없음", # 안 지어냄(브레인)
|
||||
"cycle": entry.get("cycle") or "",
|
||||
"latest_published": latest.isoformat() if latest else None,
|
||||
"our_edition": ours.isoformat() if ours else None,
|
||||
"checked_at": entry.get("checked_on") or None,
|
||||
"outdated": None if ours is None or latest is None else latest > ours,
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
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,
|
||||
"source": dataset,
|
||||
"effective_date": edition_of(doc),
|
||||
**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 = edition_of(doc)
|
||||
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", "scope")}
|
||||
common = {
|
||||
"source": v.get("scope"),
|
||||
"fuel": variable,
|
||||
"effective_date": edition_of(doc), # 전국 · 지역 판 기준일이 다름
|
||||
**head,
|
||||
}
|
||||
if "records" not in v:
|
||||
rows.append(
|
||||
{
|
||||
"@id": f"{v.get('scope')}/{variable}",
|
||||
"@source": name,
|
||||
**common,
|
||||
"price_krw_per_l": v.get("value"),
|
||||
}
|
||||
)
|
||||
for r in v.get("records") or []:
|
||||
rows.append(
|
||||
{
|
||||
"@id": f"{v.get('scope')}/{variable}/{r['sido_code']}",
|
||||
"@source": name,
|
||||
**common,
|
||||
"region_code": r["sido_code"],
|
||||
"region_name": r["sido_name"],
|
||||
"price_krw_per_l": r["value"],
|
||||
}
|
||||
)
|
||||
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": edition_of(doc),
|
||||
**_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": edition_of(doc),
|
||||
"base": v.get("base"),
|
||||
**item,
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
_BUILDERS = {"labor": _labor, "material": _material, "oil": _oil, "rate": _rate}
|
||||
|
||||
|
||||
class DuplicateRowError(ValueError):
|
||||
"""열쇠가 같은 줄인데 값이 다름 — 합치면 틀린 값이 섬(브레인 조건 · 품셈 개정으로 갈리는 날)."""
|
||||
|
||||
|
||||
def merge_same_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""열쇠가 같고 **값도 같은** 줄은 한 줄로 — 원문 두 줄(coef 「암괴…점토」)이 그 자리.
|
||||
|
||||
값이 갈리면 멈춤: 원문이 개정돼 두 줄이 달라지는 날 조용히 한 줄로 합치면 안 됨.
|
||||
"""
|
||||
body = lambda r: { # noqa: E731 — 견줄 알맹이(붙인 사유·줄 수는 뺌)
|
||||
k: v
|
||||
for k, v in r.items()
|
||||
if not k.startswith("@") and k not in ("source_rows", "duplicate_note")
|
||||
}
|
||||
first: dict[str, dict[str, Any]] = {}
|
||||
out = []
|
||||
for row in rows:
|
||||
kept = first.get(row["@id"])
|
||||
if kept is None:
|
||||
first[row["@id"]] = row
|
||||
out.append(row)
|
||||
continue
|
||||
if body(kept) != body(row):
|
||||
raise DuplicateRowError(f"열쇠가 같은데 값이 다름: {row['@id']}")
|
||||
kept["source_rows"] = kept.get("source_rows", 1) + 1
|
||||
kept["duplicate_note"] = (
|
||||
"원문 두 줄 · 값 같음 — 한 줄로 보임(원문 오기 가능성 · 임의 보정 없음)"
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
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))
|
||||
if kind in work_items.KINDS:
|
||||
return work_items.base_rows(kind)
|
||||
spec_of = base_tables.SPEC.get(kind)
|
||||
if spec_of:
|
||||
doc, name = _doc(spec_of["file"])
|
||||
return merge_same_rows(spec_of["build"](doc, name))
|
||||
return _BUILDERS[kind]()
|
||||
|
||||
|
||||
def _machine_inputs() -> machine.Inputs:
|
||||
"""기계가 끌어 쓰는 밑값 — **덮개를 얹은** 노임(건설)·전국평균 유가 + 그 판들의 기준일."""
|
||||
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']}"
|
||||
]
|
||||
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["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]),
|
||||
)
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
class SortError(ValueError):
|
||||
"""모르는 열로 세우라 함 — 조용히 원본 차례로 돌아가지 않음(브레인 ④)."""
|
||||
|
||||
|
||||
#: 품셈 번호 꼴(「9-3-2」·「12-17」) — 글자로 세우면 「10-1」이 「2-1」 앞에 섬(서브 훑기 · 브레인).
|
||||
_NUMBER_PATH = re.compile(r"^\d+(?:-\d+)+$")
|
||||
|
||||
|
||||
def _order_key(value: Any) -> tuple:
|
||||
"""섞인 칸도 세움 — 수 먼저, 품셈 번호 꼴은 **마디마다 수로**, 그 뒤 글자."""
|
||||
if isinstance(value, bool):
|
||||
return (2, str(value))
|
||||
if isinstance(value, (int, float)):
|
||||
return (0, float(value))
|
||||
if isinstance(value, (dict, list)):
|
||||
return (2, json.dumps(value, ensure_ascii=False, default=str))
|
||||
if isinstance(value, str) and _NUMBER_PATH.match(value):
|
||||
return (1, tuple(int(part) for part in value.split("-")))
|
||||
return (2, str(value))
|
||||
|
||||
|
||||
def _sorted(rows: list[dict[str, Any]], column: str, desc: bool) -> list[dict[str, Any]]:
|
||||
"""⚠ 빈 칸은 오름·내림 **어느 쪽이든 끝** — 미공표 노임 14 · 손료계수 없는 기계 226 이
|
||||
첫 쪽을 통째로 채우면 표를 못 씀(브레인 ②)."""
|
||||
filled = [r for r in rows if r.get(column) not in (None, "")]
|
||||
empty = [r for r in rows if r.get(column) in (None, "")]
|
||||
return sorted(filled, key=lambda r: _order_key(r[column]), reverse=desc) + empty
|
||||
|
||||
|
||||
#: 돌쌓기 표준경사는 B06 횡단설계가 같은 값으로 기울기를 그림 — 아홉 중 **그림에도 닿는** 하나(브레인).
|
||||
_SLOPE_NOTICE = (
|
||||
"⚠ **B06 횡단설계가 아직 여기서 고친 값을 안 읽음**(원본 값으로 벽 기울기를 그림) — 지금은 고쳐도 횡단 도면이 안 바뀜."
|
||||
" 읽게 되면: 고치면 횡단 도면 모양도 함께 바뀜"
|
||||
)
|
||||
#: 같은 「1:0.3」 이 두 자리에 있음(코덱스 검증 2) — 값은 같으나 **축과 근거가 다름**.
|
||||
#: masonry_slope = 품셈 13-4-4 [주]⑪ 표준경사(직고 · 메/찰 · 성토/절토 축 · 원문표)
|
||||
#: masonry_class.face_slope = 교본 7-3 돌흙막이(구조물 **형식**별 기본값 · 판정이 안 되는 자리의 종전값)
|
||||
_CLASS_NOTICE = (
|
||||
"전면 기울기 기본값은 **교본 7-3 형식별 값** — 품셈 표준경사(직고·성토/절토 축)는 「돌쌓기 표준경사」 표에 따로 있음"
|
||||
" · 같은 1:0.3 이라도 **축과 근거가 다름**(판정이 안 되는 자리에서만 이 값으로 섬)"
|
||||
)
|
||||
_EXTRA_NOTICE = {
|
||||
"rate": _RATE_NOTICE,
|
||||
"masonry_slope": _SLOPE_NOTICE,
|
||||
"masonry_class": _CLASS_NOTICE,
|
||||
}
|
||||
|
||||
|
||||
def notice(kind: str) -> list[str]:
|
||||
"""표 수준 알림 — 화면이 따로 가지면 문구가 두 벌이 됨(브레인 ③)."""
|
||||
if kind in work_items.KINDS:
|
||||
return work_items.notice(kind)
|
||||
return [_SCOPE_NOTICE] + ([_EXTRA_NOTICE[kind]] if kind in _EXTRA_NOTICE else [])
|
||||
|
||||
|
||||
def spec(kind: str, columns: list[str]) -> dict[str, Any]:
|
||||
"""표 수준 — editable · locked · formula."""
|
||||
if kind == "machine":
|
||||
return machine.spec(columns)
|
||||
if kind in work_items.KINDS:
|
||||
return work_items.spec(columns)
|
||||
if kind in base_tables.SPEC:
|
||||
spec_of = base_tables.SPEC[kind]
|
||||
editable = [c for c in spec_of["editable"] if c in columns]
|
||||
axis = {a for r in base_rows(kind) for a in r.get("@axis", ())}
|
||||
told = spec_of.get("locked") or {} # 계산값처럼 까닭이 따로인 칸
|
||||
return {
|
||||
"editable": editable,
|
||||
"locked": {
|
||||
c: told.get(c) or (_KEY if c in axis else _SOURCE)
|
||||
for c in columns
|
||||
if c not in editable
|
||||
},
|
||||
"formula": {k: v for k, v in (spec_of.get("formula") or {}).items() if k in columns},
|
||||
}
|
||||
editable = {
|
||||
"labor": ["daily_wage_krw"],
|
||||
"material": ["price_krw"],
|
||||
"oil": ["price_krw_per_l"],
|
||||
"rate": [c for c in _RATE_EDITABLE if c in columns],
|
||||
}[kind]
|
||||
keys = {"labor": {"source", "occupation_code"}, "material": {"item_code"}}.get(kind, set())
|
||||
if kind == "oil":
|
||||
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}
|
||||
return {
|
||||
"editable": editable,
|
||||
"locked": locked,
|
||||
"formula": {},
|
||||
}
|
||||
|
||||
|
||||
#: 갈래 둘 — 기초단가(원가 = 얼마인가) · 품셈 기준(수량 = 얼마나 드나). 화면은 이것으로 상자를 세움.
|
||||
BASE_PRICE_GROUP, PUMSEM_GROUP = "base_price", "pumsem_basis"
|
||||
|
||||
|
||||
def kind_group(kind: str) -> str:
|
||||
if kind in work_items.KINDS:
|
||||
return work_items.GROUP
|
||||
return PUMSEM_GROUP if kind in base_tables.SPEC else BASE_PRICE_GROUP
|
||||
|
||||
|
||||
def kind_label(kind: str) -> str:
|
||||
"""kind 한글 이름 — 이름표 `merged_tables` 에서만(코드에 박지 않음 · 없으면 영문 그대로)."""
|
||||
merged = tables.load_labels().get("merged_tables") or []
|
||||
named = next((t for t in merged if t.get("key") in (kind, f"{kind}s")), {})
|
||||
return named.get("name_ko") or kind
|
||||
|
||||
|
||||
def group_label(group: str) -> str:
|
||||
"""상자 이름 — 이름표 `base_price_groups[갈래].name_ko` 에서만(없으면 영문 그대로 · 화면은 `group_label` 을 읽음)."""
|
||||
named = (tables.load_labels().get("base_price_groups") or {}).get(group) or {}
|
||||
return named.get("name_ko") or group
|
||||
|
||||
|
||||
def kinds() -> list[dict[str, Any]]:
|
||||
"""화면이 상자를 세울 목록 — **서버가 냄**(화면이 제 코드에 들면 새 kind 가 조용히 안 뜸).
|
||||
|
||||
갈래 · 갈래 한글 이름 · kind · 한글 이름(이름표) · 줄 수.
|
||||
"""
|
||||
return [
|
||||
{
|
||||
"group": kind_group(kind),
|
||||
"group_label": group_label(kind_group(kind)),
|
||||
"kind": kind,
|
||||
"label": kind_label(kind),
|
||||
"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()
|
||||
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,
|
||||
"label": named.get("name_ko") or key,
|
||||
"unit": named.get("unit") or "",
|
||||
"hidden": named.get("visible") is False,
|
||||
}
|
||||
|
||||
|
||||
#: 뜻 칸에서 뺄 자리 — 판·공표 날짜는 갱신 때마다 바뀌어 「뜻 바뀜」과 구별이 안 됨
|
||||
_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`)은 뺌."""
|
||||
return {k: v for k, v in row.items() if k not in ("@source", "@axis")}
|
||||
|
||||
|
||||
#: 찾기만 같게 보는 글자 — 원문이 물결표·가운뎃점을 여러 벌로 씀(∼ 26 · ～ 8 · ․ · ㆍ) · 자간 공백(「굴 삭 기」).
|
||||
#: ⚠ **찾기에만** 씀 — 열쇠·칸 값은 원문 그대로(손질하면 갈래 열쇠가 갈림 · 브레인 2026-09-17).
|
||||
_SEARCH_SAME = str.maketrans(
|
||||
{"∼": "~", "～": "~", "〜": "~", "․": "·", "ㆍ": "·", "･": "·", "‧": "·"}
|
||||
)
|
||||
|
||||
|
||||
def fold(text: str) -> str:
|
||||
"""찾기 비교용 — 소문자 · 물결표·가운뎃점 한 벌 · 공백 없앰."""
|
||||
return "".join(str(text or "").lower().translate(_SEARCH_SAME).split())
|
||||
|
||||
|
||||
def table(
|
||||
kind: str,
|
||||
page: int = 1,
|
||||
size: int = tables.DEFAULT_PAGE_SIZE,
|
||||
q: str = "",
|
||||
sort: str = "",
|
||||
desc: bool = False,
|
||||
filters: dict[str, str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
all_rows = rows(kind)
|
||||
columns = _columns(kind, all_rows)
|
||||
needle = fold(q)
|
||||
hits = (
|
||||
[
|
||||
r
|
||||
for r in all_rows
|
||||
if needle in fold(json.dumps(public(r), ensure_ascii=False, default=str))
|
||||
]
|
||||
if needle
|
||||
else all_rows
|
||||
)
|
||||
if kind in work_items.KINDS: # 서버가 준 거르기(부문·줄 구실) — 고른 값만 옴
|
||||
hits = work_items.apply_filters(kind, hits, filters or {})
|
||||
if sort: # ⚠ 거른 뒤에 세움 — 거꾸로 하면 차례가 깨짐(브레인 ③)
|
||||
if sort not in columns:
|
||||
raise SortError(f"없는 열로 세울 수 없음: {sort}")
|
||||
hits = _sorted(hits, sort, desc)
|
||||
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],
|
||||
# 못 고치는 칸(계산값)도 **세우기는 됨** — 화면은 이 목록으로 눌릴 제목을 가림
|
||||
"sortable": list(columns),
|
||||
"rows": [public(r) for r in hits[start : start + size]],
|
||||
"total": len(hits),
|
||||
**spec(kind, columns),
|
||||
"notice": notice(kind),
|
||||
"source": sources(kind),
|
||||
**(work_items.extra(kind) if kind in work_items.KINDS else {}),
|
||||
}
|
||||
|
||||
|
||||
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:
|
||||
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))
|
||||
Reference in New Issue
Block a user