Aislo/B06_wf3_ProfileCross/B06_wf3_ProfileCross_Repository.py

"""B06 종횡단 결과의 aiomysql Raw SQL 접근.

longitudinal_sections(종단면 1건), cross_sections(측점별 다건) 테이블에
메타데이터와 상대 경로를 기록한다. 상세 샘플 데이터는 파일에 저장하고 DB에는
요약 data(JSON)와 경로만 기록한다.
"""

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

import aiomysql

_STAGE_ROOT = "B06_wf3_ProfileCross"


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"B06 산출물 경로는 {_STAGE_ROOT} 아래여야 합니다.")
    return normalized.as_posix()


async def get_confirmed_route_context(
    connection: aiomysql.Connection, project_id: UUID
) -> dict[str, Any] | None:
    """프로젝트의 최신 확정 경로와 연결된 지표면 좌표계를 조회한다.

    surface_models.crs_epsg가 NULL이면(분석에 사용한 입력 파일에 좌표계가
    없던 경우) 같은 프로젝트 input_files의 감지된 좌표계로 폴백한다.
    """
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT r.id AS route_id,
                   COALESCE(
                       sm.crs_epsg,
                       (SELECT f.crs_epsg
                        FROM input_files f
                        WHERE f.project_id = r.project_id AND f.crs_epsg IS NOT NULL
                        ORDER BY f.id DESC
                        LIMIT 1)
                   ) AS crs_epsg
            FROM routes r
            LEFT JOIN surface_models sm ON sm.id = r.surface_model_id
            WHERE r.project_id = %s AND r.status = 'CONFIRMED'
            ORDER BY r.computed_at DESC, r.id DESC
            LIMIT 1
            """,
            (str(project_id),),
        )
        row = await cursor.fetchone()
    if not row:
        return None
    return {
        "route_id": int(row["route_id"]),
        "crs_epsg": int(row["crs_epsg"]) if row["crs_epsg"] is not None else None,
    }


async def delete_sections_for_route(connection: aiomysql.Connection, route_id: int) -> None:
    """경로 재생성 전에 기존 종횡단 레코드를 삭제한다 (멱등 재실행)."""
    async with connection.cursor() as cursor:
        await cursor.execute("DELETE FROM cross_sections WHERE route_id = %s", (route_id,))
        await cursor.execute("DELETE FROM longitudinal_sections WHERE route_id = %s", (route_id,))


async def create_longitudinal_section(
    connection: aiomysql.Connection,
    *,
    project_id: UUID,
    route_id: int,
    data: dict[str, Any] | None,
    longitudinal_file_path: str,
    status: str = "DRAFT",
) -> int:
    """종단면 메타데이터를 저장하고 생성된 ID를 반환한다."""
    file_rel = _validate_stage_path(longitudinal_file_path)
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            INSERT INTO longitudinal_sections (
                project_id, route_id, computed_at, data, longitudinal_file_path, status
            )
            VALUES (%s, %s, CURRENT_TIMESTAMP, %s, %s, %s)
            """,
            (
                str(project_id),
                route_id,
                json.dumps(data, ensure_ascii=False) if data is not None else None,
                file_rel,
                status,
            ),
        )
        new_id = cursor.lastrowid
    if not new_id:
        raise RuntimeError("longitudinal_sections 레코드 생성 결과에 ID가 없습니다.")
    return int(new_id)


async def insert_cross_sections(
    connection: aiomysql.Connection,
    *,
    project_id: UUID,
    route_id: int,
    sections: list[dict[str, Any]],
) -> int:
    """측점별 횡단면 레코드를 일괄 저장하고 저장 건수를 반환한다.

    각 section dict: {chainage_m, sequence_num, data, cross_section_file_path, status?}
    """
    if not sections:
        return 0
    rows = []
    for section in sections:
        file_rel = _validate_stage_path(section["cross_section_file_path"])
        data = section.get("data")
        rows.append(
            (
                str(project_id),
                route_id,
                section.get("chainage_m"),
                section.get("sequence_num"),
                json.dumps(data, ensure_ascii=False) if data is not None else None,
                file_rel,
                section.get("status", "DRAFT"),
            )
        )
    async with connection.cursor() as cursor:
        await cursor.executemany(
            """
            INSERT INTO cross_sections (
                project_id, route_id, chainage_m, sequence_num,
                data, cross_section_file_path, status
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            """,
            rows,
        )
    return len(rows)


async def get_longitudinal_section(
    connection: aiomysql.Connection, project_id: UUID, route_id: int
) -> dict[str, Any] | None:
    """경로의 종단면 메타데이터를 조회한다 (없으면 None)."""
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT id, data, longitudinal_file_path, status, computed_at
            FROM longitudinal_sections
            WHERE project_id = %s AND route_id = %s
            ORDER BY id DESC
            LIMIT 1
            """,
            (str(project_id), route_id),
        )
        row = await cursor.fetchone()
    if not row:
        return None
    data = row[1]
    if isinstance(data, str):
        data = json.loads(data)
    return {
        "id": int(row[0]),
        "data": data if isinstance(data, dict) else None,
        "longitudinal_file_path": row[2],
        "status": row[3],
        "computed_at": row[4].isoformat() if row[4] else None,
    }


async def count_cross_sections(connection: aiomysql.Connection, route_id: int) -> int:
    """경로에 저장된 횡단면 개수를 반환한다."""
    async with connection.cursor() as cursor:
        await cursor.execute("SELECT COUNT(*) FROM cross_sections WHERE route_id = %s", (route_id,))
        row = await cursor.fetchone()
    return int(row[0]) if row else 0


async def confirm_sections_for_route(connection: aiomysql.Connection, route_id: int) -> None:
    """경로의 종횡단면 상태를 CONFIRMED로 변경한다."""
    async with connection.cursor() as cursor:
        await cursor.execute(
            "UPDATE longitudinal_sections SET status = 'CONFIRMED' WHERE route_id = %s",
            (route_id,),
        )
        await cursor.execute(
            "UPDATE cross_sections SET status = 'CONFIRMED' WHERE route_id = %s",
            (route_id,),
        )