From 21af3aa866aee3e9ef25ada3b2212d29059025a0 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sat, 17 Oct 2026 21:48:26 +0900 Subject: [PATCH] =?UTF-8?q?feat(db=5Fmanagement):=20=EB=B3=B5=EC=A0=9C=20?= =?UTF-8?q?=EB=8F=84=EA=B5=AC=20--storage-only=20=E2=80=94=20DB=20?= =?UTF-8?q?=EB=8A=94=20=EC=9D=BD=EA=B8=B0=EB=A7=8C=20=C2=B7=20storage=20?= =?UTF-8?q?=ED=8F=B4=EB=8D=94=20=EB=B3=B5=EC=82=AC=20+=20=ED=8C=8C?= =?UTF-8?q?=EC=9D=BC=20=EC=95=88=20id=20=EB=B0=94=EA=BF=88=EB=A7=8C=20(402?= =?UTF-8?q?-1b)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 폴더 복사 · id 바꿈을 _copy_folder · _fix_files 로 떼어 통째 복제와 같이 씀 - 지표면 id 짝은 DB 의 원본 · 새 프로젝트 행 차례로 · 새 폴더에 파일이 있으면 멈춤 - 시험: 통째 복제 폴더를 지우고 --storage-only 로 다시 만든 결과 283 파일 md5 같음 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VmC34fsxxbweEG3RUjs3Mn --- db_management/tools_clone_project.py | 136 ++++++++++++++++++--------- 1 file changed, 91 insertions(+), 45 deletions(-) diff --git a/db_management/tools_clone_project.py b/db_management/tools_clone_project.py index 19bcb3356..f38317b82 100644 --- a/db_management/tools_clone_project.py +++ b/db_management/tools_clone_project.py @@ -20,6 +20,9 @@ 쓰는 법 ./venv/Scripts/python.exe db_management/tools_clone_project.py <원본 UUID> "<새 이름>" [--owner <사용자 id>] [--members ] (없으면 원본 주인 · 참여자 그대로) + 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: