feat(B05,B06): 자동저장을 걷어내고 초기값 스냅샷으로 초기화한다
CLAUDE.md 5장(조작·데이터 흐름 정책)을 코드에 반영한다. 조작분은 세션에만 쌓고 영구저장은 [저장]·[확정]에서만 하며, [초기화]는 재계산이 아니라 초기값 복원이다. 자동저장 폐지 - B06 조정창 기준벽 구간값: 800ms 디바운스 PUT을 없애고 세션(b06:culvertopt)에 담는다. flushCulvertOptions()를 B06 [저장]·[확정]과 B05 [저장]이 부른다. - B05 구조물: 조작 즉시 PUT + 서버 재조회로 화면을 덮어쓰던 것을 세션 (b05:structures) 적재로 바꾼다. 저장 전에도 고르고 지울 수 있도록 식별자를 crypto.randomUUID()로 미리 발급한다(서버는 빈 값일 때만 새로 발급). - B05에서 만지고 B06에서 확정하는 경로를 위해 flushPendingStructures()를 공용화. 초기값 스냅샷 - common_util_initial_snapshot: 자동설계 체인 성공 직후 정본 파일 4트리와 routes+자식 4표를 initial_snapshot/에 뜬다. 이후 읽기 전용. - reset_route_design: 스냅샷이 있으면 DELETE와 같은 트랜잭션에서 행을 되세우고 파일을 되돌린다. 없으면 종전 재계산 폴백. 응답에 restored를 더한다. 조작 응답 - 등고선 재적용·B06 진입 정합·[모두 적용]에서 전체 화면 오버레이 제거. - 재계산이 design을 갈아끼울 때 extra_spans를 보존한다(다른 조작값과 동일). 테스트: tmp/tests/test_initial_snapshot.py 5건 추가. 245 passed·8 failed(기존 실패 — 기슭막이 이관 때 placement가 interval→point로 바뀐 것을 테스트 미반영).
This commit is contained in:
@@ -0,0 +1,146 @@
|
||||
"""초기값 스냅샷 — 자동설계 체인이 만든 첫 결과를 그대로 떠 두고 [초기화]가 되돌린다.
|
||||
|
||||
CLAUDE.md 5장(조작·데이터 흐름 정책)의 **초기값** 층이다. 자동설계 체인이 끝난 직후
|
||||
한 번 찍고, 그 뒤로는 읽기 전용이다 — 어떤 저장 경로도 이 폴더에 쓰지 않는다.
|
||||
|
||||
[초기화]가 재계산이 아니라 복원이어야 하는 이유: [저장]·[확정]이 새 행을 만들지 않고
|
||||
최신 `routes` 행을 제자리 갱신하고 종·횡단 정본 파일도 덮어쓰므로, 초기 상태는 따로
|
||||
떠 두지 않으면 남지 않는다. 재계산으로 되살리려 해도 `structures.json`·
|
||||
`pipe_points.json` 편집분이 그대로 남아 초기값과 다른 결과가 나온다(2026-08-29).
|
||||
"""
|
||||
|
||||
import json
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import aiomysql
|
||||
|
||||
# 스냅샷 폴더는 워크플로우 단계가 아니므로 PROJECT_STORAGE_LAYOUT_V2에 넣지 않는다.
|
||||
SNAPSHOT_DIRNAME = "initial_snapshot"
|
||||
_DB_DUMP_NAME = "db.json"
|
||||
|
||||
# 프로젝트 루트 기준 상대 경로 — 정본 파일이 사는 자리 전부.
|
||||
_FILE_TREES = (
|
||||
"B05_Profile/route",
|
||||
"B06_Section/longitudinal",
|
||||
"B06_Section/cross_sections",
|
||||
"B04_PreProcess/drainage/edits",
|
||||
)
|
||||
|
||||
# `routes.id`를 참조하는 표는 스키마상 이 넷이 전부다(001_create_schema.sql:539~556).
|
||||
_CHILD_TABLES = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections")
|
||||
|
||||
|
||||
def snapshot_dir(project_root: Path) -> Path:
|
||||
return Path(project_root) / SNAPSHOT_DIRNAME
|
||||
|
||||
|
||||
def has_initial_snapshot(project_root: Path) -> bool:
|
||||
return (snapshot_dir(project_root) / _DB_DUMP_NAME).is_file()
|
||||
|
||||
|
||||
def _copy_tree(source: Path, target: Path) -> None:
|
||||
if not source.is_dir():
|
||||
return
|
||||
if target.exists():
|
||||
shutil.rmtree(target)
|
||||
shutil.copytree(source, target)
|
||||
|
||||
|
||||
async def _dump_rows(
|
||||
connection: aiomysql.Connection, table: str, route_id: int
|
||||
) -> list[dict[str, Any]]:
|
||||
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
# 표 이름은 이 모듈의 상수에서만 오므로 자리표시자 대상이 아니다.
|
||||
await cursor.execute(f"SELECT * FROM {table} WHERE route_id = %s", (route_id,)) # noqa: S608
|
||||
return list(await cursor.fetchall())
|
||||
|
||||
|
||||
def _json_safe(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""TIMESTAMP 등 JSON이 모르는 값을 문자열로 낮춘다."""
|
||||
return [
|
||||
{key: (value if _is_json_native(value) else str(value)) for key, value in row.items()}
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
def _is_json_native(value: Any) -> bool:
|
||||
return value is None or isinstance(value, (bool, int, float, str, list, dict))
|
||||
|
||||
|
||||
async def save_initial_snapshot(
|
||||
connection: aiomysql.Connection, project_root: Path, route_id: int
|
||||
) -> None:
|
||||
"""자동설계 체인 성공 직후 한 번 부른다. 이미 있으면 덮어쓰지 않는다."""
|
||||
root = Path(project_root)
|
||||
target = snapshot_dir(root)
|
||||
if has_initial_snapshot(root):
|
||||
return
|
||||
target.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
for tree in _FILE_TREES:
|
||||
_copy_tree(root / tree, target / tree.replace("/", "__"))
|
||||
|
||||
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute("SELECT * FROM routes WHERE id = %s", (route_id,))
|
||||
route = await cursor.fetchone()
|
||||
if not route:
|
||||
return
|
||||
dump: dict[str, Any] = {"routes": _json_safe([dict(route)])}
|
||||
for table in _CHILD_TABLES:
|
||||
dump[table] = _json_safe(await _dump_rows(connection, table, route_id))
|
||||
(target / _DB_DUMP_NAME).write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8")
|
||||
|
||||
|
||||
def restore_snapshot_files(project_root: Path) -> None:
|
||||
"""스냅샷 파일을 작업본 자리에 되돌린다. 스냅샷에 없던 트리는 비워 둔다."""
|
||||
root = Path(project_root)
|
||||
source = snapshot_dir(root)
|
||||
for tree in _FILE_TREES:
|
||||
_copy_tree(source / tree.replace("/", "__"), root / tree)
|
||||
|
||||
|
||||
async def restore_initial_snapshot(
|
||||
connection: aiomysql.Connection, project_root: Path, project_id: str
|
||||
) -> int | None:
|
||||
"""`routes`와 자식 4표를 스냅샷 값으로 다시 세우고 새 route id를 돌려준다.
|
||||
|
||||
호출자가 기존 `routes` 행을 지운 **뒤에** 부른다(자식은 FK CASCADE로 함께 지워진다).
|
||||
파일 복원은 트랜잭션 밖이라 `restore_snapshot_files()`를 따로 부른다.
|
||||
"""
|
||||
dump_path = snapshot_dir(Path(project_root)) / _DB_DUMP_NAME
|
||||
if not dump_path.is_file():
|
||||
return None
|
||||
dump = json.loads(dump_path.read_text(encoding="utf-8"))
|
||||
route = (dump.get("routes") or [None])[0]
|
||||
if not route:
|
||||
return None
|
||||
|
||||
route = dict(route)
|
||||
route.pop("id", None)
|
||||
route["project_id"] = project_id
|
||||
new_id = await _insert_row(connection, "routes", route)
|
||||
|
||||
for table in _CHILD_TABLES:
|
||||
for row in dump.get(table, []):
|
||||
child = dict(row)
|
||||
child.pop("id", None)
|
||||
child["route_id"] = new_id
|
||||
if "project_id" in child:
|
||||
child["project_id"] = project_id
|
||||
await _insert_row(connection, table, child)
|
||||
return new_id
|
||||
|
||||
|
||||
async def _insert_row(connection: aiomysql.Connection, table: str, row: dict[str, Any]) -> int:
|
||||
columns = list(row.keys())
|
||||
placeholders = ", ".join(["%s"] * len(columns))
|
||||
names = ", ".join(f"`{name}`" for name in columns)
|
||||
async with connection.cursor() as cursor:
|
||||
# 표 이름은 상수, 열 이름은 스냅샷이 뜬 실제 스키마에서 온다.
|
||||
await cursor.execute(
|
||||
f"INSERT INTO {table} ({names}) VALUES ({placeholders})", # noqa: S608
|
||||
tuple(row[name] for name in columns),
|
||||
)
|
||||
return int(cursor.lastrowid)
|
||||
Reference in New Issue
Block a user