eomsangdonandClaude Opus 5
a8d635f2cc
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn
620 lines
26 KiB
Python
620 lines
26 KiB
Python
"""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 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):
|
||
"""모르는 열로 세우라 함 — 조용히 원본 차례로 돌아가지 않음(브레인 ④)."""
|
||
|
||
|
||
def _order_key(value: Any) -> tuple:
|
||
"""섞인 칸도 세움 — 수 먼저, 그 뒤 글자."""
|
||
if isinstance(value, bool):
|
||
return (1, str(value))
|
||
if isinstance(value, (int, float)):
|
||
return (0, float(value))
|
||
if isinstance(value, (dict, list)):
|
||
return (1, json.dumps(value, ensure_ascii=False, default=str))
|
||
return (1, 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))
|