eomsangdonandClaude Opus 5
6fe4056d98
- Z01_MasterData_Repository_BasePrices.py 신설 — master_base_price·master_revision Raw SQL(읽기·줄 수·판번호·저장·되돌리기) · 판번호 줄을 잠그고 다시 봄 · 틀리면 통째로 되돌림
- GET /base-prices/{kind} 가 DB 를 읽음 · 응답에 revision · 줄마다 changed_columns·is_added 추가
- PUT /base-prices/{kind} {base_revision, edits, added, deleted} — 판 다르면 409 · 하나라도 틀리면 아무것도 안 씀 · edits 500 상한 · 모르는 칸 거절
- POST /base-prices/{kind}/reset {row_keys} — 주입 줄은 data := seed(지운 줄도 되살림) · 추가 줄은 지움
- 기계 시간당 단가는 DB 노임·유가로 읽을 때마다 셈 · 제원은 줄 값에서(추가한 기종도 섬)
- 옛 덮개 층 삭제 — Z01_MasterData_Overrides.py · 칸 단위 PUT · /overrides·/overrides/clear
- 새 줄 열쇠는 서버가 줌(added/{난수}) · 주입 data 에 '@' 칸이 없어도 축 잠금은 파일 원본으로 섬
- 시험: DB 가짜(helper_z01_fake_repo.py) 로 API 손질 · 저장소 SQL 흐름은 가짜 커넥션으로 잼
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V1MKKZKpUHTPKb513FneU8
674 lines
28 KiB
Python
674 lines
28 KiB
Python
"""Z01 기초단가 다섯 — 노임·기계·자재·유가·요율을 kind 마다 **한 표**로(2026-09-15 사용자 지시 · 브레인 계약).
|
||
|
||
표 수준(한 번): columns · total · editable · locked{열: 왜 못 고치나} · formula{열: 식} · notice(요율만)
|
||
줄 수준: `@id`(자료 제 열쇠) · `changed_columns`(초기값과 다른 칸) · `is_added`(관리자 추가 줄) · `@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`)의 재료 · 공종 축 둘만 아직 파일 읽기.
|
||
"""
|
||
|
||
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
|
||
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_Repository_BasePrices as repo
|
||
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(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")]
|
||
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
|
||
]
|
||
return machine.Inputs(
|
||
wages={
|
||
str(r["occupation_code"]): None
|
||
if r.get("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,
|
||
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},
|
||
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 _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
|
||
|
||
|
||
async def kinds() -> list[dict[str, Any]]:
|
||
"""화면이 상자를 세울 목록 — **서버가 냄**(화면이 제 코드에 들면 새 kind 가 조용히 안 뜸).
|
||
|
||
갈래 · 갈래 한글 이름 · kind · 한글 이름(이름표) · 줄 수(DB · 공종 축은 파일).
|
||
"""
|
||
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),
|
||
}
|
||
for kind in KINDS
|
||
]
|
||
|
||
|
||
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,
|
||
}
|
||
|
||
|
||
_INNER = ("@source", "@axis", "@data", "@added", "@changed")
|
||
|
||
|
||
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}
|
||
|
||
|
||
#: 찾기만 같게 보는 글자 — 원문이 물결표·가운뎃점을 여러 벌로 씀(∼ 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,
|
||
all_rows: list[dict[str, Any]],
|
||
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)` 가 준 전체(DB 읽기와 셈을 떼어 둠)."""
|
||
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),
|
||
"changed_columns": r.get("@changed", []),
|
||
"is_added": r.get("@added", False),
|
||
}
|
||
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 {}),
|
||
}
|
||
|
||
|
||
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)
|
||
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))
|