Aislo/B05_Profile/B05_Profile_Repository.py

"""B05 경로 설계 결과의 aiomysql Raw SQL 접근.

routes(경로), route_points(렌더링 샘플), route_statistics(통계) 테이블에
메타데이터와 상대 경로를 기록한다. 전체 폴리라인은 route_data_path의
GeoJSON 파일에 저장하고 DB에는 경로만 기록한다.
"""

import json
from pathlib import PurePosixPath
from typing import Any
from uuid import UUID

import aiomysql

_STAGE_ROOT = "B05_Profile"


def _validate_stage_path(relative_path: str) -> str:
    normalized = PurePosixPath(relative_path.replace("\\", "/"))
    if normalized.is_absolute() or ".." in normalized.parts:
        raise ValueError("DB에는 프로젝트 루트 기준 상대 경로만 저장할 수 있습니다.")
    if not normalized.parts or normalized.parts[0] != _STAGE_ROOT:
        raise ValueError(f"B05 산출물 경로는 {_STAGE_ROOT} 아래여야 합니다.")
    return normalized.as_posix()


async def create_route(
    connection: aiomysql.Connection,
    *,
    project_id: UUID,
    surface_model_id: int | None,
    total_length_m: float | None,
    start_chainage_m: float | None,
    end_chainage_m: float | None,
    grade_percent: list[float] | None,
    constraints: dict[str, Any] | None,
    algorithm_params: dict[str, Any] | None,
    route_data_path: str,
    status: str = "DRAFT",
) -> int:
    """경로 메타데이터를 저장하고 생성된 ID를 반환한다."""
    route_rel = _validate_stage_path(route_data_path)
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            INSERT INTO routes (
                project_id, surface_model_id, status,
                start_chainage_m, end_chainage_m, total_length_m,
                grade_percent, constraints, algorithm_params,
                route_data_path, computed_at
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP)
            """,
            (
                str(project_id),
                surface_model_id,
                status,
                start_chainage_m,
                end_chainage_m,
                total_length_m,
                json.dumps(grade_percent) if grade_percent is not None else None,
                json.dumps(constraints, ensure_ascii=False) if constraints is not None else None,
                json.dumps(algorithm_params, ensure_ascii=False)
                if algorithm_params is not None
                else None,
                route_rel,
            ),
        )
        route_id = cursor.lastrowid
    if not route_id:
        raise RuntimeError("routes 레코드 생성 결과에 ID가 없습니다.")
    return int(route_id)


async def insert_route_points(
    connection: aiomysql.Connection,
    route_id: int,
    points: list[dict[str, Any]],
) -> int:
    """경로 렌더링 샘플 포인트를 일괄 저장하고 저장 건수를 반환한다.

    각 point dict: {x, y, z, chainage_m, elevation_m, slope_percent, sequence_num}
    """
    if not points:
        return 0
    rows = [
        (
            route_id,
            point.get("x"),
            point.get("y"),
            point.get("z"),
            point.get("chainage_m"),
            point.get("elevation_m"),
            point.get("slope_percent"),
            point.get("sequence_num"),
        )
        for point in points
    ]
    async with connection.cursor() as cursor:
        await cursor.executemany(
            """
            INSERT INTO route_points (
                route_id, model_x, model_y, model_z,
                chainage_m, elevation_m, slope_percent, sequence_num
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            """,
            rows,
        )
    return len(rows)


async def create_route_statistics(
    connection: aiomysql.Connection,
    *,
    route_id: int,
    min_slope: float | None,
    max_slope: float | None,
    mean_slope: float | None,
    cut_volume_m3: float | None = None,
    fill_volume_m3: float | None = None,
    tree_cutting_volume: float | None = None,
    cost_score: float | None = None,
) -> int:
    """경로 통계를 저장하고 생성된 ID를 반환한다."""
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            INSERT INTO route_statistics (
                route_id, min_slope, max_slope, mean_slope,
                cut_volume_m3, fill_volume_m3, tree_cutting_volume, cost_score
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            """,
            (
                route_id,
                min_slope,
                max_slope,
                mean_slope,
                cut_volume_m3,
                fill_volume_m3,
                tree_cutting_volume,
                cost_score,
            ),
        )
        stat_id = cursor.lastrowid
    if not stat_id:
        raise RuntimeError("route_statistics 레코드 생성 결과에 ID가 없습니다.")
    return int(stat_id)


async def get_latest_route(
    connection: aiomysql.Connection, project_id: UUID
) -> dict[str, Any] | None:
    """프로젝트의 최신 경로를 조회한다 (없으면 None)."""
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT r.id, r.status, r.surface_model_id, r.total_length_m,
                   r.route_data_path, r.constraints, r.algorithm_params, r.computed_at,
                   rs.min_slope, rs.max_slope, rs.mean_slope, rs.cost_score
            FROM routes r
            LEFT JOIN route_statistics rs ON rs.route_id = r.id
            WHERE r.project_id = %s
            ORDER BY r.computed_at DESC, r.id DESC
            LIMIT 1
            """,
            (str(project_id),),
        )
        row = await cursor.fetchone()
    if not row:
        return None
    result = dict(row)
    result["id"] = int(result["id"])
    result["computed_at"] = result["computed_at"].isoformat() if result["computed_at"] else None
    for key in ("constraints", "algorithm_params"):
        if isinstance(result.get(key), str):
            result[key] = json.loads(result[key])
    return result


async def get_route_points(
    connection: aiomysql.Connection,
    route_id: int,
) -> list[dict[str, Any]]:
    """최신 경로의 DB 렌더 좌표를 순서대로 조회한다."""
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT model_x AS x, model_y AS y, model_z AS z,
                   chainage_m, elevation_m, slope_percent, sequence_num
            FROM route_points
            WHERE route_id = %s
            ORDER BY sequence_num ASC, id ASC
            """,
            (route_id,),
        )
        rows = await cursor.fetchall()
    return [dict(row) for row in rows]


