Merge remote-tracking branch 'origin/sub_laptop_6' into sub_laptop_8

B09_Estimation/B09_Estimation_Router.py

@@ -69,6 +69,7 @@ from B08_Quantity.B08_Quantity_Milestone_Calc import compute_milestone
 from B08_Quantity.B08_Quantity_Router_Earthwork import earthwork_table
 from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
 from B08_Quantity.B08_Quantity_Router_Material import material_sheets
+from B08_Quantity.B08_Quantity_Router_Material import once as material_once
 from B08_Quantity.B08_Quantity_Router_Milestone import read_data as read_milestone
 from B08_Quantity.B08_Quantity_Router_Milestone import site_info
 from B09_Estimation import B09_Estimation_Bill_Library as library
@@ -95,6 +96,7 @@ from common_util.common_util_auth import verify_session
 from common_util.common_util_project_settings import reset_section
 from common_util.common_util_storage import resolve_stored_project_path
 from common_util.common_util_workflow_state import mark_stage
+from M01_MasterData import M01_MasterData_Store_Cache as m01_cache
 from M02_MasterTemplete import M02_Layer_Access as access
 from M02_MasterTemplete.M02_Item_Snapshot import importer
 from M02_MasterTemplete.M02_MasterTemplete_Router_Items import _call, _root
