"""[저장] · [확정] 전부 아니면 없음(PLAN 472-6 · 브레인 A) — 떠 두고 실패면 되돌림. 차례마다 안쪽 창구가 저마다 커넥션 · 파일을 쓰므로 DB 트랜잭션 하나로 못 꿴다. 대신 시작 전에 - DB 행 — `projects` 한 줄 · `project_workflow_stages` · `routes` 와 자식 넷(`route_points` · `route_statistics` · `longitudinal_sections` · `cross_sections`) - 노선 바뀐 저장(6-B)이면 지표면 모델 확정 상태(`surface_models.status` · 설계 지형 바꿈) - storage — 저장 차례가 쓰는 폴더(`STORAGE_SCOPE` · 관 차례 · 노선 저장이면 배수 폴더 · 추천 [적용]이면 초기값 사본도) · 뒤에서 도는 미리 셈이 뜨는 사이 지운 파일은 건너뜀(`_derived`) 를 떠 두고, 하나라도 실패하면 둘 다 되돌린다. 되돌림 쓰기는 바뀐 것만(시놀로지 동기 — 안 바뀐 파일 · 표는 안 씀). 같은 프로젝트 저장이 겹치면 뒤 요청이 기다림(`project_lock`). """ from __future__ import annotations import asyncio import shutil import tempfile from dataclasses import dataclass, field from pathlib import Path from typing import Any import aiomysql #: 저장 차례가 쓰는 storage 자리(프로젝트 뿌리 기준) — 추천 노선 미리 셈(`route_recommend`)은 뒤에서 #: 따로 도는 곳이라 안 넣음. STORAGE_SCOPE = ( "B05_Profile/route", "B05_Profile/corridor", "B06_Section", "B07_DesignDetail", "_derived", "templates", "workflow.json", "project_manifest.json", ) #: 관 · 관 옵션 차례 · [확정](배수유역 다시 셈)이 쓰는 자리. PIPE_SCOPE = ("B04_PreProcess/drainage",) #: 노선 초기화(추천 [적용])가 다시 뜨는 초기값 사본. INITIAL_SCOPE = ("initial_snapshot",) #: 노선 아래 표(노선 행이 지워지면 함께 지워짐 · CASCADE). ROUTE_CHILDREN = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections") _LOCKS: dict[str, asyncio.Lock] = {} def project_lock(project_id: str) -> asyncio.Lock: """같은 프로젝트 저장 한 번에 하나 — 뒤 요청은 앞 저장이 끝날 때까지 기다림. ponytail: 프로세스 메모리 잠금 · 서버 여러 대면 DB 잠금으로.""" return _LOCKS.setdefault(str(project_id), asyncio.Lock()) # ── storage ──────────────────────────────────────────────────────────────── def _stamp(path: Path) -> tuple[int, int] | None: """(크기, 바뀐 때) · 그 사이 지워졌으면 None.""" try: stat = path.stat() except FileNotFoundError: return None return stat.st_size, stat.st_mtime_ns def _files(root: Path, scope: tuple[str, ...]) -> dict[str, Path]: out: dict[str, Path] = {} for relative in scope: target = root / relative if target.is_file(): out[relative] = target elif target.is_dir(): for path in target.rglob("*"): if path.is_file(): out[path.relative_to(root).as_posix()] = path return out @dataclass class StorageSnapshot: """떠 둔 파일 — 사본 폴더(임시) · 파일마다 (크기, 바뀐 때).""" root: Path scope: tuple[str, ...] copy_dir: Path stamps: dict[str, tuple[int, int]] = field(default_factory=dict) def restore(self) -> int: """바뀐 · 새로 생긴 · 지워진 파일만 되돌림 → 되쓴 수(지움 포함).""" written = 0 now = _files(self.root, self.scope) for relative, path in now.items(): if relative not in self.stamps: path.unlink(missing_ok=True) written += 1 elif _stamp(path) != self.stamps[relative]: shutil.copy2(self.copy_dir / relative, path) written += 1 for relative in self.stamps.keys() - now.keys(): target = self.root / relative target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(self.copy_dir / relative, target) written += 1 return written def changed(self) -> list[str]: """떠 둔 뒤 바뀐 자리(시험 · 로그).""" now = _files(self.root, self.scope) moved = [r for r, p in now.items() if self.stamps.get(r) != _stamp(p)] return sorted(moved + list(self.stamps.keys() - now.keys())) def discard(self) -> None: shutil.rmtree(self.copy_dir, ignore_errors=True) def snapshot_storage(root: Path, scope: tuple[str, ...]) -> StorageSnapshot: copy_dir = Path(tempfile.mkdtemp(prefix="aislo_save_")) snap = StorageSnapshot(Path(root), scope, copy_dir) for relative, path in _files(snap.root, scope).items(): target = copy_dir / relative target.parent.mkdir(parents=True, exist_ok=True) try: # 뒤 미리 셈(추천 노선 · `_derived` 원자 쓰기 · 정리)이 그 사이 지움 = 없던 파일 shutil.copy2(path, target) except FileNotFoundError: continue stamp = _stamp(path) if stamp is not None: snap.stamps[relative] = stamp return snap # ── DB ───────────────────────────────────────────────────────────────────── async def _select(cursor: Any, sql: str, args: tuple | list) -> list[dict[str, Any]]: await cursor.execute(sql, args) return [dict(row) for row in await cursor.fetchall()] async def snapshot_rows( connection: aiomysql.Connection, project_id: str, surface: bool = False ) -> dict[str, list]: """프로젝트 한 줄 · 단계 · 노선과 자식 넷의 행(값 그대로) · `surface` = 지표면 모델 확정 상태도.""" async with connection.cursor(aiomysql.DictCursor) as cursor: rows = { "projects": await _select(cursor, "SELECT * FROM projects WHERE id=%s", (project_id,)), "project_workflow_stages": await _select( cursor, "SELECT * FROM project_workflow_stages WHERE project_id=%s ORDER BY id", (project_id,), ), "routes": await _select( cursor, "SELECT * FROM routes WHERE project_id=%s ORDER BY id", (project_id,) ), } ids = [row["id"] for row in rows["routes"]] or [-1] marks = ",".join(["%s"] * len(ids)) for table in ROUTE_CHILDREN: rows[table] = await _select( cursor, f"SELECT * FROM {table} WHERE route_id IN ({marks}) ORDER BY id", ids ) if surface: rows["surface_models"] = await _select( cursor, "SELECT id, status FROM surface_models WHERE project_id=%s ORDER BY id", (project_id,), ) return rows async def _insert(cursor: Any, table: str, rows: list[dict[str, Any]]) -> None: if not rows: return columns = list(rows[0]) names = ",".join(f"`{name}`" for name in columns) marks = ",".join(["%s"] * len(columns)) await cursor.executemany( f"INSERT INTO {table} ({names}) VALUES ({marks})", [[row[name] for name in columns] for row in rows], ) async def restore_rows( connection: aiomysql.Connection, project_id: str, saved: dict[str, list] ) -> list[str]: """바뀐 표만 떠 둔 행으로 — 한 트랜잭션 → 되돌린 표 이름.""" now = await snapshot_rows(connection, project_id, surface="surface_models" in saved) tree = ("routes", *ROUTE_CHILDREN) redo_tree = any(now[table] != saved[table] for table in tree) redo_stages = now["project_workflow_stages"] != saved["project_workflow_stages"] redo_project = now["projects"] != saved["projects"] and bool(saved["projects"]) redo_surface = now.get("surface_models") != saved.get("surface_models") if not (redo_tree or redo_stages or redo_project or redo_surface): return [] await connection.begin() try: async with connection.cursor() as cursor: if redo_tree: await cursor.execute("DELETE FROM routes WHERE project_id=%s", (project_id,)) for table in tree: await _insert(cursor, table, saved[table]) if redo_stages: await cursor.execute( "DELETE FROM project_workflow_stages WHERE project_id=%s", (project_id,) ) await _insert(cursor, "project_workflow_stages", saved["project_workflow_stages"]) if redo_project: row = saved["projects"][0] columns = [name for name in row if name != "id"] sets = ",".join(f"`{name}`=%s" for name in columns) await cursor.execute( f"UPDATE projects SET {sets} WHERE id=%s", [row[name] for name in columns] + [project_id], ) if redo_surface: # 확정 하나 = 다른 모델 COMPLETE(`confirm_surface_model` 짝) · 상태만 await cursor.executemany( "UPDATE surface_models SET status=%s WHERE id=%s AND project_id=%s", [(row["status"], row["id"], project_id) for row in saved["surface_models"]], ) await connection.commit() except Exception: await connection.rollback() raise return [ *(tree if redo_tree else ()), *(("project_workflow_stages",) if redo_stages else ()), *(("projects",) if redo_project else ()), *(("surface_models",) if redo_surface else ()), ]