Aislo/common_util/common_util_design_sets.py

"""지형마다 설계 한 벌(765 · 요구 605 · 606-1) — 활성 벌은 제자리(DB · 파일) · 쉬는 벌은 묶음.

묶음 = `<프로젝트>/design_sets/<las|sheet>/`
  db.json           노선과 자식 넷 · 단계 2 이후 행 · projects.status
  files/            벌 몫 파일(프로젝트 뿌리 기준 상대 경로 그대로)
  initial_snapshot/ 그 벌의 초기값([초기화])
  meta.json         지형 · 지문(필터 · 표현 · 평활 · 모델 파일 내용) · 옮긴 자리 · 못 지은 까닭
벌 몫 = B05 노선 정본 · 코리도 · B06~B09 · 배수 결과 중 노선 몫 · 초기값.
공유(안 옮김 · 안 지움) = B05 비용면 · 예정노선 · 추천 고름 · 배수 공유 캐시
  (도엽 좌표 · 하강장 · 강우) · 그 밖 전부.
떠 두기 = 벌 몫을 묶음으로 옮기고(같은 볼륨 = 이름 바꿈) DB 노선 지움 · 단계 2 이후 처음.
꺼내 끼우기 = 새 노선 id 로 넣고 자식 다시 잇기 · 코리도 이름(`rekey_id_files`) · 묶음 지움.
지형 칩 = 맞바꿈(`swap_design_set`) — 실패 = `save_guard` 로 통째 되돌림 · 묶음 없음 · 지문 어긋남 =
그 자리에서 자동 설계 체인으로 한 번 지음. 계산 자리 = ③ 서버 단독.
"""

from __future__ import annotations

import asyncio
import json
import shutil
import time
from pathlib import Path
from typing import Any
from uuid import UUID

import aiomysql

from common_util.common_util_initial_snapshot import (
    DRAINAGE_SHARED_PREFIXES,
    SNAPSHOT_DIRNAME,
    _json_safe,
)
from common_util.common_util_surface_confirmation import (
    AnalysisTerrain,
    analysis_terrain_params,
    analysis_terrain_sources,
    design_terrain,
    get_surface_confirmation_params,
)

SETS_DIRNAME = "design_sets"
MODELS_DIR = Path("B04_PreProcess") / "models"
TERRAINS: tuple[AnalysisTerrain, AnalysisTerrain] = ("las", "sheet")
#: 통째로 벌 몫인 폴더
_SET_DIRS = (
    "B05_Profile/corridor",
    "B06_Section",
    "B07_DesignDetail",
    "B08_Quantity",
    "B09_Estimation",
    SNAPSHOT_DIRNAME,
)
#: 공유를 뺀 나머지가 벌 몫인 폴더 → 공유 이름 앞머리
_SPLIT_DIRS = {
    "B05_Profile/route": ("cost_surface_", "expected_route", "initial_choice"),
    "B04_PreProcess/drainage": DRAINAGE_SHARED_PREFIXES,
}
_ROUTE_CHILDREN = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections")
_STAGE_COLUMNS = (
    "state",
    "progress_percent",
    "params",
    "message",
    "started_at",
    "completed_at",
)
#: 벌이 되는 단계(노선 · 횡단 · 상세 · 수량 · 내역)
FIRST_SET_STAGE = 2


def sets_dir(root: Path) -> Path:
    return Path(root) / SETS_DIRNAME


def bundle_dir(root: Path, terrain: str) -> Path:
    return sets_dir(root) / terrain


def set_entries(root: Path) -> list[str]:
    """지금 제자리에 있는 벌 몫(프로젝트 뿌리 기준 상대 경로 · 폴더 또는 파일)."""
    root = Path(root)
    out = [name for name in _SET_DIRS if (root / name).exists()]
    for folder, shared in _SPLIT_DIRS.items():
        base = root / folder
        if base.is_dir():
            out += [
                f"{folder}/{path.name}"
                for path in sorted(base.iterdir())
                if not path.name.startswith(shared)
            ]
    return out


def _bundle_path(bundle: Path, relative: str) -> Path:
    if relative == SNAPSHOT_DIRNAME:
        return bundle / SNAPSHOT_DIRNAME
    return bundle / "files" / relative


def _move(source: Path, target: Path) -> None:
    if target.is_dir():
        shutil.rmtree(target)
    elif target.exists():
        target.unlink()
    target.parent.mkdir(parents=True, exist_ok=True)
    shutil.move(str(source), str(target))


def read_meta(root: Path, terrain: str) -> dict[str, Any]:
    try:
        return json.loads((bundle_dir(root, terrain) / "meta.json").read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}


