chore(db): 프로젝트 통째 복제 도구를 저장소에 둠 (창별 검증 프로젝트용)

⚠ 네 창이 **같은 원본에서 같은 방법으로** 떠야 수치를 견줄 수 있는데,
쓰던 스크립트가 `tmp/` 에 있어 다른 창이 못 봤음(2026-09-08 랩탑 두 창이 막힘).
원본은 SELECT 만 함 — 원본 행·파일에 쓰지 않음.

뜨는 것 — DB 열 표(projects·input_files·processed_point_cloud·surface_models·
routes·route_points·route_statistics·longitudinal_sections·cross_sections·
project_workflow_stages)와 저장소 폴더 통째. 자동증가 id 는 새로 받고 참조를 다시 이음.
파일 안에 박힌 옛 UUID 와 초기값 스냅숏의 surface_model_id 도 치환함.

⚠ DB 만 뜨거나 파일만 떠서는 화면이 안 섬 — 수량은 DB(측점·횡단)를 보고
구조물·관 정본은 파일(structures.json·pipe_points.json)이라 둘 다 필요함.

쓰는 법: ./venv/Scripts/python.exe db_management/tools_clone_project.py <원본 UUID> "<새 이름>"

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

db_management/tools_clone_project.py

@@ -0,0 +1,267 @@
+"""프로젝트 통째 복제 — **검증용 프로젝트를 창마다 하나씩 뜨는 도구** (2026-09-08).
+
+⚠ **왜 저장소에 두나** — 네 창이 **같은 원본에서 같은 방법으로** 떠야 수치를 견줄 수 있다.
+  `tmp/` 는 창마다 따로라 서로 못 봄(2026-09-08 랩탑 두 창이 막힌 자리).
+
+⚠ **원본은 읽기만 한다** — SELECT 뿐이고 원본 행·파일에 쓰지 않는다.
+  ⚠ 다만 **복제 중에 원본이 바뀌면 창마다 복제본이 달라진다** — 원본을 얼린 뒤 뜰 것.
+
+무엇을 뜨나
+  DB   projects · input_files · processed_point_cloud · surface_models · routes ·
+       route_points · route_statistics · longitudinal_sections · cross_sections ·
+       project_workflow_stages
+       ⚠ 자동증가 id 는 **새로 받고** 참조(외래키)를 새 id 로 다시 이어 붙인다.
+  파일  저장소 폴더 통째(robocopy). 경로가 프로젝트 루트 기준 상대라 그대로 쓰인다.
+  치환  파일 안에 박힌 **옛 프로젝트 UUID**, 초기값 스냅숏의 **surface_model_id**.
+
+⚠ **DB 만 뜨거나 파일만 떠서는 안 된다** — 수량(B08)은 DB(측점·횡단)를 보고,
+  구조물·관 정본은 파일(`structures.json`·`pipe_points.json`)이라 **둘 다** 있어야 화면이 선다.
+
+쓰는 법
+    ./venv/Scripts/python.exe db_management/tools_clone_project.py <원본 UUID> "<새 이름>"
+"""
+
+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 config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER
+
+
+def _args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="프로젝트를 통째로 복제한다(원본은 읽기만).")
+    parser.add_argument("source_id", help="원본 프로젝트 UUID")
+    parser.add_argument("new_name", help="새 프로젝트 이름")
+    return parser.parse_args()
+
+
+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)
+
+
+async def main() -> None:
+    connection = await aiomysql.connect(
+        host=DB_HOST,
+        port=DB_PORT,
+        user=DB_USER,
+        password=DB_PASSWORD,
+        db=DB_NAME,
+        charset="utf8mb4",
+    )
+    try:
+        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,),
+            )
+            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 를 건드리지 않은 채로 끝난다.
+        print(f"파일 복사 시작: {source_root.name} -> {new_id}")
+        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}초")
+
+        # 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
+            await _insert(cursor, "projects", project)
+
+            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)
+
+            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(file_map)} · 지표면 {len(model_map)}"
+            f" · 노선 {len(route_map)}"
+            f" · 정점 {len(points)} · 횡단 {len(crosses)} · 단계 {len(stages)}"
+        )
+
+        # 3) 파일 안에 박힌 옛 프로젝트 id 치환 + 초기값 스냅샷의 지표면 모델 id 교정.
+        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)
+
+        print(f"\n새 프로젝트: {NEW_NAME}\n  id = {new_id}\n  저장소 = {new_storage}")
+    finally:
+        connection.close()
+
+
+asyncio.run(main())