diff --git a/B06_Section/B06_Section_Repository_Bulk.py b/B06_Section/B06_Section_Repository_Bulk.py index 72ebf0944..4cf3262e1 100644 --- a/B06_Section/B06_Section_Repository_Bulk.py +++ b/B06_Section/B06_Section_Repository_Bulk.py @@ -42,7 +42,23 @@ async def merge_cross_section_designs( 행이 없는 측점은 `project_id` 가 오면 새로 만든다 — 구조물(비정규) 측점은 B05 확정이 파일만 쓰고 DB 행을 안 만들기 때문이다(한 행짜리 함수와 같은 규칙). """ - if not entries: + return await merge_cross_section_design_batches( + connection, route_id=route_id, batches=[(entries, replace, project_id)] + ) + + +async def merge_cross_section_design_batches( + connection: aiomysql.Connection, + *, + route_id: int, + batches: list[tuple[list[tuple[float, dict[str, Any]]], bool, UUID | None]], +) -> int: + """여러 벌(`(entries, replace, project_id)`)을 차례로 얹되 읽기 한 번 · 쓰기 한 번(342-2). + + 앞 벌이 만든 새 행도 뒤 벌이 찾는다(따로 부르던 때와 같은 결과). 값이 그대로인 행은 쓰지 + 않는다 — 같은 입력의 [저장]·[확정]이 행을 다시 쓰지 않게. + """ + if not any(entries for entries, _, _ in batches): return 0 async with connection.cursor() as cursor: @@ -56,8 +72,11 @@ async def merge_cross_section_designs( for row_id, _chainage, raw in rows: data = json.loads(raw) if isinstance(raw, str) else raw existing[int(row_id)] = data if isinstance(data, dict) else {} + original = dict(existing) # 행 dict 는 제자리에서 안 고침(아래가 늘 사본을 만듦) — 대조 기준 # 같은 측점이 여러 행이면 뒤에 온 것(=큰 id)을 쓴다 — 한 행짜리 함수의 `ORDER BY id DESC`. ordered = [(float(chainage), int(row_id)) for row_id, chainage, _ in rows] + # 새 행 = 음수 id · 뒤 벌이 찾음(있는 행이 먼저 · 새 행끼리는 먼저 만든 것) · 끝에 INSERT + fresh: dict[int, tuple[str, float]] = {} def find(chainage_m: float) -> int | None: best: int | None = None @@ -68,40 +87,44 @@ async def merge_cross_section_designs( # 행마다 한 벌 — 같은 행에 여러 줄이 오면 앞 줄 결과 위에 얹음(`CASE id` 는 첫 WHEN 만 씀 · # 318-4 옛 조정값 비움 뒤 새 값) - updates: dict[int, str] = {} - inserts: list[tuple[str, int, float, str]] = [] - for chainage_m, payload in entries: - if not payload and not replace: - continue - # 얹기의 None = 그 열쇠 지움(JSON merge-patch · 318-4 관 종류 바꿈 옛 조정값) - drop = [] if replace else [key for key, value in payload.items() if value is None] - if drop: - payload = {key: value for key, value in payload.items() if value is not None} - row_id = find(chainage_m) - if row_id is None: - if project_id is None or not (payload or replace): + for entries, replace, project_id in batches: + for chainage_m, payload in entries: + if not payload and not replace: continue - inserts.append( - ( - str(project_id), - route_id, - chainage_m, - json.dumps({"design": payload}, ensure_ascii=False), - ) - ) - continue - data = dict(existing[row_id]) - if replace: - data["design"] = payload - else: - design = data.get("design") - design = dict(design) if isinstance(design, dict) else {} - design.update(payload) - for key in drop: - design.pop(key, None) - data["design"] = design - existing[row_id] = data - updates[row_id] = json.dumps(data, ensure_ascii=False) + # 얹기의 None = 그 열쇠 지움(JSON merge-patch · 318-4 관 종류 바꿈 옛 조정값) + drop = [] if replace else [key for key, value in payload.items() if value is None] + if drop: + payload = {key: value for key, value in payload.items() if value is not None} + row_id = find(chainage_m) + if row_id is None: + if project_id is None or not (payload or replace): + continue + row_id = -(len(fresh) + 1) + fresh[row_id] = (str(project_id), chainage_m) + ordered.append((chainage_m, row_id)) + existing[row_id] = {"design": payload} + continue + data = dict(existing[row_id]) + if replace: + data["design"] = payload + else: + design = data.get("design") + design = dict(design) if isinstance(design, dict) else {} + design.update(payload) + for key in drop: + design.pop(key, None) + data["design"] = design + existing[row_id] = data + + updates = { + row_id: json.dumps(data, ensure_ascii=False) + for row_id, data in existing.items() + if row_id > 0 and data != original[row_id] + } + inserts = [ + (owner, route_id, chainage_m, json.dumps(existing[row_id], ensure_ascii=False)) + for row_id, (owner, chainage_m) in fresh.items() + ] written = 0 async with connection.cursor() as cursor: diff --git a/B06_Section/B06_Section_Server_Calc_Prebuild.py b/B06_Section/B06_Section_Server_Calc_Prebuild.py index 2a965c06d..262bf28cc 100644 --- a/B06_Section/B06_Section_Server_Calc_Prebuild.py +++ b/B06_Section/B06_Section_Server_Calc_Prebuild.py @@ -23,6 +23,7 @@ from __future__ import annotations import asyncio +import hashlib import json import logging import time @@ -36,8 +37,8 @@ from B06_Section.B06_Section_Repository import ( get_longitudinal_section, merge_longitudinal_section_data, ) -from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs -from common_util.common_util_node_bundle import run_bundle_json +from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_design_batches +from common_util.common_util_node_bundle import bundle_stale, run_bundle_json from common_util.common_util_project_settings import ( haul_equipment_limits, mixed_conversion_factors, @@ -73,6 +74,37 @@ _AREA_KEYS = ( ) +# 입력 지문 → Node 결과(342-2 · 틀 = B07 덧그림 기억) — 같은 입력이면 Node 를 다시 안 돌림. +# 지문 = Node 에 넘기는 입력 **전부** + 번들 판(mtime) · TS 가 번들보다 새로우면 버림. +# ponytail: 프로세스 메모리 · 최근 몇 벌만. 서버 여러 대면 `_derived/` 파일로. +_NODE_CACHE: dict[str, Any] = {} +_NODE_CACHE_LIMIT = 8 + + +def _node_fingerprint(node_input: dict[str, Any]) -> str: + """Node 입력 전부 + 번들 판 → 지문. 입력 한 칸이라도 바뀌면 다른 값.""" + digest = hashlib.sha1( + json.dumps(node_input, sort_keys=True, default=str).encode("utf-8"), + usedforsecurity=False, + ) + digest.update(str(BUNDLE.stat().st_mtime_ns if BUNDLE.is_file() else 0).encode()) + return digest.hexdigest() + + +def _node_output(node_input: dict[str, Any]) -> Any: + """같은 입력이면 지난 결과 · 아니면 Node(면적 · 유토곡선)를 돌려 기억한다.""" + key = _node_fingerprint(node_input) + if key in _NODE_CACHE and not bundle_stale(BUNDLE): + return _NODE_CACHE[key] + output = run_bundle_json(BUNDLE, _NPM_SCRIPT, node_input) + if output is None: # 실패는 기억하지 않음 — 다음 [저장]이 다시 셈 + return None + if len(_NODE_CACHE) >= _NODE_CACHE_LIMIT: + _NODE_CACHE.pop(next(iter(_NODE_CACHE))) + _NODE_CACHE[_node_fingerprint(node_input)] = output # 셈 사이 번들을 다시 지었으면 새 판으로 + return output + + async def haul_inputs_for(project_id: Any) -> dict[str, Any]: """B08 이 낸 **구조물 몫**(채집석 공제·구조물 잔토)을 받아 온다. @@ -268,9 +300,7 @@ async def _recompute(project_id: UUID | str, route_id: int) -> int: marks.append(("포장·세월교 보정", time.perf_counter())) output = await asyncio.to_thread( - run_bundle_json, - BUNDLE, - _NPM_SCRIPT, + _node_output, { "detail": detail, "context": _mass_haul_context( @@ -303,43 +333,42 @@ async def _recompute(project_id: UUID | str, route_id: int) -> int: if not fixed and not rows and not mass_haul and not extra_walls: return 0 + stored = (longitudinal_row or {}).get("data") or {} + if isinstance(mass_haul, dict): + # balloon 위치는 **사용자가 끌어 옮긴 화면값**이다 — 서버가 만들지 않으므로 + # 저장분에서 떼어 새 유토곡선에 도로 붙인다(2026-09-06). 위 병렬 조회 한 번을 씀(342-2). + offsets = (stored.get("mass_haul") or {}).get("balloon_offsets") + if offsets is not None: # 사본에 붙임 — Node 결과는 지문 기억이 쥐고 있음 + mass_haul = {**mass_haul, "balloon_offsets": offsets} + if mass_haul == stored.get("mass_haul"): + mass_haul = None # 같은 값 — 다시 안 씀 + + area_entries: list[tuple[float, dict[str, Any]]] = [] + for row in rows if isinstance(rows, list) else []: + patch: dict[str, Any] = { + key: float(row[key]) for key in _AREA_KEYS if isinstance(row.get(key), (int, float)) + } + if patch: + area_entries.append((float(row["chainage_m"]), patch)) + updated = 0 async with pool.acquire() as connection: - # balloon 위치는 **사용자가 끌어 옮긴 화면값**이다 — 서버가 만들지 않으므로 - # 저장분에서 떼어 새 유토곡선에 도로 붙인다(2026-09-06). - if isinstance(mass_haul, dict): - existing = await get_longitudinal_section(connection, project_uuid, route_id) - stored = (existing or {}).get("data") or {} - offsets = (stored.get("mass_haul") or {}).get("balloon_offsets") - if offsets is not None: - mass_haul["balloon_offsets"] = offsets - await connection.begin() try: - # 행마다 쓰면 원격 DB 왕복이 행 수만큼 난다(측정: 22행 670ms) — 한 문장으로 묶는다. - await merge_cross_section_designs( + # 행 읽기 한 번 · 바뀐 행만 한 문장(342-2) — 보정 설계 갈아 끼움 → 면적 · 벽 얹기 차례. + updated = await merge_cross_section_design_batches( connection, route_id=route_id, - entries=[(float(item.get("chainage_m") or 0.0), item["design"]) for item in fixed], - replace=True, - project_id=project_uuid, + batches=[ + ( + [(float(item.get("chainage_m") or 0.0), item["design"]) for item in fixed], + True, + project_uuid, + ), + (area_entries, False, None), + (extra_walls, False, None), + ], ) - area_entries: list[tuple[float, dict[str, Any]]] = [] - for row in rows if isinstance(rows, list) else []: - patch: dict[str, Any] = { - key: float(row[key]) - for key in _AREA_KEYS - if isinstance(row.get(key), (int, float)) - } - if patch: - area_entries.append((float(row["chainage_m"]), patch)) - updated = await merge_cross_section_designs( - connection, route_id=route_id, entries=area_entries, replace=False - ) - if extra_walls: - await merge_cross_section_designs( - connection, route_id=route_id, entries=extra_walls, replace=False - ) if isinstance(mass_haul, dict): await merge_longitudinal_section_data( connection, route_id=route_id, data_patch={"mass_haul": mass_haul} @@ -351,7 +380,7 @@ async def _recompute(project_id: UUID | str, route_id: int) -> int: marks.append(("정본 저장", time.perf_counter())) _log_steps("서버 재계산 내부", marks) logger.info( - "서버 재계산: route_id=%s 설계 보정 %s곳, 구조물 면적 %s곳, 유토곡선 %s", + "서버 재계산: route_id=%s 설계 보정 %s곳, 바뀐 행 %s곳, 유토곡선 %s", route_id, len(fixed), updated, diff --git a/resources/tester/test_342_2_confirm_fingerprint.py b/resources/tester/test_342_2_confirm_fingerprint.py new file mode 100644 index 000000000..7b5580067 --- /dev/null +++ b/resources/tester/test_342_2_confirm_fingerprint.py @@ -0,0 +1,273 @@ +# -*- coding: utf-8 -*- +"""342-2 [확정] · [저장] 서버 재계산 — 입력 지문 · 바뀐 행만 · 행 읽기 한 번(부록 4 R23). + +① 지문 = Node 에 넘기는 입력 **전부** + 번들 판 — 같은 입력이면 Node 를 다시 안 돌림 · + 입력 어느 칸(깊은 칸 · 문맥 · 번들)이 바뀌어도 다시 돌림. +② Node 에 가는 입력과 지문이 같은 한 벌 — 지문 밖으로 새는 입력 길이 없음. +③ 행 쓰기 = 읽기 한 번 · 바뀐 행만 한 문장 · 앞 벌이 만든 새 행을 뒤 벌이 찾음. +④ 같은 입력으로 두 번 = 두 번째는 Node 0 번 · 행 쓰기 0 · 유토곡선 쓰기 0. +""" + +from __future__ import annotations + +import asyncio +import copy +import json +import os +import sys +from pathlib import Path +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import B06_Section.B06_Section_Router as b06_router # noqa: E402 +import B06_Section.B06_Section_Router_Design as b06_design # noqa: E402 +import B06_Section.B06_Section_Server_Calc_Prebuild as prebuild # noqa: E402 +from B06_Section.B06_Section_Repository_Bulk import ( # noqa: E402 + merge_cross_section_design_batches, +) + +DETAIL = { + "longitudinal": {"length_m": 40.0, "stations": [{"chainage_m": 0.0}]}, + "cross_sections": [ + {"chainage_m": 0.0, "design": {"ground_type": "soil", "slope": {"cut": 1.0}}}, + {"chainage_m": 20.0, "design": {"ground_type": "rock"}}, + ], +} +CONTEXT = {"earthwork_conversion": {"soil": {"L": 1.25, "C": 0.9}}, "haul_equipment_limits": []} + + +@pytest.fixture +def node(monkeypatch, tmp_path): + bundle = tmp_path / "bundle.js" + bundle.write_text("//", encoding="utf-8") + calls: list = [] + + def fake_run(path, script, payload): + calls.append(payload) + return {"areas": [], "mass_haul": {"total": len(calls)}} + + monkeypatch.setattr(prebuild, "BUNDLE", bundle) + monkeypatch.setattr(prebuild, "run_bundle_json", fake_run) + monkeypatch.setattr(prebuild, "bundle_stale", lambda path: False) + monkeypatch.setattr(prebuild, "_NODE_CACHE", {}) + return SimpleNamespace(calls=calls, bundle=bundle) + + +def _input(): + return {"detail": copy.deepcopy(DETAIL), "context": copy.deepcopy(CONTEXT)} + + +def test_같은_입력이면_Node_한_번(node): + first = prebuild._node_output(_input()) + second = prebuild._node_output(_input()) + assert len(node.calls) == 1 + assert first == second + + +@pytest.mark.parametrize( + "change", + [ + lambda data: data["detail"]["cross_sections"][0]["design"]["slope"].update(cut=1.5), + lambda data: data["detail"]["cross_sections"].append({"chainage_m": 40.0}), + lambda data: data["detail"]["longitudinal"].update(length_m=40.5), + lambda data: data["context"]["earthwork_conversion"]["soil"].update(C=0.85), + lambda data: data.update(context={**data["context"], "structure_spoil_m3": 3.0}), + ], + ids=["깊은 설계 칸", "측점 더함", "종단", "환산계수", "문맥 새 칸"], +) +def test_입력_어느_칸이_바뀌어도_다시(node, change): + prebuild._node_output(_input()) + changed = _input() + change(changed) + prebuild._node_output(changed) + assert len(node.calls) == 2 + + +def test_번들이_바뀌면_다시(node): + prebuild._node_output(_input()) + stat = node.bundle.stat() + os.utime(node.bundle, ns=(stat.st_atime_ns, stat.st_mtime_ns + 10_000_000)) + prebuild._node_output(_input()) + assert len(node.calls) == 2 + + +def test_TS_가_번들보다_새로우면_다시(node, monkeypatch): + prebuild._node_output(_input()) + monkeypatch.setattr(prebuild, "bundle_stale", lambda path: True) + prebuild._node_output(_input()) + assert len(node.calls) == 2 + + +def test_실패는_기억_안_함(node, monkeypatch): + monkeypatch.setattr(prebuild, "run_bundle_json", lambda *a: node.calls.append(a) or None) + assert prebuild._node_output(_input()) is None + assert prebuild._node_output(_input()) is None + assert len(node.calls) == 2 + + +# ── ③ 행 쓰기 ────────────────────────────────────────────────────────────── + + +class _Cursor: + def __init__(self, log, rows): + self.log, self.rows = log, rows + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def execute(self, sql, params=None): + self.log.append((" ".join(sql.split()), list(params or []))) + + async def fetchall(self): + return self.rows + + +class _Connection: + def __init__(self, rows): + self.log: list = [] + self.rows = rows + + def cursor(self, *args): + return _Cursor(self.log, self.rows) + + async def begin(self): + return None + + async def commit(self): + return None + + async def rollback(self): + return None + + +def _rows(): + return [ + (1, 0.0, {"design": {"ground_type": "soil", "cut_area_m2": 2.0}}), + (2, 20.0, json.dumps({"design": {"ground_type": "rock"}})), + ] + + +def _kinds(log): + return [sql.split()[0] for sql, _ in log] + + +def test_읽기_한_번_바뀐_행만(tmp_path): + connection = _Connection(_rows()) + written = asyncio.run( + merge_cross_section_design_batches( + connection, + route_id=7, + batches=[ + ([(0.0, {"ground_type": "soil", "cut_area_m2": 2.0})], True, None), # 그대로 + ([(20.0, {"cut_area_m2": 3.0})], False, None), # 바뀜 + ([(40.0, {"ground_type": "soil"})], True, uuid4()), # 새 행 + ([(40.0, {"cut_area_m2": 1.0})], False, None), # 앞 벌의 새 행에 얹음 + ], + ) + ) + assert _kinds(connection.log) == ["SELECT", "UPDATE", "INSERT"] + update = connection.log[1][1] + assert update[0] == 2 and json.loads(update[1])["design"] == { + "ground_type": "rock", + "cut_area_m2": 3.0, + } + insert = connection.log[2][1] + assert json.loads(insert[3]) == {"design": {"ground_type": "soil", "cut_area_m2": 1.0}} + assert written == 2 + + +def test_모두_그대로면_쓰기_없음(): + connection = _Connection(_rows()) + written = asyncio.run( + merge_cross_section_design_batches( + connection, + route_id=7, + batches=[ + ([(0.0, {"cut_area_m2": 2.0})], False, None), + ([(20.0, {"ground_type": "rock"})], True, None), + ], + ) + ) + assert _kinds(connection.log) == ["SELECT"] + assert written == 0 + + +# ── ②④ 재계산 두 번 ────────────────────────────────────────────────────────── + + +class _Pool: + def __init__(self, connection): + self.connection = connection + + def acquire(self): + connection = self.connection + + class _Acquire: + async def __aenter__(self): + return connection + + async def __aexit__(self, *exc): + return False + + return _Acquire() + + +def test_같은_입력_두_번이면_Node_0_쓰기_0(node, monkeypatch, tmp_path): + rows = [(1, 0.0, {"design": DETAIL["cross_sections"][0]["design"]})] + connection = _Connection(rows) + stored = {"data": {"mass_haul": None}} + written_haul: list = [] + + async def detail(project_id, route_id): + return SimpleNamespace(model_dump=lambda mode: copy.deepcopy(DETAIL)) + + async def run_with(fn, *args): + if fn is prebuild.get_longitudinal_section: + return copy.deepcopy(stored) + if fn is prebuild.get_project_storage_relative_path: + return "p" + return {"source_filter": "csf", "method": "dtm", "smooth": False} + + async def write_haul(connection, *, route_id, data_patch): + written_haul.append(data_patch) + stored["data"]["mass_haul"] = data_patch["mass_haul"] + + def fake_run(path, script, payload): + node.calls.append(payload) + # Node 에 간 입력이 곧 지문 입력 — 다른 입력 길 없음 + assert set(payload) == {"detail", "context"} + return {"areas": [{"chainage_m": 0.0, "cut_area_m2": 5.0}], "mass_haul": {"total": 9.0}} + + monkeypatch.setattr(prebuild, "run_bundle_json", fake_run) + monkeypatch.setattr(b06_router, "get_section_detail", detail) + monkeypatch.setattr(prebuild, "run_with_connection", run_with) + monkeypatch.setattr(prebuild, "resolve_stored_project_path", lambda path: str(tmp_path)) + monkeypatch.setattr(prebuild, "get_db_pool", lambda: _Pool(connection)) + monkeypatch.setattr(prebuild, "merge_longitudinal_section_data", write_haul) + monkeypatch.setattr(prebuild, "quantity_settings", lambda root: {}) + monkeypatch.setattr(prebuild, "mixed_conversion_factors", lambda settings: CONTEXT) + monkeypatch.setattr(prebuild, "haul_equipment_limits", lambda settings: []) + monkeypatch.setattr(b06_design, "enforce_design_lines", lambda *a: None) + monkeypatch.setattr(prebuild, "_enforce_stored_designs", lambda *a: None) + + asyncio.run(prebuild.recompute_server_side(uuid4(), 3)) + assert len(node.calls) == 1 + assert _kinds(connection.log) == ["SELECT", "UPDATE"] + assert len(written_haul) == 1 + # 첫 번째가 쓴 값을 행에 반영(가짜 DB) — 두 번째는 같은 입력 + rows[0] = (1, 0.0, json.loads(connection.log[1][1][1])) + connection.log.clear() + + asyncio.run(prebuild.recompute_server_side(uuid4(), 3)) + assert len(node.calls) == 1 # Node 다시 안 돎 + assert _kinds(connection.log) == ["SELECT"] # 바뀐 행 없음 — 쓰기 0 + assert len(written_haul) == 1 # 유토곡선 같은 값 — 안 씀