def _write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, default=str), encoding="utf-8")


def note_unbuilt(root: Path, terrain: str, reason: str) -> None:
    """그 지형 벌을 못 지음 — 묶음 비우고 meta.json 에 까닭만."""
    bundle = bundle_dir(root, terrain)
    shutil.rmtree(bundle, ignore_errors=True)
    _write_json(bundle / "meta.json", {"terrain": terrain, "built": False, "reason": reason})


def discard_sets(root: Path) -> bool:
    """묶음 모두 버림(전처리 다시 · 입력 바뀜)."""
    target = sets_dir(root)
    if not target.exists():
        return False
    shutil.rmtree(target, ignore_errors=True)
    return True


def other_terrain(params: dict[str, Any], root: Path) -> AnalysisTerrain | None:
    """설계 지형 말고 고를 수 있는 지형 — 없으면 None."""
    design = design_terrain(params)
    sources = analysis_terrain_sources(params, Path(root) / MODELS_DIR)
    return next((name for name in sources if name != design), None)


# ── 지문 ───────────────────────────────────────────────────────────────────

_FINGERPRINT_KEYS = ("terrain", "source_filter", "method", "smooth", "model_digest")


async def set_fingerprint(
    connection: aiomysql.Connection, project_id: str, root: Path, terrain: str
) -> dict[str, Any] | None:
    """그 지형 벌이 기댄 입력 — 지형 · 필터 · 표현 · 평활 · 지표면 모델 id · 모델 파일 내용 지문.
    그 지형 모델이 없으면 None."""
    from B04_PreProcess.B04_PreProcess_Repository import list_surface_models
    from common_util.common_util_derived_store import file_digest

    params = await get_surface_confirmation_params(connection, project_id)
    target = analysis_terrain_params(params, terrain)  # type: ignore[arg-type]
    if design_terrain(target) != terrain:
        return None
    for model in await list_surface_models(connection, UUID(project_id)):
        generation = model.get("generation_params") or {}
        if (
            model["status"] in ("COMPLETE", "CONFIRMED")
            and str(model["model_type"]).lower() == str(target["method"]).lower()
            and str(generation.get("source_filter", "")).lower()
            == str(target["source_filter"]).lower()
        ):
            path = model.get("model_file_path")
            digest = await asyncio.to_thread(file_digest, Path(root) / path if path else None)
            return {
                "terrain": terrain,
                "source_filter": str(target["source_filter"]),
                "method": str(target["method"]),
                "smooth": bool(target.get("smooth")),
                "model_id": int(model["id"]),
                "model_digest": digest,
            }
    return None


def bundle_ready(root: Path, terrain: str, fingerprint: dict[str, Any] | None) -> bool:
    """묶음이 있고 지금 입력과 같은 지문인가(모델 id 는 안 봄 — 복제 · 갈음에 바뀜)."""
    meta = read_meta(root, terrain)
    if not fingerprint or not meta.get("built"):
        return False
    if not (bundle_dir(root, terrain) / "db.json").is_file():
        return False
    return all(meta.get(key) == fingerprint.get(key) for key in _FINGERPRINT_KEYS)


# ── DB ─────────────────────────────────────────────────────────────────────


async def _select(cursor: Any, sql: str, args: Any) -> list[dict[str, Any]]:
    await cursor.execute(sql, args)
    return _json_safe([dict(row) for row in await cursor.fetchall()])


async def _dump_db(connection: aiomysql.Connection, project_id: str) -> dict[str, Any]:
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        dump: dict[str, Any] = {
            "routes": await _select(
                cursor, "SELECT * FROM routes WHERE project_id=%s ORDER BY id", (project_id,)
            ),
            "stages": await _select(
                cursor,
                "SELECT stage_no, " + ", ".join(_STAGE_COLUMNS) + " FROM project_workflow_stages"
                " WHERE project_id=%s AND stage_no >= %s ORDER BY stage_no",
                (project_id, FIRST_SET_STAGE),
            ),
            "project_status": (
                await _select(cursor, "SELECT status FROM projects WHERE id=%s", (project_id,))
                or [{}]
            )[0].get("status"),
        }
        ids = [row["id"] for row in dump["routes"]] or [-1]
        marks = ",".join(["%s"] * len(ids))
        for table in _ROUTE_CHILDREN:
            dump[table] = await _select(
                cursor, f"SELECT * FROM {table} WHERE route_id IN ({marks}) ORDER BY id", ids
            )
    return dump


