feat(528-4 · 532-1): 공용 프로젝트 복사 · E01 안정성평가 생성 복사

- 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 <noreply@anthropic.com>

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

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

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}")

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())

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"] == "임도 안정성평가"
 

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,)))