diff --git a/Z01_MasterData/Z01_MasterData_BasePrices.py b/Z01_MasterData/Z01_MasterData_BasePrices.py new file mode 100644 index 00000000..bbbcadb1 --- /dev/null +++ b/Z01_MasterData/Z01_MasterData_BasePrices.py @@ -0,0 +1,347 @@ +"""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 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_Overrides as overrides +from Z01_MasterData import Z01_MasterData_Tables as tables + +KINDS = ("labor", "machine", "material", "oil", "rate") +_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", +) +_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 None: + raise FileNotFoundError(f"이름표에 원본 자리가 없음: {file_id}") + return path + + +@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 _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, + "dataset_id": dataset, + "effective_date": doc.get("effective_date"), + **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 = doc.get("effective_date") + 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")} + common = { + "scope": v.get("scope"), + "fuel": variable, + "effective_date": doc.get("effective_date"), # 전국 · 지역 판 기준일이 다름 + **head, + } + if "records" not in v: + rows.append( + { + "@id": f"{v.get('scope')}/{variable}", + "@source": name, + **common, + "value": v.get("value"), + } + ) + for r in v.get("records") or []: + rows.append( + { + "@id": f"{v.get('scope')}/{variable}/{r['sido_code']}", + "@source": name, + **common, + **r, + } + ) + 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": doc.get("effective_date"), + **_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": doc.get("effective_date"), + "base": v.get("base"), + **item, + } + ) + return rows + + +_BUILDERS = {"labor": _labor, "material": _material, "oil": _oil, "rate": _rate} + + +def base_rows(kind: str) -> list[dict[str, Any]]: + """덮개 얹기 전 원본 줄 — 기계는 입력 칸까지(계산 칸은 덮개를 얹은 뒤 셈).""" + if kind == "machine": + doc, name = _doc("mach_base") + return machine.base_rows(name, doc.get("effective_date")) + return _BUILDERS[kind]() + + +def _machine_inputs() -> machine.Inputs: + """기계가 끌어 쓰는 밑값 — **덮개를 얹은** 노임(건설)·전국평균 유가 + 그 판들의 기준일.""" + labor = [r for r in overrides.overlay("labor", _labor()) if r["dataset_id"] == "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["value"])) for r in oil}, + oil_dates={r["fuel"]: r.get("date") or r["effective_date"] for r in oil}, + operating_date=_doc("pum_const")[0].get("effective_date"), + ) + + +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) + + +def spec(kind: str, columns: list[str]) -> dict[str, Any]: + """표 수준 — editable · locked · formula (· notice).""" + if kind == "machine": + return machine.spec(columns) + editable = { + "labor": ["daily_wage_krw"], + "material": ["price_krw"], + "oil": ["value"], + "rate": [c for c in _RATE_EDITABLE if c in columns], + }[kind] + keys = {"labor": {"dataset_id", "occupation_code"}, "material": {"item_code"}}.get(kind, set()) + if kind == "oil": + keys = {"scope", "fuel", "sido_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} + out: dict[str, Any] = {"editable": editable, "locked": locked, "formula": {}} + if kind == "rate": + out["notice"] = _RATE_NOTICE + return out + + +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 == "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('sido_name') or '전국평균'}" + 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]: + """열 이름 — 이름표 `synthetic_tables['@base_prices/{kind}']` → `columns[열key]` → 영문 key.""" + labels = tables.load_labels() + own = (labels.get("synthetic_tables") or {}).get(f"@base_prices/{kind}") or {} + named = next((c for c in own.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, + } + + +def public(row: dict[str, Any]) -> dict[str, Any]: + """내보낼 줄 — 안쪽 칸(`@source`)은 뺌.""" + return {k: v for k, v in row.items() if k != "@source"} + + +def table( + kind: str, page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE, q: str = "" +) -> 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 + ) + 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], + "rows": [public(r) for r in hits[start : start + size]], + "total": len(hits), + **spec(kind, columns), + } + + +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 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) + 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"])) + item["column_label"] = column_meta(each, item["column"])["label"] + 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)) diff --git a/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py b/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py new file mode 100644 index 00000000..ed1e3bfb --- /dev/null +++ b/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py @@ -0,0 +1,213 @@ +"""Z01 기초단가 — 기계 한 표: 기종 한 줄 + 시간당 단가(계산값) · 산출근거(2026-09-15 브레인 계약). + +기계단가는 취득가 613 · 손료계수 387 · 운전경비(연료·잡재료·조종원 — 건설품셈 8-4) · 노임 · 유가가 계산으로 물림. + 고치는 칸 = 취득가 · 손료계수(이 표) — 노임·유가는 그 표에서 고치고 여기는 **덮개 얹은 값**을 끌어 씀 + 계산 칸 = 손료 · 연료 · 조종원 · 시간당 단가 — editable 에서 빼고 locked 에 까닭 +⚠ 셈은 B09 함수를 그대로 부름(`load_machine_catalog` · `load_operating_records` · `hourly_machine_cost`) — 식을 두 벌로 안 둠. + `hourly_cost_of`(B09) 와 같은 값인지 `test_z01_base_prices.py` 가 대조. 반만 선 값(연료·조종원 빠짐)은 시간당 단가를 안 냄. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from decimal import Decimal +from functools import lru_cache +from typing import Any + +from B09_Estimation.B09_Estimation_MachineCost import ( + OPERATOR_WAGE_FORMULA, + hourly_machine_cost, + load_machine_catalog, +) +from B09_Estimation.B09_Estimation_MachineOperating import FUEL_VARIABLES, load_operating_records +from Z01_MasterData.Z01_MasterData_Tables import RESOURCES + +#: B09 셈이 여는 원본 셋 — 바뀌면 캐시를 새로 읽음 +_INPUT_FILES = ("mach_base_2026.json", "pum_const_2026.json", "labor_const_2026-01-01.json") +EDITABLE = ["price_thousand_krw", "loss_coefficient_per_hour"] +COMPUTED = ("loss_krw_per_hour", "fuel_krw_per_hour", "operator_krw_per_hour", "hourly_krw") +#: 조종원 식 글자 — B09 식 문자열(좌→우 차례)을 사람이 읽게만 바꿈 +_WAGE_TEXT = OPERATOR_WAGE_FORMULA.replace("*", " × ").replace("/", " ÷ ") +_FORMULA = { + "loss_krw_per_hour": "손료 = 취득가(천원) × 1,000 × 시간당 손료계수 (건설품셈 8-1-6)", + "fuel_krw_per_hour": "연료 = 주연료(ℓ/hr) × (1 + 잡재료%) × 전국평균 유가 (건설품셈 8-4 · 8-1-7)", + "operator_krw_per_hour": ( + f"조종원 = 일 노임 × 인원 × {_WAGE_TEXT}" + " · 원 미만 절사 (건설품셈 8-1-2 · 건협 임금적용요령 4-나)" + ), + "hourly_krw": "시간당 사용료 = 손료 + 연료 + 조종원", +} +_LOCKED_COMPUTED = "계산값 — 밑값(취득가·손료계수는 이 표 · 노임·유가는 그 표)을 고치면 따라 바뀜" +_LOCKED_OPERATING = "건설품셈 8-4 운전경비표(로직)에서 옴 — 기초단가가 아님" +_LOCKED = { + "machine_code": "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음", + "machine_name": "공표 원문 칸 — 원본 갱신으로만 바뀜", + "specification": "공표 원문 칸 — 원본 갱신으로만 바뀜", + "economic_life_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜", + "annual_standard_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜", + "fuel_kind": _LOCKED_OPERATING, + "fuel_liters_per_hour": _LOCKED_OPERATING, + "misc_material_percent": _LOCKED_OPERATING, + "operator_occupation_code": _LOCKED_OPERATING, + "operator_person_days": _LOCKED_OPERATING, + "fuel_price_krw_per_l": "유가 표에서 고침 — 여기는 덮개 얹은 전국평균을 끌어 씀", + "operator_daily_wage_krw": "노임 표에서 고침 — 여기는 덮개 얹은 건설 노임을 끌어 씀", + "hourly_note": "계산이 왜 안 섰는지 — 밑값이 비면 섬", + "effective_date": "원본 판 기준일 — 원본 갱신으로만 바뀜", + "operating_effective_date": "운전경비 원단위가 나온 건설품셈 판 기준일", + "fuel_price_date": "끌어 쓴 유가 판 기준일 — 유가 표 판", + "operator_wage_date": "끌어 쓴 노임 판 기준일 — 노임 표 판", + **dict.fromkeys(COMPUTED, _LOCKED_COMPUTED), +} + + +def _num(value: Decimal | int | None) -> int | float | None: + if value is None: + return None + value = Decimal(value) + return int(value) if value == value.to_integral_value() else float(value) + + +def _fmt(value: Decimal | int | float) -> str: + text = f"{Decimal(str(value)):,.4f}".rstrip("0").rstrip(".") + return text + + +@dataclass(frozen=True) +class Inputs: + """기계 셈이 다른 표에서 끌어 오는 값 — 덮개를 얹은 뒤 · 판 기준일과 함께.""" + + wages: dict[str, Decimal | None] + wage_date: str | None + oil: dict[str, Decimal] + oil_dates: dict[str, str | None] + operating_date: str | None + + +@lru_cache(maxsize=2) +def _cached(_mtimes: tuple[int, ...]): + return load_machine_catalog(), {r.machine_code: r for r in load_operating_records().records} + + +def _inputs(): + folder = RESOURCES / "data_cost_input_value" + return _cached(tuple((folder / name).stat().st_mtime_ns for name in _INPUT_FILES)) + + +def base_rows(source_name: str, effective_date: str | None) -> list[dict[str, Any]]: + """기종 한 줄 — 원본 칸(카탈로그가 되살린 이름·규격 · 손료계수 포함)과 운전경비 원단위.""" + catalog, records = _inputs() + rows = [] + for code, m in catalog.machines.items(): + record = records.get(code) + rows.append( + { + "@id": code, + "@source": source_name, + "machine_code": code, + "effective_date": effective_date, + "machine_name": m.name, + "specification": m.specification, + "price_thousand_krw": _num(m.price_thousand_krw), + "loss_coefficient_per_hour": _num(m.loss_coefficient_per_hour), + "economic_life_hours": m.economic_life_hours, + "annual_standard_hours": m.annual_standard_hours, + "fuel_kind": record.fuel_kind if record else None, + "fuel_liters_per_hour": _num(record.fuel_liters_per_hour) if record else None, + "misc_material_percent": _num(record.misc_material_percent) if record else None, + "operator_occupation_code": record.operator_occupation_code if record else None, + "operator_person_days": _num(record.operator_person_days) if record else None, + } + ) + return rows + + +def compute(row: dict[str, Any], inputs: Inputs) -> dict[str, Any]: + """덮개 얹은 줄 → 시간당 단가 칸을 붙인 새 줄. B09 `hourly_cost_of` 와 같은 길.""" + catalog, records = _inputs() + code = row["@id"] + coef = row["loss_coefficient_per_hour"] + record = records.get(code) + out = { + **row, + "operating_effective_date": inputs.operating_date if record is not None else None, + "fuel_price_krw_per_l": None, + "fuel_price_date": None, + "operator_daily_wage_krw": None, + "operator_wage_date": None, + **dict.fromkeys(COMPUTED), + } + if coef is None: + out["hourly_note"] = "손료계수 없음 — 시간당 단가 안 섬" + return out + spec = replace( + catalog.machines[code], + price_thousand_krw=Decimal(str(row["price_thousand_krw"])), + loss_coefficient_per_hour=Decimal(str(coef)), + ) + liters = fuel_price = wage = None + if record is not None: + liters = record.fuel_liters_per_hour + if liters is not None: + variable = FUEL_VARIABLES.get(record.fuel_kind, "") + fuel_price = inputs.oil.get(variable) + out["fuel_price_date"] = inputs.oil_dates.get(variable) + if record.misc_material_percent is not None: + liters = liters * (Decimal(1) + record.misc_material_percent / Decimal(100)) + wage = inputs.wages.get(record.operator_occupation_code) + out["operator_daily_wage_krw"] = _num(wage) + out["operator_wage_date"] = inputs.wage_date + if wage is not None and record.operator_person_days is not None: + wage = wage * record.operator_person_days # 인원 칸이 비면 B09 처럼 일 노임 그대로 + cost = hourly_machine_cost( + spec, + fuel_liters_per_hour=liters if fuel_price is not None else None, + fuel_price_per_liter=fuel_price, + operator_daily_wage=wage, + ) + money = cost.money + out["fuel_price_krw_per_l"] = _num(fuel_price) + out["loss_krw_per_hour"] = _num(money.expense) + formula = { + "loss_krw_per_hour": ( + f"{_fmt(money.expense)} = {_fmt(spec.price_thousand_krw)}천원 × 1,000 × {spec.loss_coefficient_per_hour}" + ) + } + if fuel_price is not None: + out["fuel_krw_per_hour"] = _num(money.material) + misc = record.misc_material_percent if record else None + formula["fuel_krw_per_hour"] = ( + f"{_fmt(money.material)} = {_fmt(record.fuel_liters_per_hour)}ℓ" + + (f" × (1 + {_fmt(misc)}%)" if misc is not None else "") + + f" × {_fmt(fuel_price)}원/ℓ" + ) + if wage is not None: + out["operator_krw_per_hour"] = _num(money.labor) + persons = record.operator_person_days + formula["operator_krw_per_hour"] = ( + f"{_fmt(money.labor)} = {_fmt(out['operator_daily_wage_krw'])}원" + + (f" × {_fmt(persons)}인" if persons is not None else "") + + f" × {_WAGE_TEXT} (원 미만 절사)" + ) + if cost.gaps: + out["hourly_note"] = " · ".join(cost.gaps) + else: + out["hourly_krw"] = _num(money.total) + out["hourly_note"] = None + formula["hourly_krw"] = ( + f"{_fmt(money.total)} = {_fmt(money.expense)} + {_fmt(money.material)} + {_fmt(money.labor)}" + ) + out["@formula"] = formula + return out + + +def spec(columns: list[str]) -> dict[str, Any]: + return { + "editable": list(EDITABLE), + "locked": { + c: _LOCKED.get(c, "공표 원문 칸 — 원본 갱신으로만 바뀜") + for c in columns + if c not in EDITABLE + }, + "formula": dict(_FORMULA), + } diff --git a/Z01_MasterData/Z01_MasterData_Overrides.py b/Z01_MasterData/Z01_MasterData_Overrides.py new file mode 100644 index 00000000..5beeb6db --- /dev/null +++ b/Z01_MasterData/Z01_MasterData_Overrides.py @@ -0,0 +1,166 @@ +"""Z01 기초단가 덮개 — 사람이 고친 값만 담는 층(2026-09-15 브레인 틀 · 사용자 지시 「기초단가를 고칠 수 있게」). + + 원본 resources/data_*/… 읽기만 — 주기 갱신은 이 파일을 통째로 갈아끼움 + 덮개 resources/data_master_override/{kind}.json 고친 것만 — 주기가 달라 kind 마다 한 파일 + 읽기 원본 줄 위에 덮개를 얹음 · 줄 수준 `@overrides[열] = {original, value, state, current}` + +state 셋(판정은 **여기 한 곳** — 화면이 따로 재지 않음): + edited 고쳐짐 — 원본 그대로(덮개를 쓸 때 본 원본 = 지금 원본) + source_changed 고쳐짐 — 원본이 바뀜 · ⚠ 덮개 값을 그대로 씀(자동으로 원본을 쓰면 금액이 말없이 바뀜 — 사람이 정함) + orphan 주인 없음 — 갱신으로 그 줄·칸이 사라짐 · 버리지 않고 모아 보기에 남김 +되돌리기(「새 원본 값으로 받기」 도 같음) = 값 null → 그 칸을 덮개에서 뺌. +""" + +from __future__ import annotations + +import json +import math +import threading +from datetime import datetime +from decimal import Decimal +from pathlib import Path +from typing import Any + +from common_util.common_util_json import atomic_write_json + +OVERRIDE_DIR = Path(__file__).resolve().parent.parent / "resources" / "data_master_override" +EDITED, SOURCE_CHANGED, ORPHAN = "edited", "source_changed", "orphan" +STATE_ORDER = (SOURCE_CHANGED, ORPHAN, EDITED) # 모아 보기 차례 — 확인할 것이 맨 위 +_LOCK = threading.Lock() # ponytail: 한 프로세스 잠금 — 서버를 여럿 띄우면 파일 잠금이 필요 + + +def _path(kind: str) -> Path: + return OVERRIDE_DIR / f"{kind}.json" + + +def load(kind: str) -> dict[str, dict[str, dict[str, Any]]]: + """`@id` → 열 → {value, original, source, at, by}. 파일이 없으면 빈 덮개.""" + path = _path(kind) + if not path.is_file(): + return {} + return json.loads(path.read_text(encoding="utf-8")).get("overrides") or {} + + +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 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 mark(original: Any, value: Any, current: Any) -> dict[str, Any]: + state = EDITED if same(original, current) else SOURCE_CHANGED + return {"original": original, "value": value, "state": state, "current": current} + + +def overlay(kind: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """원본 줄(바꾸지 않음) 위에 덮개를 얹은 **새 줄들**.""" + store = load(kind) + out = [] + for row in rows: + entries = store.get(row["@id"]) + if not entries: + out.append(row) + continue + row = dict(row) + marks = {} + for column, entry in entries.items(): + if column not in row: + continue # 주인 없는 칸 — 모아 보기에만 + marks[column] = mark(entry.get("original"), entry.get("value"), row[column]) + row[column] = entry.get("value") + if marks: + row["@overrides"] = marks + out.append(row) + return out + + +def write(kind: str, base_row: dict[str, Any], values: dict[str, Any], by: Any) -> None: + """고친 값을 덮개에 씀 — null 이거나 지금 원본과 같으면 그 칸을 뺌. 원본은 안 건드림.""" + row_id = base_row["@id"] + with _LOCK: + store = load(kind) + entries = dict(store.get(row_id) or {}) + for column, value in values.items(): + current = base_row.get(column) + if value is None or same(value, current): + entries.pop(column, None) + continue + entries[column] = { + "value": value, + "original": current, # 이 값을 보고 고침 — 갱신 뒤 원본과 달라지면 source_changed + "source": base_row.get("@source", ""), + "at": datetime.now().astimezone().isoformat(timespec="seconds"), + "by": by, + } + if entries: + store[row_id] = entries + else: + store.pop(row_id, None) + _save(kind, store) + + +def _save(kind: str, store: dict[str, dict[str, dict[str, Any]]]) -> None: + atomic_write_json( + _path(kind), + { + "schema_version": "1.0", + "kind": kind, + "note": "사람이 고친 기초단가만 — 원본 파일은 그대로 · Z01 마스터 데이터 화면이 씀", + "overrides": store, + }, + ) + + +def clear(kind: str, state: str, base_rows: list[dict[str, Any]]) -> int: + """한 kind 의 그 state 덮개를 한꺼번에 뺌 — 품목 퇴출로 여럿이 주인을 잃는 갱신 뒤(브레인 ②). 뺀 수.""" + with _LOCK: + doomed = {(i["row_id"], i["column"]) for i in items(kind, base_rows) if i["state"] == state} + if not doomed: + return 0 + store = load(kind) + for row_id, column in doomed: + store[row_id].pop(column, None) + if not store[row_id]: + del store[row_id] + _save(kind, store) + return len(doomed) + + +def items(kind: str, base_rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """모아 보기 한 kind 몫 — 주인 없는 것까지 · 차례는 부르는 쪽이 STATE_ORDER 로.""" + by_id = {r["@id"]: r for r in base_rows} + out = [] + for row_id, entries in load(kind).items(): + row = by_id.get(row_id) + for column, entry in entries.items(): + if row is None or column not in row: + current, state = None, ORPHAN + else: + current = row[column] + state = mark(entry.get("original"), entry.get("value"), current)["state"] + out.append( + { + "kind": kind, + "row_id": row_id, + "column": column, + "original": entry.get("original"), + "current": current, + "value": entry.get("value"), + "state": state, + } + ) + return out diff --git a/Z01_MasterData/Z01_MasterData_Router.py b/Z01_MasterData/Z01_MasterData_Router.py index e3e1e315..31b9e53c 100644 --- a/Z01_MasterData/Z01_MasterData_Router.py +++ b/Z01_MasterData/Z01_MasterData_Router.py @@ -1,19 +1,32 @@ -"""Z01 마스터 데이터 라우터 — 읽기 전용(고치기는 다음 차례 · 2026-09-15 브레인). +"""Z01 마스터 데이터 라우터 — 읽기 + 기초단가 고치기(2026-09-15 브레인). GET /api/master-data/tree 갈래 → 파일 → 표 GET /api/master-data/rows?file=&table=&page=&size=&q= 표 줄(쪽 나누기 · 검색) + GET /api/master-data/base-prices/{kind}?page=&size=&q= 기초단가 한 표(labor|machine|material|oil|rate) + PUT /api/master-data/base-prices/{kind}/{row_id} {values:{열:값}} → 덮개에만 씀 · null = 되돌리기 + GET /api/master-data/overrides 고친 것·원본 바뀐 것·주인 없는 것(서버 정렬) ⚠ 권한은 등록하는 쪽(`main.py` · 랩탑 서브)이 `dependencies=[verify_session, require_system_admin]` 로 붙임. """ from __future__ import annotations -from fastapi import APIRouter, HTTPException +from typing import Any, Literal +from fastapi import APIRouter, Depends, HTTPException +from fastapi.responses import JSONResponse +from pydantic import BaseModel + +from common_util.common_util_auth import verify_session +from Z01_MasterData import Z01_MasterData_BasePrices as base_prices from Z01_MasterData import Z01_MasterData_Tables as tables router = APIRouter(prefix="/api/master-data", tags=["Z01 MasterData"]) +class BasePriceEdit(BaseModel): + values: dict[str, Any] + + @router.get("/tree") def get_tree() -> dict: return tables.tree() @@ -27,3 +40,44 @@ def get_rows( if result is None: raise HTTPException(status_code=404, detail="없는 파일이나 표입니다.") return result + + +def _kind(kind: str) -> str: + if kind not in base_prices.KINDS: + raise HTTPException(status_code=404, detail=f"없는 기초단가: {kind}") + return kind + + +@router.get("/base-prices/{kind}") +def get_base_prices( + kind: str, page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE, q: str = "" +) -> dict: + return base_prices.table(_kind(kind), page=page, size=size, q=q) + + +@router.put("/base-prices/{kind}/{row_id:path}") +def put_base_price( + kind: str, row_id: str, body: BasePriceEdit, session: dict = Depends(verify_session) +) -> JSONResponse: + status, payload = base_prices.edit(_kind(kind), row_id, body.values, session.get("user_id")) + if status != 200: + raise HTTPException(status_code=status, detail=payload["message"]) + return JSONResponse(content=payload) + + +@router.get("/overrides") +def get_overrides( + kind: str = "", state: str = "", page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE +) -> dict: + return base_prices.override_items(kind and _kind(kind), state, page=page, size=size) + + +class OverrideClear(BaseModel): + kind: str + state: Literal["edited", "source_changed", "orphan"] + + +@router.post("/overrides/clear") +def clear_overrides(body: OverrideClear, _session: dict = Depends(verify_session)) -> dict: + """한 kind 의 그 state 덮개를 한꺼번에 뺌 — kind·state 둘 다 있어야 함(통째로 비우기 막음).""" + return {"removed": base_prices.clear_overrides(_kind(body.kind), body.state)} diff --git a/resources/tester/test_z01_base_prices.py b/resources/tester/test_z01_base_prices.py new file mode 100644 index 00000000..5c904de5 --- /dev/null +++ b/resources/tester/test_z01_base_prices.py @@ -0,0 +1,411 @@ +"""Z01 기초단가 다섯(노임·기계·자재·유가·요율) — 고칠 수 있게 · 2026-09-15 브레인 계약(사용자 지시). + +틀(브레인 못박음): 원본 `resources/data_*` 는 **읽기만** · 고친 값은 덮개 +`resources/data_master_override/{kind}.json`(주기가 달라 kind 마다 한 파일)에만 · 읽을 때 원본 위에 얹음 · +덮개 줄마다 original(원래 값)·source(원본 파일) — 원본이 바뀌면 「원본 바뀜」 · 줄이 사라지면 「주인 없음」(버리지 않음) · +기계 시간당 단가는 계산값 — editable 에서 빼고 왜 못 고치나(locked) · 밑값(취득가·손료계수·노임·유가)을 고치면 따라 바뀜. +""" + +from __future__ import annotations + +import hashlib +import json +from decimal import Decimal +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from Z01_MasterData import Z01_MasterData_Overrides as overrides +from Z01_MasterData import Z01_MasterData_Router as router_module + +ROOT = Path(__file__).resolve().parents[2] +SOURCES = [ + ROOT / "resources/data_cost_input_value" / name + for name in ( + "labor_const_2026-01-01.json", + "labor_mfg_2026-07-01.json", + "mach_base_2026.json", + "mat_price_public_2026-08-14.json", + "oil_2026-08-14.json", + "oil_regional_2026-09-09.json", + "rates_2026.json", + ) +] + + +@pytest.fixture +def store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + folder = tmp_path / "data_master_override" + monkeypatch.setattr(overrides, "OVERRIDE_DIR", folder) + return folder + + +@pytest.fixture +def client(store: Path) -> TestClient: + from common_util.common_util_auth import verify_session + + app = FastAPI() + app.include_router(router_module.router) + app.dependency_overrides[verify_session] = lambda: {"user_id": 42, "role": "SYSTEM_ADMIN"} + return TestClient(app) + + +def _get(client: TestClient, kind: str, **params) -> dict: + res = client.get(f"/api/master-data/base-prices/{kind}", params=params) + assert res.status_code == 200, res.text + return res.json() + + +def _row(client: TestClient, kind: str, row_id: str) -> dict: + rows = _get(client, kind, q=row_id.rsplit("/", 1)[-1], size=500)["rows"] + return next(r for r in rows if r["@id"] == row_id) + + +def _put(client: TestClient, kind: str, row_id: str, values: dict): + return client.put(f"/api/master-data/base-prices/{kind}/{row_id}", json={"values": values}) + + +def _sha(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> None: + expected = { + "labor": (261, ["daily_wage_krw"]), + "material": (6999, ["price_krw"]), + "oil": (36, ["value"]), + "machine": (613, ["price_thousand_krw", "loss_coefficient_per_hour"]), + } + for kind, (total, editable) in expected.items(): + table = _get(client, kind, size=500) + assert table["total"] == total and table["editable"] == editable, kind + ids = [r["@id"] for r in _get(client, kind, size=500)["rows"]] + assert len(ids) == len(set(ids)), kind + assert all("@overrides" not in r for r in table["rows"]), ( + kind + ) # 안 고친 줄엔 칸 자체가 없음 + columns = {c["key"] for c in table["columns"]} + assert not {k for k in columns if k.startswith("@")}, kind # 줄 수준 칸은 열 목록 밖 + assert set(table["locked"]) <= columns and not set(table["locked"]) & set(editable), kind + assert _row(client, "labor", "labor_const/1001")["daily_wage_krw"] == 215907 + assert _row(client, "labor", "labor_mfg/1")["occupation_name"] == "CAD설계사(기계)" + assert _row(client, "oil", "national_average/oil_diesel")["value"] == 1846.39 + assert _row(client, "oil", "regional/oil_diesel/01")["sido_name"] == "서울" + rate = _get(client, "rate", size=500) + assert rate["total"] == 218 and "rate_percent" in rate["editable"] + by_id = {r["@id"]: r for r in rate["rows"]} + assert by_id["rate_sanjae"]["rate_percent"] == 3.56 + goyong = ( + "rate_goyong/brackets/estimated_amount_bracket=gte_140_billion;grade=1" # 구간 칸 이름 차례 + ) + assert by_id[goyong]["rate_percent"] == 1.57 and by_id[goyong]["base"] == "total_labor_cost" + material = _get(client, "material", q="육각볼트", size=5) + assert 0 < material["total"] < 6999 and len(material["rows"]) == 5 + + +def test_기계는_시간당_단가를_계산값으로_잠그고_산출근거를_실음(client: TestClient) -> None: + table = _get(client, "machine", q="0101-0007", size=5) + assert "hourly_krw" in table["locked"] and "hourly_krw" in table["formula"] + row = next(r for r in table["rows"] if r["@id"] == "0101-0007") + parts = row["loss_krw_per_hour"] + row["fuel_krw_per_hour"] + row["operator_krw_per_hour"] + assert row["hourly_krw"] == pytest.approx(parts) + assert ( + "hourly_krw" in row["@formula"] + and f"{int(row['hourly_krw']):,}" in row["@formula"]["hourly_krw"] + ) + res = _put(client, "machine", "0101-0007", {"hourly_krw": 1}) + assert res.status_code == 400 and table["locked"]["hourly_krw"] in res.text + + +def test_기계_계산값은_B09_시간당_사용료와_같음(client: TestClient) -> None: + from B09_Estimation.B09_Estimation_MachineOperating import ( + hourly_cost_of, + load_operating_records, + ) + + codes = sorted(r.machine_code for r in load_operating_records().records)[::9] + rows = {r["@id"]: r for r in _get(client, "machine", size=500, page=1)["rows"]} + rows |= {r["@id"]: r for r in _get(client, "machine", size=500, page=2)["rows"]} + checked = 0 + from B09_Estimation.B09_Estimation_MachineCost import MachineCostError + + for code in codes: + try: + cost = hourly_cost_of(code) + except MachineCostError: # 손료계수 없는 기종 — B09 도 못 셈 + assert rows[code]["hourly_krw"] is None and rows[code]["hourly_note"], code + continue + if cost.gaps: + assert rows[code]["hourly_krw"] is None, code # 반만 선 값은 안 냄 + continue + assert rows[code]["hourly_krw"] == pytest.approx(float(cost.money.total)), code + checked += 1 + assert checked >= 5 + + +def test_고친_값은_덮개에만_쓰고_원본은_안_건드림(client: TestClient, store: Path) -> None: + before = {p: _sha(p) for p in SOURCES} + res = _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000}) + assert res.status_code == 200, res.text + row = res.json() + assert row["daily_wage_krw"] == 230000 + assert row["@overrides"] == { + "daily_wage_krw": { + "original": 215907, + "value": 230000, + "state": overrides.EDITED, + "current": 215907, + } + } + assert ( + _row(client, "labor", "labor_const/1001")["@overrides"]["daily_wage_krw"]["value"] == 230000 + ) + assert _put(client, "oil", "national_average/oil_diesel", {"value": 1900}).status_code == 200 + assert sorted(p.name for p in store.iterdir()) == [ + "labor.json", + "oil.json", + ] # kind 마다 한 파일 + saved = json.loads((store / "labor.json").read_text(encoding="utf-8")) + entry = saved["overrides"]["labor_const/1001"]["daily_wage_krw"] + assert ( + entry["original"] == 215907 + and entry["source"] == "labor_const_2026-01-01.json" + and entry["by"] == 42 + ) + assert {p: _sha(p) for p in SOURCES} == before + + +def test_밑값을_고치면_기계_계산값이_따라_바뀜(client: TestClient) -> None: + base = _row(client, "machine", "0101-0007") + operator = base["operator_occupation_code"] + wage = _row(client, "labor", f"labor_const/{operator}")["daily_wage_krw"] + _put(client, "labor", f"labor_const/{operator}", {"daily_wage_krw": wage + 80000}) + _put(client, "oil", "national_average/oil_diesel", {"value": 2000}) + _put(client, "machine", "0101-0007", {"price_thousand_krw": base["price_thousand_krw"] * 2}) + after = _row(client, "machine", "0101-0007") + assert after["operator_krw_per_hour"] > base["operator_krw_per_hour"] + assert after["fuel_krw_per_hour"] > base["fuel_krw_per_hour"] + assert after["loss_krw_per_hour"] == pytest.approx(base["loss_krw_per_hour"] * 2) + price = base["price_thousand_krw"] + assert after["@overrides"] == { + "price_thousand_krw": { + "original": price, + "value": price * 2, + "state": overrides.EDITED, + "current": price, + } + } # 노임·유가 덮개는 그 표 줄에 서고 기계 줄은 계산만 바뀜 + + +def test_null_은_되돌리기_같은_값도_덮개를_안_남김(client: TestClient, store: Path) -> None: + _put(client, "material", "10023392", {"price_krw": 30}) + reverted = _put(client, "material", "10023392", {"price_krw": None}).json() + assert reverted["price_krw"] == 22 and "@overrides" not in reverted + assert _put(client, "material", "10023392", {"price_krw": 22}).status_code == 200 + saved = json.loads((store / "material.json").read_text(encoding="utf-8")) + assert saved["overrides"] == {} + + +def test_원본이_바뀐_덮개는_덮개_값을_쓰고_세게_드러냄_주인_없는_덮개도_남김( + client: TestClient, store: Path +) -> None: + store.mkdir() + (store / "labor.json").write_text( + json.dumps( + { + "schema_version": "1.0", + "kind": "labor", + "overrides": { + "labor_const/1001": { + "daily_wage_krw": { + "value": 199000, + "original": 200000, + "source": "labor_const_2025-07-01.json", + } + }, + "labor_const/9999": { + "daily_wage_krw": { + "value": 1, + "original": 1, + "source": "labor_const_2025-07-01.json", + } + }, + }, + } + ), + encoding="utf-8", + ) + # 브레인 ③ — 자동으로 원본을 쓰면 금액이 말없이 바뀜 → 사람이 고친 값 유지 · 갈렸다는 것만 또렷이 + row = _row(client, "labor", "labor_const/1001") + assert row["daily_wage_krw"] == 199000 + mark = row["@overrides"]["daily_wage_krw"] + assert mark == { + "original": 200000, + "value": 199000, + "state": overrides.SOURCE_CHANGED, + "current": 215907, + } + _put(client, "labor", "labor_const/1002", {"daily_wage_krw": 170000}) + listed = client.get("/api/master-data/overrides").json() + assert [ + (i["row_id"], i["state"]) for i in listed["items"] + ] == [ # 서버가 정렬 — 원본 바뀜이 맨 위 + ("labor_const/1001", overrides.SOURCE_CHANGED), + ("labor_const/9999", overrides.ORPHAN), + ("labor_const/1002", overrides.EDITED), + ] + first = listed["items"][0] + assert set(first) == { + "kind", + "row_id", + "row_label", + "column", + "column_label", + "original", + "current", + "value", + "state", + } + assert first["row_label"] == "작업반장" and ( + first["original"], + first["current"], + first["value"], + ) == (200000, 215907, 199000) + assert listed["items"][1]["current"] is None + saved = json.loads((store / "labor.json").read_text(encoding="utf-8"))["overrides"] + assert "labor_const/9999" in saved # 다른 줄을 고쳐도 주인 없는 덮개는 안 지움 + # 같은 값을 다시 넣으면 「지금 원본을 보고 고침」 으로 굳음 · null 이면 새 원본 값으로 받기 + kept = _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 199000}).json() + assert kept["@overrides"]["daily_wage_krw"] == { + "original": 215907, + "value": 199000, + "state": overrides.EDITED, + "current": 215907, + } + taken = _put(client, "labor", "labor_const/1001", {"daily_wage_krw": None}).json() + assert taken["daily_wage_krw"] == 215907 and "@overrides" not in taken + + +def test_빈_칸에_새로_넣은_값도_덮개_원본에_값이_생기면_원본_바뀜( + client: TestClient, store: Path +) -> None: + blank = next(r for r in _get(client, "labor", size=500)["rows"] if r["daily_wage_krw"] is None) + row = _put(client, "labor", blank["@id"], {"daily_wage_krw": 150000}).json() + assert row["@overrides"]["daily_wage_krw"] == { + "original": None, + "value": 150000, + "state": overrides.EDITED, + "current": None, + } + saved = json.loads((store / "labor.json").read_text(encoding="utf-8")) + saved["overrides"][blank["@id"]]["daily_wage_krw"]["original"] = ( + 140000 # 옛 판엔 값이 있었다고 치면 + ) + (store / "labor.json").write_text(json.dumps(saved), encoding="utf-8") + assert ( + _row(client, "labor", blank["@id"])["@overrides"]["daily_wage_krw"]["state"] + == overrides.SOURCE_CHANGED + ) + + +def test_요율_표는_법정값_경고를_실음(client: TestClient) -> None: + rate = _get(client, "rate", size=1) + assert "갱신" in rate["notice"] and "주인" in rate["notice"] + assert "notice" not in _get(client, "material", size=1) + + +def test_막는_자리(client: TestClient) -> None: + assert _put(client, "wage", "x", {"daily_wage_krw": 1}).status_code == 404 + assert _put(client, "labor", "labor_const/0000", {"daily_wage_krw": 1}).status_code == 404 + assert _put(client, "labor", "labor_const/1001", {"occupation_name": "x"}).status_code == 400 + locked = _put(client, "labor", "labor_const/1001", {"hours_per_day": 9}) # 숫자여도 잠긴 칸 + assert locked.status_code == 400 and "공표 원문" in locked.text + for bad in (True, -1, "220000", float("nan")): + body = json.dumps({"values": {"daily_wage_krw": bad}}).encode() + res = client.put( + "/api/master-data/base-prices/labor/labor_const/1001", + content=body, + headers={"content-type": "application/json"}, + ) + assert res.status_code in (400, 422), bad + fee = next( + r + for r in _get(client, "rate", size=500)["rows"] + if r["@id"].startswith("rate_performance_guarantee_fee/brackets/") + ) + assert ( + _put(client, "rate", fee["@id"], {"rate_percent": 1}).status_code == 400 + ) # 그 줄엔 요율 칸이 없음 + assert Decimal(str(_row(client, "rate", "rate_sanjae")["rate_percent"])) == Decimal("3.56") + + +def test_합친_표는_줄마다_판_기준일을_실음(client: TestClient) -> None: + """브레인 ① — 노임 두 판(건설 01-01 · 제조 07-01)·유가 두 판(전국 08-14 · 지역 09-09)이 한 표에 섞임.""" + assert _row(client, "labor", "labor_const/1001")["effective_date"] == "2026-01-01" + assert _row(client, "labor", "labor_mfg/1")["effective_date"] == "2026-07-01" + assert _row(client, "oil", "national_average/oil_diesel")["effective_date"] == "2026-08-14" + assert _row(client, "oil", "regional/oil_diesel/01")["effective_date"] == "2026-09-09" + assert _row(client, "material", "10023392")["effective_date"] == "2026-08-14" + assert _row(client, "rate", "rate_sanjae")["effective_date"] == "2026-04-13" + m = _row(client, "machine", "0101-0007") + assert (m["effective_date"], m["operating_effective_date"]) == ("2026-01-01", "2026-01-01") + assert (m["fuel_price_date"], m["operator_wage_date"]) == ("2026-08-14", "2026-01-01") + no_record = next(r for r in _get(client, "machine", size=500)["rows"] if r["fuel_kind"] is None) + assert no_record["operating_effective_date"] is None and no_record["hourly_krw"] is None + + +def test_주인_없는_덮개는_모아서_한꺼번에_다룸(client: TestClient, store: Path) -> None: + """브레인 ② — 자재·기계는 품목 퇴출이 상시라 갱신 한 번에 여럿이 주인을 잃음.""" + store.mkdir() + gone = { + f"9999000{i}": {"price_krw": {"value": 1, "original": 1, "source": "old.json"}} + for i in range(3) + } + (store / "material.json").write_text( + json.dumps({"schema_version": "1.0", "kind": "material", "overrides": gone}), + encoding="utf-8", + ) + _put(client, "material", "10023392", {"price_krw": 30}) + _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000}) + listed = client.get( + "/api/master-data/overrides", params={"state": overrides.ORPHAN, "kind": "material"} + ).json() + assert listed["total"] == 3 and {i["state"] for i in listed["items"]} == {overrides.ORPHAN} + assert listed["counts"] == { + "material": {overrides.ORPHAN: 3, overrides.EDITED: 1}, + "labor": {overrides.EDITED: 1}, + } + paged = client.get("/api/master-data/overrides", params={"size": 2, "page": 2}).json() + assert paged["total"] == 5 and len(paged["items"]) == 2 + everything = client.get("/api/master-data/overrides").json()[ + "items" + ] # labor 가 kind 차례로 앞이어도 + assert [i["state"] for i in everything] == [overrides.ORPHAN] * 3 + [overrides.EDITED] * 2 + res = client.post( + "/api/master-data/overrides/clear", json={"kind": "material", "state": overrides.ORPHAN} + ) + assert res.status_code == 200 and res.json()["removed"] == 3 + left = client.get("/api/master-data/overrides").json() + assert [(i["kind"], i["state"]) for i in left["items"]] == [ # 같은 state 안은 kind 차례 + ("labor", overrides.EDITED), + ("material", overrides.EDITED), + ] + assert ( + client.post("/api/master-data/overrides/clear", json={"kind": "material"}).status_code + == 422 + ) # 거름 없이 통째로 못 비움 + + +def test_줄은_있는데_칸이_사라진_덮개도_주인_없음(client: TestClient, store: Path) -> None: + rates = _get(client, "rate", size=500)["rows"] + fee = next(r for r in rates if r["@id"].startswith("rate_performance_guarantee_fee/brackets/")) + store.mkdir() + entry = {"rate_percent": {"value": 1, "original": 0.5, "source": "old.json"}} + (store / "rate.json").write_text( + json.dumps({"overrides": {fee["@id"]: entry}}), encoding="utf-8" + ) + items = client.get("/api/master-data/overrides", params={"kind": "rate"}).json()["items"] + assert [(i["row_id"], i["state"]) for i in items] == [(fee["@id"], overrides.ORPHAN)] + assert "@overrides" not in _row(client, "rate", fee["@id"]) # 없는 칸을 새로 만들지 않음