async def _clear_db(connection: aiomysql.Connection, project_id: str) -> None:
    """노선(자식은 CASCADE) 지움 · 단계 2 이후 처음 — 한 트랜잭션."""
    from common_util.common_util_workflow_state import reset_stages_after_input_change

    await connection.begin()
    try:
        async with connection.cursor() as cursor:
            await cursor.execute("DELETE FROM routes WHERE project_id=%s", (project_id,))
            await reset_stages_after_input_change(cursor, project_id, FIRST_SET_STAGE)
        await connection.commit()
    except Exception:
        await connection.rollback()
        raise


async def _insert(cursor: Any, table: str, row: dict[str, Any]) -> int:
    names = ", ".join(f"`{name}`" for name in row)
    marks = ", ".join(["%s"] * len(row))
    await cursor.execute(f"INSERT INTO {table} ({names}) VALUES ({marks})", tuple(row.values()))
    return int(cursor.lastrowid)


async def _load_db(
    connection: aiomysql.Connection, project_id: str, dump: dict[str, Any]
) -> dict[int, int]:
    """노선 새 id 로 넣고 자식 다시 잇기 · 단계 · 상태 되씀 — 한 트랜잭션 → 옛 id → 새 id."""
    route_map: dict[int, int] = {}
    await connection.begin()
    try:
        async with connection.cursor() as cursor:
            for route in dump.get("routes", []):
                row = {**route, "project_id": project_id}
                old = int(row.pop("id"))
                route_map[old] = await _insert(cursor, "routes", row)
            for table in _ROUTE_CHILDREN:
                for child in dump.get(table, []):
                    row = dict(child)
                    row.pop("id", None)
                    row["route_id"] = route_map[int(row["route_id"])]
                    if "project_id" in row:
                        row["project_id"] = project_id
                    await _insert(cursor, table, row)
            for stage in dump.get("stages", []):
                sets = ", ".join(f"`{name}`=%s" for name in _STAGE_COLUMNS)
                await cursor.execute(
                    f"UPDATE project_workflow_stages SET {sets}"
                    " WHERE project_id=%s AND stage_no=%s",
                    [stage.get(name) for name in _STAGE_COLUMNS] + [project_id, stage["stage_no"]],
                )
            if dump.get("project_status") is not None:
                await cursor.execute(
                    "UPDATE projects SET status=%s WHERE id=%s",
                    (dump["project_status"], project_id),
                )
        await connection.commit()
    except Exception:
        await connection.rollback()
        raise
    return route_map


# ── 떠 두기 · 꺼내 끼우기 ───────────────────────────────────────────────────


def _clear_files(root: Path) -> None:
    for relative in set_entries(root):
        path = Path(root) / relative
        if path.is_dir():
            shutil.rmtree(path, ignore_errors=True)
        else:
            path.unlink(missing_ok=True)


async def clear_active_set(connection: aiomysql.Connection, root: Path, project_id: str) -> None:
    """제자리 벌 비움(못 지은 벌 · 반쯤 지은 결과) — 공유 캐시는 그대로."""
    await _clear_db(connection, project_id)
    await asyncio.to_thread(_clear_files, root)


def _park_files(root: Path, work: Path) -> list[str]:
    entries = set_entries(root)
    for relative in entries:
        _move(Path(root) / relative, _bundle_path(work, relative))
    return entries


async def park_active_set(
    connection: aiomysql.Connection,
    root: Path,
    project_id: str,
    fingerprint: dict[str, Any],
) -> dict[str, Any]:
    """제자리 벌을 `design_sets/<지형>/` 으로 떠 두고 제자리를 비움 → meta."""
    root = Path(root)
    terrain = str(fingerprint["terrain"])
    dump = await _dump_db(connection, project_id)
    work = sets_dir(root) / f".{terrain}_new"
    await asyncio.to_thread(shutil.rmtree, work, True)
    _write_json(work / "db.json", dump)
    entries = await asyncio.to_thread(_park_files, root, work)
    meta = {
        **fingerprint,
        "built": True,
        "entries": entries,
        "routes": [row["id"] for row in dump["routes"]],
        "parked_at": time.strftime("%Y-%m-%d %H:%M:%S"),
    }
    _write_json(work / "meta.json", meta)
    await _clear_db(connection, project_id)
    bundle = bundle_dir(root, terrain)
    await asyncio.to_thread(shutil.rmtree, bundle, True)
    work.rename(bundle)
    return meta


def _unpark_files(root: Path, bundle: Path, entries: list[str], route_map: dict) -> None:
    from common_util.common_util_project_copy import rekey_id_files

    for relative in entries:
        source = _bundle_path(bundle, relative)
        if source.exists():
            _move(source, Path(root) / relative)
    rekey_id_files(Path(root), {}, route_map)
    shutil.rmtree(bundle, ignore_errors=True)


