diff --git a/common_util/common_util_project_copy.py b/common_util/common_util_project_copy.py index 4cfe2e059..960bfc58f 100644 --- a/common_util/common_util_project_copy.py +++ b/common_util/common_util_project_copy.py @@ -4,9 +4,11 @@ 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 로 다시 이어 + longitudinal_sections · cross_sections · project_workflow_stages · D 의 plan_networks · + plan_route_pieces · plan_route_years · plan_route_assignees)을 새 id 로 다시 이어 한 트랜잭션에 넣고 저장소 폴더 통째(robocopy) — 복제 도구(`db_management/ - tools_clone_project.py`)의 본체. + tools_clone_project.py`)의 본체. 종류(B · D · E)는 `project_kind` 로 판별 — 같은 길로 + 뜨고(빈 표는 건너뜀) E 의 원본 칸(source_project_id 등)은 projects 행째 그대로. stability DB 행은 넣지 않고 E 프로젝트 저장소 `E01_StabRegister/source/` 에 동결 스냅샷 (`db.json`) + 해석에 쓰는 파일만(B05 노선 · 구조물 · B06 종횡단 · 쓰인 DTM 하나 · 최신 노선 corridor · 관 편집분) — 원본 LAS · 전처리 점군 · B07~B09 · 초기값 안 가져옴. @@ -32,6 +34,10 @@ 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"] +KIND_LABELS = {"ROAD": "B", "PLAN": "D", "STABILITY": "E"} +#: 자동증가 id 없이 project_id 만 바꿔 넣는 D 표 +_PLAN_TABLES = ("plan_networks", "plan_route_pieces", "plan_route_years") +_PLAN_ALL = (*_PLAN_TABLES, "plan_route_assignees") MANIFEST_NAME = "copy_manifest.json" #: E 프로젝트 안 복사본 자리 — E 엔진은 `E01_StabRegister_Source` 로만 읽음 STABILITY_SOURCE_DIR = "E01_StabRegister/source" @@ -168,9 +174,54 @@ async def _source_rows(cursor, source_id: str) -> dict[str, list]: if not rows["projects"]: raise LookupError(f"원본 프로젝트가 없음: {source_id}") rows.update(await _route_children(cursor, [route["id"] for route in rows["routes"]])) + for table in _PLAN_ALL: + rows[table] = await _rows( + cursor, f"SELECT * FROM {table} WHERE project_id = %s", (source_id,) + ) return rows +async def _uncovered(cursor, source_id: str, covered) -> dict[str, int]: + """project_id 칸이 있는데 복제 목록 밖인 표 중 원본 행이 있는 것 → 표 → 행 수.""" + await cursor.execute( + "SELECT TABLE_NAME AS name FROM information_schema.COLUMNS" + " WHERE TABLE_SCHEMA = DATABASE() AND COLUMN_NAME = 'project_id'" + ) + found = {} + for row in await cursor.fetchall(): + if row["name"] in covered: + continue + await cursor.execute( + f"SELECT COUNT(*) AS n FROM `{row['name']}` WHERE project_id = %s", (source_id,) + ) + count = (await cursor.fetchone())["n"] + if count: + found[row["name"]] = count + return found + + +async def dry_run(connection, source_id: str) -> dict[str, Any]: + """쓰지 않고 셈만 — 종류 · 표마다 행 수 · 목록 밖 표 · 폴더 크기.""" + async with connection.cursor(aiomysql.DictCursor) as cursor: + rows = await _source_rows(cursor, source_id) + skipped = await _uncovered(cursor, source_id, rows) + source = rows["projects"][0] + root = Path(resolve_stored_project_path(source["storage_path"])) + files = [path for path in root.rglob("*") if path.is_file()] if root.is_dir() else [] + report = { + "project_id": source_id, + "name": source["name"], + "kind": KIND_LABELS.get(source.get("project_kind"), source.get("project_kind")), + "tables": {name: len(items) for name, items in rows.items() if items}, + "skipped_tables": skipped, + "folder": str(root), + "files": len(files), + "bytes": sum(path.stat().st_size for path in files), + } + print(json.dumps(report, ensure_ascii=False, indent=1)) + return report + + 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") @@ -281,6 +332,17 @@ async def _insert_full( 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) + + for table in _PLAN_TABLES: + for row in rows[table]: + await _insert(cursor, table, {**row, "project_id": new_id}) + for row in rows["plan_route_assignees"]: + row = dict(row) + row.pop("id") + row["project_id"] = new_id + # 업체가 이은 임도는 원본 D 몫 — 복제본은 잇지 않은 채로 시작 + row["linked_project_id"] = None + await _insert(cursor, "plan_route_assignees", row) await connection.commit() print( f"DB 복제 완료: 참여자 {len(member_ids)} · 입력파일 {len(file_map)}" @@ -288,6 +350,7 @@ async def _insert_full( f" · 노선 {len(route_map)}" f" · 정점 {len(rows['route_points'])} · 횡단 {len(rows['cross_sections'])}" f" · 단계 {len(rows['project_workflow_stages'])}" + f" · 예정노선 {' · '.join(f'{t} {len(rows[t])}' for t in _PLAN_ALL)}" ) return model_map @@ -303,7 +366,12 @@ async def clone_full( """통째 복제 → 새 project_id. 새 폴더 = 원본과 같은 회사/사용자 폴더.""" async with connection.cursor(aiomysql.DictCursor) as cursor: rows = await _source_rows(cursor, source_id) + skipped = await _uncovered(cursor, source_id, rows) source = rows["projects"][0] + kind = KIND_LABELS.get(source.get("project_kind"), source.get("project_kind")) + print(f"원본 종류: {kind}") + if skipped: + print("⚠ 복제 목록 밖 표(안 뜸):", skipped) new_id = str(uuid4()) source_root = Path(resolve_stored_project_path(source["storage_path"])) new_storage = "/".join(source["storage_path"].split("/")[:-1] + [new_id]) @@ -319,10 +387,12 @@ async def clone_full( new_root, { "scope": "full", + "kind": kind, "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()}, + "skipped_tables": skipped, "files": _tree_files(new_root), }, ) diff --git a/db_management/tools_clone_project.py b/db_management/tools_clone_project.py index 9e816cdb4..229f2f9c4 100644 --- a/db_management/tools_clone_project.py +++ b/db_management/tools_clone_project.py @@ -11,7 +11,8 @@ 무엇을 뜨나 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 + cross_sections · project_workflow_stages · (D) plan_networks · plan_route_pieces · + plan_route_years · plan_route_assignees — 종류(B · D · E)는 project_kind 로 판별 ⚠ 자동증가 id 는 **새로 받고** 참조(외래키)를 새 id 로 다시 이어 붙인다. 파일 저장소 폴더 통째(robocopy). 경로가 프로젝트 루트 기준 상대라 그대로 쓰인다. 치환 파일 안에 박힌 **옛 프로젝트 UUID**, 초기값 스냅숏의 **surface_model_id**. @@ -22,6 +23,8 @@ 쓰는 법 ./venv/Scripts/python.exe db_management/tools_clone_project.py <원본 UUID> "<새 이름>" [--owner <사용자 id>] [--members ] (없으면 원본 주인 · 참여자 그대로) + 셈만(쓰지 않음 — 종류 · 표마다 행 수 · 목록 밖 표 · 폴더 크기) + ./venv/Scripts/python.exe db_management/tools_clone_project.py <원본 UUID> - --dry-run storage 만(DB 는 이미 복제됨 — 다른 PC 의 storage 를 맞출 때) ./venv/Scripts/python.exe db_management/tools_clone_project.py \\ <원본 UUID> <새 UUID> --storage-only @@ -36,22 +39,32 @@ sys.path.append(str(Path(__file__).resolve().parent.parent)) import aiomysql -from common_util.common_util_project_copy import clone_full, clone_storage_only +from common_util.common_util_project_copy import clone_full, clone_storage_only, dry_run from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER 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)") + parser.add_argument( + "new_name", + nargs="?", + help="새 프로젝트 이름(--storage-only 면 새 프로젝트 UUID · --dry-run 이면 없어도 됨)", + ) parser.add_argument( "--storage-only", action="store_true", help="DB 는 읽기만 · storage 폴더 복사와 파일 안 id 바꿈만", ) + parser.add_argument( + "--dry-run", action="store_true", help="쓰지 않고 표마다 행 수 · 폴더 크기만" + ) parser.add_argument("--owner", type=int, help="새 주인 사용자 id(없으면 원본 주인)") parser.add_argument("--members", help="참여자 사용자 id 쉼표 목록(없으면 원본 참여자)") - return parser.parse_args(argv) + args = parser.parse_args(argv) + if not args.dry_run and not args.new_name: + parser.error("new_name 필요") + return args async def main(argv: list[str] | None = None) -> str | None: @@ -65,6 +78,9 @@ async def main(argv: list[str] | None = None) -> str | None: charset="utf8mb4", ) try: + if args.dry_run: + await dry_run(connection, args.source_id) + return None if args.storage_only: await clone_storage_only(connection, args.source_id, args.new_name) return None diff --git a/resources/tester/test_573_1_clone_de.py b/resources/tester/test_573_1_clone_de.py new file mode 100644 index 000000000..4659db613 --- /dev/null +++ b/resources/tester/test_573_1_clone_de.py @@ -0,0 +1,228 @@ +"""573-1 복제 도구 D · E 갈래 — 임시 storage 사본에서(요구 404 · 405). + +- D(예정노선) — plan_networks · plan_route_pieces · plan_route_years · plan_route_assignees · + plan_base_json · 지표면 · 지형층 행 수 같음 · 자동증가 id 새로 · 지형층 → 새 지표면 · + 업체 연결 끊김 · 지형 models_plan · _derived 파일 그대로 +- E(안정성평가) — 원본 칸(source_project_id 등) 그대로 · 단계 행 여섯 상태 그대로 · + E05 보고서 · _derived 파일 수 같음 +- 둘 다 — 종류 판별 · 이름 그대로 · 파일 안 옛 UUID 없음 · 원본 행 · 파일 그대로 · + --dry-run 은 쓰지 않음 · 만든 복제본은 끝에 hard_delete_project(임시 사본 폴더째) +""" + +import asyncio +import hashlib +import importlib.util +import json +import os +import shutil +import sys +from contextlib import contextmanager +from pathlib import Path + +import aiomysql +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from helper_temp_storage import _LINKED, 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 + +#: 가이드 예정노선 · 가이드 안정성평가(읽기만) +PLAN_ID = "87d47800-9936-4a56-9850-276fbcb34cac" +STAB_ID = "088e173d-a8ab-4c51-badb-f5a95afc43bb" +ROOT = Path(__file__).resolve().parents[2] +TABLES = ( + "surface_models", + "project_members", + "project_workflow_stages", + *cp._PLAN_ALL, +) + + +async def _query(sql, args=()): + connection = await aiomysql.connect( + host=DB_HOST, + port=DB_PORT, + user=DB_USER, + password=DB_PASSWORD, + db=DB_NAME, + charset="utf8mb4", + autocommit=True, + ) + try: + async with connection.cursor(aiomysql.DictCursor) as cursor: + await cursor.execute(sql, args) + return await cursor.fetchall() + finally: + connection.close() + + +def _q(sql, args=()): + return asyncio.run(_query(sql, args)) + + +def _project(project_id): + rows = _q("SELECT * FROM projects WHERE id = %s", (project_id,)) + return rows[0] if rows else None + + +def _counts(project_id): + return { + table: _q(f"SELECT COUNT(*) AS n FROM {table} WHERE project_id = %s", (project_id,))[0]["n"] + for table in TABLES + } + + +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("*") + if path.is_file() and path.suffix == ".json" + } + + +def _files(root: Path) -> set[str]: + """정션 자리(시험 복사가 건너뜀) 빼고.""" + names = {p.relative_to(root).as_posix() for p in root.rglob("*") if p.is_file()} + return {n for n in names if not n.startswith(tuple(f"{link}/" for link in _LINKED))} + + +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))], + ) + + +async def _hard_delete(project_id, actor): + from common_util.common_util_project_delete import hard_delete_project + from config.config_db import close_db_pool, init_db_pool + + await init_db_pool() + try: + return await hard_delete_project(project_id, actor) + finally: + await close_db_pool() + + +@contextmanager +def _clone(source_id, name, monkeypatch): + """임시 storage 사본에서 복제 → (원본 사본 뿌리, 새 id) · 끝에 지움.""" + source = _project(source_id) + if not source: + pytest.skip("원본 프로젝트가 DB 에 없음") + shared = Path(resolve_stored_project_path(source["storage_path"])) + with temp_project_storage(shared) as root: + monkeypatch.setattr(cp, "_copy_folder", _copytree) + before_rows = _counts(source_id) + before_files = _hashes(root) + new_id = None + try: + # dry-run 은 쓰지 않음 + assert asyncio.run(_cli().main([source_id, "--dry-run"])) is None + assert not (root.parent / name).exists() + new_id = asyncio.run(_cli().main([source_id, name])) + yield source, root, new_id + # 원본 그대로 + assert _counts(source_id) == before_rows + assert _hashes(root) == before_files + finally: + if new_id: + assert asyncio.run(_hard_delete(new_id, source["user_id"])) + assert _project(new_id) is None + assert not (root.parent / new_id).exists() + + +def _common(source, root, new_id): + new = _project(new_id) + assert new_id != source["id"] and new["name"].startswith("시험_573_1") + assert new["project_kind"] == source["project_kind"] + assert new["storage_path"].endswith(new_id) + assert _counts(new_id) == _counts(source["id"]) + new_root = root.parent / new_id + manifest = json.loads((new_root / cp.MANIFEST_NAME).read_text(encoding="utf-8")) + assert manifest["kind"] == cp.KIND_LABELS[source["project_kind"]] + assert manifest["skipped_tables"] == {} + assert _files(new_root) - {cp.MANIFEST_NAME} == _files(root) + 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 + return new, new_root + + +def test_clone_plan(monkeypatch): + with _clone(PLAN_ID, "시험_573_1_D", monkeypatch) as (source, root, new_id): + new, new_root = _common(source, root, new_id) + assert json.loads(new["plan_base_json"]) == json.loads(source["plan_base_json"]) + assert (new_root / "B04_PreProcess" / "models_plan").is_dir() + assert (new_root / "_derived").is_dir() + + def rows(table, project_id, order): + return _q( + f"SELECT * FROM {table} WHERE project_id = %s ORDER BY {order}", (project_id,) + ) + + old_net, new_net = ( + rows("plan_networks", PLAN_ID, "project_id"), + rows("plan_networks", new_id, "project_id"), + ) + assert json.loads(new_net[0]["network_json"]) == json.loads(old_net[0]["network_json"]) + for table, order in (("plan_route_pieces", "piece_id"), ("plan_route_years", "year")): + old, now = rows(table, PLAN_ID, order), rows(table, new_id, order) + drop = ("project_id", "updated_at") + assert [{k: v for k, v in r.items() if k not in drop} for r in now] == [ + {k: v for k, v in r.items() if k not in drop} for r in old + ] + old_a = rows("plan_route_assignees", PLAN_ID, "id") + new_a = rows("plan_route_assignees", new_id, "id") + assert not {r["id"] for r in old_a} & {r["id"] for r in new_a} + assert [(r["piece_id"], r["year"], r["company_name"]) for r in new_a] == [ + (r["piece_id"], r["year"], r["company_name"]) for r in old_a + ] + assert all(r["linked_project_id"] is None for r in new_a) + + old_m = _q("SELECT id FROM surface_models WHERE project_id = %s", (PLAN_ID,)) + new_m = _q("SELECT id FROM surface_models WHERE project_id = %s", (new_id,)) + assert {r["id"] for r in new_m}.isdisjoint({r["id"] for r in old_m}) + layers = _q( + "SELECT t.surface_model_id FROM terrain_layers t JOIN surface_models s" + " ON s.id = t.surface_model_id WHERE s.project_id = %s", + (new_id,), + ) + assert len(layers) == 1 + + +def test_clone_stability(monkeypatch): + with _clone(STAB_ID, "시험_573_1_E", monkeypatch) as (source, root, new_id): + new, new_root = _common(source, root, new_id) + for column in ("source_project_id", "source_copied_at", "source_fingerprint"): + assert new[column] == source[column] + + def stages(project_id): + return _q( + "SELECT stage_no, stage_key, state FROM project_workflow_stages" + " WHERE project_id = %s ORDER BY stage_no", + (project_id,), + ) + + assert len(stages(new_id)) == 6 and stages(new_id) == stages(STAB_ID) + assert (new_root / "E05_StabReport" / "report.json").is_file() + inner = json.loads( + (new_root / cp.STABILITY_SOURCE_DIR / cp.MANIFEST_NAME).read_text(encoding="utf-8") + ) + assert inner["project_id"] == new_id + assert inner["source_project_id"] == source["source_project_id"]