"""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")} 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 = 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 ) 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))