Aislo/common_util/common_util_project_copy.py

"""프로젝트 복사 — 통째 복제(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}")