From 94a7fd580845fbb28742c96e20b03097d28ac3c1 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 25 Oct 2026 19:47:12 +0900 Subject: [PATCH] =?UTF-8?q?feat(528-4=20=C2=B7=20532-1):=20=EA=B3=B5?= =?UTF-8?q?=EC=9A=A9=20=ED=94=84=EB=A1=9C=EC=A0=9D=ED=8A=B8=20=EB=B3=B5?= =?UTF-8?q?=EC=82=AC=20=C2=B7=20E01=20=EC=95=88=EC=A0=95=EC=84=B1=ED=8F=89?= =?UTF-8?q?=EA=B0=80=20=EC=83=9D=EC=84=B1=20=EB=B3=B5=EC=82=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - common_util_project_copy — 복제 도구 본체 이전 · 범위 full | stability · 파일 안 UUID 치환 · copy_manifest.json - tools_clone_project — CLI 껍데기(동작 그대로 · main(argv)) - E01 생성 = stability 복사(E01_StabRegister/source/ · 동결 db.json + B05 route · B06 종횡단 · 쓰인 DTM 하나 · 최신 corridor · 관 편집분) · 지문 · 복사 시각 기록 · 실패 = 만든 E 지움 - E01_StabRegister_Source — 복사본 읽기(측점 · 횡단 · 지형 격자 · corridor · 구조물) - 시험 test_532_1_project_copy(임시 storage) · test_528 E01 생성에 복사 응답 반영 Co-Authored-By: Claude Opus 5.5 --- E01_StabRegister/E01_StabRegister_Router.py | 41 +- E01_StabRegister/E01_StabRegister_Source.py | 134 ++++++ common_util/common_util_project_copy.py | 470 ++++++++++++++++++++ db_management/tools_clone_project.py | 296 +----------- resources/tester/test_528_project_groups.py | 8 +- resources/tester/test_532_1_project_copy.py | 203 +++++++++ 6 files changed, 872 insertions(+), 280 deletions(-) create mode 100644 E01_StabRegister/E01_StabRegister_Source.py create mode 100644 common_util/common_util_project_copy.py create mode 100644 resources/tester/test_532_1_project_copy.py diff --git a/E01_StabRegister/E01_StabRegister_Router.py b/E01_StabRegister/E01_StabRegister_Router.py index 0d89c5822..38ad6d57d 100644 --- a/E01_StabRegister/E01_StabRegister_Router.py +++ b/E01_StabRegister/E01_StabRegister_Router.py @@ -1,8 +1,13 @@ """E01 안정성평가 생성 API (요구 345 · 347 · 요구서 4-1). -대상 = 같은 회사 임도(ROAD) 중 B06 횡단 단계(stage 3) 완료. 자료 복사는 528-4 — 지금은 행만. +대상 = 같은 회사 임도(ROAD) 중 B06 횡단 단계(stage 3) 완료. 생성 때 해석에 쓰는 자료만 +`E01_StabRegister/source/` 로 동결 복사(532-1 · 원본이 바뀌어도 불변) +— 실패하면 만든 행 · 폴더를 지움. """ +import logging +from datetime import datetime +from pathlib import Path from typing import Any import aiomysql @@ -10,9 +15,14 @@ from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel, Field from common_util.common_util_auth import require_company +from common_util.common_util_project_copy import copy_project +from common_util.common_util_project_delete import hard_delete_project from common_util.common_util_project_kind import create_kind_project +from common_util.common_util_storage import resolve_stored_project_path from config.config_db import get_db_pool +logger = logging.getLogger(__name__) + router = APIRouter(prefix="/api/e01", tags=["E01_StabRegister"]) _SOURCE_SQL = """ @@ -84,4 +94,31 @@ async def post_stab_project( project_year=source["project_year"], source_project_id=source["project_id"], ) - return {"project_id": project_id} + try: + manifest = await _copy_source(project_id, source["project_id"]) + except Exception: + logger.exception("E01 복사 실패 — 만든 프로젝트를 지움: %s", project_id) + await hard_delete_project(project_id, int(session["user_id"]), request) + raise HTTPException(status_code=500, detail="원본 자료 복사에 실패했습니다.") from None + return {"project_id": project_id, "source_fingerprint": manifest["source_fingerprint"]} + + +async def _copy_source(project_id: str, source_id: str) -> dict[str, Any]: + """원본 → E 복사본 + 지문 · 복사 시각 기록.""" + async with get_db_pool().acquire() as connection: + async with connection.cursor() as cursor: + await cursor.execute("SELECT storage_path FROM projects WHERE id = %s", (project_id,)) + (storage_path,) = await cursor.fetchone() + manifest = await copy_project( + connection, + source_id, + scope="stability", + new_id=project_id, + project_root=Path(resolve_stored_project_path(storage_path)), + ) + async with connection.cursor() as cursor: + await cursor.execute( + "UPDATE projects SET source_fingerprint = %s, source_copied_at = %s WHERE id = %s", + (manifest["source_fingerprint"], datetime.utcnow(), project_id), + ) + return manifest diff --git a/E01_StabRegister/E01_StabRegister_Source.py b/E01_StabRegister/E01_StabRegister_Source.py new file mode 100644 index 000000000..ba4895d65 --- /dev/null +++ b/E01_StabRegister/E01_StabRegister_Source.py @@ -0,0 +1,134 @@ +"""E 안정성평가 복사본 읽기 — storage 를 아는 유일한 자리 (요구 345 · 532-1). + +복사본 = E 프로젝트 저장소 `E01_StabRegister/source/`(`common_util_project_copy.copy_stability` +이 생성 때 한 번 만듦 · 그 뒤 불변). E 엔진 · 화면은 이 함수들로만 받음 — 원본 임도 프로젝트를 +읽지 않음. +""" + +from __future__ import annotations + +import json +from functools import lru_cache +from pathlib import Path +from typing import Any + +import aiomysql + +from common_util.common_util_project_copy import ( + MANIFEST_NAME, + STABILITY_DB_NAME, + STABILITY_SOURCE_DIR, +) +from common_util.common_util_storage import resolve_stored_project_path +from config.config_db import get_db_pool + + +def source_dir(project_root: Path | str) -> Path: + """E 프로젝트 루트 → 복사본 폴더.""" + return Path(project_root) / STABILITY_SOURCE_DIR + + +async def project_source_dir(project_id: str) -> Path: + """E project_id → 복사본 폴더(없으면 LookupError).""" + async with ( + get_db_pool().acquire() as connection, + connection.cursor(aiomysql.DictCursor) as cursor, + ): + await cursor.execute( + "SELECT storage_path FROM projects" + " WHERE id = %s AND project_kind = 'STABILITY' AND deleted_at IS NULL", + (project_id,), + ) + row = await cursor.fetchone() + if not row: + raise LookupError(f"안정성평가 프로젝트가 아님: {project_id}") + root = source_dir(resolve_stored_project_path(row["storage_path"])) + if not (root / MANIFEST_NAME).is_file(): + raise LookupError(f"복사본이 없음: {project_id}") + return root + + +@lru_cache(maxsize=8) +def _read_json(path: str, mtime_ns: int) -> Any: + return json.loads(Path(path).read_text(encoding="utf-8")) + + +def _json(path: Path) -> Any: + return _read_json(str(path), path.stat().st_mtime_ns) + + +def manifest(root: Path) -> dict[str, Any]: + """복사 목록 — 원본 id · 복사 시각 · 지문 · 파일.""" + return _json(root / MANIFEST_NAME) + + +def db_snapshot(root: Path) -> dict[str, Any]: + """동결 DB 행 — project · project_workflow_stages · surface_params · route · + route_points · route_statistics · longitudinal_sections · cross_sections.""" + return _json(root / STABILITY_DB_NAME) + + +def _cross_rows(root: Path) -> list[dict[str, Any]]: + rows = db_snapshot(root)["cross_sections"] + return sorted(rows, key=lambda row: (float(row["chainage_m"] or 0.0), row["sequence_num"] or 0)) + + +def stations(root: Path) -> list[dict[str, Any]]: + """측점 목록(누가 차례) — chainage_m · label · 중심 좌표 · 방위각 · 오르막 쪽 · 지반 유형.""" + out = [] + for row in _cross_rows(root): + data = json.loads(row["data"]) if isinstance(row["data"], str) else row["data"] + cross = _json(root / row["cross_section_file_path"]) + out.append( + { + "chainage_m": float(row["chainage_m"]), + "label": cross.get("label"), + "kind": cross.get("kind"), + "center_x": cross.get("center_x"), + "center_y": cross.get("center_y"), + "center_z": cross.get("center_z"), + "azimuth_deg": cross.get("azimuth_deg"), + "uphill_side": cross.get("uphill_side"), + "ground_type": (data.get("design") or {}).get("ground_type"), + } + ) + return out + + +def cross_section(root: Path, chainage_m: float) -> dict[str, Any]: + """한 측점 횡단 — 지반선(samples: offset_m · x · y · z · valid) · 좌표틀 · B06 설계값 + (`design`: design_line · road_edges · section_mode · 절성토 경사 · 소단 · ground_type · + rock_boundary_offset_m(암 경계 = 지반선을 그만큼 내린 선 · 암 측점만) …).""" + for row in _cross_rows(root): + if abs(float(row["chainage_m"]) - float(chainage_m)) < 1e-6: + data = json.loads(row["data"]) if isinstance(row["data"], str) else row["data"] + cross = _json(root / row["cross_section_file_path"]) + return { + "chainage_m": float(row["chainage_m"]), + "label": cross.get("label"), + "frame": cross.get("frame"), + "uphill_side": cross.get("uphill_side"), + "samples": cross.get("samples") or [], + "design": data.get("design") or {}, + } + raise KeyError(f"측점 없음: {chainage_m}") + + +def terrain_grid_path(root: Path) -> Path: + """복사한 지형 격자(DTM npz — 원본 설계가 쓴 그 하나).""" + found = sorted((root / "B04_PreProcess" / "models").glob("dtm_*.npz")) + if len(found) != 1: + raise FileNotFoundError(f"복사한 지형 격자가 하나가 아님: {[p.name for p in found]}") + return found[0] + + +def corridor_path(root: Path) -> Path | None: + """최신 노선 3D 노선면(없으면 None).""" + found = sorted((root / "B05_Profile" / "corridor").glob("corridor_*.json")) + return found[0] if found else None + + +def structures_path(root: Path) -> Path | None: + """노선 구조물 정본(벽 등 · 없으면 None).""" + path = root / "B05_Profile" / "route" / "structures.json" + return path if path.is_file() else None diff --git a/common_util/common_util_project_copy.py b/common_util/common_util_project_copy.py new file mode 100644 index 000000000..12783a998 --- /dev/null +++ b/common_util/common_util_project_copy.py @@ -0,0 +1,470 @@ +"""프로젝트 복사 — 통째 복제(full) · 안정성평가용 동결 복사(stability) (요구 345 · 528-4 · 532-1). + +원본은 읽기만 — SELECT 와 파일 읽기뿐. 복사 목록은 `copy_manifest.json` 에 남김. + +full DB 행(projects · project_members · input_files · processed_point_cloud · + surface_models · terrain_layers · routes · route_points · route_statistics · + longitudinal_sections · cross_sections · project_workflow_stages)을 새 id 로 다시 이어 + 한 트랜잭션에 넣고 저장소 폴더 통째(robocopy) — 복제 도구(`db_management/ + tools_clone_project.py`)의 본체. +stability DB 행은 넣지 않고 E 프로젝트 저장소 `E01_StabRegister/source/` 에 동결 스냅샷 + (`db.json`) + 해석에 쓰는 파일만(B05 노선 · 구조물 · B06 종횡단 · 쓰인 DTM 하나 · + 최신 노선 corridor · 관 편집분) — 원본 LAS · 전처리 점군 · B07~B09 · 초기값 안 가져옴. + +둘 다 복사한 JSON 안의 옛 프로젝트 UUID 를 새 UUID 로 바꿈. +""" + +from __future__ import annotations + +import hashlib +import json +import subprocess +import time +from datetime import datetime +from pathlib import Path +from typing import Any, Literal +from uuid import uuid4 + +import aiomysql + +from common_util.common_util_storage import resolve_stored_project_path +from common_util.common_util_surface_confirmation import get_surface_confirmation_params + +Scope = Literal["full", "stability"] +MANIFEST_NAME = "copy_manifest.json" +#: E 프로젝트 안 복사본 자리 — E 엔진은 `E01_StabRegister_Source` 로만 읽음 +STABILITY_SOURCE_DIR = "E01_StabRegister/source" +STABILITY_DB_NAME = "db.json" +#: stability 가 통째로 뜨는 폴더(프로젝트 루트 기준) +_STABILITY_TREES = ( + "B05_Profile/route", + "B06_Section/longitudinal", + "B06_Section/cross_sections", + "B04_PreProcess/drainage/edits", +) + + +async def _rows(cursor, sql, args): + await cursor.execute(sql, args) + return await cursor.fetchall() + + +async def _insert(cursor, table, row: dict) -> int: + names = ", ".join(f"`{key}`" for key in row) + holders = ", ".join(["%s"] * len(row)) + await cursor.execute(f"INSERT INTO `{table}` ({names}) VALUES ({holders})", tuple(row.values())) + return int(cursor.lastrowid) + + +def _copy_folder(source_root: Path, new_root: Path) -> None: + print(f"파일 복사 시작: {source_root.name} -> {new_root.name}") + started = time.perf_counter() + result = subprocess.run( + [ + "robocopy", + str(source_root), + str(new_root), + "/E", + "/NFL", + "/NDL", + "/NJH", + "/NJS", + "/MT:8", + ], + capture_output=True, + text=True, + ) + if result.returncode >= 8: + raise RuntimeError(f"robocopy 실패 (코드 {result.returncode})") + print(f"파일 복사 완료: {time.perf_counter() - started:.0f}초") + + +def replace_project_id(root: Path, source_id: str, new_id: str) -> list[str]: + """root 아래 JSON 안의 옛 프로젝트 UUID → 새 UUID · 바꾼 파일(상대 경로).""" + replaced = [] + for path in root.rglob("*.json"): + try: + text = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + if source_id not in text: + continue + path.write_text(text.replace(source_id, new_id), encoding="utf-8") + replaced.append(path.relative_to(root).as_posix()) + return replaced + + +def _fix_files(new_root: Path, source_id: str, new_id: str, model_map: dict) -> None: + print("프로젝트 id 치환:", replace_project_id(new_root, source_id, new_id)) + + snapshot = new_root / "initial_snapshot" / "db.json" + if snapshot.is_file(): + dump = json.loads(snapshot.read_text(encoding="utf-8")) + changed = False + for route in dump.get("routes", []): + old_model = route.get("surface_model_id") + if old_model in model_map: + route["surface_model_id"] = model_map[old_model] + changed = True + if changed: + snapshot.write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8") + print("초기값 스냅샷 지표면 id 교정:", changed) + + +def _write_manifest(root: Path, manifest: dict[str, Any]) -> None: + (root / MANIFEST_NAME).write_text( + json.dumps(manifest, ensure_ascii=False, indent=1, default=str), encoding="utf-8" + ) + + +def _tree_files(root: Path) -> dict[str, int]: + return { + path.relative_to(root).as_posix(): path.stat().st_size + for path in sorted(root.rglob("*")) + if path.is_file() + } + + +async def _source_rows(cursor, source_id: str) -> dict[str, list]: + """복제 대상 DB 행 — 표 이름 → 행 목록(부모 → 자식 차례).""" + one = "WHERE project_id = %s ORDER BY id" + rows = { + "projects": await _rows(cursor, "SELECT * FROM projects WHERE id = %s", (source_id,)), + "input_files": await _rows(cursor, f"SELECT * FROM input_files {one}", (source_id,)), + "processed_point_cloud": await _rows( + cursor, f"SELECT * FROM processed_point_cloud {one}", (source_id,) + ), + "surface_models": await _rows(cursor, f"SELECT * FROM surface_models {one}", (source_id,)), + "project_members": await _rows( + cursor, "SELECT * FROM project_members WHERE project_id = %s", (source_id,) + ), + "terrain_layers": await _rows( + cursor, + "SELECT t.* FROM terrain_layers t JOIN surface_models s" + " ON s.id = t.surface_model_id WHERE s.project_id = %s ORDER BY t.id", + (source_id,), + ), + "routes": await _rows(cursor, f"SELECT * FROM routes {one}", (source_id,)), + "project_workflow_stages": await _rows( + cursor, + "SELECT * FROM project_workflow_stages WHERE project_id = %s ORDER BY stage_no", + (source_id,), + ), + } + if not rows["projects"]: + raise LookupError(f"원본 프로젝트가 없음: {source_id}") + rows.update(await _route_children(cursor, [route["id"] for route in rows["routes"]])) + return rows + + +async def _route_children(cursor, route_ids: list[int]) -> dict[str, list]: + children: dict[str, list] = { + name: [] for name in ("route_points", "route_statistics", "longitudinal_sections") + } + children["cross_sections"] = [] + if route_ids: + holders = ", ".join(["%s"] * len(route_ids)) + for table in children: + children[table] = await _rows( + cursor, + f"SELECT * FROM {table} WHERE route_id IN ({holders}) ORDER BY id", + route_ids, + ) + return children + + +async def _insert_full( + connection, rows: dict[str, list], new_id: str, new_name: str, new_storage: str, owner, members +) -> dict[str, int]: + """DB 행을 새 id 로 다시 이어 한 트랜잭션에 넣음 → 지표면 id 짝(옛 → 새).""" + await connection.begin() + async with connection.cursor(aiomysql.DictCursor) as cursor: + project = dict(rows["projects"][0]) + project["id"] = new_id + project["name"] = new_name + project["storage_path"] = new_storage + if owner is not None: + project["user_id"] = owner + await _insert(cursor, "projects", project) + + member_ids = ( + list(members) + if members is not None + else [row["user_id"] for row in rows["project_members"]] + ) + for user in member_ids: + await _insert(cursor, "project_members", {"project_id": new_id, "user_id": user}) + + file_map: dict[int, int] = {} + for row in rows["input_files"]: + row = dict(row) + old = row.pop("id") + row["project_id"] = new_id + file_map[old] = await _insert(cursor, "input_files", row) + + cloud_map: dict[int, int] = {} + for row in rows["processed_point_cloud"]: + row = dict(row) + old = row.pop("id") + row["project_id"] = new_id + row["input_file_id"] = file_map.get(row["input_file_id"], row["input_file_id"]) + cloud_map[old] = await _insert(cursor, "processed_point_cloud", row) + + model_map: dict[int, int] = {} + for row in rows["surface_models"]: + row = dict(row) + old = row.pop("id") + row["project_id"] = new_id + row["source_file_id"] = file_map.get(row["source_file_id"], row["source_file_id"]) + row["processed_cloud_id"] = cloud_map.get( + row["processed_cloud_id"], row["processed_cloud_id"] + ) + model_map[old] = await _insert(cursor, "surface_models", row) + + for row in rows["terrain_layers"]: + row = dict(row) + row.pop("id") + row["surface_model_id"] = model_map[row["surface_model_id"]] + await _insert(cursor, "terrain_layers", row) + + route_map: dict[int, int] = {} + for row in rows["routes"]: + row = dict(row) + old = row.pop("id") + row["project_id"] = new_id + row["surface_model_id"] = model_map.get( + row["surface_model_id"], row["surface_model_id"] + ) + route_map[old] = await _insert(cursor, "routes", row) + + for table in ( + "route_points", + "route_statistics", + "longitudinal_sections", + "cross_sections", + ): + for row in rows[table]: + row = dict(row) + row.pop("id") + row["route_id"] = route_map.get(row["route_id"], row["route_id"]) + if "project_id" in row: + row["project_id"] = new_id + await _insert(cursor, table, row) + + for row in rows["project_workflow_stages"]: + row = dict(row) + row.pop("id") + row["project_id"] = new_id + params = row.get("params") + if params: + # 단계 설정에 박힌 옛 id 를 새 id 로 바꾼다(입력 파일·지표면 모델). + data = json.loads(params) + if data.get("input_file_id") is not None: + old_file = int(data["input_file_id"]) + data["input_file_id"] = str(file_map.get(old_file, old_file)) + if data.get("surface_model_id") is not None: + old_model = int(data["surface_model_id"]) + data["surface_model_id"] = model_map.get(old_model, old_model) + row["params"] = json.dumps(data, ensure_ascii=False) + await _insert(cursor, "project_workflow_stages", row) + await connection.commit() + print( + f"DB 복제 완료: 참여자 {len(member_ids)} · 입력파일 {len(file_map)}" + f" · 지표면 {len(model_map)} · 지형층 {len(rows['terrain_layers'])}" + f" · 노선 {len(route_map)}" + f" · 정점 {len(rows['route_points'])} · 횡단 {len(rows['cross_sections'])}" + f" · 단계 {len(rows['project_workflow_stages'])}" + ) + return model_map + + +async def clone_full( + connection, + source_id: str, + new_name: str, + *, + owner: int | None = None, + members: list[int] | None = None, +) -> str: + """통째 복제 → 새 project_id. 새 폴더 = 원본과 같은 회사/사용자 폴더.""" + async with connection.cursor(aiomysql.DictCursor) as cursor: + rows = await _source_rows(cursor, source_id) + source = rows["projects"][0] + new_id = str(uuid4()) + source_root = Path(resolve_stored_project_path(source["storage_path"])) + new_storage = "/".join(source["storage_path"].split("/")[:-1] + [new_id]) + new_root = source_root.parent / new_id + + # 1) 파일 먼저 - 실패하면 DB 를 건드리지 않은 채로 끝난다. + _copy_folder(source_root, new_root) + # 2) DB - 한 트랜잭션으로. + model_map = await _insert_full(connection, rows, new_id, new_name, new_storage, owner, members) + # 3) 파일 안에 박힌 옛 프로젝트 id 치환 + 초기값 스냅샷의 지표면 모델 id 교정. + _fix_files(new_root, source_id, new_id, model_map) + _write_manifest( + new_root, + { + "scope": "full", + "source_project_id": source_id, + "project_id": new_id, + "copied_at": datetime.now().isoformat(timespec="seconds"), + "tables": {name: len(items) for name, items in rows.items()}, + "files": _tree_files(new_root), + }, + ) + print(f"\n새 프로젝트: {new_name}\n id = {new_id}\n 저장소 = {new_storage}") + return new_id + + +async def clone_storage_only(connection, source_id: str, new_id: str) -> None: + """DB 는 SELECT 만 — 원본 · 새 프로젝트 행(이미 복제됨)으로 폴더 자리 · 지표면 id 짝.""" + paths = {} + models = {} + async with connection.cursor(aiomysql.DictCursor) as cursor: + for project_id in (source_id, new_id): + row = await _rows( + cursor, "SELECT storage_path FROM projects WHERE id = %s", (project_id,) + ) + paths[project_id] = row[0]["storage_path"] + rows = await _rows( + cursor, + "SELECT id FROM surface_models WHERE project_id = %s ORDER BY id", + (project_id,), + ) + models[project_id] = [item["id"] for item in rows] + # 복제는 지표면 행을 원본 id 차례대로 넣는다 — 차례로 짝 지음. + model_map = dict(zip(models[source_id], models[new_id], strict=True)) + new_root = Path(resolve_stored_project_path(paths[new_id])) + if any(path.is_file() for path in new_root.rglob("*")): + raise RuntimeError(f"새 폴더에 파일이 이미 있음: {new_root}") + _copy_folder(Path(resolve_stored_project_path(paths[source_id])), new_root) + _fix_files(new_root, source_id, new_id, model_map) + print(f"storage 만 복제 완료: {new_root}") + + +async def _stability_rows(connection, source_id: str) -> dict[str, Any]: + """E 가 읽는 DB 행 — 원본 행 · 단계 · 지표면 확정값 · 최신 노선과 그 자식 행.""" + async with connection.cursor(aiomysql.DictCursor) as cursor: + project = await _rows(cursor, "SELECT * FROM projects WHERE id = %s", (source_id,)) + if not project: + raise LookupError(f"원본 프로젝트가 없음: {source_id}") + stages = await _rows( + cursor, + "SELECT * FROM project_workflow_stages WHERE project_id = %s ORDER BY stage_no", + (source_id,), + ) + # 최신 노선 = B05 `get_latest_route` 와 같은 차례 + route = await _rows( + cursor, + "SELECT * FROM routes WHERE project_id = %s ORDER BY computed_at DESC, id DESC LIMIT 1", + (source_id,), + ) + models = await _rows( + cursor, "SELECT * FROM surface_models WHERE project_id = %s ORDER BY id", (source_id,) + ) + children = await _route_children(cursor, [route[0]["id"]] if route else []) + return { + "project": project[0], + "project_workflow_stages": stages, + "surface_params": await get_surface_confirmation_params(connection, source_id), + "route": route[0] if route else None, + "surface_models": models, + **children, + } + + +def _stability_files(source_root: Path, dump: dict[str, Any]) -> list[str]: + """복사할 파일(프로젝트 루트 기준 상대 경로).""" + files = [ + path.relative_to(source_root).as_posix() + for tree in _STABILITY_TREES + for path in sorted((source_root / tree).rglob("*")) + if path.is_file() + ] + params = dump["surface_params"] + models = source_root / "B04_PreProcess" / "models" + dtm = models / f"dtm_{params['source_filter']}.npz" + smooth = dtm.with_name(f"dtm_{params['source_filter']}_smooth.npz") + # 쓰인 지형 하나 = `build_surface_sampler` 가 여는 그 npz + used = smooth if params.get("smooth") and smooth.is_file() else dtm + for path in (used, models / "manifest.json"): + if path.is_file(): + files.append(path.relative_to(source_root).as_posix()) + if dump["route"]: + corridor = ( + source_root / "B05_Profile" / "corridor" / f"corridor_{dump['route']['id']:04d}.json" + ) + if corridor.is_file(): + files.append(corridor.relative_to(source_root).as_posix()) + return files + + +async def copy_stability(connection, source_id: str, new_id: str, project_root: Path) -> dict: + """E 프로젝트 루트 아래 `E01_StabRegister/source/` 로 동결 복사 → manifest.""" + started = time.perf_counter() + dump = await _stability_rows(connection, source_id) + source_root = Path(resolve_stored_project_path(dump["project"]["storage_path"])) + target = Path(project_root) / STABILITY_SOURCE_DIR + if target.exists(): + raise RuntimeError(f"복사본 자리가 이미 있음: {target}") + files = _stability_files(source_root, dump) + + digest = hashlib.sha1() + sizes: dict[str, int] = {} + for relative in files: + data = (source_root / relative).read_bytes() + digest.update(relative.encode("utf-8") + b"\0" + hashlib.sha1(data).digest()) + out = target / relative + out.parent.mkdir(parents=True, exist_ok=True) + out.write_bytes(data) + sizes[relative] = len(data) + replaced = replace_project_id(target, source_id, new_id) + + db_text = json.dumps(dump, ensure_ascii=False, default=str) + digest.update(STABILITY_DB_NAME.encode("utf-8") + b"\0" + db_text.encode("utf-8")) + (target / STABILITY_DB_NAME).write_text(db_text, encoding="utf-8") + sizes[STABILITY_DB_NAME] = len(db_text.encode("utf-8")) + manifest = { + "scope": "stability", + "source_project_id": source_id, + "project_id": new_id, + "copied_at": datetime.now().isoformat(timespec="seconds"), + "source_fingerprint": digest.hexdigest(), + "route_id": dump["route"]["id"] if dump["route"] else None, + "tables": { + "cross_sections": len(dump["cross_sections"]), + "route_points": len(dump["route_points"]), + "longitudinal_sections": len(dump["longitudinal_sections"]), + }, + "uuid_replaced": replaced, + "total_bytes": sum(sizes.values()), + "elapsed_s": round(time.perf_counter() - started, 2), + "files": sizes, + } + _write_manifest(target, manifest) + return manifest + + +async def copy_project( + connection, + source_id: str, + *, + scope: Scope, + new_name: str | None = None, + new_id: str | None = None, + project_root: Path | None = None, + owner: int | None = None, + members: list[int] | None = None, +) -> dict[str, Any]: + """범위별 복사 — full = 새 프로젝트 행까지(new_name) · stability = 만든 E 프로젝트(new_id · + project_root) 안 복사본만.""" + if scope == "full": + if not new_name: + raise ValueError("full 복사는 new_name 필요") + new = await clone_full(connection, source_id, new_name, owner=owner, members=members) + return {"scope": scope, "project_id": new} + if scope == "stability": + if not new_id or project_root is None: + raise ValueError("stability 복사는 new_id · project_root 필요") + return await copy_stability(connection, source_id, new_id, project_root) + raise ValueError(f"모르는 복사 범위: {scope}") diff --git a/db_management/tools_clone_project.py b/db_management/tools_clone_project.py index f38317b82..9e816cdb4 100644 --- a/db_management/tools_clone_project.py +++ b/db_management/tools_clone_project.py @@ -1,5 +1,7 @@ """프로젝트 통째 복제 — **검증용 프로젝트를 창마다 하나씩 뜨는 도구** (2026-09-08). +본체 = `common_util/common_util_project_copy.py`(full 범위) · 이 파일은 CLI 껍데기. + ⚠ **왜 저장소에 두나** — 네 창이 **같은 원본에서 같은 방법으로** 떠야 수치를 견줄 수 있다. `tmp/` 는 창마다 따로라 서로 못 봄(2026-09-08 랩탑 두 창이 막힌 자리). @@ -27,22 +29,18 @@ import argparse import asyncio -import json -import subprocess import sys -import time from pathlib import Path -from uuid import uuid4 sys.path.append(str(Path(__file__).resolve().parent.parent)) import aiomysql -from common_util.common_util_storage import resolve_stored_project_path +from common_util.common_util_project_copy import clone_full, clone_storage_only from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER -def _args() -> argparse.Namespace: +def _args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description="프로젝트를 통째로 복제한다(원본은 읽기만).") parser.add_argument("source_id", help="원본 프로젝트 UUID") parser.add_argument("new_name", help="새 프로젝트 이름(--storage-only 면 새 프로젝트 UUID)") @@ -53,104 +51,11 @@ def _args() -> argparse.Namespace: ) parser.add_argument("--owner", type=int, help="새 주인 사용자 id(없으면 원본 주인)") parser.add_argument("--members", help="참여자 사용자 id 쉼표 목록(없으면 원본 참여자)") - return parser.parse_args() + return parser.parse_args(argv) -ARGS = _args() -SOURCE_ID = ARGS.source_id -NEW_NAME = ARGS.new_name - - -async def _rows(cursor, sql, args): - await cursor.execute(sql, args) - return await cursor.fetchall() - - -async def _insert(cursor, table, row: dict) -> int: - names = ", ".join(f"`{key}`" for key in row) - holders = ", ".join(["%s"] * len(row)) - await cursor.execute(f"INSERT INTO `{table}` ({names}) VALUES ({holders})", tuple(row.values())) - return int(cursor.lastrowid) - - -def _copy_folder(source_root: Path, new_root: Path) -> None: - print(f"파일 복사 시작: {source_root.name} -> {new_root.name}") - started = time.perf_counter() - result = subprocess.run( - [ - "robocopy", - str(source_root), - str(new_root), - "/E", - "/NFL", - "/NDL", - "/NJH", - "/NJS", - "/MT:8", - ], - capture_output=True, - text=True, - ) - if result.returncode >= 8: - raise RuntimeError(f"robocopy 실패 (코드 {result.returncode})") - print(f"파일 복사 완료: {time.perf_counter() - started:.0f}초") - - -def _fix_files(new_root: Path, new_id: str, model_map: dict) -> None: - replaced = [] - for path in new_root.rglob("*.json"): - try: - text = path.read_text(encoding="utf-8") - except (OSError, UnicodeDecodeError): - continue - if SOURCE_ID not in text: - continue - path.write_text(text.replace(SOURCE_ID, new_id), encoding="utf-8") - replaced.append(str(path.relative_to(new_root))) - print("프로젝트 id 치환:", replaced) - - snapshot = new_root / "initial_snapshot" / "db.json" - if snapshot.is_file(): - dump = json.loads(snapshot.read_text(encoding="utf-8")) - changed = False - for route in dump.get("routes", []): - old_model = route.get("surface_model_id") - if old_model in model_map: - route["surface_model_id"] = model_map[old_model] - changed = True - if changed: - snapshot.write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8") - print("초기값 스냅샷 지표면 id 교정:", changed) - - -async def storage_only(connection) -> None: - """DB 는 SELECT 만 — 원본 · 새 프로젝트 행(이미 복제됨)으로 폴더 자리 · 지표면 id 짝.""" - new_id = NEW_NAME - paths = {} - models = {} - async with connection.cursor(aiomysql.DictCursor) as cursor: - for project_id in (SOURCE_ID, new_id): - row = await _rows( - cursor, "SELECT storage_path FROM projects WHERE id = %s", (project_id,) - ) - paths[project_id] = row[0]["storage_path"] - rows = await _rows( - cursor, - "SELECT id FROM surface_models WHERE project_id = %s ORDER BY id", - (project_id,), - ) - models[project_id] = [item["id"] for item in rows] - # 복제는 지표면 행을 원본 id 차례대로 넣는다 — 차례로 짝 지음. - model_map = dict(zip(models[SOURCE_ID], models[new_id], strict=True)) - new_root = Path(resolve_stored_project_path(paths[new_id])) - if any(path.is_file() for path in new_root.rglob("*")): - raise RuntimeError(f"새 폴더에 파일이 이미 있음: {new_root}") - _copy_folder(Path(resolve_stored_project_path(paths[SOURCE_ID])), new_root) - _fix_files(new_root, new_id, model_map) - print(f"storage 만 복제 완료: {new_root}") - - -async def main() -> None: +async def main(argv: list[str] | None = None) -> str | None: + args = _args(argv) connection = await aiomysql.connect( host=DB_HOST, port=DB_PORT, @@ -160,183 +65,20 @@ async def main() -> None: charset="utf8mb4", ) try: - if ARGS.storage_only: - await storage_only(connection) - return - async with connection.cursor(aiomysql.DictCursor) as cursor: - source = (await _rows(cursor, "SELECT * FROM projects WHERE id = %s", (SOURCE_ID,)))[0] - input_files = await _rows( - cursor, "SELECT * FROM input_files WHERE project_id = %s ORDER BY id", (SOURCE_ID,) - ) - clouds = await _rows( - cursor, - "SELECT * FROM processed_point_cloud WHERE project_id = %s ORDER BY id", - (SOURCE_ID,), - ) - models = await _rows( - cursor, - "SELECT * FROM surface_models WHERE project_id = %s ORDER BY id", - (SOURCE_ID,), - ) - members = await _rows( - cursor, "SELECT * FROM project_members WHERE project_id = %s", (SOURCE_ID,) - ) - layers = await _rows( - cursor, - "SELECT t.* FROM terrain_layers t JOIN surface_models s" - " ON s.id = t.surface_model_id WHERE s.project_id = %s ORDER BY t.id", - (SOURCE_ID,), - ) - routes = await _rows( - cursor, "SELECT * FROM routes WHERE project_id = %s ORDER BY id", (SOURCE_ID,) - ) - stages = await _rows( - cursor, - "SELECT * FROM project_workflow_stages WHERE project_id = %s ORDER BY stage_no", - (SOURCE_ID,), - ) - route_ids = [route["id"] for route in routes] - points: list = [] - statistics: list = [] - longitudinal: list = [] - crosses: list = [] - if route_ids: - holders = ", ".join(["%s"] * len(route_ids)) - points = await _rows( - cursor, - f"SELECT * FROM route_points WHERE route_id IN ({holders}) ORDER BY id", - route_ids, - ) - statistics = await _rows( - cursor, - f"SELECT * FROM route_statistics WHERE route_id IN ({holders}) ORDER BY id", - route_ids, - ) - longitudinal = await _rows( - cursor, - "SELECT * FROM longitudinal_sections" - f" WHERE route_id IN ({holders}) ORDER BY id", - route_ids, - ) - crosses = await _rows( - cursor, - f"SELECT * FROM cross_sections WHERE route_id IN ({holders}) ORDER BY id", - route_ids, - ) - - new_id = str(uuid4()) - source_root = Path(resolve_stored_project_path(source["storage_path"])) - new_storage = "/".join(source["storage_path"].split("/")[:-1] + [new_id]) - new_root = source_root.parent / new_id - - # 1) 파일 먼저 - 실패하면 DB 를 건드리지 않은 채로 끝난다. - _copy_folder(source_root, new_root) - - # 2) DB - 한 트랜잭션으로. - await connection.begin() - async with connection.cursor(aiomysql.DictCursor) as cursor: - project = dict(source) - project["id"] = new_id - project["name"] = NEW_NAME - project["storage_path"] = new_storage - if ARGS.owner is not None: - project["user_id"] = ARGS.owner - await _insert(cursor, "projects", project) - - member_ids = ( - [int(user) for user in ARGS.members.split(",") if user.strip()] - if ARGS.members is not None - else [row["user_id"] for row in members] - ) - for user in member_ids: - await _insert(cursor, "project_members", {"project_id": new_id, "user_id": user}) - - file_map: dict[int, int] = {} - for row in input_files: - row = dict(row) - old = row.pop("id") - row["project_id"] = new_id - file_map[old] = await _insert(cursor, "input_files", row) - - cloud_map: dict[int, int] = {} - for row in clouds: - row = dict(row) - old = row.pop("id") - row["project_id"] = new_id - row["input_file_id"] = file_map.get(row["input_file_id"], row["input_file_id"]) - cloud_map[old] = await _insert(cursor, "processed_point_cloud", row) - - model_map: dict[int, int] = {} - for row in models: - row = dict(row) - old = row.pop("id") - row["project_id"] = new_id - row["source_file_id"] = file_map.get(row["source_file_id"], row["source_file_id"]) - row["processed_cloud_id"] = cloud_map.get( - row["processed_cloud_id"], row["processed_cloud_id"] - ) - model_map[old] = await _insert(cursor, "surface_models", row) - - for row in layers: - row = dict(row) - row.pop("id") - row["surface_model_id"] = model_map[row["surface_model_id"]] - await _insert(cursor, "terrain_layers", row) - - route_map: dict[int, int] = {} - for row in routes: - row = dict(row) - old = row.pop("id") - row["project_id"] = new_id - row["surface_model_id"] = model_map.get( - row["surface_model_id"], row["surface_model_id"] - ) - route_map[old] = await _insert(cursor, "routes", row) - - for table, rows in ( - ("route_points", points), - ("route_statistics", statistics), - ("longitudinal_sections", longitudinal), - ("cross_sections", crosses), - ): - for row in rows: - row = dict(row) - row.pop("id") - row["route_id"] = route_map.get(row["route_id"], row["route_id"]) - if "project_id" in row: - row["project_id"] = new_id - await _insert(cursor, table, row) - - for row in stages: - row = dict(row) - row.pop("id") - row["project_id"] = new_id - params = row.get("params") - if params: - # 단계 설정에 박힌 옛 id 를 새 id 로 바꾼다(입력 파일·지표면 모델). - data = json.loads(params) - if data.get("input_file_id") is not None: - old_file = int(data["input_file_id"]) - data["input_file_id"] = str(file_map.get(old_file, old_file)) - if data.get("surface_model_id") is not None: - old_model = int(data["surface_model_id"]) - data["surface_model_id"] = model_map.get(old_model, old_model) - row["params"] = json.dumps(data, ensure_ascii=False) - await _insert(cursor, "project_workflow_stages", row) - await connection.commit() - print( - f"DB 복제 완료: 참여자 {len(member_ids)} · 입력파일 {len(file_map)}" - f" · 지표면 {len(model_map)} · 지형층 {len(layers)}" - f" · 노선 {len(route_map)}" - f" · 정점 {len(points)} · 횡단 {len(crosses)} · 단계 {len(stages)}" + if args.storage_only: + await clone_storage_only(connection, args.source_id, args.new_name) + return None + members = ( + [int(user) for user in args.members.split(",") if user.strip()] + if args.members is not None + else None + ) + return await clone_full( + connection, args.source_id, args.new_name, owner=args.owner, members=members ) - - # 3) 파일 안에 박힌 옛 프로젝트 id 치환 + 초기값 스냅샷의 지표면 모델 id 교정. - _fix_files(new_root, new_id, model_map) - - print(f"\n새 프로젝트: {NEW_NAME}\n id = {new_id}\n 저장소 = {new_storage}") finally: connection.close() -asyncio.run(main()) +if __name__ == "__main__": + asyncio.run(main()) diff --git a/resources/tester/test_528_project_groups.py b/resources/tester/test_528_project_groups.py index b88997378..ee5021cee 100644 --- a/resources/tester/test_528_project_groups.py +++ b/resources/tester/test_528_project_groups.py @@ -138,7 +138,12 @@ def _e01(monkeypatch, rows): return "stab-id" monkeypatch.setattr(e01, "_sources", fake_sources) + async def fake_copy(project_id, source_id): + seen["copied"] = (project_id, source_id) + return {"source_fingerprint": "f" * 40} + monkeypatch.setattr(e01, "create_kind_project", fake_create) + monkeypatch.setattr(e01, "_copy_source", fake_copy) payload = e01.CreateStabRequest(source_project_id="s" * 36) return seen, lambda: asyncio.run(e01.post_stab_project(None, payload, USER)) @@ -154,7 +159,8 @@ def test_e01_requires_b06(monkeypatch): call() assert e.value.status_code == 404 seen, call = _e01(monkeypatch, [{**src, "project_year": 2026, "section_complete": 1}]) - assert call() == {"project_id": "stab-id"} + assert call() == {"project_id": "stab-id", "source_fingerprint": "f" * 40} + assert seen["copied"] == ("stab-id", "s" * 36) assert seen["kind"] == "STABILITY" and seen["source_project_id"] == "s" * 36 assert seen["name"] == "임도 안정성평가" diff --git a/resources/tester/test_532_1_project_copy.py b/resources/tester/test_532_1_project_copy.py new file mode 100644 index 000000000..0d7ac1031 --- /dev/null +++ b/resources/tester/test_532_1_project_copy.py @@ -0,0 +1,203 @@ +"""528-4 · 532-1 공용 프로젝트 복사 — 임시 storage 사본에서(요구 345 · 351). + +- stability — 필요한 것만(LAS · 전처리 점군 · B07~B09 · 초기값 없음 · 지형 하나) · 지문 40자 · + 복사본만으로 측점 수 · 설계선 = DB · 원본을 고쳐도 복사본 불변 · 고친 원본 = 다른 지문 +- full(CLI 껍데기) — 새 id · 표마다 행 수 같음 · 지표면 id 다시 이음 · 파일 안 UUID 치환 · + 원본 그대로(만든 행은 끝에 지움 · 연쇄 삭제) +""" + +import asyncio +import hashlib +import importlib.util +import json +import os +import shutil +import sys +import tempfile +from pathlib import Path + +import aiomysql +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from helper_temp_storage import temp_project_storage # noqa: E402 + +from common_util import common_util_project_copy as cp # noqa: E402 +from common_util.common_util_storage import resolve_stored_project_path # noqa: E402 +from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER # noqa: E402 +from E01_StabRegister import E01_StabRegister_Source as src # noqa: E402 + +#: 검증_sub1 — B06 까지 설계된 임도(읽기만) +SOURCE_ID = "916ea58c-0e35-4982-bf55-d8558eeb78b5" +ROOT = Path(__file__).resolve().parents[2] + + +async def _connect(): + return await aiomysql.connect( + host=DB_HOST, + port=DB_PORT, + user=DB_USER, + password=DB_PASSWORD, + db=DB_NAME, + charset="utf8mb4", + autocommit=True, + ) + + +async def _query(sql, args=()): + connection = await _connect() + try: + async with connection.cursor(aiomysql.DictCursor) as cursor: + await cursor.execute(sql, args) + return await cursor.fetchall() + finally: + connection.close() + + +@pytest.fixture(scope="module") +def source_root(): + rows = asyncio.run(_query("SELECT storage_path FROM projects WHERE id = %s", (SOURCE_ID,))) + if not rows: + pytest.skip("검증 원본 프로젝트가 DB 에 없음") + shared = Path(resolve_stored_project_path(rows[0]["storage_path"])) + with temp_project_storage(shared) as root: + yield root + + +async def _stability(target: Path) -> dict: + connection = await _connect() + try: + return await cp.copy_project( + connection, SOURCE_ID, scope="stability", new_id="e" * 36, project_root=target + ) + finally: + connection.close() + + +def _hashes(root: Path) -> dict[str, str]: + return { + path.relative_to(root).as_posix(): hashlib.sha1(path.read_bytes()).hexdigest() + for path in root.rglob("*.json") + if path.is_file() + } + + +@pytest.fixture +def work(): + """pytest tmp_path 정리가 이 PC 임시 폴더 권한에 걸려 tempfile 로.""" + path = Path(tempfile.mkdtemp(prefix="aislo_532_")) + yield path + shutil.rmtree(path, ignore_errors=True) + + +def test_stability_copy(source_root, work): + before = _hashes(source_root) + manifest = asyncio.run(_stability(work / "e1")) + root = src.source_dir(work / "e1") + + assert manifest["scope"] == "stability" and manifest["source_project_id"] == SOURCE_ID + assert len(manifest["source_fingerprint"]) == 40 + assert manifest["total_bytes"] < 40 * 1024 * 1024 + banned = ("B03_FileInput", "B04_PreProcess/processed", "B07_", "B08_", "B09_", "initial_") + assert not [name for name in manifest["files"] if name.startswith(banned)] + assert src.terrain_grid_path(root).is_file() # 지형 하나뿐 + assert src.corridor_path(root) is not None + # 파일 안 옛 UUID = 없음(동결 db.json · manifest 의 원본 기록 빼고) + for path in root.rglob("*.json"): + if path.name not in (cp.STABILITY_DB_NAME, cp.MANIFEST_NAME): + assert SOURCE_ID not in path.read_text(encoding="utf-8"), path + + # 복사본만으로 측점 · 설계선 = DB 최신 노선 + rows = asyncio.run( + _query( + "SELECT chainage_m, data FROM cross_sections WHERE route_id = %s ORDER BY chainage_m", + (manifest["route_id"],), + ) + ) + stations = src.stations(root) + assert len(stations) == len(rows) > 0 + for station, row in zip(stations, rows, strict=True): + assert station["chainage_m"] == pytest.approx(row["chainage_m"]) + cross = src.cross_section(root, station["chainage_m"]) + assert cross["design"]["design_line"] == json.loads(row["data"])["design"]["design_line"] + assert len(cross["samples"]) > 10 + + # 원본 그대로 · 원본을 고쳐도 복사본 불변 · 같은 원본 = 같은 지문 + assert _hashes(source_root) == before + first = stations[0]["chainage_m"] + kept = src.cross_section(root, first) + cross_file = next((source_root / "B06_Section" / "cross_sections").glob("*.json")) + data = json.loads(cross_file.read_text(encoding="utf-8")) + for sample in data["samples"]: + sample["z"] = -999.0 + cross_file.write_text(json.dumps(data), encoding="utf-8") # 임시 사본만 고침 + assert src.cross_section(root, first) == kept + again = asyncio.run(_stability(work / "e2")) + assert again["source_fingerprint"] != manifest["source_fingerprint"] + + +def _cli(): + spec = importlib.util.spec_from_file_location( + "tools_clone_project", ROOT / "db_management" / "tools_clone_project.py" + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _copytree(source: Path, target: Path) -> None: + """robocopy 대신 — 시험 사본의 정션(B03 · 전처리 점군)은 따라가지 않음.""" + shutil.copytree( + source, + target, + ignore=lambda folder, names: [n for n in names if os.path.isjunction(Path(folder, n))], + ) + + +COUNT_TABLES = ( + "routes", + "surface_models", + "input_files", + "project_workflow_stages", + "project_members", + "cross_sections", + "longitudinal_sections", +) + + +def _counts(project_id): + return { + table: asyncio.run( + _query(f"SELECT COUNT(*) AS n FROM {table} WHERE project_id = %s", (project_id,)) + )[0]["n"] + for table in COUNT_TABLES + } + + +def test_full_clone_cli(source_root, monkeypatch): + monkeypatch.setattr(cp, "_copy_folder", _copytree) + before = _hashes(source_root) + new_id = asyncio.run(_cli().main([SOURCE_ID, "시험_528_4_복제"])) + try: + assert new_id and new_id != SOURCE_ID + assert _counts(new_id) == _counts(SOURCE_ID) + models = { + r["id"] + for r in asyncio.run( + _query("SELECT id FROM surface_models WHERE project_id = %s", (new_id,)) + ) + } + routes = asyncio.run( + _query("SELECT surface_model_id FROM routes WHERE project_id = %s", (new_id,)) + ) + assert {r["surface_model_id"] for r in routes} <= models + new_root = source_root.parent / new_id + assert (new_root / cp.MANIFEST_NAME).is_file() + for path in new_root.rglob("*.json"): + if path.name != cp.MANIFEST_NAME: + assert SOURCE_ID not in path.read_text(encoding="utf-8"), path + assert _hashes(source_root) == before + finally: + if new_id: + asyncio.run(_query("DELETE FROM projects WHERE id = %s", (new_id,)))