@@ -128,17 +130,19 @@ async def _b08_for(project_id: UUID, notes: list[str], reuse: bool) -> dict[str,
 
 
 async def _b08(project_id: UUID, notes: list[str]) -> dict[str, Any]:
-    """B08 수량 원천 — 못 읽으면 빈칸 + 알림(B09 는 B08 를 고치지 않음)."""
+    """B08 수량 원천 — 못 읽으면 빈칸 + 알림(B09 는 B08 를 고치지 않음) ·
+    자재 집계표 한 번(토공 채집석 공제가 같은 답 · `material_once`)."""
     out: dict[str, Any] = {}
-    for key, name, read in (
-        ("materials", "자재 집계표", material_sheets),
-        ("earthwork", "토공집계표", earthwork_table),
-    ):
-        try:
-            out[key] = await read(project_id)
-        except Exception as error:  # noqa: BLE001 — 원천 하나가 막혀도 나머지 표는 그림
-            notes.append(f"B08 {name} 못 읽음 — {getattr(error, 'detail', error)}")
-            out[key] = None
+    with material_once():
+        for key, name, read in (
+            ("materials", "자재 집계표", material_sheets),
+            ("earthwork", "토공집계표", earthwork_table),
+        ):
+            try:
+                out[key] = await read(project_id)
+            except Exception as error:  # noqa: BLE001 — 원천 하나가 막혀도 나머지 표는 그림
+                notes.append(f"B08 {name} 못 읽음 — {getattr(error, 'detail', error)}")
+                out[key] = None
     # 구조물 집계표 = 자재 집계표가 이미 부른 값(열 · 계 · 조합) — 두 번 안 부름
     out["summary"] = (out["materials"] or {}).get("source")
     # 총괄 · 콘크리트 표 = B08 답의 `sheets` 한 층 아래(재료비목록표 · 운반 수량이 읽음 · 167-2)
@@ -146,12 +150,15 @@ async def _b08(project_id: UUID, notes: list[str]) -> dict[str, Any]:
     return out
 
 
-def compute_sheets(root: Path, inputs: dict[str, Any], user: dict[str, Any]) -> dict[str, Any]:
+def compute_sheets(
+    root: Path, inputs: dict[str, Any], user: dict[str, Any], memo: dict[str, Any] | None = None
+) -> dict[str, Any]:
     """짝 셈 한 벌 — 바탕(144) → 산근(145-1 · 서버 단독) → 대가(145-2) → 내역(145-3) → 총괄(145-4).
 
-    화면 짝 = `B09_Estimation_UI_Shell.computeSheets`(산근만 서버 답 그대로)."""
+    화면 짝 = `B09_Estimation_UI_Shell.computeSheets`(산근만 서버 답 그대로) ·
+    memo = 두 번 셈의 앞 산근 풀이(`Sangeun.collect`)."""
     base = compute_base(inputs, user)
-    sg = sangeun.build(root, base, user)  # 145-1 — 굳힌 단가로 풀이(+ 166-2 손 줄 로직)
+    sg = sangeun.build(root, base, user, memo)  # 145-1 — 굳힌 단가로 풀이(+ 166-2 손 줄 로직)
     by_key = {row["key"]: row for row in sg["sg_list"]}
     labor = compute_labor(
         inputs.get("daega"), {**base, "gaps": []}, by_key, user.get("daega_cover")
@@ -171,13 +178,15 @@ def compute_sheets(root: Path, inputs: dict[str, Any], user: dict[str, Any]) ->
 def compute_twice(root: Path, inputs: dict[str, Any], user: dict[str, Any]) -> tuple[dict, dict]:
     """183-1 두 번 셈 — 1차 셈으로 쓰는 자재만 남기고(`used_materials`) 재료비목록표 수량 =
     1차 역전개 총수량(`material_qty` = 재료비수량금액집계표 수량) → 2차 셈 · 집계표.
-    `inputs` 를 고쳐 둠(화면 짝이 거른 입력 · 수량을 그대로 받음)."""
-    first = compute_sheets(root, inputs, user)
+    `inputs` 를 고쳐 둠(화면 짝이 거른 입력 · 수량을 그대로 받음) ·
+    2차 산근 = 1차가 본 단가가 같으면 1차 풀이 그대로(`memo`)."""
+    memo: dict[str, Any] = {}
+    first = compute_sheets(root, inputs, user, memo)
     summary = compute_summary(first, hopyo_lines(first))
     keep = used_materials(first, inputs, user)
     inputs["materials"] = [m for m in inputs.get("materials") or [] if m["key"] in keep]
     inputs["material_qty"] = {row["key"]: row["qty"] for row in summary["sum_material"]}
-    sheets = compute_sheets(root, inputs, user)
+    sheets = compute_sheets(root, inputs, user, memo)
     return sheets, compute_summary(sheets, hopyo_lines(sheets))
 
 
@@ -218,11 +227,27 @@ async def _demand(root: Path, draft: dict[str, Any] | None = None) -> list[str]:
 
 
 async def payload(
-    project_id: UUID, root: Path, draft: dict[str, Any] | None = None, reuse_b08: bool = False
+    project_id: UUID,
+    root: Path,
+    draft: dict[str, Any] | None = None,
+    reuse_b08: bool = False,
+    keys: list[str] | None = None,
+) -> dict[str, Any]:
+    """셈 답 한 벌 — keys = 방금 센 단가 수요(같은 저장 값 · [저장]) · 마스터 자국은 한 번만."""
+    with m01_cache.steady():
+        return await _payload(project_id, root, draft, reuse_b08, keys)
+
+
+async def _payload(
+    project_id: UUID,
+    root: Path,
+    draft: dict[str, Any] | None,
+    reuse_b08: bool,
+    keys: list[str] | None,
 ) -> dict[str, Any]:
     notes: list[str] = []
     b08 = await _b08_for(project_id, notes, reuse_b08)
-    keys = await _demand(root, draft)
+    keys = await _demand(root, draft) if keys is None else keys
     if not keys:
         notes.append(
             "연결된 구조물 조합 사본 없음 — 노무 · 경비 · 중기 단가 줄은 사본이 연결되면 생김"
@@ -268,6 +293,11 @@ async def estimation(project_id: UUID) -> dict[str, Any]:
 @router.put("/{project_id}/estimation")
 async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) -> dict[str, Any]:
     root = await _project_root(project_id)
+    with m01_cache.steady():
+        return await _save(project_id, root, body)
+
+
+async def _save(project_id: UUID, root: Path, body: dict[str, Any]) -> dict[str, Any]:
     user = await asyncio.to_thread(store.write_user, root, body.get("user"))
     calc = await asyncio.to_thread(calc_user, root)  # 263-2 운반 값 사본 = 지금 거리
     if (rows := calc["bill"].get("rows")) != ((user.get("bill") or {}).get("rows") or []):
@@ -279,8 +309,11 @@ async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) ->
     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)
-    await mark_stage(str(project_id), 6, "save")  # [저장](B1)
-    return await payload(project_id, root)
+    # [저장](B1) 단계 표시는 셈과 함께 · 수요 = 방금 센 것(저장 값 · 사본 그대로)
+    out, _ = await asyncio.gather(
+        payload(project_id, root, keys=keys), mark_stage(str(project_id), 6, "save")
+    )
+    return out
 
 
 @router.post("/{project_id}/estimation/complete")

B09_Estimation/B09_Estimation_Sangeun.py

@@ -30,6 +30,7 @@
 
 from __future__ import annotations
 
+import copy
 import hashlib
 import json
 import re
@@ -148,13 +149,49 @@ def rows_of(folder: Path, whole: Any = None) -> list[dict[str, Any]]:
 
 
 # ── 풀이 ──────────────────────────────────────────────────────────────
