feat(db_management): 복제 도구 --storage-only — DB 는 읽기만 · storage 폴더 복사 + 파일 안 id 바꿈만 (402-1b)

- 폴더 복사 · id 바꿈을 _copy_folder · _fix_files 로 떼어 통째 복제와 같이 씀
- 지표면 id 짝은 DB 의 원본 · 새 프로젝트 행 차례로 · 새 폴더에 파일이 있으면 멈춤
- 시험: 통째 복제 폴더를 지우고 --storage-only 로 다시 만든 결과 283 파일 md5 같음

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VmC34fsxxbweEG3RUjs3Mn

db_management/tools_clone_project.py

@@ -20,6 +20,9 @@
 쓰는 법
     ./venv/Scripts/python.exe db_management/tools_clone_project.py <원본 UUID> "<새 이름>"
         [--owner <사용자 id>] [--members <id,id>]   (없으면 원본 주인 · 참여자 그대로)
+    storage 만(DB 는 이미 복제됨 — 다른 PC 의 storage 를 맞출 때)
+    ./venv/Scripts/python.exe db_management/tools_clone_project.py \\
+        <원본 UUID> <새 UUID> --storage-only
 """
 
 import argparse
@@ -42,7 +45,12 @@ 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="새 프로젝트 이름")
+    parser.add_argument("new_name", help="새 프로젝트 이름(--storage-only 면 새 프로젝트 UUID)")
+    parser.add_argument(
+        "--storage-only",
+        action="store_true",
+        help="DB 는 읽기만 · storage 폴더 복사와 파일 안 id 바꿈만",
+    )
     parser.add_argument("--owner", type=int, help="새 주인 사용자 id(없으면 원본 주인)")
     parser.add_argument("--members", help="참여자 사용자 id 쉼표 목록(없으면 원본 참여자)")
     return parser.parse_args()
@@ -65,6 +73,83 @@ async def _insert(cursor, table, row: dict) -> int:
     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:
     connection = await aiomysql.connect(
         host=DB_HOST,
@@ -75,6 +160,9 @@ 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(
@@ -142,26 +230,7 @@ async def main() -> None:
         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}초")
+        _copy_folder(source_root, new_root)
 
         # 2) DB - 한 트랜잭션으로.
         await connection.begin()
@@ -263,30 +332,7 @@ async def main() -> None:
         )
 
         # 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)
+        _fix_files(new_root, new_id, model_map)
 
         print(f"\n새 프로젝트: {NEW_NAME}\n  id = {new_id}\n  저장소 = {new_storage}")
     finally: