eomsangdonandClaude Opus 5
e1c71c3e8b
- GET /work-items(상태별 개수) · GET/PUT /work-items/{key} · POST …/reset 신설, 시스템관리자 전용 묶음에 등록
- 저장은 한 트랜잭션 — 낡은 revision 409 · 없는 자원 열쇠·남의 줄 id 422 · 실패 시 전부 되돌림
- 빠진 줄은 초기값 줄이면 숨김(초기값으로 때 살림) · 관리자 추가 줄이면 지움
- 검사·줄 비교·되돌리기 규칙은 WorkItems.py 순수 함수 · SQL 은 Repository_WorkItems.py
- 옛 기초단가 표의 공종 축 보기(JSON 읽기)는 그대로 둠 — BasePrices.py(sub_laptop_2 몫)가 동기로 부름
- 시험 15개(test_z01_work_item_edit.py) — DB 없이 가짜 커넥션으로 409·422·롤백 확인
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016WRFJmmhPi4exAzHgPZHMT
427 lines
18 KiB
Python
427 lines
18 KiB
Python
"""Z01 공종 축 — 산림·건설 두 kind 를 기초단가와 **같은 계약**으로(2026-09-17 브레인 · PLAN_공종축 8-4).
|
||
|
||
줄 = **품셈 표가 붙은 마디**(`axis_policy.json` `flat_list_filter: pum_tables > 0` — 잎으로 거르면 부모에 붙은 표가 안 뜸) ·
|
||
갈래(`variant_keys`)마다 한 줄로 폄 — 줄 수 = 편 수(`stats.units`) · 공종 수 = `stats.work_items`.
|
||
이름 = 뿌리부터 전체 경로(「토공 › 토사깍기 › 기계」 · 건설은 부문부터) — 잎 이름만으론 겹침.
|
||
계산 규칙 = 표 수준 `rules`(부모 마디 choose_one · sum_steps) + 줄 칸 `rule`·`step_weight`
|
||
— 빠지면 발파암(절취 0.1 + 깍기 0.9 + 집토 1) 금액이 틀어짐.
|
||
`@id` = 불변 열쇠(FW-·CW-) — 없는 줄은 목차 코드로 서고 알림에 드러냄.
|
||
⚠ 읽기만 — 고칠 칸 없음(원문 값 풀기 8-7 뒤). 원본 `work_item_code`(FP-)는 B08·B09 가 씀 — 안 건드림.
|
||
|
||
공종 편집(DB 정본 · 2026-09-18 PLAN 1-0) — 맨 아래 「공종 편집」 절. `master_work_item`·`_row` 를
|
||
고치는 규칙만 순수 함수로(DB 없이 시험) · SQL 은 `Z01_MasterData_Repository_WorkItems`.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from decimal import Decimal
|
||
from functools import lru_cache
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from Z01_MasterData import Z01_MasterData_Tables as tables
|
||
|
||
GROUP = "work_item"
|
||
#: kind → 원본 품셈 dataset_id — 공종 축 파일 `dataset_version.dataset_id` 로 가림(파일 이름에 기대지 않음)
|
||
SOURCE_OF = {
|
||
"work_item_forest": "pum_forest",
|
||
"work_item_const": "pum_const",
|
||
} # 이름표 `work_item_axis` 와 같은 이름
|
||
FOLDER = tables.RESOURCES / "data_work_item_master"
|
||
#: 불변 열쇠 칸 — ⚠ 데스크탑 서브 자료가 오면 칸 이름을 맞출 것(여기 한 곳)
|
||
KEY_FIELD = "work_item_key"
|
||
PATH_JOIN = " › "
|
||
_LOCKED = "공종 축 뼈대 — 품셈 원문으로만 바뀜(값 고치기는 원문 값 풀기 뒤)"
|
||
|
||
|
||
@lru_cache(maxsize=8)
|
||
def _read(path: str, _mtime_ns: int) -> dict[str, Any]:
|
||
return json.loads(Path(path).read_text(encoding="utf-8"))
|
||
|
||
|
||
def doc(kind: str) -> tuple[dict[str, Any], str] | None:
|
||
"""그 품셈의 공종 축 끝 판(문서 · 파일 이름) — 없으면 None(건설은 아직 없음)."""
|
||
found = []
|
||
for path in sorted(FOLDER.glob("*work_item_master_*.json")): # 건설은 `const_` 앞꼬리
|
||
d = _read(str(path), path.stat().st_mtime_ns)
|
||
if (d.get("dataset_version") or {}).get("dataset_id") == SOURCE_OF[kind]:
|
||
found.append((d, path.name))
|
||
return max(found, key=lambda each: each[0].get("effective_date") or "", default=None)
|
||
|
||
|
||
#: 자료가 있는 품셈만 kind 로 섬 — 빈 상자는 이름만 뜨고 안 열림 · 건설은 자료가 오면 서버 재시작 뒤 저절로
|
||
KINDS = tuple(kind for kind in SOURCE_OF if doc(kind))
|
||
|
||
|
||
def _key(item: dict[str, Any]) -> str:
|
||
return item.get(KEY_FIELD) or item["work_item_code"]
|
||
|
||
|
||
def _tree(d: dict[str, Any]) -> tuple[list[dict], dict[str, dict], dict[str, list[dict]]]:
|
||
items = d.get("work_items") or []
|
||
children: dict[str, list[dict]] = {}
|
||
for item in items:
|
||
if item.get("parent_code"):
|
||
children.setdefault(item["parent_code"], []).append(item)
|
||
return items, {i["work_item_code"]: i for i in items}, children
|
||
|
||
|
||
def _path(item: dict[str, Any] | None, by: dict[str, dict]) -> str:
|
||
names = []
|
||
while item:
|
||
names.append(item["name"])
|
||
item = by.get(item.get("parent_code"))
|
||
return PATH_JOIN.join(reversed(names))
|
||
|
||
|
||
def _shown_path(item: dict[str, Any], by: dict[str, dict]) -> str:
|
||
"""목록에 보일 경로 — 조각마다 자간 공백 다듬기(「토목부문 › 측 량」→「측량」). 열쇠와 무관."""
|
||
from B08_Quantity.B08_Quantity_Build_WorkItemMaster_Keys import tidy_spacing
|
||
|
||
return PATH_JOIN.join(tidy_spacing(part) for part in _path(item, by).split(PATH_JOIN))
|
||
|
||
|
||
def base_rows(kind: str) -> list[dict[str, Any]]:
|
||
found = doc(kind)
|
||
if found is None:
|
||
return []
|
||
d, name = found
|
||
items, by, children = _tree(d)
|
||
# 계산 규칙 이름 — 이름표 `value_overrides["work_item_master::<방식>"]`(「아래를 단계로 다 더함」) · 없으면 영문
|
||
named = tables.load_labels().get("value_overrides") or {}
|
||
linked, roles = linked_keys(), _role_names()
|
||
rows = []
|
||
for item in items:
|
||
if not item.get("tables"):
|
||
continue
|
||
parent = by.get(item.get("parent_code")) or {}
|
||
weight = next(
|
||
(s["weight"] for s in parent.get("steps") or [] if s["code"] == item["work_item_code"]),
|
||
None,
|
||
)
|
||
key = _key(item)
|
||
# 갈래마다 한 줄 — 일위대가가 갈래마다 달라 집는 단위가 갈래임(교목굴취 나무높이 9 · 2026-09-17 브레인 ③).
|
||
# 열쇠 = `공종 열쇠#갈래 번호`(`FW-00105#01`) — **갈래 글자를 열쇠에 담지 않는다.**
|
||
# 담으면 원문 자간·물결표가 다듬어지는 날 열쇠가 갈린다(2026-09-17 브레인 · 장부가 번호를 준다).
|
||
for variant in item.get("variants") or [None]:
|
||
rows.append(
|
||
{
|
||
"@id": f"{key}#{variant['variant_key']}" if variant else key,
|
||
"@key": key, # 공종 열쇠 — 거르기·줄 수는 공종 단위로
|
||
"@source": name,
|
||
"work_item_code": item["work_item_code"], # 목차 코드 — 열쇠 아님, 칸으로만
|
||
"number": item["number"],
|
||
"name": _shown_path(item, by), # 자간 공백만 다듬음 — 원문 이름은 마스터에
|
||
"variant": variant["name"] if variant else None, # 원문 그대로
|
||
"variant_clean": variant["name_clean"] if variant else None, # 다듬은 것
|
||
"pum_tables": [t["pum_table_id"] for t in item["tables"]],
|
||
"rule": (named.get(f"work_item_master::{parent.get('parent_mode')}") or {}).get(
|
||
"name_ko"
|
||
)
|
||
or parent.get("parent_mode"),
|
||
"step_weight": weight,
|
||
# 「임도가 쓰는 것」 — 연결고리 표가 확정으로 이은 공종(판정은 그 표 · 감추지 않고 표시만)
|
||
"link": LINKED if key in linked else "",
|
||
# 줄 구실(총칙·공종) — 이름은 axis_policy.json · 표시 칸(거르기 상자로 고름)
|
||
"axis_role": roles.get(str(item.get("axis_role") or ""), ""),
|
||
"effective_date": d.get("pum_edition") or d.get("effective_date"),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
LINK_FOLDER = tables.RESOURCES / "data_work_item_link"
|
||
LINKED = "연결됨"
|
||
|
||
|
||
def linked_keys() -> frozenset[str]:
|
||
"""연결고리 표(`work_item_link_*.json` 끝 판 · 코덱스)가 **확정**으로 이은 열쇠(산림·건설).
|
||
|
||
⚠ 목록을 코드에 안 둠 — 판정은 그 표 · 미판정은 안 이음(표의 `unconfirmed_action: do_not_link`).
|
||
"""
|
||
found = sorted(LINK_FOLDER.glob("work_item_link_*.json")) if LINK_FOLDER.is_dir() else []
|
||
if not found:
|
||
return frozenset()
|
||
links = _read(str(found[-1]), found[-1].stat().st_mtime_ns).get("links") or []
|
||
return frozenset(
|
||
key
|
||
for link in links
|
||
if link.get("status") == "확정"
|
||
for key in (link.get("forest_key"), link.get("const_key"))
|
||
if key
|
||
)
|
||
|
||
|
||
def rules(kind: str) -> list[dict[str, Any]]:
|
||
"""부모 마디 계산 규칙 — choose_one(아래 중 하나) · sum_steps(아래를 무게대로 더해야 한 단위)."""
|
||
found = doc(kind)
|
||
if found is None:
|
||
return []
|
||
items, by, children = _tree(found[0])
|
||
out = []
|
||
for item in items:
|
||
mode = item.get("parent_mode")
|
||
if not mode:
|
||
continue
|
||
rule = {
|
||
"@id": _key(item),
|
||
"work_item_code": item["work_item_code"],
|
||
"name": _shown_path(item, by),
|
||
"mode": mode,
|
||
"children": [_key(c) for c in children.get(item["work_item_code"], [])],
|
||
}
|
||
if mode == "sum_steps":
|
||
rule["steps"] = [
|
||
{"@id": _key(by[s["code"]]), "weight": s["weight"]} for s in item["steps"]
|
||
]
|
||
rule["basis"] = item.get("steps_basis")
|
||
out.append(rule)
|
||
return out
|
||
|
||
|
||
def spec(columns: list[str]) -> dict[str, Any]:
|
||
return {"editable": [], "locked": dict.fromkeys(columns, _LOCKED), "formula": {}}
|
||
|
||
|
||
def notice(kind: str) -> list[str]:
|
||
lines = ["공종 축은 읽기만 — 고칠 칸 없음(품셈 원문 값 풀기 뒤에 열림)"]
|
||
found = doc(kind)
|
||
if found and any(KEY_FIELD not in i for i in found[0].get("work_items") or []):
|
||
lines.append(
|
||
"불변 열쇠(FW-·CW-) 아직 없음 — 목차 코드(FP-)를 열쇠로 씀 · 품셈 개정으로 번호가 밀리면 딴 줄을 가리킬 수 있음"
|
||
)
|
||
return lines
|
||
|
||
|
||
#: 거르기 — **자료에 있는 가름만**(건설 부문 · 줄 구실 · 연결고리 표). 어느 줄을 거를지는 서버 한 곳 ·
|
||
#: 화면은 고른 값만 보냄 · 연결은 기본 「전부」(감추지 않음 — 「어디에 쓸지 모르니 건설 전체」 · 브레인) ·
|
||
#: 줄 구실만 기본 「공종」(총칙 기본 감춤 · 서브 2ee6a7ed · 브레인).
|
||
FILTER_KEYS = ("link", "division", "axis_role")
|
||
ALL = ""
|
||
#: 값을 안 보냈을 때 거는 값 — **잣대가 정한다**(`axis_policy.json` 의 `list_default`).
|
||
#: ⚠ 코드에 박지 않는다 — 총칙 범위는 기관이 정하는 것이라 관리자가 고칠 수 있어야 한다.
|
||
#: 잣대를 못 읽을 때만 쓰는 대비값(총칙 줄은 기본 감춤 · 「전부」를 고르면 보임 · 2026-09-17 브레인).
|
||
FALLBACK_DEFAULTS = {"axis_role": "work_item"}
|
||
|
||
|
||
def _facets(kind: str) -> dict[str, dict[str, str]]:
|
||
found = doc(kind)
|
||
items = found[0].get("work_items") or [] if found else []
|
||
linked = linked_keys()
|
||
return {
|
||
_key(i): {
|
||
"link": LINKED if _key(i) in linked else ALL,
|
||
"division": str(i.get("division") or ""),
|
||
"axis_role": str(i.get("axis_role") or ""),
|
||
}
|
||
for i in items
|
||
}
|
||
|
||
|
||
def _policy() -> dict[str, Any]:
|
||
"""줄 구실 잣대 — `axis_policy.json`(데스크탑 서브) 한 곳에서."""
|
||
path = FOLDER / "axis_policy.json"
|
||
return _read(str(path), path.stat().st_mtime_ns) if path.is_file() else {}
|
||
|
||
|
||
def _role_names() -> dict[str, str]:
|
||
"""거르기 상자에 뜰 이름. 목록에서는 「총칙」이 아니라 **공종이 아님**을 드러내는 이름으로 적는다."""
|
||
roles = _policy().get("roles") or {}
|
||
return {
|
||
key: str(role.get("list_label") or role.get("name_ko") or key)
|
||
for key, role in roles.items()
|
||
}
|
||
|
||
|
||
def _defaults() -> dict[str, str]:
|
||
"""안 고른 가름에 걸 값 — 잣대의 `list_default`. 못 읽으면 대비값."""
|
||
wanted = _policy().get("list_default") or {}
|
||
key = str(wanted.get("filter") or "")
|
||
return {key: str(wanted.get("value") or ALL)} if key else dict(FALLBACK_DEFAULTS)
|
||
|
||
|
||
def filters(kind: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""거르기 상자 — 값이 둘 이상 갈리는 가름만 · 기본은 **잣대**(`list_default`)가 정한다 — 그 값이 표에 있을 때만."""
|
||
facets = _facets(kind)
|
||
labels = {
|
||
"link": ("연결", {LINKED: "연결된 것만"}),
|
||
"division": ("부문", {}),
|
||
"axis_role": ("줄 구실", _role_names()),
|
||
}
|
||
out = []
|
||
for key in FILTER_KEYS:
|
||
counts: dict[str, int] = {}
|
||
for row in rows:
|
||
value = facets.get(row["@key"], {}).get(key, ALL)
|
||
counts[value] = counts.get(value, 0) + 1
|
||
values = [v for v in counts if v != ALL]
|
||
if len(counts) < 2: # 다 같은 값이면 거를 것이 없음(연결은 「연결됨」 · 빈칸 둘로 갈림)
|
||
continue
|
||
label, names = labels[key]
|
||
default = _defaults().get(key, ALL)
|
||
out.append(
|
||
{
|
||
"key": key,
|
||
"label": label,
|
||
"options": [{"value": ALL, "label": "전부", "rows": len(rows)}]
|
||
+ [{"value": v, "label": names.get(v, v), "rows": counts[v]} for v in values],
|
||
"default": default if default in values else ALL,
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def apply_filters(
|
||
kind: str, rows: list[dict[str, Any]], picked: dict[str, str]
|
||
) -> list[dict[str, Any]]:
|
||
facets = _facets(kind)
|
||
# 안 보낸 가름은 상자 기본값으로 — 상자에 보이는 값과 걸린 값이 같게(찾기 결과가 아니라 표 전체로 정함)
|
||
defaults = {f["key"]: f["default"] for f in filters(kind, base_rows(kind))}
|
||
for key in FILTER_KEYS:
|
||
value = picked.get(key, defaults.get(key, ALL))
|
||
if value != ALL:
|
||
rows = [r for r in rows if facets.get(r["@key"], {}).get(key) == value]
|
||
return rows
|
||
|
||
|
||
def extra(kind: str) -> dict[str, Any]:
|
||
"""표 수준 덧붙임 — 계산 규칙 · 줄 수 · 갈래 편 수(쪽 나누기와 무관하게 전체)."""
|
||
rows = base_rows(kind)
|
||
return {
|
||
"filters": filters(kind, rows),
|
||
"rules": rules(kind),
|
||
"stats": {
|
||
"work_items": len({r["@key"] for r in rows}),
|
||
"units": len(rows),
|
||
},
|
||
}
|
||
|
||
|
||
# ── 공종 편집 — DB 표 `master_work_item`·`master_work_item_row` 를 고치는 규칙(PLAN 1-0 계약) ──
|
||
|
||
STATUSES = ("auto_ok", "partial", "needs_input", "not_item", "reviewed")
|
||
#: 관리자가 고치는 칸 — [초기값으로] 도 이 칸만 seed 에서 되살림
|
||
HEAD_FIELDS = ("name", "basis_quantity", "basis_unit", "status")
|
||
ROW_FIELDS = (
|
||
"variant",
|
||
"resource_kind",
|
||
"resource_key",
|
||
"resource_name",
|
||
"resource_spec",
|
||
"amount",
|
||
"amount_unit",
|
||
"alternative_amount",
|
||
"group_ratio_pct",
|
||
"raw_cell",
|
||
"pum_table_id",
|
||
)
|
||
_HIDDEN = ("id", "work_item_id", "seed", "source", "notes", "is_deleted")
|
||
|
||
|
||
def revision_scope(key: str) -> str:
|
||
return f"work_item:{key}"
|
||
|
||
|
||
def _same(a: Any, b: Any) -> bool:
|
||
"""칸 값이 같은가 — 숫자는 값으로(DB `1.500000` = 요청 `1.5` = seed `1.5`) · 빈 글자 = NULL."""
|
||
numbers = (int, float, Decimal)
|
||
if isinstance(a, numbers) and isinstance(b, numbers):
|
||
return Decimal(str(a)) == Decimal(str(b))
|
||
return (None if a == "" else a) == (None if b == "" else b)
|
||
|
||
|
||
def changed_columns(current: dict[str, Any], seed: dict[str, Any] | None, fields) -> list[str]:
|
||
"""초기값과 다른 칸 — 관리자 추가 줄(seed 없음)은 빈 목록(`is_added` 로 드러남)."""
|
||
if seed is None:
|
||
return []
|
||
return [f for f in fields if f in seed and not _same(current.get(f), seed[f])]
|
||
|
||
|
||
def detail(head: dict[str, Any], rows: list[dict[str, Any]], revision: int) -> dict[str, Any]:
|
||
"""GET 응답 — 머리(+revision) · 구성 줄 · 원문 표 · [주]. seed 는 안 내보내고 고친 칸 이름만."""
|
||
seed = head.get("seed")
|
||
return {
|
||
"item": {
|
||
**{k: v for k, v in head.items() if k not in _HIDDEN},
|
||
"revision": revision,
|
||
"changed_columns": changed_columns(head, seed, HEAD_FIELDS),
|
||
},
|
||
"rows": [
|
||
{
|
||
**{k: v for k, v in row.items() if k not in _HIDDEN or k == "id"},
|
||
"is_added": row.get("seed") is None,
|
||
"changed_columns": changed_columns(row, row.get("seed"), ROW_FIELDS),
|
||
}
|
||
for row in rows
|
||
],
|
||
"source": head.get("source") or [],
|
||
"notes": head.get("notes") or [],
|
||
}
|
||
|
||
|
||
def check_rows(
|
||
rows: list[dict[str, Any]], existing_ids: set[int], known_keys: set[str]
|
||
) -> list[str]:
|
||
"""[저장] 서버 검사 중 DB 와 맞춰 볼 것 — 모양(숫자 ≥0 · 종류 · 모르는 칸)은 요청 모델이 봄."""
|
||
errors, seen = [], set()
|
||
for number, row in enumerate(rows, 1):
|
||
row_id = row.get("id")
|
||
if row_id is not None:
|
||
if row_id not in existing_ids:
|
||
errors.append(f"{number}번째 줄: 이 공종에 없는 줄(id {row_id})")
|
||
elif row_id in seen:
|
||
errors.append(f"{number}번째 줄: 같은 줄(id {row_id})이 두 번 옴")
|
||
seen.add(row_id)
|
||
key = row.get("resource_key")
|
||
if key is not None and key not in known_keys:
|
||
errors.append(f"{number}번째 줄: 기초단가에 없는 자원 열쇠 {key}")
|
||
return errors
|
||
|
||
|
||
def plan_rows(
|
||
existing: list[dict[str, Any]], incoming: list[dict[str, Any]]
|
||
) -> dict[str, list[Any]]:
|
||
"""전체 줄 보내기 → 쓸 것. existing = 지금 살아 있는 줄 · incoming 순서 = sort_order.
|
||
|
||
id 있음 → 수정(바뀐 줄만) · id 없음 → 추가 · 빠진 줄 → 초기값 줄은 숨김(`is_deleted` ·
|
||
[초기값으로] 가 살림) · 관리자 추가 줄은 지움.
|
||
"""
|
||
by_id = {row["id"]: row for row in existing}
|
||
updates, inserts = [], []
|
||
for order, row in enumerate(incoming):
|
||
values = {f: row.get(f) for f in ROW_FIELDS} | {"sort_order": order}
|
||
old = by_id.get(row.get("id"))
|
||
if old is None:
|
||
inserts.append(values)
|
||
elif any(not _same(old.get(f), v) for f, v in values.items()):
|
||
updates.append({"id": old["id"], **values})
|
||
kept = {row.get("id") for row in incoming}
|
||
gone = [row for row in existing if row["id"] not in kept]
|
||
return {
|
||
"updates": updates,
|
||
"inserts": inserts,
|
||
"hide": [row["id"] for row in gone if row.get("seed") is not None],
|
||
"drop": [row["id"] for row in gone if row.get("seed") is None],
|
||
}
|
||
|
||
|
||
def reset_plan(head: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""[초기값으로] — 머리·줄을 seed 로(seed 에 있는 칸만) · 숨긴 초기값 줄도 살림 · 추가 줄은 지움.
|
||
|
||
rows = 숨긴 줄까지 전부. 쓰기 모양은 `plan_rows` 와 같음(+ `head`).
|
||
"""
|
||
seed = head.get("seed") or {}
|
||
restore = (*ROW_FIELDS, "sort_order")
|
||
return {
|
||
"head": {f: seed[f] for f in HEAD_FIELDS if f in seed},
|
||
"updates": [
|
||
{"id": row["id"], **{f: row["seed"][f] for f in restore if f in row["seed"]}}
|
||
for row in rows
|
||
if row.get("seed") is not None
|
||
],
|
||
"inserts": [],
|
||
"hide": [],
|
||
"drop": [row["id"] for row in rows if row.get("seed") is None],
|
||
}
|