+class _Seen(dict):
+    """읽은 키를 적는 단가 표 — 두 번 셈(183-1)에서 첫 풀이가 본 값만 견줌."""
+
+    def __init__(self, *args: Any) -> None:
+        super().__init__(*args)
+        self.seen: set[Any] = set()
+
+    def __contains__(self, key: object) -> bool:
+        self.seen.add(key)
+        return super().__contains__(key)
+
+    def __getitem__(self, key: Any) -> Any:
+        self.seen.add(key)
+        return super().__getitem__(key)
+
+    def get(self, key: Any, default: Any = None) -> Any:
+        self.seen.add(key)
+        return super().get(key, default)
+
+    def same(self, other: dict) -> bool:
+        """본 키마다 있음 · 값이 `other` 와 같음."""
+        return all(
+            dict.__contains__(self, k) == dict.__contains__(other, k)
+            and dict.get(self, k) == dict.get(other, k)
+            for k in self.seen
+        )
+
+
 class _Prices:
     """B09 표 단가 — 노무 · 중기(노 · 재 · 경) · 자재 적용 단가(키 → 값)."""
 
     def __init__(self, sheets: dict[str, Any]) -> None:
-        self.labor = {r["key"]: num(r.get("price")) for r in sheets.get("labor") or []}
-        self.machine = {r["key"]: r for r in sheets.get("machine") or []}
-        self.material = {r["key"]: num(r.get("applied")) for r in sheets.get("compare") or []}
+        self.labor = _Seen({r["key"]: num(r.get("price")) for r in sheets.get("labor") or []})
+        self.machine = _Seen({r["key"]: r for r in sheets.get("machine") or []})
+        self.material = _Seen(
+            {r["key"]: num(r.get("applied")) for r in sheets.get("compare") or []}
+        )
+
+    def same(self, other: _Prices) -> bool:
+        """첫 풀이가 본 단가가 `other` 에서도 같음 — 같으면 풀이도 같음."""
+        return all(
+            getattr(self, k).same(getattr(other, k)) for k in ("labor", "machine", "material")
+        )
 
 
 def _unit(code: str) -> tuple[str, Decimal]:
@@ -538,10 +575,14 @@ def collect(
     entries: list[tuple[str, Path]],
     sheets: dict[str, Any],
     hands: list[tuple[str, str, dict[str, Any]]] | None = None,
+    memo: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """`entries` = [(구조물 id, 사본 폴더)] → 블록 · 목록 · 「연결 없음」 ·
-    `hands` = 손 줄 로직(166-2 · 사본 뒤 번호 · 마스터 로직)."""
+    `hands` = 손 줄 로직(166-2 · 사본 뒤 번호 · 마스터 로직) ·
+    memo = 같은 사본 · 손 줄의 앞 풀이(두 번 셈) — 앞 풀이가 본 단가가 같으면 그 풀이 사본."""
     solver = Solver(sheets)
+    if memo and memo["prices"].same(solver.prices):
+        return copy.deepcopy(memo["sg"])
     for owner, folder in entries:
         whole = item_prices.load_prices(read_json(folder / PRICE_FILE) or {})
         for src in rows_of(folder, whole):
@@ -558,20 +599,27 @@ def collect(
         for b in blocks
         if b["sum"] is None
     ]
-    return {
+    out = {
         "sg": blocks,
         "sg_list": listing,
         "sg_gaps": gaps,
         "sg_notes": [NO_SOURCE] + ([] if blocks else [EMPTY]),
         "sg_hand": hand_keys,
     }
+    if memo is not None:
+        memo.update(prices=solver.prices, sg=copy.deepcopy(out))
+    return out
 
 
 def build(
-    project_root: str | Path, sheets: dict[str, Any], user: dict[str, Any] | None = None
+    project_root: str | Path,
+    sheets: dict[str, Any],
+    user: dict[str, Any] | None = None,
+    memo: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """연결된 사본 전부(구조물 id 차례 → 설계내역서 서고 사본 · 702-3)의 산근 +
-    설계내역서 손 줄 로직(166-2 · 뒤 번호) — 차례 = `Labor_Source.daega_inputs`."""
+    설계내역서 손 줄 로직(166-2 · 뒤 번호) — 차례 = `Labor_Source.daega_inputs` ·
+    memo = 같은 root · user 두 번 셈의 앞 풀이(`collect`)."""
     from B09_Estimation.B09_Estimation_Bill_Library import copies
 
     base = structures_dir(project_root)
@@ -583,4 +631,4 @@ def build(
     entries += [
         (f"서고 {(m.get('원본') or {}).get('name') or f.name}", f) for f, m in copies(project_root)
     ]
-    return collect(entries, sheets, hand_sources(user))
+    return collect(entries, sheets, hand_sources(user), memo)