diff --git a/Z01_MasterData/Z01_MasterData_BasePrices.py b/Z01_MasterData/Z01_MasterData_BasePrices.py index b8bd4a3a..0b5c2d8e 100644 --- a/Z01_MasterData/Z01_MasterData_BasePrices.py +++ b/Z01_MasterData/Z01_MasterData_BasePrices.py @@ -3,7 +3,7 @@ 표 수준(한 번): columns · total · editable · locked{열: 왜 못 고치나} · formula{열: 식} · notice(요율만) 줄 수준(해당할 때만): `@id`(자료 제 열쇠) · `@overrides`(고친 줄) · `@formula`(계산값 줄) `@id` — 줄 차례가 아니라 자료 열쇠(브레인 승인): 노임 `{dataset_id}/{occupation_code}` · 기계 machine_code · - 자재 item_code · 유가 `{scope}/{변수}[/{sido_code}]` · 요율 `{변수}[/{목록}/{구간칸=값;…}]`(구간 칸 이름 차례 · 값·식 칸 뺌) + 자재 item_code · 유가 `{scope}/{변수}[/{sido_code}]`(원본 값 그대로 · 열 이름은 이름표 것) · 요율 `{변수}[/{목록}/{구간칸=값;…}]`(구간 칸 이름 차례 · 값·식 칸 뺌) 원본 파일 자리는 이름표 파일이 가리키는 곳(`Z01_MasterData_Tables`) — 원본은 읽기만, 고친 값은 덮개(`Z01_MasterData_Overrides`). """ @@ -35,6 +35,10 @@ _RATE_EDITABLE = ( "manager_thresholds.default_estimated_amount_krw", "manager_thresholds.civil_main_work_estimated_amount_krw", ) +_SCOPE_NOTICE = ( + "여기서 고친 값은 **새로 만드는 프로젝트부터** 쓰임 — 이미 만든 프로젝트 금액은 안 바뀜" + "(프로젝트는 만들 때 뜬 사본을 봄)" +) _RATE_NOTICE = "법이 정한 값 — 고치면 원본 갱신 때 구간 이름이 바뀌어 고친 값이 주인을 잃기 쉬움(주인 없음으로 남음)" @@ -98,7 +102,7 @@ def sources(kind: str) -> list[dict[str, Any]]: "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 "", + "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, @@ -125,7 +129,7 @@ def _labor() -> list[dict[str, Any]]: row = { "@id": f"{dataset}/{r['occupation_code']}", "@source": name, - "dataset_id": dataset, + "source": dataset, "effective_date": edition_of(doc), **r, } @@ -148,9 +152,9 @@ def _oil() -> list[dict[str, Any]]: 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")} + head = {k: x for k, x in v.items() if k not in ("records", "value", "scope")} common = { - "scope": v.get("scope"), + "source": v.get("scope"), "fuel": variable, "effective_date": edition_of(doc), # 전국 · 지역 판 기준일이 다름 **head, @@ -161,7 +165,7 @@ def _oil() -> list[dict[str, Any]]: "@id": f"{v.get('scope')}/{variable}", "@source": name, **common, - "value": v.get("value"), + "price_krw_per_l": v.get("value"), } ) for r in v.get("records") or []: @@ -170,7 +174,9 @@ def _oil() -> list[dict[str, Any]]: "@id": f"{v.get('scope')}/{variable}/{r['sido_code']}", "@source": name, **common, - **r, + "region_code": r["sido_code"], + "region_name": r["sido_name"], + "price_krw_per_l": r["value"], } ) return rows @@ -238,7 +244,7 @@ def base_rows(kind: str) -> list[dict[str, Any]]: def _machine_inputs() -> machine.Inputs: """기계가 끌어 쓰는 밑값 — **덮개를 얹은** 노임(건설)·전국평균 유가 + 그 판들의 기준일.""" - labor = [r for r in overrides.overlay("labor", _labor()) if r["dataset_id"] == "labor_const"] + 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']}" ] @@ -250,7 +256,7 @@ def _machine_inputs() -> machine.Inputs: 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={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]), ) @@ -272,29 +278,35 @@ def _columns(kind: str, all_rows: list[dict[str, Any]]) -> list[str]: return list(keys) +def notice(kind: str) -> list[str]: + """표 수준 알림 — 화면이 따로 가지면 문구가 두 벌이 됨(브레인 ③).""" + return [_SCOPE_NOTICE] + ([_RATE_NOTICE] if kind == "rate" else []) + + def spec(kind: str, columns: list[str]) -> dict[str, Any]: - """표 수준 — editable · locked · formula (· notice).""" + """표 수준 — editable · locked · formula.""" if kind == "machine": return machine.spec(columns) editable = { "labor": ["daily_wage_krw"], "material": ["price_krw"], - "oil": ["value"], + "oil": ["price_krw_per_l"], "rate": [c for c in _RATE_EDITABLE if c in columns], }[kind] - keys = {"labor": {"dataset_id", "occupation_code"}, "material": {"item_code"}}.get(kind, set()) + keys = {"labor": {"source", "occupation_code"}, "material": {"item_code"}}.get(kind, set()) if kind == "oil": - keys = {"scope", "fuel", "sido_code"} + 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} - out: dict[str, Any] = {"editable": editable, "locked": locked, "formula": {}} - if kind == "rate": - out["notice"] = _RATE_NOTICE - return out + return { + "editable": editable, + "locked": locked, + "formula": {}, + } def row_label(kind: str, row: dict[str, Any] | None) -> str: @@ -307,7 +319,7 @@ def row_label(kind: str, row: dict[str, Any] | None) -> str: 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 '전국평균'}" + return f"{row.get('source_product_name') or row.get('fuel')} {row.get('region_name') or '전국평균'}" named = (tables.load_labels().get("value_overrides") or {}).get( f"rates::variables/{row['variable']}" ) or {} @@ -316,10 +328,12 @@ def row_label(kind: str, row: dict[str, Any] | None) -> str: def column_meta(kind: str, key: str) -> dict[str, Any]: - """열 이름 — 이름표 `synthetic_tables['@base_prices/{kind}']` → `columns[열key]` → 영문 key.""" + """열 이름 — 이름표 `merged_tables[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) + 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, @@ -356,6 +370,7 @@ def table( "rows": [public(r) for r in hits[start : start + size]], "total": len(hits), **spec(kind, columns), + "notice": notice(kind), "source": sources(kind), } diff --git a/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py b/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py index ed1e3bfb..350a84c8 100644 --- a/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py +++ b/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py @@ -25,17 +25,17 @@ 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") +COMPUTED = ("hourly_loss_krw", "hourly_fuel_krw", "hourly_operator_krw", "hourly_total_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": ( + "hourly_loss_krw": "손료 = 취득가(천원) × 1,000 × 시간당 손료계수 (건설품셈 8-1-6)", + "hourly_fuel_krw": "연료 = 주연료(ℓ/hr) × (1 + 잡재료%) × 전국평균 유가 (건설품셈 8-4 · 8-1-7)", + "hourly_operator_krw": ( f"조종원 = 일 노임 × 인원 × {_WAGE_TEXT}" " · 원 미만 절사 (건설품셈 8-1-2 · 건협 임금적용요령 4-나)" ), - "hourly_krw": "시간당 사용료 = 손료 + 연료 + 조종원", + "hourly_total_krw": "시간당 사용료 = 손료 + 연료 + 조종원", } _LOCKED_COMPUTED = "계산값 — 밑값(취득가·손료계수는 이 표 · 노임·유가는 그 표)을 고치면 따라 바뀜" _LOCKED_OPERATING = "건설품셈 8-4 운전경비표(로직)에서 옴 — 기초단가가 아님" @@ -45,11 +45,11 @@ _LOCKED = { "specification": "공표 원문 칸 — 원본 갱신으로만 바뀜", "economic_life_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜", "annual_standard_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜", - "fuel_kind": _LOCKED_OPERATING, - "fuel_liters_per_hour": _LOCKED_OPERATING, + "fuel_type": _LOCKED_OPERATING, + "fuel_rate_l_per_hour": _LOCKED_OPERATING, "misc_material_percent": _LOCKED_OPERATING, "operator_occupation_code": _LOCKED_OPERATING, - "operator_person_days": _LOCKED_OPERATING, + "operator_person_per_day": _LOCKED_OPERATING, "fuel_price_krw_per_l": "유가 표에서 고침 — 여기는 덮개 얹은 전국평균을 끌어 씀", "operator_daily_wage_krw": "노임 표에서 고침 — 여기는 덮개 얹은 건설 노임을 끌어 씀", "hourly_note": "계산이 왜 안 섰는지 — 밑값이 비면 섬", @@ -112,11 +112,11 @@ def base_rows(source_name: str, effective_date: str | None) -> list[dict[str, An "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, + "fuel_type": record.fuel_kind if record else None, + "fuel_rate_l_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, + "operator_person_per_day": _num(record.operator_person_days) if record else None, } ) return rows @@ -167,24 +167,24 @@ def compute(row: dict[str, Any], inputs: Inputs) -> dict[str, Any]: ) money = cost.money out["fuel_price_krw_per_l"] = _num(fuel_price) - out["loss_krw_per_hour"] = _num(money.expense) + out["hourly_loss_krw"] = _num(money.expense) formula = { - "loss_krw_per_hour": ( + "hourly_loss_krw": ( 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) + out["hourly_fuel_krw"] = _num(money.material) misc = record.misc_material_percent if record else None - formula["fuel_krw_per_hour"] = ( + formula["hourly_fuel_krw"] = ( 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) + out["hourly_operator_krw"] = _num(money.labor) persons = record.operator_person_days - formula["operator_krw_per_hour"] = ( + formula["hourly_operator_krw"] = ( f"{_fmt(money.labor)} = {_fmt(out['operator_daily_wage_krw'])}원" + (f" × {_fmt(persons)}인" if persons is not None else "") + f" × {_WAGE_TEXT} (원 미만 절사)" @@ -192,9 +192,9 @@ def compute(row: dict[str, Any], inputs: Inputs) -> dict[str, Any]: if cost.gaps: out["hourly_note"] = " · ".join(cost.gaps) else: - out["hourly_krw"] = _num(money.total) + out["hourly_total_krw"] = _num(money.total) out["hourly_note"] = None - formula["hourly_krw"] = ( + formula["hourly_total_krw"] = ( f"{_fmt(money.total)} = {_fmt(money.expense)} + {_fmt(money.material)} + {_fmt(money.labor)}" ) out["@formula"] = formula diff --git a/resources/tester/test_z01_base_prices.py b/resources/tester/test_z01_base_prices.py index 7e27ab9c..22c58a84 100644 --- a/resources/tester/test_z01_base_prices.py +++ b/resources/tester/test_z01_base_prices.py @@ -75,7 +75,7 @@ def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> No expected = { "labor": (261, ["daily_wage_krw"]), "material": (6999, ["price_krw"]), - "oil": (36, ["value"]), + "oil": (36, ["price_krw_per_l"]), "machine": (613, ["price_thousand_krw", "loss_coefficient_per_hour"]), } for kind, (total, editable) in expected.items(): @@ -91,8 +91,8 @@ def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> No 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"] == "서울" + assert _row(client, "oil", "national_average/oil_diesel")["price_krw_per_l"] == 1846.39 + assert _row(client, "oil", "regional/oil_diesel/01")["region_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"]} @@ -107,16 +107,16 @@ def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> No 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"] + assert "hourly_total_krw" in table["locked"] and "hourly_total_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) + parts = row["hourly_loss_krw"] + row["hourly_fuel_krw"] + row["hourly_operator_krw"] + assert row["hourly_total_krw"] == pytest.approx(parts) assert ( - "hourly_krw" in row["@formula"] - and f"{int(row['hourly_krw']):,}" in row["@formula"]["hourly_krw"] + "hourly_total_krw" in row["@formula"] + and f"{int(row['hourly_total_krw']):,}" in row["@formula"]["hourly_total_krw"] ) - res = _put(client, "machine", "0101-0007", {"hourly_krw": 1}) - assert res.status_code == 400 and table["locked"]["hourly_krw"] in res.text + res = _put(client, "machine", "0101-0007", {"hourly_total_krw": 1}) + assert res.status_code == 400 and table["locked"]["hourly_total_krw"] in res.text def test_기계_계산값은_B09_시간당_사용료와_같음(client: TestClient) -> None: @@ -135,12 +135,12 @@ def test_기계_계산값은_B09_시간당_사용료와_같음(client: TestClien try: cost = hourly_cost_of(code) except MachineCostError: # 손료계수 없는 기종 — B09 도 못 셈 - assert rows[code]["hourly_krw"] is None and rows[code]["hourly_note"], code + assert rows[code]["hourly_total_krw"] is None and rows[code]["hourly_note"], code continue if cost.gaps: - assert rows[code]["hourly_krw"] is None, code # 반만 선 값은 안 냄 + assert rows[code]["hourly_total_krw"] is None, code # 반만 선 값은 안 냄 continue - assert rows[code]["hourly_krw"] == pytest.approx(float(cost.money.total)), code + assert rows[code]["hourly_total_krw"] == pytest.approx(float(cost.money.total)), code checked += 1 assert checked >= 5 @@ -162,7 +162,10 @@ def test_고친_값은_덮개에만_쓰고_원본은_안_건드림(client: TestC 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 ( + _put(client, "oil", "national_average/oil_diesel", {"price_krw_per_l": 1900}).status_code + == 200 + ) assert sorted(p.name for p in store.iterdir()) == [ "labor.json", "oil.json", @@ -182,12 +185,12 @@ def test_밑값을_고치면_기계_계산값이_따라_바뀜(client: TestClien 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, "oil", "national_average/oil_diesel", {"price_krw_per_l": 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) + assert after["hourly_operator_krw"] > base["hourly_operator_krw"] + assert after["hourly_fuel_krw"] > base["hourly_fuel_krw"] + assert after["hourly_loss_krw"] == pytest.approx(base["hourly_loss_krw"] * 2) price = base["price_thousand_krw"] assert after["@overrides"] == { "price_thousand_krw": { @@ -313,8 +316,11 @@ def test_빈_칸에_새로_넣은_값도_덮개_원본에_값이_생기면_원 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) + assert any("갱신" in line and "주인" in line for line in rate["notice"]) + for kind in ("labor", "machine", "material", "oil", "rate"): # 브레인 ③ — 문구는 서버가 한 벌로 + notice = _get(client, kind, size=1)["notice"] + assert any("새로 만드는 프로젝트" in line and "이미 만든" in line for line in notice), kind + assert len(rate["notice"]) == 2 and len(_get(client, "material", size=1)["notice"]) == 1 def test_막는_자리(client: TestClient) -> None: @@ -353,8 +359,8 @@ def test_합친_표는_줄마다_판_기준일을_실음(client: TestClient) -> 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 + no_record = next(r for r in _get(client, "machine", size=500)["rows"] if r["fuel_type"] is None) + assert no_record["operating_effective_date"] is None and no_record["hourly_total_krw"] is None def test_주인_없는_덮개는_모아서_한꺼번에_다룸(client: TestClient, store: Path) -> None: @@ -473,3 +479,20 @@ def test_모아_보기_항목에_단위(client: TestClient) -> None: _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000}) (item,) = client.get("/api/master-data/overrides").json()["items"] assert item["unit"] == "원" and item["column_label"] == "일 노임" + + +def test_열_이름은_이름표에서만_옴(client: TestClient) -> None: + """브레인 ② — 이름·판정은 한 곳에만. 합친 표 열 이름은 이름표 `merged_tables` 가 정본.""" + labels = json.loads( + (ROOT / "resources/data_master_labels/labels_2026-01-01.json").read_text(encoding="utf-8") + ) + merged = {t["key"]: t for t in labels["merged_tables"]} + for kind in ("labor", "material", "oil"): + named = {c["key"]: c for c in merged[kind]["columns"]} + for column in _get(client, kind, size=1)["columns"]: + if column["key"] in named: + assert column["label"] == named[column["key"]]["name_ko"], (kind, column["key"]) + assert column["unit"] == named[column["key"]]["unit"] + assert not [c for c in _get(client, kind, size=1)["columns"] if c["label"] == c["key"]], ( + kind + )