feat(Z01): 공종 편집 저장소·API — 목록·상세·전체 줄 저장·초기값으로 (PLAN 1-3)

- 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
This commit is contained in:
2026-09-18 23:47:33 +09:00
co-authored by Claude Opus 5
parent 9629430aff
commit e1c71c3e8b
5 changed files with 702 additions and 0 deletions
+132
View File
@@ -7,11 +7,15 @@
— 빠지면 발파암(절취 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
@@ -292,3 +296,131 @@ def extra(kind: str) -> dict[str, Any]:
"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],
}