async def confirm_route(connection: aiomysql.Connection, route_id: int) -> None:
    """경로 상태를 CONFIRMED로 변경한다."""
    async with connection.cursor() as cursor:
        await cursor.execute(
            "UPDATE routes SET status = 'CONFIRMED' WHERE id = %s",
            (route_id,),
        )


async def update_longitudinal_grade_summary(
    connection: aiomysql.Connection,
    *,
    route_id: int,
    grade_summary: dict[str, Any] | None,
) -> None:
    """계획선 편집 저장 시 longitudinal_sections.data의 grade_summary만 갱신한다.

    측점·반폭 등 다른 옵션 스냅샷은 그대로 두어야 하므로 data 전체를 덮어쓰지 않고
    읽어서 해당 키만 바꿔 다시 쓴다(단일 소스 유지).
    """
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT id, data FROM longitudinal_sections
            WHERE route_id = %s ORDER BY id DESC LIMIT 1
            """,
            (route_id,),
        )
        row = await cursor.fetchone()
        if not row:
            return
        stored = row[1]
        if isinstance(stored, str):
            stored = json.loads(stored)
        data = stored if isinstance(stored, dict) else {}
        data["grade_summary"] = grade_summary
        await cursor.execute(
            "UPDATE longitudinal_sections SET data = %s WHERE id = %s",
            (json.dumps(data, ensure_ascii=False), int(row[0])),
        )


async def sync_cross_section_rows(
    connection: aiomysql.Connection, route_id: int, chainages: list[float]
) -> int:
    """종단 정본 측점에 횡단 행을 맞춘다 — 정본에 없는 측점의 행(옮긴 · 지운 구조물 측점의 옛 자리).

    0.5 m 안에 행 없는 새 측점이 있으면 그리로 옮긴다(옛 소수 → 정수 m · 사용자 설계 값 그대로 ·
    면적은 저장 뒤 서버 재계산) · 나머지는 지운다 — 남기면 수량(B08 토적)이 옛 자리 측점을 그대로
    셈(272). 같은 측점 = 0.01 m 안(`_rowless_station_chainages` 와 같은 폭). 측점 목록이 비면
    아무것도 안 건드린다. 옮기거나 지운 행 수.
    """
    if not chainages:
        return 0
    async with connection.cursor() as cursor:
        await cursor.execute(
            "SELECT id, chainage_m FROM cross_sections WHERE route_id = %s", (route_id,)
        )
        rows = [(int(row[0]), float(row[1])) for row in await cursor.fetchall()]
        orphans = [
            (row_id, at) for row_id, at in rows if all(abs(at - c) >= 0.01 for c in chainages)
        ]
        free = [c for c in chainages if all(abs(c - at) >= 0.01 for _, at in rows)]
        pairs = sorted(
            (abs(c - at), row_id, c) for row_id, at in orphans for c in free if abs(c - at) <= 0.5
        )
        moved: dict[int, float] = {}
        for _gap, row_id, chainage in pairs:
            if row_id not in moved and chainage not in moved.values():
                moved[row_id] = chainage
        for row_id, chainage in moved.items():
            await cursor.execute(
                "UPDATE cross_sections SET chainage_m = %s WHERE id = %s", (chainage, row_id)
            )
        dropped = [row_id for row_id, _ in orphans if row_id not in moved]
        if dropped:
            marks = ", ".join(["%s"] * len(dropped))
            await cursor.execute(f"DELETE FROM cross_sections WHERE id IN ({marks})", dropped)
    return len(orphans)


async def get_surface_crs_epsg(
    connection: aiomysql.Connection, project_id: UUID, surface_model_id: int
) -> int | None:
    """종횡단 메타데이터용 좌표계를 조회한다.

    지표면 모델 crs_epsg가 NULL이면 같은 프로젝트 input_files의 감지된
    좌표계로 폴백한다 (B06 get_confirmed_route_context와 동일 규칙).
    """
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT COALESCE(
                       (SELECT sm.crs_epsg FROM surface_models sm WHERE sm.id = %s),
                       (SELECT f.crs_epsg
                        FROM input_files f
                        WHERE f.project_id = %s AND f.crs_epsg IS NOT NULL
                        ORDER BY f.id DESC
                        LIMIT 1)
                   ) AS crs_epsg
            """,
            (surface_model_id, str(project_id)),
        )
        row = await cursor.fetchone()
    return int(row["crs_epsg"]) if row and row["crs_epsg"] is not None else None