From f127c0179e75983629b94f179a08a288458a4a53 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Thu, 15 Oct 2026 12:34:58 +0900 Subject: [PATCH 1/3] =?UTF-8?q?feat(=EB=85=B8=EC=84=A0=EC=B6=94=EC=B2=9C):?= =?UTF-8?q?=20327-1=20=EC=B6=94=EC=B2=9C=20=EA=B2=B0=EA=B3=BC=20=EC=98=81?= =?UTF-8?q?=EA=B5=AC=20=EC=A0=80=EC=9E=A5=20=E2=80=94=20=EC=9E=85=EB=A0=A5?= =?UTF-8?q?=20=EC=A7=80=EB=AC=B8=20=ED=82=A4=20=C2=B7=20=EA=B0=99=EC=9D=80?= =?UTF-8?q?=20=EC=9E=85=EB=A0=A5=EC=9D=B4=EB=A9=B4=20=EC=A0=80=EC=9E=A5?= =?UTF-8?q?=EB=B3=B8=20=C2=B7=20=EB=B0=94=EB=80=90=20=EC=9E=85=EB=A0=A5?= =?UTF-8?q?=EB=A7=8C=20=EC=85=88=ED=95=B4=20=EB=8D=94=ED=95=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 새 B05_Profile_Engine_RouteInitial_Store(ALGO_VERSION · request_fingerprint · 읽기 · 쓰기 · 묶음 잠금) · 자리 B05_Profile/route_recommend/<키>.json · 덮어쓰기 없음 · 상한 없음 - 엔드포인트 몸통 compute_initial(찾기 → 셈 · 더함 · 덮개 수 꺼낼 때 다시 셈 · stored 표시) · precompute_defaults(기본 다섯 알약) - resolve_inputs 로 기본값 풀이 한 곳 · 머리글 「파일 안 씀」 정정 - 시험 test_224_3_recommend_store Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_014utrjfoW831Vc5wJUd13kL --- .../B05_Profile_Engine_RouteInitial.py | 26 +- ...B05_Profile_Engine_RouteInitial_Profile.py | 5 +- .../B05_Profile_Engine_RouteInitial_Store.py | 128 ++++++++++ B05_Profile/B05_Profile_Router_Replan.py | 2 +- .../B05_Profile_Router_RouteInitial.py | 110 +++++++-- .../tester/test_224_3_recommend_store.py | 231 ++++++++++++++++++ 6 files changed, 470 insertions(+), 32 deletions(-) create mode 100644 B05_Profile/B05_Profile_Engine_RouteInitial_Store.py create mode 100644 resources/tester/test_224_3_recommend_store.py diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial.py b/B05_Profile/B05_Profile_Engine_RouteInitial.py index ba7278c84..13a1e0fd5 100644 --- a/B05_Profile/B05_Profile_Engine_RouteInitial.py +++ b/B05_Profile/B05_Profile_Engine_RouteInitial.py @@ -3,8 +3,9 @@ 예상노선 추종 · 영선 노선 · 토공 최소 · 기울기 순한 · 안전 우선 · 계류 횡단 최소 · 공사비 어림 최소 · 비중. -계산 자리 = 서버 단독(③). 결과는 화면 캐시용 — 이 모듈은 파일을 쓰지 않는다. 적용은 화면이 -받은 노드로 기존 `POST /route/replan`([확인])을 부른다. +계산 자리 = 서버 단독(③). 이 모듈은 파일을 쓰지 않는다 — 결과는 라우터가 입력 지문 키로 +`B05_Profile/route_recommend/` 에 남긴다(`..._Store` · PLAN 327). 적용은 화면이 받은 노드로 +기존 `POST /route/replan`([확인])을 부른다. 갈래(`mode`): · follow — 예상노선 추적 = 전처리 초기값(`planned_route_initial.csv`)과 같은 규칙 · 같은 선. @@ -182,6 +183,20 @@ def write_initial_choice( return choice +def resolve_inputs( + max_grade: float, + stream_offset_m: float | None, + grade_smooth_m: float | None, + target_grade: float | None, +) -> tuple[float, float, float]: + """(계류 이격, 기울기 잴 지반 폭, 영선 목표 기울기) — 비운 값은 기본값 · 저장 키와 같은 풀이.""" + stream_offset = STREAM_OFFSET_M if stream_offset_m is None else float(stream_offset_m) + grade_smooth = GRADE_SMOOTH_M if grade_smooth_m is None else float(grade_smooth_m) + # 영선 노선 목표 기울기(비율) — 없으면 상한 · 상한 넘게 넣어도 상한(제약 안에서만) + goal_grade = max_grade if target_grade is None else min(float(target_grade), max_grade) + return stream_offset, grade_smooth, goal_grade + + def resolve_weights(mode: str, weights: dict[str, float] | None) -> dict[str, float]: """갈래의 비중(합 100). weighted 가 아니면 갈래가 정한 값.""" if mode != "weighted": @@ -240,10 +255,9 @@ def generate_initial_route( if len(expected) < 2: raise ValueError("예상노선 정점이 2개 미만입니다.") used = resolve_weights(mode, weights) - stream_offset = STREAM_OFFSET_M if stream_offset_m is None else float(stream_offset_m) - grade_smooth = GRADE_SMOOTH_M if grade_smooth_m is None else float(grade_smooth_m) - # 영선 노선 목표 기울기(비율) — 없으면 상한 · 상한 넘게 넣어도 상한(제약 안에서만) - goal_grade = max_grade if target_grade is None else min(float(target_grade), max_grade) + stream_offset, grade_smooth, goal_grade = resolve_inputs( + max_grade, stream_offset_m, grade_smooth_m, target_grade + ) def measure( outline: Any, *, bound: float | None, follow: bool = False, weighted: bool = False diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py b/B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py index 13ec86a8e..de8e9fc8f 100644 --- a/B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py +++ b/B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py @@ -1,8 +1,9 @@ """노선 추천 후보의 균형 종단(PLAN 63-4 · 63-2 안 A). 후보 노선 하나에 종단을 세워 절 · 성토를 **횡단 ㎥** 로 맞춘 계획선과 그 토량을 낸다. -편집 창 [노선 추천]의 점수 · 미리 보기에만 쓴다 — 파일 · 캐시 · initial_snapshot · 확정 종단을 -바꾸지 않는다. 확정 체인의 종단 규칙(1차 계획고 = 지반고 · 2026-09-02 사용자 확정)은 그대로다. +편집 창 [노선 추천]의 점수 · 미리 보기에만 쓴다 — 작업본 파일 · initial_snapshot · 확정 종단을 +바꾸지 않는다(추천 결과째 저장본에 남는 것은 라우터 몫 · PLAN 327). 확정 체인의 종단 규칙 +(1차 계획고 = 지반고 · 2026-09-02 사용자 확정)은 그대로다. 계산 자리 = 서버 단독(③) — 화면은 받은 측점 · 지반고 · 계획고를 그리기만 한다. diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py b/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py new file mode 100644 index 000000000..5f4cd944b --- /dev/null +++ b/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py @@ -0,0 +1,128 @@ +"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3). + +같은 입력이면 같은 결과 — 셈한 추천을 입력 지문 키로 프로젝트 폴더에 남기고, 같은 입력이 오면 +셈하지 않고 꺼낸다. 바뀐 입력만 새로 셈해 더한다(덮어쓰기 없음 · 상한 없음). + +층 = 서버 계산 캐시(데이터 3층 밖 · 사용자 편집값 아님 — 자동저장 금지와 무관). 자리 +`B05_Profile/route_recommend/<키>.json` — initial_snapshot 대상(`B05_Profile/route`) 밖이라 +[초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다. + +키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표 +기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · `ALGO_VERSION`. 관 지점은 뺀다 — 덮개 수는 꺼낼 때 +다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 올린다(옛 파일은 다른 키라 안 쓰임). +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +from contextlib import asynccontextmanager +from datetime import datetime +from pathlib import Path +from typing import Any, AsyncIterator + +from B05_Profile.B05_Profile_Engine_RouteInitial import ( + DEFAULT_WEIGHTS, + Terrain, + _pool_key, + resolve_inputs, + resolve_weights, +) +from common_util.common_util_json import atomic_write_json + +ALGO_VERSION = 1 +STORE_DIR = "B05_Profile/route_recommend" +# 업로드 체인 끝에 미리 셈할 기본 다섯 알약(화면 차례) · 추적은 묶음이 선 뒤라 `similar` 가 채워짐 +DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow") + +# 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음) +_LOCKS: dict[str, asyncio.Lock] = {} + + +def store_dir(project_root: Path) -> Path: + return Path(project_root) / STORE_DIR + + +def request_fingerprint( + expected: list[tuple[float, float]], + terrain: Terrain, + streams: list[list[tuple[float, float]]], + *, + mode: str, + weights: dict[str, float] | None, + max_grade: float, + min_radius_m: float, + stream_offset_m: float | None, + grade_smooth_m: float | None, + target_grade: float | None, + with_profile: bool, + design_speed: int, +) -> tuple[str, str]: + """(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키.""" + pool = _pool_key( + expected, + terrain, + streams, + max_grade, + min_radius_m, + *resolve_inputs(max_grade, stream_offset_m, grade_smooth_m, target_grade), + ).hex() + used = resolve_weights(mode, weights) + tail = (mode, sorted(used.items()), bool(with_profile), int(design_speed), ALGO_VERSION) + key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest() + return key, pool + + +def is_default( + mode: str, weights: dict[str, float] | None, criteria: dict[str, Any] | None, terrain: Any +) -> bool: + """기본값 결과 — 빈 제약 · 50/25/25 · 설계 지형. 표시일 뿐 지우는 데 쓰지 않음.""" + empty = not any(value is not None for value in (criteria or {}).values()) + plain = mode != "weighted" or resolve_weights(mode, weights) == DEFAULT_WEIGHTS + return empty and plain and terrain is None + + +def read_stored(project_root: Path, key: str) -> dict[str, Any] | None: + """저장본 한 벌 — 없거나 못 읽거나 판이 다르면 None.""" + try: + data = json.loads((store_dir(project_root) / f"{key}.json").read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + if not isinstance(data, dict) or data.get("version") != ALGO_VERSION: + return None + return data + + +def write_stored( + project_root: Path, + key: str, + pool: str, + request: dict[str, Any], + result: dict[str, Any], + default: bool, +) -> bool: + """새 키만 적는다 — 이미 있으면 그대로 두고 False(덮어쓰기 없음).""" + path = store_dir(project_root) / f"{key}.json" + if path.exists(): + return False + atomic_write_json( + path, + { + "version": ALGO_VERSION, + "key": key, + "pool": pool, + "default": default, + "created_at": datetime.now().isoformat(timespec="seconds"), + "request": request, + "result": result, + }, + ) + return True + + +@asynccontextmanager +async def pool_lock(pool: str) -> AsyncIterator[None]: + # ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로 + async with _LOCKS.setdefault(pool, asyncio.Lock()): + yield diff --git a/B05_Profile/B05_Profile_Router_Replan.py b/B05_Profile/B05_Profile_Router_Replan.py index 10e2f4559..e3ae05f53 100644 --- a/B05_Profile/B05_Profile_Router_Replan.py +++ b/B05_Profile/B05_Profile_Router_Replan.py @@ -26,7 +26,7 @@ GET /api/projects/{id}/route/plan → 예상노선·계획노선 정점(사업지 좌표계) POST /api/projects/{id}/route/replan → 고친 계획노선으로 갈아 끼우고 재계산 POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산 - POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(파일 안 씀) + POST /api/projects/{id}/route/initial → 초기 계획노선 갈래 하나를 만들어 돌려줌(추천 저장본만 씀) """ import asyncio diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index 88e90e6b1..5bb6f494e 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial.py @@ -30,14 +30,35 @@ async def initial_route( ) -> dict[str, Any] | JSONResponse: """초기 계획노선 갈래 하나를 만들어 비교값과 함께 돌려준다(PLAN 20장). - **파일을 쓰지 않는다** — 화면 캐시용. 적용은 받은 `nodes` 로 `POST /route/replan`([확인]). + 작업본은 쓰지 않는다 — 추천 저장본(`..._Store` · PLAN 327)만. 적용은 받은 `nodes` 로 + `POST /route/replan`([확인]). + """ + return await compute_initial(project_id, request) + + +async def compute_initial( + project_id: UUID, request: replan.RouteInitialRequest +) -> dict[str, Any] | JSONResponse: + """같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로. + + 응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음). """ from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct from B05_Profile.B05_Profile_Engine_RouteInitial import ( generate_initial_route, load_terrain_and_streams, ) - from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import project_pipes + from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import ( + cover_shortfall, + project_pipes, + ) + from B05_Profile.B05_Profile_Engine_RouteInitial_Store import ( + is_default, + pool_lock, + read_stored, + request_fingerprint, + write_stored, + ) from common_util.common_util_surface_confirmation import ( analysis_terrain_params, get_surface_confirmation_params, @@ -81,37 +102,62 @@ async def initial_route( async with pool.acquire() as connection: selection = await get_surface_confirmation_params(connection, str(project_id)) selection = analysis_terrain_params(selection, request.terrain) + inputs = { + "mode": request.mode, + "weights": weights, + "max_grade": max_grade, + "min_radius_m": radius_m, + "stream_offset_m": stream_offset_m, + "grade_smooth_m": grade_smooth_m, + "target_grade": None if target_pct is None else target_pct / 100.0, + "design_speed": design_speed or 20, + "with_profile": request.with_profile, + } started = time.perf_counter() try: terrain, streams = await asyncio.to_thread( load_terrain_and_streams, project_root, selection ) - pipes = ( - await asyncio.to_thread(project_pipes, project_root) if request.with_profile else None - ) - result = await asyncio.to_thread( - functools.partial( - generate_initial_route, - [(x, y) for x, y in expected], - terrain, - streams, - mode=request.mode, - weights=weights, - max_grade=max_grade, - min_radius_m=radius_m, - stream_offset_m=stream_offset_m, - grade_smooth_m=grade_smooth_m, - target_grade=None if target_pct is None else target_pct / 100.0, - design_speed=design_speed or 20, - pipes=pipes, - with_profile=request.with_profile, - ) + route = [(x, y) for x, y in expected] + key, pool_key = await asyncio.to_thread( + functools.partial(request_fingerprint, route, terrain, streams, **inputs) ) + async with pool_lock(pool_key): + stored = await asyncio.to_thread(read_stored, project_root, key) + if stored is not None: + result = stored["result"] + else: + # 관 지점은 키 밖 — 덮개 수 없이 셈해 남기고 아래에서 지금 관으로 센다. + result = await asyncio.to_thread( + functools.partial( + generate_initial_route, route, terrain, streams, pipes=None, **inputs + ) + ) + await asyncio.to_thread( + write_stored, + project_root, + key, + pool_key, + request.model_dump(), + result, + is_default( + request.mode, + weights, + criteria.model_dump() if criteria else None, + request.terrain, + ), + ) + profile = result["metrics"].get("profile") + if request.with_profile and profile and "error" not in profile: + pipes = await asyncio.to_thread(project_pipes, project_root) + if pipes: + profile.update(cover_shortfall(profile, result["planned"], pipes)) except (FileNotFoundError, ValueError) as error: return JSONResponse(status_code=409, content={"status": "error", "message": str(error)}) logger.info( - "초기 계획노선 %s: project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm", + "초기 계획노선 %s(%s): project_id=%s %.1fs 연장 %.0fm 이격 평균 %.1fm", request.mode, + "저장본" if stored is not None else "셈", project_id, time.perf_counter() - started, result["metrics"]["length_m"], @@ -122,10 +168,28 @@ async def initial_route( "status": "success", "project_id": str(project_id), **result, + "stored": stored is not None, "current_initial": current, } +async def precompute_defaults(project_id: UUID) -> None: + """업로드 체인 끝 예약(PLAN 327-2) — 기본 다섯 알약을 미리 셈해 남김 · 있으면 꺼내기만.""" + from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES + + started = time.perf_counter() + for mode in DEFAULT_MODES: + # 화면 기본과 같은 몸통 — 빈 제약 · 50/25/25 · 설계 지형 · 추적만 종단 끔(`withProfileOf`) + request = replan.RouteInitialRequest(mode=mode, with_profile=mode != "follow") + outcome = await compute_initial(project_id, request) + if isinstance(outcome, JSONResponse): + logger.warning( + "추천 미리 셈 멈춤(%s %s): project_id=%s", mode, outcome.status_code, project_id + ) + return + logger.info("추천 미리 셈 끝: project_id=%s %.1fs", project_id, time.perf_counter() - started) + + def _set_base_pvi(path: Path, stations: list[float], plan: list[float]) -> int: """종단 정본의 1차 변화점(`base_pvi`) 계획고를 `stations` · `plan` 직선 보간으로 바꿔 쓴다.""" import json diff --git a/resources/tester/test_224_3_recommend_store.py b/resources/tester/test_224_3_recommend_store.py new file mode 100644 index 000000000..528f34605 --- /dev/null +++ b/resources/tester/test_224_3_recommend_store.py @@ -0,0 +1,231 @@ +"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3) — `B05_Profile_Engine_RouteInitial_Store`. + + ./venv/Scripts/python.exe -m pytest resources/tester/test_224_3_recommend_store.py -q + +같은 입력 = 같은 키 · 하나만 바뀌어도 다른 키 · 두 번째 요청 셈 0회 · 덮어쓰기 없음 · +[초기화](스냅샷 복원) 뒤에도 남음 · 자료 교체 때 지워짐 · 덮개 수는 꺼낼 때 지금 관으로. +엔드포인트 몸통(`compute_initial`)은 DB · 지형 읽기만 갈아 끼우고 그대로 돌린다. +""" + +import asyncio +import json +import shutil +from contextlib import asynccontextmanager +from pathlib import Path +from uuid import uuid4 + +import numpy as np +import pytest + +from B05_Profile import B05_Profile_Router_Replan as replan +from B05_Profile import B05_Profile_Router_RouteInitial as router +from B05_Profile.B05_Profile_Engine_RouteInitial import STREAM_OFFSET_M, Terrain +from B05_Profile.B05_Profile_Engine_RouteInitial_Store import ( + DEFAULT_MODES, + request_fingerprint, + store_dir, + write_stored, +) +from common_util.common_util_initial_snapshot import ( + _FILE_TREES, + restore_snapshot_files, + snapshot_dir, +) +from common_util.common_util_project_reset import purge_output_folders + + +def _terrain(lift: float = 0.0) -> Terrain: + x = np.arange(0.0, 40.0, 2.0) + y = np.arange(0.0, 20.0, 2.0) + xx, yy = np.meshgrid(x, y) + return Terrain(x, y, 0.1 * xx + lift, np.ones(xx.shape, dtype=bool)) + + +EXPECTED = [(0.0, 10.0), (20.0, 10.0), (38.0, 10.0)] +STREAMS = [[(5.0, 0.0), (5.0, 18.0)]] +BASE = { + "mode": "weighted", + "weights": None, + "max_grade": 0.14, + "min_radius_m": 12.0, + "stream_offset_m": None, + "grade_smooth_m": None, + "target_grade": None, + "with_profile": True, + "design_speed": 20, +} + + +def _key(expected=EXPECTED, terrain=None, streams=STREAMS, **changes) -> str: + return request_fingerprint(expected, terrain or _terrain(), streams, **{**BASE, **changes})[0] + + +def test_same_input_same_key() -> None: + assert _key() == _key() + # 비운 값 = 기본값 · 비중 None = 50/25/25 — 화면이 채워 보내도 같은 키 + assert _key() == _key(stream_offset_m=STREAM_OFFSET_M) + assert _key() == _key(weights={"expected": 50.0, "earthwork": 25.0, "grade": 25.0}) + # 비중 갈래가 아니면 비중은 갈래가 정함 — 보낸 비중은 키에 안 듦 + assert _key(mode="earthwork") == _key(mode="earthwork", weights={"expected": 1.0}) + + +@pytest.mark.parametrize( + "change", + [ + {"mode": "earthwork"}, + {"weights": {"expected": 40.0, "earthwork": 30.0, "grade": 30.0}}, + {"max_grade": 0.12}, + {"min_radius_m": 15.0}, + {"stream_offset_m": 50.0}, + {"grade_smooth_m": 3.0}, + {"target_grade": 0.08}, + {"with_profile": False}, + {"design_speed": 30}, + {"expected": [(0.0, 10.0), (20.0, 11.0), (38.0, 10.0)]}, + {"terrain": _terrain(lift=0.5)}, + {"streams": []}, + ], + ids=lambda change: next(iter(change)), +) +def test_one_change_other_key(change: dict) -> None: + assert _key(**change) != _key() + + +def test_no_overwrite(tmp_path: Path) -> None: + assert write_stored(tmp_path, "k", "p", {}, {"n": 1}, True) + assert not write_stored(tmp_path, "k", "p", {}, {"n": 2}, False) + data = json.loads((store_dir(tmp_path) / "k.json").read_text(encoding="utf-8")) + assert data["result"] == {"n": 1} and data["default"] is True + assert set(data) == {"version", "key", "pool", "default", "created_at", "request", "result"} + + +def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None: + (tmp_path / "B05_Profile/route").mkdir(parents=True) + (tmp_path / "B05_Profile/route/initial_choice.json").write_text("{}", encoding="utf-8") + write_stored(tmp_path, "k", "p", {}, {}, True) + # 스냅샷 = 작업본 트리 사본(`save_initial_snapshot` 과 같은 자리) + for tree in _FILE_TREES: + source = tmp_path / tree + if source.is_dir(): + shutil.copytree(source, snapshot_dir(tmp_path) / tree.replace("/", "__")) + restore_snapshot_files(tmp_path) + assert (store_dir(tmp_path) / "k.json").is_file() + purge_output_folders(tmp_path) # 자료 교체 + assert not store_dir(tmp_path).exists() + + +class _Calls: + def __init__(self) -> None: + self.count = 0 + + +@pytest.fixture +def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls: + """`compute_initial` 의 DB · 파일 읽기를 갈아 끼움 · 셈은 횟수만 세는 가짜.""" + import B05_Profile.B05_Profile_Engine_Grade as grade + import B05_Profile.B05_Profile_Engine_RouteInitial as engine + import B05_Profile.B05_Profile_Engine_RouteInitial_Profile as profile + import common_util.common_util_surface_confirmation as surface + + calls = _Calls() + + async def project_paths(_pid): + return tmp_path, "x" + + async def plan_criteria(_pid): + return 12.0, 12.0, 5.0 + + async def road_settings(_pid): + return "간선", 20, "일반" + + async def surface_params(_connection, _pid): + return {} + + class _Pool: + @asynccontextmanager + async def acquire(self): + yield None + + def generate(expected, terrain, streams, *, pipes, **_inputs): + assert pipes is None # 관은 키 밖 — 덮개 없이 셈해 남김 + calls.count += 1 + return { + "planned": [[x, y] for x, y in expected], + "metrics": { + "length_m": 38.0, + "offset_avg_m": 0.0, + "profile": { + "stations_m": [0.0, 38.0], + "plan_m": [1.0, 1.0], + "ground_m": [0.0, 0.0], + }, + }, + } + + monkeypatch.setattr(replan, "_project_paths", project_paths) + monkeypatch.setattr(replan, "_plan_criteria", plan_criteria) + monkeypatch.setattr(replan, "_road_settings", road_settings) + monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED]) + monkeypatch.setattr(router, "get_db_pool", lambda: _Pool()) + monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params) + monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection) + monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0) + monkeypatch.setattr(engine, "load_terrain_and_streams", lambda *_a: (_terrain(), STREAMS)) + monkeypatch.setattr(engine, "generate_initial_route", generate) + # 덮개 여유 2 m 관 하나가 노선 위 — 계획고 − 지반고 1 m 라 모자람 1 + monkeypatch.setattr(profile, "project_pipes", lambda _root: [(19.0, 10.0, 2.0)]) + calls.root = tmp_path + return calls + + +def _ask(**fields) -> dict: + request = replan.RouteInitialRequest(**fields) + return asyncio.run(router.compute_initial(uuid4(), request)) + + +def test_second_request_computes_nothing(wired: _Calls) -> None: + first = _ask(mode="earthwork") + assert wired.count == 1 and first["stored"] is False + second = _ask(mode="earthwork") + assert wired.count == 1 and second["stored"] is True + assert second["planned"] == first["planned"] + # 덮개 수는 꺼낼 때마다 지금 관으로 · 저장본에는 없음 + assert second["metrics"]["profile"]["cover_short"] == 1 + files = list(store_dir(wired.root).glob("*.json")) + assert len(files) == 1 + saved = json.loads(files[0].read_text(encoding="utf-8")) + assert "cover_short" not in saved["result"]["metrics"]["profile"] + assert saved["default"] is True and saved["request"]["mode"] == "earthwork" + # 제약 하나 바꾸면 파일 하나 더 · 옛 파일 그대로 · 기본값 표시 꺼짐 + changed = _ask(mode="earthwork", criteria={"min_radius_m": 15.0}) + assert changed["stored"] is False and wired.count == 2 + assert len(list(store_dir(wired.root).glob("*.json"))) == 2 + assert json.loads(files[0].read_text(encoding="utf-8")) == saved + + +def test_same_pool_in_flight_waits(wired: _Calls) -> None: + async def both() -> list[dict]: + request = replan.RouteInitialRequest(mode="gentle") + project = uuid4() + return await asyncio.gather( + router.compute_initial(project, request), router.compute_initial(project, request) + ) + + results = asyncio.run(both()) + assert wired.count == 1 + assert sorted(result["stored"] for result in results) == [False, True] + + +def test_precompute_defaults_then_hits(wired: _Calls) -> None: + asyncio.run(router.precompute_defaults(uuid4())) + assert wired.count == len(DEFAULT_MODES) == 5 + saved = [ + json.loads(path.read_text(encoding="utf-8")) + for path in store_dir(wired.root).glob("*.json") + ] + assert sorted(item["request"]["mode"] for item in saved) == sorted(DEFAULT_MODES) + assert all(item["default"] for item in saved) + # 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 추적은 종단 끔) — 셈 없이 저장본 + hit = _ask(mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25}) + follow = _ask(mode="follow", with_profile=False) + assert hit["stored"] and follow["stored"] and wired.count == 5 From be86e3ee2c97c25d2b434cec532b43e50ede1f7b Mon Sep 17 00:00:00 2001 From: umsangdon Date: Thu, 15 Oct 2026 12:37:17 +0900 Subject: [PATCH 2/3] =?UTF-8?q?feat(=EB=85=B8=EC=84=A0=EC=B6=94=EC=B2=9C):?= =?UTF-8?q?=20327-2=20=EB=AF=B8=EB=A6=AC=20=EC=85=88=20=C2=B7=20=ED=91=9C?= =?UTF-8?q?=EC=8B=9C=20=E2=80=94=20=EC=97=85=EB=A1=9C=EB=93=9C=20=EC=B2=B4?= =?UTF-8?q?=EC=9D=B8=20=EC=8A=A4=EB=83=85=EC=83=B7=20=EB=92=A4=20=EA=B8=B0?= =?UTF-8?q?=EB=B3=B8=20=EB=8B=A4=EC=84=AF=20=EC=95=8C=EC=95=BD=20=EC=98=88?= =?UTF-8?q?=EC=95=BD=20=C2=B7=20=EC=B6=94=EC=B2=9C=20=EC=B0=BD=20=EC=83=81?= =?UTF-8?q?=ED=83=9C=20=EC=A4=84=20=E3=80=8C=EC=A0=80=EC=9E=A5=EB=90=9C=20?= =?UTF-8?q?=EA=B2=B0=EA=B3=BC=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - B03 자동 설계 체인: 초기값 스냅샷 뒤 precompute_defaults 예약(기다리지 않음 · 실패는 비치명적) - 추천 창: 서버 stored 를 받아 미리 보기 줄에 「(저장된 결과)」 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_014utrjfoW831Vc5wJUd13kL --- B03_FileInput/B03_FileInput_Service_Chain.py | 12 ++++++++++++ .../B05_Profile_UI_RouteEdit_Initial_Model.ts | 3 +++ B05_Profile/B05_Profile_UI_RouteEdit_Recommend.ts | 4 +++- 3 files changed, 18 insertions(+), 1 deletion(-) diff --git a/B03_FileInput/B03_FileInput_Service_Chain.py b/B03_FileInput/B03_FileInput_Service_Chain.py index 703c27d4e..90791b126 100644 --- a/B03_FileInput/B03_FileInput_Service_Chain.py +++ b/B03_FileInput/B03_FileInput_Service_Chain.py @@ -350,6 +350,18 @@ async def run_auto_design_chain( logger.exception("초기값 스냅샷 실패: project_id=%s", project_id) mark_design_failed(project_root, f"초기값 스냅샷 저장 실패: {exc}") + # 노선 추천 기본 다섯 알약 미리 셈(PLAN 327-2) — 예약만 · 체인이 [초기화] 요청 안에서도 + # 불리므로 기다리지 않는다. 같은 입력이면 저장본이 있어 셈 없이 끝난다. + try: + from B03_FileInput.B03_FileInput_Router_Helpers import _schedule_background_task + from B05_Profile.B05_Profile_Router_RouteInitial import precompute_defaults + + _schedule_background_task( + precompute_defaults(project_id), task_name=f"추천 노선 미리 셈 {project_id}" + ) + except Exception: # noqa: BLE001 — 미리 셈이 체인을 막지는 않는다 + logger.exception("추천 노선 미리 셈 예약 실패: project_id=%s", project_id) + return { "route_id": route_id, "length_m": float(solve_result.total_length_m or 0.0), diff --git a/B05_Profile/B05_Profile_UI_RouteEdit_Initial_Model.ts b/B05_Profile/B05_Profile_UI_RouteEdit_Initial_Model.ts index 6e63790ff..7199fa9b8 100644 --- a/B05_Profile/B05_Profile_UI_RouteEdit_Initial_Model.ts +++ b/B05_Profile/B05_Profile_UI_RouteEdit_Initial_Model.ts @@ -189,6 +189,8 @@ export interface InitialChoice { similar?: SimilarRoute[] | null; /** 제약 안에서 갈 수 있는 길의 띠 폭 어림(m) — 서버 `band_width_m`. */ bandWidthM?: number | null; + /** 서버 추천 저장본에서 꺼낸 결과인가(계획서 327) — 서버 `stored`. */ + stored?: boolean; /** 균형 종단(계획서 63-4) — 모달 미리 보기에만. 캐시에는 안 적는다(`writeInitialChoice`). */ balance?: BalanceProfile | null; } @@ -336,6 +338,7 @@ export function normalizeRouteInitial( terrainBoundPct: body.relaxed === true ? num(body.terrain_bound_pct) : null, similar: parseSimilar(body.similar), bandWidthM: num(body.band_width_m), + stored: body.stored === true, balance: parseBalance(m.profile), criteria: body.criteria ? { diff --git a/B05_Profile/B05_Profile_UI_RouteEdit_Recommend.ts b/B05_Profile/B05_Profile_UI_RouteEdit_Recommend.ts index 6d8a4f854..d42d94aef 100644 --- a/B05_Profile/B05_Profile_UI_RouteEdit_Recommend.ts +++ b/B05_Profile/B05_Profile_UI_RouteEdit_Recommend.ts @@ -462,7 +462,9 @@ export function openRecommendModal(params: RecommendParams): void { preview = choice; if (choice.balance) seen.set(label, choice.balance); else seen.delete(label); // 종단 없는 갈래(예상노선 추적) - stateEl.textContent = `미리 보기: ${choiceLabel(logic, sentWeights)} — [적용]을 눌러야 편집 창에 놓입니다.`; + stateEl.textContent = `미리 보기: ${choiceLabel(logic, sentWeights)}${ + choice.stored ? " (저장된 결과)" : "" + } — [적용]을 눌러야 편집 창에 놓입니다.`; } catch (error) { if (closed || mine !== seq) return; stateEl.textContent = "셈하지 못했습니다."; From 27c62ce22094358319bdf3e89e8e780af85430d4 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Thu, 15 Oct 2026 12:37:18 +0900 Subject: [PATCH 3/3] =?UTF-8?q?feat(=EC=9D=BC=EC=9C=84=EB=8C=80=EA=B0=80?= =?UTF-8?q?=EC=8B=A4=EB=AC=B4):=20329-1=20=C2=B7=20330-1=20=C2=B7=20330-2?= =?UTF-8?q?=20=EC=84=9C=EB=B2=84=20=E2=80=94=20=EC=97=91=EC=85=80=EA=BC=B4?= =?UTF-8?q?=20=EA=B2=A9=EC=9E=90=20=C2=B7=20=ED=96=89=20=EB=B2=94=EC=9C=84?= =?UTF-8?q?=20=C2=B7=20=EB=8B=A8=EA=B0=80=EC=82=B0=EC=B6=9C=20=EA=B7=BC?= =?UTF-8?q?=EA=B1=B0=20=EB=A7=9E=EC=B6=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 329-1 새 M02_Hopyo_Grid(xls formatting_info 값 · 병합 · 열 폭 · 행 높이 · 서식표 · 틀고정 / xlsx 값만) · read_book 호표 · 줄 「행」 · Store 격자 저장 · 읽기 · 지움(실무자료/격자/<키>.json) · GET hopyo-grid/{key}(옛 파일 404) · Match 요약 .get - 330-1 새 M02_Hopyo_Basis.read — 단산N 블록 이름 · 단위 · 행 · 출처(책 · 절) · 직종(우리 인력 이름 · 옛이름) · 금액 · 단산 줄 「근거」 - 330-2 Basis.ours · candidates(조항 1.0 · 앞 마디 0.5 · 건설 공통 · 토목 0.1 · 직종 0.5 · 단위 0.15 · 까닭) · match_line 이름 후보와 로직 키로 합침 · 상태 후보 그대로 - 시험 test_m02_hopyo_grid · test_m02_hopyo_basis 새로 · test_m02_hopyo_match 제3호표 구조물터파기 1순위 → 굴착(인력/토사) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Ji4nuyLVHcxrXWHTdb5nRD --- M02_MasterTemplete/M02_Hopyo_Basis.py | 231 +++++++++++++++++++++++ M02_MasterTemplete/M02_Hopyo_Excel.py | 26 ++- M02_MasterTemplete/M02_Hopyo_Grid.py | 211 +++++++++++++++++++++ M02_MasterTemplete/M02_Hopyo_Match.py | 19 +- M02_MasterTemplete/M02_Hopyo_Store.py | 30 ++- M02_MasterTemplete/M02_Labor_Router.py | 19 +- resources/tester/test_m02_hopyo_basis.py | 108 +++++++++++ resources/tester/test_m02_hopyo_grid.py | 120 ++++++++++++ resources/tester/test_m02_hopyo_match.py | 2 +- 9 files changed, 748 insertions(+), 18 deletions(-) create mode 100644 M02_MasterTemplete/M02_Hopyo_Basis.py create mode 100644 M02_MasterTemplete/M02_Hopyo_Grid.py create mode 100644 resources/tester/test_m02_hopyo_basis.py create mode 100644 resources/tester/test_m02_hopyo_grid.py diff --git a/M02_MasterTemplete/M02_Hopyo_Basis.py b/M02_MasterTemplete/M02_Hopyo_Basis.py new file mode 100644 index 000000000..e4fcf4458 --- /dev/null +++ b/M02_MasterTemplete/M02_Hopyo_Basis.py @@ -0,0 +1,231 @@ +"""M02 일위대가 실무자료 단가산출 근거(PLAN 330) — 엑셀 「단가산출」 블록 → 출처 · 직종 · +우리 돈 로직과 견준 근거 후보. + +읽기(`read`) = B열(머리 「산출근거」) 머리 줄 「N.이름[단위]」(앞 빈칸 없음)부터 +다음 머리 앞까지 한 블록 · +출처 = 「…품셈」(오타 「품줌」) 뒤 20 글자 안 절 번호 — 앞 글에 산림 · 임도 = 산림품셈 · +그 밖 = 건설품셈 · +직종 = 「이름 : 수」 줄의 빈칸 뺀 이름이 우리 인력(LB) 이름 · 옛이름일 때만(지금 이름으로). +견줌(`candidates`) = 같은 책 조항 같음 1.0 · 앞 마디 같음 0.5(조항이 안 맞으면 후보 아님) · +건설 공통 · 토목 덤 0.1 · 직종 겹침(보통인부 뺌) 0.5 · 단위 0.15 · 직종도 뜻 낱말도 안 겹치면 버림 · +같은 점수는 글자 두 개 묶음(괄호 글 포함)이 가름. 자동 확정 없음 — 맞춤 차례는 `M02_Hopyo_Match`. +M01 · 원본 로직은 읽기만. +""" + +from __future__ import annotations + +import re +from functools import lru_cache +from typing import Any + +from M01_MasterData import M01_MasterData_Store as m01 +from M02_MasterTemplete import M02_Hopyo_Match as hopyo_match # 서로 부름 — 부를 때만 씀 +from M02_MasterTemplete import M02_Labor_Inputs as labor_inputs + +SHEET = "단가산출" +COMMON = "보통인부" +BONUS_PARTS = ("공통", "토목") +_HEAD = re.compile(r"^(\d+)\.\s*(.*?)\s*\[([^\[\]]*)\]\s*$") +_BOOK = re.compile(r"품[셈줌]") +_TRADE = re.compile(r"^[\s*·\-]*([가-힣][가-힣\s]*?)\s*:\s*[\d,]") +_TYPOS = (("깍기", "깎기"),) +_MONEY = {"합계": "계", "노무비": "노무비", "재료비": "재료비", "경비": "경비"} + + +def _flat(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")) + + +# ── 엑셀 블록 읽기 ───────────────────────────────────────────────────── +@lru_cache(maxsize=2) +def _trade_names(whole_id: int) -> dict[str, str]: + """{빈칸 뺀 이름 · 옛이름: 지금 이름} — 우리 인력(LB).""" + out: dict[str, str] = {} + for row in m01.loaded()[1].numbers.get("LB", {}).values(): + name = str(row.get("이름") or "") + out[_flat(name)] = name + for old in row.get("옛이름") or []: + out.setdefault(_flat(old), name) + return out + + +def trade_names() -> dict[str, str]: + return _trade_names(id(m01.loaded()[1])) + + +def sources(text: str) -> list[list[str]]: + """글 → [[책, 절 번호]] — 「건설공사표준품셈 3-2-1」 · 「산림사업표준품셈(9-14)」.""" + out = [] + for hit in _BOOK.finditer(text): + num = labor_inputs._NUMBER.search(text, hit.end()) + if not num or num.start() - hit.end() > 20: + continue + before = text[max(0, hit.start() - 12) : hit.start()] + book = "산림품셈" if "산림" in before or "임도" in before else "건설품셈" + if [book, num.group()] not in out: + out.append([book, num.group()]) + return out + + +def _columns(rows: list[list[Any]]) -> tuple[int, int, dict[str, int]]: + """(자료 첫 줄, 산출근거 열, {계 · 노무비 · 재료비 · 경비: 열}) — 머리 없으면 (0, 1, {}).""" + for at, row in enumerate(rows[:10]): + flat = [_flat(v) for v in row] + if "산출근거" in flat: + money = {key: flat.index(head) for head, key in _MONEY.items() if head in flat} + return at + 1, flat.index("산출근거"), money + return 0, 1, {} + + +def read(rows: list[list[Any]] | None) -> dict[int, dict[str, Any]]: + """단가산출 시트 줄 → `{N: {번, 이름, 단위, 행: [시작, 끝], 출처, 직종, 금액}}` + (행 = 엑셀 1부터).""" + if not rows: + return {} + start, col, money = _columns(rows) + names = trade_names() + blocks: dict[int, dict[str, Any]] = {} + block: dict[str, Any] | None = None + for at, row in enumerate(rows[start:], start + 1): + text = row[col] if col < len(row) else None + if not isinstance(text, str) or not text.strip(): + continue + head = None if text[:1].isspace() else _HEAD.match(text) + if head: + block = blocks.setdefault( + int(head.group(1)), + { + "번": int(head.group(1)), + "이름": head.group(2), + "단위": head.group(3).strip(), + "행": [at, at], + "출처": [], + "직종": [], + "금액": { + key: v + for key, c in money.items() + for v in [row[c] if c < len(row) else None] + if isinstance(v, (int, float)) and not isinstance(v, bool) + }, + }, + ) + continue + if block is None: + continue + block["행"][1] = at + block["출처"] += [one for one in sources(text) if one not in block["출처"]] + trade = _TRADE.match(text) + name = names.get(_flat(trade.group(1))) if trade else None + if name and name not in block["직종"]: + block["직종"].append(name) + return blocks + + +# ── 우리 쪽 조항 · 직종 ──────────────────────────────────────────────── +@lru_cache(maxsize=2) +def _ours(whole_id: int) -> dict[str, dict[str, Any]]: + """{로직 키: {책, 부문, 구분, 조항: {절 번호}, 직종: {이름}}} — 조항 = `sources()` 묶음 · + 직종 = 호표 인력(LB) 요소.""" + files, whole = m01.loaded() + lb = {str(row.get("키")): str(row.get("이름")) for row in whole.numbers.get("LB", {}).values()} + out = {} + for data in files.values(): + if data.get("그룹") != "로직": + continue + for row in data.get("줄") or []: + kind = str(row.get("구분") or "") + book, _, part = kind.partition(" ") + trades = { + lb[str(h.get("요소"))] + for h in row.get("호표") or [] + if isinstance(h, dict) and str(h.get("요소")) in lb + } + clauses = set(labor_inputs._NUMBER.findall(str(labor_inputs.sources(row) or ""))) + out[str(row.get("키"))] = { + "책": book, + "부문": part, + "구분": kind, + "조항": clauses, + "직종": trades, + } + return out + + +def ours() -> dict[str, dict[str, Any]]: + return _ours(id(m01.loaded()[1])) + + +def _head(number: str) -> str: + return number.rsplit("-", 1)[0] + + +def _clause(theirs: list[list[str]], our: dict[str, Any]) -> tuple[float, list[str], str]: + """(점수, 맞은 우리 조항 글, 꼬리표) — 같은 책만.""" + nums = [n for book, n in theirs if book == our["책"]] + same = sorted(n for n in nums if n in our["조항"]) + if same: + return 1.0, [f"{our['구분']} {n}" for n in same], "조항 같음" + heads = {_head(n) for n in nums} + near = sorted(n for n in our["조항"] if _head(n) in heads) + if near: + return 0.5, [f"{our['구분']} {n}" for n in near], "앞 마디 같음" + return 0.0, [], "" + + +def _text(line: dict[str, Any]) -> str: + text = f"{line.get('이름') or ''} {line.get('규격') or ''}" + for old, new in _TYPOS: + text = text.replace(old, new) + return text + + +def candidates(line: dict[str, Any], got: dict[str, Any]) -> list[dict[str, Any]]: + """근거 후보 — 줄 `근거.출처` 가 없으면 빈 것 · `{로직, 이름, 단위, 점수, 근거}` 점수 차례.""" + basis = line.get("근거") or {} + theirs = [list(one) for one in basis.get("출처") or []] + if not theirs: + return [] + trades = set(basis.get("직종") or []) - {COMMON} + text = _text(line) + mine = hopyo_match.words(text) # 괄호 글 뺌 — 「잔토처리(인력)」 ↔ 「철골재 철거(인력)」 + grams = hopyo_match._grams(hopyo_match.mm.plain(text)) # 괄호 글 포함 — 같은 점수 가름 + table = ours() + scored = [] + for one in got["로직"]: + our = table.get(one["로직"]) + if not our: + continue + score, clauses, tag = _clause(theirs, our) + if not score: + continue + shared = sorted(trades & our["직종"]) + if not shared and not any(a in b or b in a for a in mine for b in one["words"]): + continue + tags = [tag] + if our["책"] == "건설품셈" and our["부문"] in BONUS_PARTS: + score += 0.1 + tags.append(f"건설 {our['부문']}") + if shared: + score += 0.5 + tags.append("직종 " + " · ".join(shared)) + unit = hopyo_match.fits(line.get("단위"), one["단위"]) + if unit: + score += 0.15 + tags.append("단위 맞음") + dice = 2 * len(grams & one["grams"]) / (len(grams) + len(one["grams"])) + why = {"조항": clauses, "직종": shared, "단위": unit, "점수": round(score, 3)} + scored.append( + (score, dice, {**hopyo_match._public(one), "근거": {**why, "까닭": " · ".join(tags)}}) + ) + scored.sort(key=lambda x: (-x[0], -x[1])) + return [one for _, _, one in scored] + + +def merge(basis: list[dict[str, Any]], named: list[dict[str, Any]]) -> list[dict[str, Any]]: + """근거 후보 + 이름 후보 → 로직 키로 합침(점수 = 이름 점수 + 근거 점수) · 점수 차례.""" + if not basis: + return named + by_name = {c.get("로직"): c["점수"] for c in named} + out = [{**c, "점수": round(c["근거"]["점수"] + by_name.pop(c["로직"], 0), 3)} for c in basis] + out += [c for c in named if c.get("로직") in by_name] + return sorted(out, key=lambda c: -c["점수"]) diff --git a/M02_MasterTemplete/M02_Hopyo_Excel.py b/M02_MasterTemplete/M02_Hopyo_Excel.py index 38d11a81e..9995de2b4 100644 --- a/M02_MasterTemplete/M02_Hopyo_Excel.py +++ b/M02_MasterTemplete/M02_Hopyo_Excel.py @@ -11,6 +11,9 @@ 검산 = 줄 금액 floor(단가 × 수량 × 10) / 10 · 합계 단가 = 셋 합 · % 줄 = floor(바탕 × 율 / 100 × 10) / 10 · 호표 비목 = floor(줄 금액 합) · 총괄표 합계 — 어긋남은 글 목록(막지 않음). +행(329-1) = 호표 [제N호표 줄, 마지막 구성 줄] · 구성 줄 한 줄(엑셀 1부터) · +근거(330-1) = 단산N참조 줄에 「단가산출」 블록 `{출처, 직종, 행}`(`M02_Hopyo_Basis.read`) · +`단산` = 블록마다 `{번, 이름, 단위, 행}`. """ from __future__ import annotations @@ -20,6 +23,8 @@ import re from decimal import ROUND_FLOOR, Decimal, InvalidOperation from typing import Any +from M02_MasterTemplete import M02_Hopyo_Basis as hopyo_basis + SHEET = "일위대가" SUMMARY = "일위대가총괄표" COSTS = ("노무비", "재료비", "경비") @@ -67,8 +72,8 @@ def _out(value: Decimal | None) -> int | float | None: # ── 파일 → 시트 표(줄마다 값 목록) ──────────────────────────────────────── def _sheets(name: str, data: bytes) -> dict[str, list[list[Any]]]: - """시트 이름 → 줄 목록 — 일위대가 · 총괄표만 읽음.""" - wanted = (SHEET, SUMMARY) + """시트 이름 → 줄 목록 — 일위대가 · 총괄표 · 단가산출만 읽음.""" + wanted = (SHEET, SUMMARY, hopyo_basis.SHEET) if not data: raise ExcelError("빈 파일") try: @@ -212,7 +217,7 @@ def read_book(name: str, data: bytes) -> dict[str, Any]: start, cols = _columns(rows, ("합계", *COSTS)) hopyos: list[dict[str, Any]] = [] raw: list[list[dict[str, Any]]] = [] - for row in rows[start:]: + for at, row in enumerate(rows[start:], start + 1): head = _flat(_cell(row, cols.get("호표"))) name_ = _text(_cell(row, cols.get("품명"))) hit = _HOPYO.fullmatch(head) @@ -224,6 +229,7 @@ def read_book(name: str, data: bytes) -> dict[str, Any]: "규격": _text(_cell(row, cols.get("규격"))), "단위": _text(_cell(row, cols.get("단위"))), "금액": _money(row, cols, 1), + "행": [at, at], } ) raw.append([]) @@ -244,9 +250,15 @@ def read_book(name: str, data: bytes) -> dict[str, Any]: "참조": {"종류": ref.group(1), "번": int(ref.group(2))} if ref else None, } line["갈래"] = kind_of(line) + line["행"] = hopyos[-1]["행"][1] = at raw[-1].append(line) if not hopyos: raise ExcelError(f"{SHEET} 시트에 호표(제N호표)가 없음") + blocks = hopyo_basis.read(sheets.get(hopyo_basis.SHEET)) + for line in (one for lines in raw for one in lines): + ref = line["참조"] or {} + if ref.get("종류") == "단산" and (block := blocks.get(ref["번"])): + line["근거"] = {k: block[k] for k in ("출처", "직종", "행")} wrong: list[str] = [] totals = _summary(sheets.get(SUMMARY)) for hopyo, lines in zip(hopyos, raw, strict=True): @@ -258,7 +270,13 @@ def read_book(name: str, data: bytes) -> dict[str, Any]: {**_json(hopyo), "줄": [_json(line) for line in lines]} for hopyo, lines in zip(hopyos, raw, strict=True) ] - return {"시트": SHEET, "호표": out, "검산": {"어긋남": wrong, "총괄표": len(totals)}} + basis = [{k: b[k] for k in ("번", "이름", "단위", "행")} for b in blocks.values()] + return { + "시트": SHEET, + "호표": out, + "단산": basis, + "검산": {"어긋남": wrong, "총괄표": len(totals)}, + } def _json(value: Any) -> Any: diff --git a/M02_MasterTemplete/M02_Hopyo_Grid.py b/M02_MasterTemplete/M02_Hopyo_Grid.py new file mode 100644 index 000000000..39af897fa --- /dev/null +++ b/M02_MasterTemplete/M02_Hopyo_Grid.py @@ -0,0 +1,211 @@ +"""M02 일위대가 실무자료 엑셀꼴 격자(PLAN 329-1) — 올린 엑셀 「일위대가」 · 「단가산출」 시트 → +A00 산출근거 스프레드시트 Workbook(`A00_Common/spreadsheet/spreadsheet_types.ts` 꼴 · +읽기 전용 보기). + +xls = xlrd `formatting_info` — 값 · 병합 · 열 폭(÷256 글자) · +행 높이(÷20 pt · 시트 기본과 다른 것만) · 서식 표(글꼴 · 굵게 · 맞춤 · 줄바꿈 · 테두리 · 채움 · +숫자 형식) · 틀고정(파일 값 · 없으면 머리 4행). +xlsx = 값만 · 넓은 시트는 머리 글(앞 4 줄) 마지막 열까지. +결과 `{책, 호표: [{번, 이름, 행}], 단산: [{번, 이름, 행}]}` — 행 = 엑셀 1부터 [시작, 끝]. +""" + +from __future__ import annotations + +import io +from typing import Any + +from openpyxl.utils import get_column_letter + +from M02_MasterTemplete import M02_Hopyo_Basis as hopyo_basis +from M02_MasterTemplete import M02_Hopyo_Excel as hopyo_excel + +SHEETS = (hopyo_excel.SHEET, hopyo_basis.SHEET) +HEAD_ROWS = 4 +_HALIGN = {1: "left", 2: "center", 3: "right", 6: "centerContinuous", 7: "distributed"} +_VALIGN = {0: "top", 1: "center"} +_LINES = {1: "thin", 2: "medium", 3: "dashed", 4: "dotted", 5: "thick", 6: "double", 7: "hair"} +_SIDES = (("위", "top"), ("아래", "bottom"), ("왼", "left"), ("오른", "right")) + + +class _Styles: + """서식 표 — 같은 서식은 한 번만 · [0] = 기본 `{}`.""" + + def __init__(self) -> None: + self.table: list[dict[str, Any]] = [{}] + self._index: dict[str, int] = {"{}": 0} + + def add(self, style: dict[str, Any]) -> int: + key = repr(sorted(style.items())) + if not style: + return 0 + if key not in self._index: + self._index[key] = len(self.table) + self.table.append(style) + return self._index[key] + + +def _rgb(book: Any, index: int) -> str | None: + got = book.colour_map.get(index) + return "#{:02X}{:02X}{:02X}".format(*got) if got else None + + +def _xf_style(book: Any, xf: Any, base: Any) -> dict[str, Any]: + font = book.font_list[xf.font_index] + out: dict[str, Any] = {} + if font.name != base.name: + out["글꼴"] = font.name + if font.height != base.height: + out["크기"] = font.height / 20 + for key, on in ( + ("굵게", font.bold), + ("기울임", font.italic), + ("밑줄", font.underline_type), + ("취소선", font.struck_out), + ): + if on: + out[key] = True + color = _rgb(book, font.colour_index) + if color and color != "#000000": + out["글자색"] = color + if xf.background.fill_pattern == 1 and (fill := _rgb(book, xf.background.pattern_colour_index)): + out["채움"] = fill + align = xf.alignment + if align.hor_align in _HALIGN: + out["가로"] = _HALIGN[align.hor_align] + if align.vert_align in _VALIGN: + out["세로"] = _VALIGN[align.vert_align] + if align.text_wrapped: + out["줄바꿈"] = True + sides = {} + for name, side in _SIDES: + line = getattr(xf.border, f"{side}_line_style") + if line: + one = {"선": _LINES.get(line, "dashed")} + color = _rgb(book, getattr(xf.border, f"{side}_colour_index")) + if color and color != "#000000": + one["색"] = color + sides[name] = one + if sides: + out["테두리"] = sides + fmt = book.format_map.get(xf.format_key) + if fmt and fmt.format_str and fmt.format_str != "General": + out["형식"] = fmt.format_str + return out + + +def _value(cell: Any) -> Any: + import xlrd + + if cell.ctype == xlrd.XL_CELL_TEXT: + return cell.value or None + if cell.ctype in (xlrd.XL_CELL_NUMBER, xlrd.XL_CELL_DATE): + return int(cell.value) if cell.value == int(cell.value) else cell.value + if cell.ctype == xlrd.XL_CELL_BOOLEAN: + return bool(cell.value) + if cell.ctype == xlrd.XL_CELL_ERROR: + return xlrd.error_text_from_code.get(cell.value, "#N/A") + return None + + +def _xls_sheet(book: Any, sh: Any, styles: _Styles, sid: str) -> dict[str, Any]: + base = book.font_list[0] + by_xf: dict[int, int] = {} + cells: dict[str, dict[str, Any]] = {} + for r in range(sh.nrows): + for c in range(sh.ncols): + cell = sh.cell(r, c) + if cell.xf_index not in by_xf: + by_xf[cell.xf_index] = styles.add( + _xf_style(book, book.xf_list[cell.xf_index], base) + ) + one: dict[str, Any] = {} + if (value := _value(cell)) is not None: + one["값"] = value + if by_xf[cell.xf_index]: + one["서식"] = by_xf[cell.xf_index] + if one: + cells[f"{get_column_letter(c + 1)}{r + 1}"] = one + merged = [ + f"{get_column_letter(c0 + 1)}{r0 + 1}:{get_column_letter(c1)}{r1}" + for r0, r1, c0, c1 in sh.merged_cells + ] + cols = { + get_column_letter(c + 1): { + "폭": round(info.width / 256, 2), + **({"숨김": True} if info.hidden else {}), + } + for c, info in sh.colinfo_map.items() + if c < sh.ncols + } + rows = { + str(r + 1): {"높이": info.height / 20, **({"숨김": True} if info.hidden else {})} + for r, info in sh.rowinfo_map.items() + if r < sh.nrows and (info.height != sh.default_row_height or info.hidden) + } + frozen = sh.panes_are_frozen and (sh.horz_split_pos or sh.vert_split_pos) + freeze = {"행": sh.horz_split_pos, "열": sh.vert_split_pos} if frozen else {"행": HEAD_ROWS} + out = { + "id": sid, + "이름": sh.name.strip(), + "칸": cells, + "병합": merged, + "열": cols, + "행": rows, + "틀고정": {k: v for k, v in freeze.items() if v}, + } + if sh.default_row_height: + out["기본"] = {"행높이": sh.default_row_height / 20} + return out + + +def _xls(data: bytes) -> dict[str, Any]: + import xlrd + + book = xlrd.open_workbook(file_contents=data, formatting_info=True, logfile=io.StringIO()) + styles = _Styles() + found = {sh.name.strip(): sh for sh in book.sheets()} + sheets = [ + _xls_sheet(book, found[name], styles, f"s{at}") + for at, name in enumerate((n for n in SHEETS if n in found), 1) + ] + font = book.font_list[0] + return { + "기본": {"글꼴": font.name, "크기": font.height / 20}, + "서식": styles.table, + "시트": sheets, + } + + +def _xlsx(name: str, data: bytes) -> dict[str, Any]: + # ponytail: xlsx 는 값만(병합 · 폭 · 서식은 openpyxl 비 read_only 로 무거움) — 필요하면 그때 + found = hopyo_excel._sheets(name, data) + sheets = [] + for at, title in enumerate((n for n in SHEETS if n in found), 1): + rows = found[title] + width = max( + (c + 1 for row in rows[:HEAD_ROWS] for c, v in enumerate(row) if v not in (None, "")), + default=0, + ) + cells = { + f"{get_column_letter(c + 1)}{r + 1}": { + "값": v if isinstance(v, (int, float, bool)) else str(v) + } + for r, row in enumerate(rows) + for c, v in enumerate(row[:width]) + if v not in (None, "") + } + sheets.append({"id": f"s{at}", "이름": title, "칸": cells, "틀고정": {"행": HEAD_ROWS}}) + return {"서식": [{}], "시트": sheets} + + +def read(name: str, data: bytes, parsed: dict[str, Any]) -> dict[str, Any]: + """엑셀 → `{책: Workbook, 호표, 단산}` — `parsed` = `M02_Hopyo_Excel.read_book` 결과 + (행 범위).""" + xlsx = data[:4] == b"PK\x03\x04" or name.lower().endswith(".xlsx") + book = _xlsx(name, data) if xlsx else _xls(data) + keep = ("번", "이름", "행") + return { + "책": {"종류": "통합문서", "판": 1, "열": "", **book, "활성": "s1"}, + "호표": [{k: h.get(k) for k in keep} for h in parsed.get("호표") or []], + "단산": [{k: b.get(k) for k in keep} for b in parsed.get("단산") or []], + } diff --git a/M02_MasterTemplete/M02_Hopyo_Match.py b/M02_MasterTemplete/M02_Hopyo_Match.py index dc894c8a4..c9ccbc557 100644 --- a/M02_MasterTemplete/M02_Hopyo_Match.py +++ b/M02_MasterTemplete/M02_Hopyo_Match.py @@ -4,6 +4,8 @@ 씨앗(일위대가조합.json 메모 · 읽기만) ③ 정리 이름 같음(괄호 · 기계 낱말) ④ 후보 5(글자 두 개 묶음 점수 + 단위 맞음 + 산림품셈 덤 · 뜻 낱말이 하나도 안 겹치면 뺌 — 「잔토처리」 ↔ 「소금처리」) — ++ 근거 후보(330-2 · `M02_Hopyo_Basis` · 단산 줄 출처 · 직종 · 이름 거름 안 거침 · +로직 키로 합쳐 점수 더함) · ③ ④ 는 사용자 확정(층 기억) 뒤에만 들어감. 줄 갈래: 단가산출 · 기계 · 기타 = 로직 / 자재 · 별도계상 = 자재(M01 자재품목) / 호표 = 그 호표 / % = 안 들어감(로직 덧줄이 셈). @@ -27,6 +29,7 @@ from typing import Any from M01_MasterData import M01_MasterData_Store as m01 from M01_MasterData import M01_MasterData_Store_Combo as combo_store +from M02_MasterTemplete import M02_Hopyo_Basis as hopyo_basis from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store from M02_MasterTemplete import M02_Labor_Inputs as labor_inputs from M02_MasterTemplete import M02_Labor_Ref as refs @@ -295,9 +298,10 @@ def _exact(line: dict[str, Any], pool: str, got: dict[str, Any]) -> list[dict[st def candidates( - line: dict[str, Any], pool: str, got: dict[str, Any] | None = None, limit: int = 5 + line: dict[str, Any], pool: str, got: dict[str, Any] | None = None, limit: int | None = 5 ) -> list[dict[str, Any]]: - """③ 정리 이름 같음 먼저 · ④ 글자 두 개 묶음 점수 + 단위 맞음 + 산림품셈 덤 — 위 `limit`.""" + """③ 정리 이름 같음 먼저 · ④ 글자 두 개 묶음 점수 + 단위 맞음 + 산림품셈 덤 — 위 `limit` + (None = 모두).""" me = tidy(line.get("이름")) grams, unit = _grams(me), line.get("단위") text = mm.plain_spec(f"{line.get('이름')}{line.get('규격')}") @@ -402,11 +406,9 @@ def match_line( return out top = [{**_public(x), "점수": 9} for x in found] same = [_public(x) for x in found] - rest = [ - c - for c in candidates(line, pool, got) - if {k: v for k, v in c.items() if k != "점수"} not in same - ] + basis = hopyo_basis.candidates(line, got) if pool == "로직" else [] + named = candidates(line, pool, got, None if basis else 5) + rest = [c for c in hopyo_basis.merge(basis, named) if _public(c) not in same] out["후보"] = (top + rest)[:5] return {**out, "상태": "후보" if out["후보"] else "없음"} @@ -461,7 +463,8 @@ def view(folder: Path, key: str, no: int | None = None) -> dict[str, Any]: if no is None: got = targets() hopyos = [match_hopyo(h, memo, units, False, got) for h in doc.get("호표") or []] - brief = [{k: h[k] for k in ("번", "이름", "규격", "단위", "금액", "맞춤")} for h in hopyos] + keep = ("번", "이름", "규격", "단위", "금액", "맞춤") + brief = [{k: h.get(k) for k in keep} for h in hopyos] ready = [sum(h["맞춤"][0] for h in hopyos), sum(h["맞춤"][1] for h in hopyos)] return {**hopyo_store.brief(doc), "검산": doc.get("검산"), "호표": brief, "맞춤": ready} hopyo = next((h for h in doc.get("호표") or [] if int(h["번"]) == int(no)), None) diff --git a/M02_MasterTemplete/M02_Hopyo_Store.py b/M02_MasterTemplete/M02_Hopyo_Store.py index c632a409d..b82ad7b51 100644 --- a/M02_MasterTemplete/M02_Hopyo_Store.py +++ b/M02_MasterTemplete/M02_Hopyo_Store.py @@ -1,5 +1,5 @@ """M02 일위대가 실무자료 저장(PLAN 324-1) — 층 폴더 `실무자료/<키>.json` · -맞춤 기억 `실무자료/맞춤.json`. +맞춤 기억 `실무자료/맞춤.json` · 엑셀꼴 격자 `실무자료/격자/<키>.json`(329-1 · 파일과 같이 지움). 파일 = 읽은 결과만(원본 엑셀 안 둠) `{키, 파일, 올린때, 올린이, 시트, 호표, 검산}` · 키 `HP-######` (층마다 다음 번호) · 맞춤 기억 = `{줄: {줄 열쇠: {로직?|자재?|빼기?, 입력값, 환산, 확정}}}`. @@ -20,6 +20,8 @@ from M02_MasterTemplete import M02_MasterTemplete_Store as store DIRNAME = "실무자료" MEMORY = "맞춤" +GRID = "격자" +NO_GRID = "격자 없음 — 다시 올리면 보임" _KEY = re.compile(r"^HP-(\d{6})$") # ponytail: 프로세스 안 잠금 — 서버를 여럿 띄우면 파일 잠금으로 _LOCK = threading.Lock() @@ -42,6 +44,10 @@ def _load(path: Path) -> Any: return None +def _grid_path(folder: str | Path, key: str) -> Path: + return _dir(folder) / GRID / _path(folder, key).name + + def brief(doc: dict[str, Any]) -> dict[str, Any]: """목록 한 줄 — `{키, 파일, 올린때, 올린이, 시트, 호표수, 어긋남}`.""" keep = ("키", "파일", "올린때", "올린이", "시트") @@ -69,8 +75,14 @@ def read(folder: str | Path, key: str) -> dict[str, Any]: return doc -def save(folder: str | Path, name: str, parsed: dict[str, Any], who: Any) -> dict[str, Any]: - """읽은 결과 → 새 키로 씀 → 목록 한 줄.""" +def save( + folder: str | Path, + name: str, + parsed: dict[str, Any], + who: Any, + grid: dict[str, Any] | None = None, +) -> dict[str, Any]: + """읽은 결과(+ 격자) → 새 키로 씀 → 목록 한 줄.""" with _LOCK: have = [ int(m.group(1)) for p in _dir(folder).glob("HP-*.json") if (m := _KEY.match(p.stem)) @@ -83,16 +95,28 @@ def save(folder: str | Path, name: str, parsed: dict[str, Any], who: Any) -> dic "올린이": who, **parsed, } + if grid is not None: + book = {**grid["책"], "열": key} + atomic_write_json(_grid_path(folder, key), {**grid, "책": book}) atomic_write_json(_path(folder, key), doc) return brief(doc) +def grid(folder: str | Path, key: str) -> dict[str, Any]: + """엑셀꼴 격자 `{책, 호표, 단산}` — 없으면(옛 파일) 404.""" + doc = _load(_grid_path(folder, key)) + if not isinstance(doc, dict): + raise store.StoreError(404, NO_GRID) + return doc + + def delete(folder: str | Path, key: str) -> None: path = _path(folder, key) with _LOCK: if not path.is_file(): raise store.StoreError(404, f"없는 실무자료 「{key}」") path.unlink() + _grid_path(folder, key).unlink(missing_ok=True) # ── 맞춤 기억 ───────────────────────────────────────────────────────── diff --git a/M02_MasterTemplete/M02_Labor_Router.py b/M02_MasterTemplete/M02_Labor_Router.py index 1f028a4b1..55739aa3a 100644 --- a/M02_MasterTemplete/M02_Labor_Router.py +++ b/M02_MasterTemplete/M02_Labor_Router.py @@ -20,6 +20,7 @@ from pydantic import BaseModel from common_util.common_util_auth import verify_session from M02_MasterTemplete import M02_Hopyo_Build as hopyo_build from M02_MasterTemplete import M02_Hopyo_Excel as hopyo_excel +from M02_MasterTemplete import M02_Hopyo_Grid as hopyo_grid from M02_MasterTemplete import M02_Hopyo_Match as hopyo_match from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store from M02_MasterTemplete import M02_Labor_Doc as labor @@ -176,7 +177,11 @@ def _upload(name: str, data: bytes, who: dict[str, Any]) -> dict[str, Any]: parsed = hopyo_excel.read_book(name, data) except hopyo_excel.ExcelError as error: raise store.StoreError(400, str(error)) from error - return hopyo_store.save(store.root(), name, parsed, who) + try: + grid = hopyo_grid.read(name, data, parsed) + except Exception: # 격자는 보기용 덤 — 못 만들어도 올리기는 됨(보기 창 404) + grid = None + return hopyo_store.save(store.root(), name, parsed, who, grid) @router.get("/layers/{layer}/hopyo") @@ -205,7 +210,7 @@ def hopyo_upload( owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: - """엑셀(xls · xlsx) 올리기 — 「일위대가」 시트를 읽어 결과만 저장 · 20 MB · + """엑셀(xls · xlsx) 올리기 — 「일위대가」 시트를 읽어 결과 · 격자 저장 · 20 MB · 엉터리 · 시트 없음 400.""" folder = _root(session, layer, owner, write=True) data = file.file.read(HOPYO_MAX_BYTES + 1) @@ -225,6 +230,16 @@ def hopyo_read( return _call(folder, hopyo_match.view, folder, key) +@router.get("/layers/{layer}/hopyo-grid/{key}") +def hopyo_grid_read( + layer: Layer, key: str, owner: str | None = None, session: Session = Depends(verify_session) +) -> dict: + """엑셀꼴 격자(329-1) `{책: Workbook, 호표: [{번, 이름, 행}], 단산: [{번, 이름, 행}]}` · + 옛 파일 404.""" + folder = _root(session, layer, owner) + return _call(folder, hopyo_store.grid, folder, key) + + @router.get("/layers/{layer}/hopyo/{key}/{no}") def hopyo_one( layer: Layer, diff --git a/resources/tester/test_m02_hopyo_basis.py b/resources/tester/test_m02_hopyo_basis.py new file mode 100644 index 000000000..526bfd4cb --- /dev/null +++ b/resources/tester/test_m02_hopyo_basis.py @@ -0,0 +1,108 @@ +"""일위대가 실무자료 단가산출 근거 맞춤(PLAN 330 · `M02_Hopyo_Basis` · 맞춤 후보 `근거`). + +블록 북부청 95 · 사방 67 · 단산93 근거 [건설 3-2-1] · [특별인부] · +후보 93 → GC000083 · 21 → GC000198 · 64 → GC000143 위 둘 · 49 → GC000085 · +잔토처리에 철골재 철거 없음 · 근거 없는 줄 그대로 · 상태 「후보」(자동 확정 없음) · +master_data 그대로. +""" + +import pytest + +from M02_MasterTemplete import M02_Hopyo_Basis as hb +from M02_MasterTemplete import M02_Hopyo_Excel as hx +from M02_MasterTemplete import M02_Hopyo_Match as hm +from resources.tester import test_m02_hopyo_excel as base +from resources.tester.test_m02_hopyo_excel import ( + NORTH, + SABANG, + URL, + _post, + master_hash, +) + +client = base.client # 공용 고정물(임시 층) + + +@pytest.fixture(scope="module") +def north() -> dict: + return hx.read_book(NORTH.name, NORTH.read_bytes()) + + +def _line(book: dict, no: int) -> dict: + """단산 no 를 가리키는 첫 줄.""" + return next(x for h in book["호표"] for x in h["줄"] if x["참조"] == {"종류": "단산", "번": no}) + + +def test_블록_수_와_근거(north): + assert len(north["단산"]) == 95 + assert [b["번"] for b in north["단산"]] == list(range(1, 96)) + sabang = hx.read_book(SABANG.name, SABANG.read_bytes()) + assert len(sabang["단산"]) == 67 + assert _line(north, 93)["근거"] == { + "출처": [["건설품셈", "3-2-1"]], + "직종": ["특별인부"], + "행": [4229, 4244], + } + assert _line(north, 21)["근거"]["직종"] == ["콘크리트공", "보통인부"] # 「보 통 인부」 빈칸 뺌 + assert _line(north, 64)["근거"]["출처"][0] == ["건설품셈", "4-1-1"] + lines = [x for h in north["호표"] for x in h["줄"] if x["갈래"] == "단가산출"] + assert all("근거" in x for x in lines) # 268 줄 모두 블록 있음 + + +def test_출처_글(): + assert hb.sources("[건설공사표준품셈 3-2-1 인력굴착]") == [["건설품셈", "3-2-1"]] + assert hb.sources("건설표줌품셈(3-1-3) 자갈") == [["건설품셈", "3-1-3"]] + assert hb.sources("[2025년 산림사업표준품셈 9-14]") == [["산림품셈", "9-14"]] + assert hb.sources("흙운반 [임도표준품셈 4-2]") == [["산림품셈", "4-2"]] + assert hb.sources("[건설표준품셈 준용]") == [] + + +def _top(book: dict, no: int) -> list[dict]: + got = hm.match_line(_line(book, no), [], hm.units_of(book)) + assert got["상태"] == "후보" and got["들어감"] is False # 자동 확정 없음 + return got["후보"] + + +def test_근거_후보(north): + before = master_hash() + dig = _top(north, 93) + assert dig[0]["로직"] == "GC000083" + assert dig[0]["근거"] == { + "조항": ["건설품셈 공통 3-2-1"], + "직종": ["특별인부"], + "단위": True, + "점수": 1.75, + "까닭": "조항 같음 · 건설 공통 · 직종 특별인부 · 단위 맞음", + } + assert _top(north, 21)[0]["로직"] == "GC000198" + assert "GC000143" in [c["로직"] for c in _top(north, 64)[:2]] + assert _top(north, 49)[0]["로직"] == "GC000085" + soil = next(x for x in north["호표"][2]["줄"] if x["이름"].startswith("잔토처리")) + names = [c["이름"] for c in hm.match_line(soil, [], {})["후보"]] + assert not any("철골재" in n for n in names) + assert master_hash() == before + + +def test_근거_없는_줄_그대로(north): + bare = [ + x + for h in north["호표"] + for x in h["줄"] + if x["갈래"] == "단가산출" and not x["근거"]["출처"] + ] + assert bare + for line in bare: + plain = {k: v for k, v in line.items() if k != "근거"} + assert hm.match_line(line, [], {}) == hm.match_line(plain, [], {}) + + +def test_길_응답_근거_칸(client): + before = master_hash() + assert _post(client).status_code == 200 + res = client.get(f"{URL.format('personal')}/HP-000001/3") + lines = {x["이름"]: x for x in res.json()["호표"]["줄"]} + dig = lines["구조물터파기(인력100%)"] + assert dig["근거"]["출처"] == [["건설품셈", "3-2-1"]] + assert dig["맞춤"]["후보"][0]["근거"]["까닭"].startswith("조항 같음") + assert "근거" not in lines["임도표지석(받침,글새김포함)"] # 자재 줄 + assert master_hash() == before diff --git a/resources/tester/test_m02_hopyo_grid.py b/resources/tester/test_m02_hopyo_grid.py new file mode 100644 index 000000000..16ba2b303 --- /dev/null +++ b/resources/tester/test_m02_hopyo_grid.py @@ -0,0 +1,120 @@ +"""일위대가 실무자료 엑셀꼴 격자(PLAN 329-1 · `M02_Hopyo_Grid` · `GET hopyo-grid/{키}`). + +북부청 임도 xls — 병합 11 · 15열 · B20 「제3호표」 · `#,##0` · 호표 68 행 범위 안 겹침 · +제3호표 [20, 24] · 올리면 격자도 · 200 / 옛 파일 404 · 지우면 격자도 · 역할 · xlsx 값만 · +master_data 그대로. +""" + +import pytest + +from M02_MasterTemplete import M02_Hopyo_Excel as hx +from M02_MasterTemplete import M02_Hopyo_Grid as hg +from M02_MasterTemplete import M02_Hopyo_Store as hs +from M02_MasterTemplete import M02_Template_Layers as layers +from resources.tester import test_m02_hopyo_excel as base +from resources.tester.test_m02_hopyo_excel import ( + NORTH, + URL, + _post, + _xlsx, + master_hash, +) + +client = base.client # 공용 고정물(임시 층) +GRID = "/api/m02/layers/{}/hopyo-grid/{}" + + +@pytest.fixture(scope="module") +def north() -> dict: + data = NORTH.read_bytes() + return hg.read(NORTH.name, data, hx.read_book(NORTH.name, data)) + + +def _sheet(grid: dict, name: str) -> dict: + return next(s for s in grid["책"]["시트"] if s["이름"] == name) + + +def test_격자_모양(north): + book = north["책"] + assert (book["종류"], book["판"], book["활성"]) == ("통합문서", 1, "s1") + assert [s["이름"] for s in book["시트"]] == ["일위대가", "단가산출"] + sheet = _sheet(north, "일위대가") + assert len(sheet["병합"]) == 11 and "B1:O2" in sheet["병합"] + assert len(sheet["열"]) == 15 and sheet["열"]["C"]["폭"] == round(7208 / 256, 2) + assert sheet["틀고정"] == {"행": 4} + assert sheet["칸"]["B20"]["값"] == "제3호표" + assert sheet["칸"]["H5"]["값"] == 1040052 + styles = book["서식"] + assert styles[0] == {} and any(s.get("형식") == "#,##0" for s in styles) + assert any(s.get("테두리") for s in styles) + assert _sheet(north, "단가산출")["칸"]["B4"]["값"] == "1.유용토운반(무대)[㎥]" + + +def test_호표_단산_행_범위(north): + rows = [h["행"] for h in north["호표"]] + assert len(rows) == 68 + assert all(a[1] < b[0] for a, b in zip(sorted(rows), sorted(rows)[1:], strict=False)) + assert next(h for h in north["호표"] if h["번"] == 3) == { + "번": 3, + "이름": "임도표지석", + "행": [20, 24], + } + assert len(north["단산"]) == 95 + assert next(b for b in north["단산"] if b["번"] == 93)["행"] == [4229, 4244] + + +def test_줄_행(north): + book = hx.read_book(NORTH.name, NORTH.read_bytes()) + three = next(h for h in book["호표"] if h["번"] == 3) + assert three["행"] == [20, 24] and [x["행"] for x in three["줄"]] == [21, 22, 23, 24] + + +def test_올리기_격자_200_404_지우기(client): + before = master_hash() + assert _post(client).status_code == 200 + res = client.get(GRID.format("personal", "HP-000001")) + assert res.status_code == 200, res.text + got = res.json() + assert got["책"]["열"] == "HP-000001" and len(got["호표"]) == 68 + folder = layers.personal_dir(7, 42) + path = folder / hs.DIRNAME / hs.GRID / "HP-000001.json" + assert path.is_file() + assert [r["키"] for r in client.get(URL.format("personal")).json()["줄"]] == ["HP-000001"] + path.unlink() # 옛 파일(324) = 격자 없음 + old = client.get(GRID.format("personal", "HP-000001")) + assert old.status_code == 404 and old.json()["detail"] == "격자 없음 — 다시 올리면 보임" + assert _post(client).status_code == 200 + assert client.get(GRID.format("personal", "HP-000002")).status_code == 200 + assert client.delete(URL.format("personal") + "/HP-000002").status_code == 200 + assert not (folder / hs.DIRNAME / hs.GRID / "HP-000002.json").exists() + assert client.get(GRID.format("personal", "HP-000002")).status_code == 404 + assert client.get(GRID.format("personal", "../x")).status_code == 404 + assert client.get(URL.format("personal") + "/HP-000001/3").status_code == 200 # 길 안 부딪힘 + assert master_hash() == before + + +@pytest.mark.parametrize( + ("role", "layer", "owner", "want"), + [ + ({"role": "USER"}, "personal", None, 200), + ({"role": "USER"}, "personal", "7/43", 403), + ({"role": "USER"}, "company", None, 200), + ({"role": "USER"}, "system", None, 200), + ], +) +def test_역할_읽기(client, role, layer, owner, want): + client.who.update({"role": "SYSTEM_ADMIN"}) + assert _post(client, layer if want == 200 else "personal").status_code == 200 + client.who.update(role) + params = {"owner": owner} if owner else {} + assert client.get(GRID.format(layer, "HP-000001"), params=params).status_code == want + + +def test_xlsx_값만(): + data = _xlsx() + grid = hg.read("a.xlsx", data, hx.read_book("a.xlsx", data)) + sheet = grid["책"]["시트"][0] + assert sheet["이름"] == "일위대가" and sheet["틀고정"] == {"행": 4} + assert sheet["칸"]["B5"] == {"값": "제 1 호표"} and sheet["칸"]["E6"] == {"값": 0.5} + assert "K6" not in sheet["칸"] # 머리 글(앞 4 줄) 마지막 열 J 까지 + assert grid["호표"] == [{"번": 1, "이름": "표지석", "행": [5, 6]}] and grid["단산"] == [] diff --git a/resources/tester/test_m02_hopyo_match.py b/resources/tester/test_m02_hopyo_match.py index 0f1224240..69f8cd397 100644 --- a/resources/tester/test_m02_hopyo_match.py +++ b/resources/tester/test_m02_hopyo_match.py @@ -105,7 +105,7 @@ def test_이름_같음_후보_호표_퍼센트(client): assert all(c["규격"] == "25-21-120" for c in remicon["후보"][:3]) # 규격 같은 자재 먼저 dig = _hopyo(client, 3)["구조물터파기(인력100%)"] assert dig["상태"] == "후보" and len(dig["후보"]) == 5 - assert dig["후보"][0]["이름"].startswith("구조물터파기") + assert dig["후보"][0]["이름"] == "굴착(인력/토사)" # 330-2 근거(건설 3-2-1 · 특별인부) 1순위 ref = _hopyo(client, 45)["고화매트재료비"] assert ref["상태"] == "호표" and ref["대상"] == {"호표": 46, "단위": "㎡"} and ref["들어감"] pct = _hopyo(client, 46)["잡재료비"]