diff --git a/B09_Estimation/B09_Estimation_Linked.py b/B09_Estimation/B09_Estimation_Linked.py new file mode 100644 index 000000000..8bf747390 --- /dev/null +++ b/B09_Estimation/B09_Estimation_Linked.py @@ -0,0 +1,71 @@ +"""B09 연결된 사본 → 원가 수요 자원 키(PLAN 144 · 143-1 원천) — 서버 단독 · 읽기만. + +연결된 사본 = `M02_Item_Snapshot.linked_snapshots`(지정 ∩ 같은 항목 사본 · B08 · B09 공용) · +사본마다 `일위대가.json` 담은 로직을 그 사본 단가 묶음(`단가.json`) · 입력값 · 고른자재로 돌리며 +실제로 부른 자원 줄(인력 LB · 기계 EQ · 자재 MT · 유가 MO)만 모음 — 단가 묶음 통째(고르기 후보 전부)는 +안 씀 · 값은 쓰지 않음(단가는 원가 한 벌 `price_book.json` 이 맡음). +구조물 조합 사본 없으면 그 구조물은 없음 · 일위대가 없는 사본은 자원 없음. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from M02_MasterTemplete import M02_Item_Prices as item_prices +from M02_MasterTemplete.M02_Item_Snapshot import ( + PICKS_FILE, + VALUES_FILE, + linked_snapshots, + structures_dir, +) +from M02_MasterTemplete.M02_Item_Source import LABOR_FILE, PRICE_FILE, read_json + +#: 원가 단가 목록에 오르는 자원 표 +RESOURCE_TABLES = ("LB", "EQ", "MT", "MO") + +mf = item_prices.mf + + +def used_keys(folder: Path) -> list[str]: + """사본 하나가 담은 로직을 돌릴 때 부른 자원 키 — 부른 차례 · 겹침 없음.""" + labor = read_json(folder / LABOR_FILE) + if not isinstance(labor, dict): + return [] + snapshot = item_prices.load_prices(read_json(folder / PRICE_FILE) or {}) + values = read_json(folder / VALUES_FILE) or {} + picks = read_json(folder / PICKS_FILE) or {} + seen: list[str] = [] + base_get = snapshot.get + + def get(ref: str) -> dict[str, Any]: + row = base_get(ref) + key = str(row.get("키") or "") + if key[:2] in RESOURCE_TABLES and key not in seen: + seen.append(key) + return row + + snapshot.get = get # 이 사본 묶음 한 벌에만 — 마스터 캐시는 안 건드림 + for key in item_prices.mcb.logic_keys(labor.get("줄") or {}): + if not key: + continue + given = {name: (one or {}).get("값") for name, one in (values.get(key) or {}).items()} + try: + with mf.mp.choosing((picks or {}).get(key)): + mf.run(snapshot, key, given) + except (mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError): + continue # 못 돈 로직 — 부른 데까지만 + return seen + + +def linked_resource_keys(project_root: str | Path) -> list[str]: + """연결된 사본 전부가 부른 자원 키 — 구조물 id 차례 · 겹침 없음.""" + base = structures_dir(project_root) + out: list[str] = [] + for _owner, linked in sorted(linked_snapshots(project_root).items()): + if "일위대가" not in linked.get("부분", []): + continue + for key in used_keys(base / linked["폴더"]): + if key not in out: + out.append(key) + return out diff --git a/B09_Estimation/B09_Estimation_Prices.py b/B09_Estimation/B09_Estimation_Prices.py index 03ecb38b0..967a0e423 100644 --- a/B09_Estimation/B09_Estimation_Prices.py +++ b/B09_Estimation/B09_Estimation_Prices.py @@ -4,8 +4,7 @@ [단가 갱신] 때만 새로 · 대가 · 산근도 이 단가로 셈(한 직종 · 한 자재 = 한 단가). 줄 = 수요 따라(사용자 승인 ① · 10-9) — 엑셀 목록을 기본으로 깔지 않음: - 연결된 사본이 쓰는 자원 키(`linked_resource_keys` — 143-1 「연결된 사본」 원천 · 145-1 · 2 호표가 - 들어오면 그 결과를 돌려줌 · 그전엔 빈 목록 = 빈 표 「연결 없음」) + 연결된 사본이 부른 자원 키(`B09_Estimation_Linked` — 143-1 「연결된 사본」 원천 · 없으면 빈 표) + 기계가 부르는 운전사(조종원직종 → 인력 원문번호) · 연료(주연료종류 → 유가 「종류:전국평균」) + B08 총괄자재수량집계표 자재(M01 짝 없음 → 출처 빈칸 · 적용 단가는 사용자 입력) 운전사 시간 계수 = M01 CC000336(시간환산 · 상여계수 · 휴지계수 분자 · 분모 곱). @@ -30,15 +29,6 @@ FUEL_REGION = "전국평균" MT_SOURCES = ("관급", "물가자료", "유통물가", "물가정보", "거래가격") -def linked_resource_keys(project_root: str | Path) -> list[str]: - """연결된 사본(구조물 조합 사본)이 쓰는 자원 키(LB · EQ · MT · MO) — 자리만. - - 143-1 「연결된 사본」 원천 함수 · 145-1 · 2 산근 · 대가 호표가 들어오면 그 자원 키를 돌려줌 · - 그전엔 빈 목록(구조물 조합 사본 없으면 그 구조물은 없음 · 연결 안 된 값은 비움). - """ - return [] - - def _f(value: Any) -> float | None: if isinstance(value, Decimal): return float(value) diff --git a/B09_Estimation/B09_Estimation_Router.py b/B09_Estimation/B09_Estimation_Router.py index d46f5fbe9..cb94b6692 100644 --- a/B09_Estimation/B09_Estimation_Router.py +++ b/B09_Estimation/B09_Estimation_Router.py @@ -22,6 +22,7 @@ from B08_Quantity.B08_Quantity_Router_Hydro import _project_root from B08_Quantity.B08_Quantity_Router_Material import material_sheets from B09_Estimation import B09_Estimation_Prices as prices from B09_Estimation import B09_Estimation_Store as store +from B09_Estimation.B09_Estimation_Linked import linked_resource_keys from B09_Estimation.B09_Estimation_Calc_Base import compute_base router = APIRouter(prefix="/api/projects", tags=["B09 Estimation"]) @@ -41,7 +42,7 @@ async def _b08(project_id: UUID, notes: list[str]) -> dict[str, Any]: async def payload(project_id: UUID, root: Path) -> dict[str, Any]: notes: list[str] = [] b08 = await _b08(project_id, notes) - keys = prices.linked_resource_keys(root) + keys = await asyncio.to_thread(linked_resource_keys, root) if not keys: notes.append( "연결된 구조물 조합 사본 없음 — 노무 · 경비 · 중기 단가 줄은 사본이 연결되면 생김" @@ -68,7 +69,7 @@ async def estimation(project_id: UUID) -> dict[str, Any]: async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) -> dict[str, Any]: root = await _project_root(project_id) await asyncio.to_thread(store.write_user, root, body.get("user")) - keys = prices.linked_resource_keys(root) + keys = await asyncio.to_thread(linked_resource_keys, root) book = await asyncio.to_thread(store.read_book, root) book = await asyncio.to_thread(prices.extend_book, book, keys) await asyncio.to_thread(store.write_book, root, book) @@ -78,7 +79,7 @@ async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) -> @router.post("/{project_id}/estimation/prices") async def refresh_prices(project_id: UUID) -> dict[str, Any]: root = await _project_root(project_id) - keys = prices.linked_resource_keys(root) + keys = await asyncio.to_thread(linked_resource_keys, root) book = await asyncio.to_thread(prices.refresh_book, keys) await asyncio.to_thread(store.write_book, root, book) return await payload(project_id, root) diff --git a/resources/tester/test_144_b09_skeleton.py b/resources/tester/test_144_b09_skeleton.py index 913befed5..d24f7ce29 100644 --- a/resources/tester/test_144_b09_skeleton.py +++ b/resources/tester/test_144_b09_skeleton.py @@ -112,3 +112,22 @@ def test_연결_없으면_빈_표와_연결_없음(tmp_path, monkeypatch): assert any("연결된 구조물 조합 사본 없음" in n for n in body["notes"]) assert body["book"] is None # GET 은 아무것도 쓰지 않음 assert not (tmp_path / "B09_Estimation").exists() + + +def test_연결된_사본이_부른_자원만(tmp_path): + """사본 단가 묶음 통째(고르기 후보 기계 100여)가 아니라 로직이 실제로 부른 줄만.""" + from B09_Estimation.B09_Estimation_Linked import used_keys + from M02_MasterTemplete import M02_Item_Prices as item_prices + + bundle = item_prices.to_json(item_prices.collect(["GC000268"])) # 기계 시간당 기계경비 + files = { + "단가.json": bundle, + "일위대가.json": {"줄": {"종류": "일위대가", "담은로직": [{"로직": "GC000268"}]}}, + "입력값.json": {"GC000268": {"기계": {"값": "0101-0019"}, "지역": {"값": "전국평균"}}}, + } + for name, data in files.items(): + (tmp_path / name).write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + assert used_keys(tmp_path) == ["EQ000004", "MO000002", "LB000048"] + (tmp_path / "입력값.json").write_text("{}", encoding="utf-8") + assert used_keys(tmp_path) == [] # 입력 빈 로직은 못 돎 — 부른 줄 없음 + assert used_keys(tmp_path / "없음") == []