async def load_set(
    connection: aiomysql.Connection, root: Path, project_id: str, terrain: str
) -> dict[int, int] | None:
    """묶음을 제자리로 꺼내 끼움(제자리는 비어 있어야 함) → 옛 노선 id → 새 id ·
    묶음 없음 = None."""
    bundle = bundle_dir(root, terrain)
    try:
        dump = json.loads((bundle / "db.json").read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    meta = read_meta(root, terrain)
    route_map = await _load_db(connection, project_id, dump)
    await asyncio.to_thread(_unpark_files, root, bundle, list(meta.get("entries") or []), route_map)
    return route_map


# ── 지형 바꿈 ──────────────────────────────────────────────────────────────


async def switch_terrain(connection: aiomysql.Connection, project_id: UUID, terrain: str) -> int:
    """설계 지형(stage 1 확정값 · 확정 모델)을 바꿈 — 한 트랜잭션 → 확정 모델 id."""
    from B04_PreProcess.B04_PreProcess_Service import switch_design_terrain

    await connection.begin()
    try:
        switched = await switch_design_terrain(connection, project_id, terrain)  # type: ignore[arg-type]
        await connection.commit()
    except Exception:
        await connection.rollback()
        raise
    if switched:
        return int(switched[0])
    async with connection.cursor() as cursor:
        await cursor.execute(
            "SELECT id FROM surface_models WHERE project_id=%s AND status='CONFIRMED'",
            (str(project_id),),
        )
        row = await cursor.fetchone()
    if not row:
        raise LookupError("확정 지표면 모델이 없습니다.")
    return int(row[0])


async def swap_design_set(project_id: UUID, root: Path, target: str) -> dict[str, Any]:
    """지형 칩 = 맞바꿈 — 지금 벌 떠 둠 → 지형 바꿈 → 고른 벌 꺼내 끼움(없음 · 묵음 = 한 번 지음).
    실패 = DB 행 · 벌 몫 · 묶음을 떠 둔 대로 되돌리고 예외 그대로. 잠금은 부르는 쪽."""
    from B03_FileInput.B03_FileInput_Service_Chain import run_auto_design_chain
    from common_util import common_util_save_guard as guard
    from common_util.common_util_initial_snapshot import (
        clear_design_failed,
        clear_designing,
        is_design_failed,
        read_design_failure,
    )
    from config.config_db import get_db_pool

    started = time.perf_counter()
    pid, root = str(project_id), Path(root)
    pool = get_db_pool()
    async with pool.acquire() as connection:
        now = design_terrain(await get_surface_confirmation_params(connection, pid))
        if now == target:
            return {"changed": False, "design": now}
        here = await set_fingerprint(connection, pid, root, now) or {"terrain": now}
        there = await set_fingerprint(connection, pid, root, target)
        if there is None:
            raise LookupError("고른 지형의 지표면 모델이 없습니다.")
        rows = await guard.snapshot_rows(connection, pid, surface=True)
    ready = bundle_ready(root, target, there)
    # 되돌림 자리 = 지금 벌 몫 + 꺼낼 벌 몫(끼운 뒤 생기는 파일도 지우게) + 묶음
    scope = tuple(
        dict.fromkeys(
            [*set_entries(root), *(read_meta(root, target).get("entries") or []), SETS_DIRNAME]
        )
    )
    files = await asyncio.to_thread(guard.snapshot_storage, root, scope)
    had_failed = is_design_failed(root)
    had_bundle = bundle_dir(root, now).exists()
    try:
        async with pool.acquire() as connection:
            await park_active_set(connection, root, pid, here)
            model_id = await switch_terrain(connection, project_id, target)
            loaded = await load_set(connection, root, pid, target) if ready else None
        if loaded is None:
            await asyncio.to_thread(shutil.rmtree, bundle_dir(root, target), True)
            summary = await run_auto_design_chain(project_id, model_id, inner=True)
            if summary is None:
                raise RuntimeError(read_design_failure(root) or "고른 지형 설계를 짓지 못했습니다.")
    except BaseException:
        files.restore()
        if not had_bundle:  # 되돌림은 파일만 — 떠 둔 빈 폴더도 걷음
            shutil.rmtree(bundle_dir(root, now), ignore_errors=True)
        async with pool.acquire() as connection:
            await guard.restore_rows(connection, pid, rows)
        if not had_failed:
            clear_design_failed(root)
        raise
    finally:
        files.discard()
        clear_designing(root)  # 지음 길(inner 체인)이 건 마커
    return {
        "changed": True,
        "design": target,
        "built": loaded is None,
        "seconds": round(time.perf_counter() - started, 2),
    }