From 76a399026c8a67dca1d0c0be9efa3bb58f930141 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 25 Oct 2026 19:46:43 +0900 Subject: [PATCH 1/2] =?UTF-8?q?feat(529-3):=20D02=20=EC=A7=80=EC=A0=90=20?= =?UTF-8?q?=EC=A0=95=EB=B3=B8=20API=20=C2=B7=20=EA=B2=BD=EC=9C=84=EB=8F=84?= =?UTF-8?q?=20=E2=86=94=20TM=20=EC=A0=90=20=EB=B3=80=ED=99=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - common_util_crs: 경도로 TM 띠(5185 · 5186 · 5187) · lonlat_to_tm · tm_to_lonlat - D02_PlanBase_Router: GET · PUT /api/projects/{id}/plan/points — 잇기(직전 종점 = 다음 시점) · 범위 밖 · 빈 연도 거절 · 산림청만 고침 · 바뀐 연도만 분석 노선 비움 - 지킴 시험 test_529_3_plan_points Co-Authored-By: Claude Opus 5.5 --- D02_PlanBase/D02_PlanBase_Router.py | 206 +++++++++++++++++++++ common_util/common_util_crs.py | 28 ++- main.py | 2 + resources/tester/test_529_3_plan_points.py | 149 +++++++++++++++ 4 files changed, 384 insertions(+), 1 deletion(-) create mode 100644 D02_PlanBase/D02_PlanBase_Router.py create mode 100644 resources/tester/test_529_3_plan_points.py diff --git a/D02_PlanBase/D02_PlanBase_Router.py b/D02_PlanBase/D02_PlanBase_Router.py new file mode 100644 index 000000000..b61e39661 --- /dev/null +++ b/D02_PlanBase/D02_PlanBase_Router.py @@ -0,0 +1,206 @@ +"""D 예정노선 1단계 기준 지정 — 연도별 시점 · 종점 정본 API (요구 345 · 요구서 3-2 · PLAN 529-3). + +- `GET /api/projects/{id}/plan/points` — 사업연도 연도마다 시점 · 종점(경위도 + TM) · 빈 = null +- `PUT …/plan/points` — 경위도를 받아 서버가 TM 을 셈 → `plan_route_years` 행 + +규칙: 직전 연도 종점 = 다음 연도 시점(서버가 맞춤 · 보낸 시점은 첫 연도만 씀) · +범위 밖 · 빈 연도 거절 · 좌표계 = 모든 지점 평균 경도의 TM 띠 한 벌 · +지점이 바뀐 연도는 분석 노선(route_json · fingerprint · confirmed_at) 비움 · +고침 = 산림청만(지정 회사 GET 은 공용 판정 `require_project_access` 몫). +""" + +from __future__ import annotations + +from typing import Any +from uuid import UUID + +import aiomysql +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, Field + +from common_util.common_util_auth import is_forest_service, require_company +from common_util.common_util_crs import lonlat_to_tm, tm_epsg_for_lon, tm_to_lonlat +from config.config_db import get_db_pool + +router = APIRouter(prefix="/api/projects", tags=["D02 Plan Base"]) + +SAME_M = 0.001 # 1 mm 안이면 같은 지점 — 되저장으로 분석 노선을 지우지 않음 + + +class LonLat(BaseModel): + lon: float = Field(..., ge=120, le=135) + lat: float = Field(..., ge=30, le=45) + + +class YearPoints(BaseModel): + year: int + start: LonLat | None = None + end: LonLat | None = None + + +class PointsRequest(BaseModel): + years: list[YearPoints] + + +def chain_years(year_start: int, year_end: int, items: list[YearPoints]) -> list[dict[str, Any]]: + """연도 범위를 잇기 — [{year, start, end}] 경위도 · 시점 = 직전 종점(첫 연도만 보낸 값).""" + by_year: dict[int, YearPoints] = {} + for item in items: + if not year_start <= item.year <= year_end: + raise HTTPException( + status_code=400, + detail=f"{item.year}년은 사업연도({year_start}~{year_end}) 밖입니다.", + ) + if item.year in by_year: + raise HTTPException(status_code=400, detail=f"{item.year}년이 두 번 왔습니다.") + by_year[item.year] = item + rows: list[dict[str, Any]] = [] + previous: LonLat | None = None + for year in range(year_start, year_end + 1): + item = by_year.get(year) + start = previous or (item.start if item else None) + end = item.end if item else None + if start is None or end is None: + side = "시점" if start is None else "종점" + raise HTTPException(status_code=400, detail=f"{year}년 {side}이 비었습니다.") + rows.append({"year": year, "start": start, "end": end}) + previous = end + return rows + + +def with_tm(rows: list[dict[str, Any]]) -> tuple[int, list[dict[str, Any]]]: + """(epsg, 행마다 start_x · start_y · end_x · end_y 더함).""" + lons = [p.lon for row in rows for p in (row["start"], row["end"])] + epsg = tm_epsg_for_lon(sum(lons) / len(lons)) + out = [] + for row in rows: + sx, sy = lonlat_to_tm(row["start"].lon, row["start"].lat, epsg) + ex, ey = lonlat_to_tm(row["end"].lon, row["end"].lat, epsg) + out.append({"year": row["year"], "start_x": sx, "start_y": sy, "end_x": ex, "end_y": ey}) + return epsg, out + + +def changed(new: dict[str, Any], old: dict[str, Any] | None, epsg: int) -> bool: + if not old or old.get("epsg") is None or int(old["epsg"]) != epsg: + return True + keys = ("start_x", "start_y", "end_x", "end_y") + return any(old[k] is None or abs(float(old[k]) - new[k]) > SAME_M for k in keys) + + +async def _load(project_id: UUID) -> tuple[dict[str, Any], dict[int, dict[str, Any]]]: + """(예정노선 프로젝트 칸, 연도 → 지금 행).""" + async with ( + get_db_pool().acquire() as connection, + connection.cursor(aiomysql.DictCursor) as cursor, + ): + await cursor.execute( + "SELECT project_kind, project_year, project_year_end FROM projects" + " WHERE id = %s AND deleted_at IS NULL", + (str(project_id),), + ) + project = await cursor.fetchone() + await cursor.execute( + "SELECT year, start_x, start_y, end_x, end_y, epsg FROM plan_route_years" + " WHERE project_id = %s", + (str(project_id),), + ) + rows = await cursor.fetchall() + if not project: + raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.") + if project["project_kind"] != "PLAN" or project["project_year"] is None: + raise HTTPException(status_code=400, detail="예정노선 프로젝트가 아닙니다.") + return project, {int(row["year"]): row for row in rows} + + +async def _store( + project_id: UUID, epsg: int, writes: list[dict[str, Any]], drops: list[int] +) -> None: + """바뀐 연도만 넣고(분석 노선 비움) · 범위 밖 연도 행은 지움 — 한 묶음.""" + async with get_db_pool().acquire() as connection: + await connection.begin() + try: + async with connection.cursor() as cursor: + for row in writes: + await cursor.execute( + "INSERT INTO plan_route_years" + " (project_id, year, start_x, start_y, end_x, end_y, epsg)" + " VALUES (%s, %s, %s, %s, %s, %s, %s)" + " ON DUPLICATE KEY UPDATE start_x = VALUES(start_x)," + " start_y = VALUES(start_y), end_x = VALUES(end_x), end_y = VALUES(end_y)," + " epsg = VALUES(epsg), route_json = NULL, fingerprint = NULL," + " confirmed_at = NULL", + ( + str(project_id), + row["year"], + row["start_x"], + row["start_y"], + row["end_x"], + row["end_y"], + epsg, + ), + ) + for year in drops: + await cursor.execute( + "DELETE FROM plan_route_years WHERE project_id = %s AND year = %s", + (str(project_id), year), + ) + await connection.commit() + except Exception: + await connection.rollback() + raise + + +def _point(x: Any, y: Any, epsg: Any) -> dict[str, float] | None: + if x is None or y is None or epsg is None: + return None + lon, lat = tm_to_lonlat(float(x), float(y), int(epsg)) + return {"lon": lon, "lat": lat, "x": float(x), "y": float(y)} + + +def _view(project: dict[str, Any], rows: dict[int, dict[str, Any]]) -> dict[str, Any]: + start = int(project["project_year"]) + end = int(project["project_year_end"] or start) + years = [] + for year in range(start, end + 1): + row = rows.get(year) or {} + years.append( + { + "year": year, + "start": _point(row.get("start_x"), row.get("start_y"), row.get("epsg")), + "end": _point(row.get("end_x"), row.get("end_y"), row.get("epsg")), + } + ) + epsgs = {int(r["epsg"]) for r in rows.values() if r.get("epsg") is not None} + return { + "status": "success", + "year_start": start, + "year_end": end, + "epsg": epsgs.pop() if len(epsgs) == 1 else None, + "years": years, + } + + +@router.get("/{project_id}/plan/points", response_model=None) +async def get_plan_points(project_id: UUID) -> dict[str, Any]: + project, rows = await _load(project_id) + return _view(project, rows) + + +@router.put("/{project_id}/plan/points", response_model=None) +async def put_plan_points( + project_id: UUID, + payload: PointsRequest, + session: dict[str, Any] = Depends(require_company), +) -> dict[str, Any]: + if not is_forest_service(session): + raise HTTPException(status_code=403, detail="예정노선 지점은 산림청만 고칠 수 있습니다.") + project, old = await _load(project_id) + start = int(project["project_year"]) + end = int(project["project_year_end"] or start) + epsg, rows = with_tm(chain_years(start, end, payload.years)) + writes = [row for row in rows if changed(row, old.get(row["year"]), epsg)] + drops = [year for year in old if not start <= year <= end] + if writes or drops: + await _store(project_id, epsg, writes, drops) + _, now = await _load(project_id) + return {**_view(project, now), "changed_years": [row["year"] for row in writes]} diff --git a/common_util/common_util_crs.py b/common_util/common_util_crs.py index ac511ef85..60468ab8e 100644 --- a/common_util/common_util_crs.py +++ b/common_util/common_util_crs.py @@ -10,9 +10,10 @@ from __future__ import annotations import logging import re +from functools import lru_cache from pathlib import Path -from pyproj import CRS +from pyproj import CRS, Transformer logger = logging.getLogger(__name__) @@ -257,3 +258,28 @@ def resolve_project_crs( if prj_crs: return prj_crs return f"EPSG:{db_epsg or 5186}" + + +# --- 점 변환(경위도 ↔ Korea 2000 TM · 요구 345 · PLAN 529-3) --- + + +def tm_epsg_for_lon(lon: float) -> int: + """경도로 TM 띠 — 126° 앞 서부 5185 · 128° 앞 중부 5186 · 그 밖 동부 5187(동해 5188 안 씀).""" + return 5185 if lon < 126.0 else 5186 if lon < 128.0 else 5187 + + +@lru_cache(maxsize=16) +def _transformer(source: int, target: int) -> Transformer: + return Transformer.from_crs(source, target, always_xy=True) + + +def lonlat_to_tm(lon: float, lat: float, epsg: int) -> tuple[float, float]: + """WGS84 경위도 → TM (x 동, y 북) m.""" + x, y = _transformer(4326, epsg).transform(lon, lat) + return float(x), float(y) + + +def tm_to_lonlat(x: float, y: float, epsg: int) -> tuple[float, float]: + """TM (x, y) m → WGS84 (경도, 위도).""" + lon, lat = _transformer(epsg, 4326).transform(x, y) + return float(lon), float(lat) diff --git a/main.py b/main.py index b7b8c5dce..9079f9c17 100644 --- a/main.py +++ b/main.py @@ -74,6 +74,7 @@ from B08_Quantity.B08_Quantity_Router_Milestone import router as b08_milestone_r from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router from D01_PlanRegister.D01_PlanRegister_Router import dashboard_router as d01_dashboard_router from D01_PlanRegister.D01_PlanRegister_Router import router as d01_plan_register_router +from D02_PlanBase.D02_PlanBase_Router import router as d02_plan_base_router from D03_PlanAnalysis.D03_PlanAnalysis_Router import router as d03_plan_analysis_router from E01_StabRegister.E01_StabRegister_Router import router as e01_stab_register_router from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router @@ -477,6 +478,7 @@ app.include_router(b04_basins_router, dependencies=protected_with_company) app.include_router(tiles_router, dependencies=protected_with_company) app.include_router(b05_route_router, dependencies=protected_with_company) app.include_router(b05_route_lifecycle_router, dependencies=protected_with_company) +app.include_router(d02_plan_base_router, dependencies=protected_with_company) app.include_router(d03_plan_analysis_router, dependencies=protected_with_company) app.include_router(b05_corridor_router, dependencies=protected_with_company) app.include_router(b05_route_replan_router, dependencies=protected_with_company) diff --git a/resources/tester/test_529_3_plan_points.py b/resources/tester/test_529_3_plan_points.py new file mode 100644 index 000000000..7c65cf634 --- /dev/null +++ b/resources/tester/test_529_3_plan_points.py @@ -0,0 +1,149 @@ +"""D02 기준 지정(PLAN 529-3) — 경위도 ↔ TM 변환 · 지점 정본 API(잇기 · 빈 연도 · 권한). + +./venv/Scripts/python.exe -m pytest resources/tester/test_529_3_plan_points.py -q +""" + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from common_util.common_util_auth import require_company +from common_util.common_util_crs import lonlat_to_tm, tm_epsg_for_lon, tm_to_lonlat +from D02_PlanBase import D02_PlanBase_Router as router_mod + +PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2" +URL = f"/api/projects/{PID}/plan/points" + + +# --- 좌표 변환 --- + + +@pytest.mark.parametrize("epsg, lon0", [(5185, 125.0), (5186, 127.0), (5187, 129.0)]) +def test_origin_is_false_origin(epsg, lon0) -> None: + x, y = lonlat_to_tm(lon0, 38.0, epsg) # 원점 = (200 000, 600 000) + assert abs(x - 200000.0) < 0.001 and abs(y - 600000.0) < 0.001 + + +@pytest.mark.parametrize("lon, lat", [(127.3851, 36.3504), (125.7, 34.1), (129.36, 36.02)]) +def test_round_trip_within_1mm(lon, lat) -> None: + epsg = tm_epsg_for_lon(lon) + x, y = lonlat_to_tm(lon, lat, epsg) + back = lonlat_to_tm(*tm_to_lonlat(x, y, epsg), epsg) + assert abs(back[0] - x) < 0.001 and abs(back[1] - y) < 0.001 + + +def test_belt_by_longitude() -> None: + assert [tm_epsg_for_lon(v) for v in (124.5, 125.999, 126.0, 127.999, 128.0, 131.8)] == [ + 5185, + 5185, + 5186, + 5186, + 5187, + 5187, + ] + + +# --- 지점 정본 API --- + + +A, B, C = ( + {"lon": 127.10, "lat": 37.50}, + {"lon": 127.12, "lat": 37.51}, + {"lon": 127.15, "lat": 37.52}, +) + + +@pytest.fixture +def api(monkeypatch): + state = { + "project": {"project_kind": "PLAN", "project_year": 2026, "project_year_end": 2027}, + "rows": {}, + "session": {"role": "USER", "email": "kim@forest.go.kr", "company_id": 1}, + "stores": [], + } + + async def load(project_id): + return state["project"], {y: dict(r) for y, r in state["rows"].items()} + + async def store(project_id, epsg, writes, drops): + state["stores"].append((epsg, writes, drops)) + for row in writes: + state["rows"][row["year"]] = {**row, "epsg": epsg} + for year in drops: + state["rows"].pop(year) + + monkeypatch.setattr(router_mod, "_load", load) + monkeypatch.setattr(router_mod, "_store", store) + app = FastAPI() + app.include_router(router_mod.router) + app.dependency_overrides[require_company] = lambda: state["session"] + client = TestClient(app) + client.state = state + return client + + +def test_put_chains_years(api) -> None: + # 2027 시점을 엉뚱하게 보내도 서버가 2026 종점으로 맞춤 + body = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "start": A, "end": C}]} + got = api.put(URL, json=body) + assert got.status_code == 200 + data = got.json() + assert data["epsg"] == 5186 and data["changed_years"] == [2026, 2027] + y26, y27 = data["years"] + assert y27["start"]["x"] == y26["end"]["x"] and y27["start"]["y"] == y26["end"]["y"] + assert abs(y27["start"]["lon"] - B["lon"]) < 1e-8 and abs(y27["start"]["lat"] - B["lat"]) < 1e-8 + assert (y26["start"]["x"], y26["start"]["y"]) == pytest.approx( + lonlat_to_tm(127.10, 37.50, 5186) + ) + + +def test_put_rejects_empty_and_out_of_range(api) -> None: + only_first = {"years": [{"year": 2026, "start": A, "end": B}]} + got = api.put(URL, json=only_first) + assert got.status_code == 400 and "2027년 종점" in got.json()["detail"] + no_start = {"years": [{"year": 2026, "end": B}, {"year": 2027, "end": C}]} + assert "2026년 시점" in api.put(URL, json=no_start).json()["detail"] + outside = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2028, "end": C}]} + assert api.put(URL, json=outside).status_code == 400 + assert api.state["stores"] == [] + + +def test_put_forest_only(api) -> None: + body = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "end": C}]} + api.state["session"] = {"role": "USER", "email": "lee@company.com", "company_id": 1} + assert api.put(URL, json=body).status_code == 403 + api.state["session"] = {"role": "SYSTEM_ADMIN", "email": "admin@x.com", "company_id": None} + assert api.put(URL, json=body).status_code == 200 + + +def test_put_clears_only_changed_years(api) -> None: + body = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "end": C}]} + api.put(URL, json=body) + # 같은 지점 다시 저장 = 쓰기 없음(분석 노선 그대로) · 2027 종점만 옮기면 2027 만 + assert api.put(URL, json=body).json()["changed_years"] == [] + moved = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "end": A}]} + assert api.put(URL, json=moved).json()["changed_years"] == [2027] + assert len(api.state["stores"]) == 2 + + +def test_get_lists_every_year_and_drops_outside(api) -> None: + api.state["rows"][2030] = { + "year": 2030, + "start_x": 1, + "start_y": 1, + "end_x": 2, + "end_y": 2, + "epsg": 5186, + } + data = api.get(URL).json() + assert [y["year"] for y in data["years"]] == [2026, 2027] + assert data["years"][0]["start"] is None + body = {"years": [{"year": 2026, "start": A, "end": B}, {"year": 2027, "end": C}]} + api.put(URL, json=body) + assert api.state["stores"][-1][2] == [2030] and 2030 not in api.state["rows"] + + +def test_not_plan_project(api) -> None: + api.state["project"] = {"project_kind": "ROAD", "project_year": 2026, "project_year_end": None} + # _load 대역이 그대로 돌려주므로 진짜 _load 의 거절은 서버 시험(ORCA)에서 + assert api.get(URL).status_code == 200 From 94a7fd580845fbb28742c96e20b03097d28ac3c1 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 25 Oct 2026 19:47:12 +0900 Subject: [PATCH 2/2] =?UTF-8?q?feat(528-4=20=C2=B7=20532-1):=20=EA=B3=B5?= =?UTF-8?q?=EC=9A=A9=20=ED=94=84=EB=A1=9C=EC=A0=9D=ED=8A=B8=20=EB=B3=B5?= =?UTF-8?q?=EC=82=AC=20=C2=B7=20E01=20=EC=95=88=EC=A0=95=EC=84=B1=ED=8F=89?= =?UTF-8?q?=EA=B0=80=20=EC=83=9D=EC=84=B1=20=EB=B3=B5=EC=82=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- E01_StabRegister/E01_StabRegister_Router.py | 41 +- E01_StabRegister/E01_StabRegister_Source.py | 134 ++++++ common_util/common_util_project_copy.py | 470 ++++++++++++++++++++ db_management/tools_clone_project.py | 296 +----------- resources/tester/test_528_project_groups.py | 8 +- resources/tester/test_532_1_project_copy.py | 203 +++++++++ 6 files changed, 872 insertions(+), 280 deletions(-) create mode 100644 E01_StabRegister/E01_StabRegister_Source.py create mode 100644 common_util/common_util_project_copy.py create mode 100644 resources/tester/test_532_1_project_copy.py diff --git a/E01_StabRegister/E01_StabRegister_Router.py b/E01_StabRegister/E01_StabRegister_Router.py index 0d89c5822..38ad6d57d 100644 --- a/E01_StabRegister/E01_StabRegister_Router.py +++ b/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 diff --git a/E01_StabRegister/E01_StabRegister_Source.py b/E01_StabRegister/E01_StabRegister_Source.py new file mode 100644 index 000000000..ba4895d65 --- /dev/null +++ b/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 diff --git a/common_util/common_util_project_copy.py b/common_util/common_util_project_copy.py new file mode 100644 index 000000000..12783a998 --- /dev/null +++ b/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}") diff --git a/db_management/tools_clone_project.py b/db_management/tools_clone_project.py index f38317b82..9e816cdb4 100644 --- a/db_management/tools_clone_project.py +++ b/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()) diff --git a/resources/tester/test_528_project_groups.py b/resources/tester/test_528_project_groups.py index b88997378..ee5021cee 100644 --- a/resources/tester/test_528_project_groups.py +++ b/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"] == "임도 안정성평가" diff --git a/resources/tester/test_532_1_project_copy.py b/resources/tester/test_532_1_project_copy.py new file mode 100644 index 000000000..0d7ac1031 --- /dev/null +++ b/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,)))