eomsangdonandClaude Opus 5.5
152516bb59
- 옮김 끝 — 파일 없는 PLAN 10 개 = 옛 칸 → plan_network.json 한 번(일회 스크립트 · 11 개 모두 파일 대 DB 대조 같음) - 저장 모듈 = 파일 · 없으면 빈 설계(옛 칸 · 표 안 읽음) · 옛 연도 노선 접기(legacy_network) · 응답 legacy 칸 걷음 - D04 조각 노선 열쇠 route_json → route(옛 칸 이름과 헷갈림 걷음 · 메모리 값) - 복제 도구 = 옛 D 표 안 뜸 · 예정노선 설계는 폴더 파일로 따라감 - 024_plan_site_fill.py 지움 — 옛 칸을 읽는 일회 도구 · 기준점 11 개 모두 채움 - SQL 025 = 단계 번호(PLAN_BASE 걷음) · plan_networks · plan_route_pieces · projects.plan_base_json · 연도 줄 노선 칸 · 업체 줄 year 지움 — 공유 DB 적용 안 함(데스크탑 임시 배포가 새 코드를 받은 뒤 · 임시 DB 두 번 돌림 확인) - 시험: 590_1 옮김 시험 셋 → 파일 없음 = 빈 설계 · 코드에 옛 칸 이름 0 · 530_2 옛 연도 노선 시험 지움(그 길 걷음) · 591_2 옛 지점 꼴 시험 지움 · 591_1 채움 도구 시험 지움(도구 지움) · 573_1 = 파일째 따라감 · 옛 표 목록 밖 허용(025 전) · 531_3 · 563_2 · 593 = 열쇠 이름 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
563 lines
24 KiB
Python
563 lines
24 KiB
Python
"""프로젝트 복사 — 통째 복제(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 · D 의
|
|
plan_route_years · plan_route_assignees)을 새 id 로 다시 이어 한 트랜잭션에 넣고
|
|
저장소 폴더 통째(robocopy · D 설계 내용 = 폴더 파일) — 복제 도구(`db_management/
|
|
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 · 초기값 안 가져옴.
|
|
|
|
둘 다 복사한 JSON 안의 옛 프로젝트 UUID 를 새 UUID 로 바꿈.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
from uuid import uuid4
|
|
|
|
import aiomysql
|
|
|
|
from B05_Profile.B05_Profile_Engine_Sections_Names import find_cross_file
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
from common_util.common_util_surface_confirmation import get_surface_confirmation_params
|
|
from common_util.common_util_time import utc_now_text
|
|
|
|
Scope = Literal["full", "stability"]
|
|
KIND_LABELS = {"ROAD": "B", "PLAN": "D", "STABILITY": "E"}
|
|
#: 자동증가 id 없이 project_id 만 바꿔 넣는 D 표
|
|
_PLAN_TABLES = ("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"
|
|
STABILITY_DB_NAME = "db.json"
|
|
#: stability 가 통째로 뜨는 폴더(프로젝트 루트 기준)
|
|
_STABILITY_TREES = (
|
|
"B05_Profile/route",
|
|
"B06_Section/longitudinal",
|
|
"B06_Section/cross_sections",
|
|
"B04_PreProcess/drainage/edits",
|
|
)
|
|
_CROSS_DIR = "B06_Section/cross_sections"
|
|
|
|
|
|
def resolve_cross_file(root: Path, row: dict[str, Any]) -> Path | None:
|
|
"""횡단 행 → root 아래 횡단 파일 — 경로 칸이 없거나(null) · 파일이 없거나 · 절대 경로(옛 꼴)면
|
|
B06 이름 규칙(`find_cross_file` · 옛 정수 m 이름 포함)으로 찾음 · 그래도 없으면 None."""
|
|
relative = row.get("cross_section_file_path")
|
|
if relative and not Path(relative).is_absolute() and (root / relative).is_file():
|
|
return root / relative
|
|
if row.get("chainage_m") is None:
|
|
return None
|
|
found = find_cross_file(root / _CROSS_DIR, float(row["chainage_m"]))
|
|
return found if found.is_file() else None
|
|
|
|
|
|
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"]]))
|
|
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")
|
|
}
|
|
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)
|
|
|
|
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)}"
|
|
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'])}"
|
|
f" · 예정노선 {' · '.join(f'{t} {len(rows[t])}' for t in _PLAN_ALL)}"
|
|
)
|
|
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)
|
|
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])
|
|
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",
|
|
"kind": kind,
|
|
"source_project_id": source_id,
|
|
"project_id": new_id,
|
|
"copied_at": utc_now_text(),
|
|
"tables": {name: len(items) for name, items in rows.items()},
|
|
"skipped_tables": skipped,
|
|
"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"))
|
|
missing = sorted(
|
|
float(row["chainage_m"] or 0.0)
|
|
for row in dump["cross_sections"]
|
|
if resolve_cross_file(target, row) is None
|
|
)
|
|
manifest = {
|
|
"scope": "stability",
|
|
"source_project_id": source_id,
|
|
"project_id": new_id,
|
|
"copied_at": utc_now_text(),
|
|
"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,
|
|
#: 횡단 파일을 못 찾은 측점(누가거리) — E 는 그 측점을 「횡단 없음」으로 건너뜀
|
|
"missing_cross_chainages": missing,
|
|
"warnings": [f"횡단 없는 측점 {len(missing)} 개 — 해석에서 빠짐"] if missing else [],
|
|
"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}")
|