Aislo/B06_Section/B06_Section_Repository.py

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

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

import json
from collections.abc import Callable
from pathlib import PurePosixPath
from typing import Any
from uuid import UUID

import aiomysql

_STAGE_ROOT = "B06_Section"


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 get_latest_section_options(
    connection: aiomysql.Connection, project_id: UUID
) -> dict[str, Any] | None:
    """프로젝트 최신 종단면 data에 저장된 생성 옵션 스냅샷을 반환한다 (없으면 None)."""
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT data
            FROM longitudinal_sections
            WHERE project_id = %s
            ORDER BY id DESC
            LIMIT 1
            """,
            (str(project_id),),
        )
        row = await cursor.fetchone()
    if not row or not row[0]:
        return None
    data = row[0]
    if isinstance(data, str):
        data = json.loads(data)
    options = data.get("options") if isinstance(data, dict) else None
    return options if isinstance(options, dict) else None


async def list_recent_company_projects(
    connection: aiomysql.Connection, company_id: int, exclude_project_id: UUID
) -> list[dict[str, Any]]:
    """같은 회사의 최근 프로젝트 5개(최근 갱신순, 현재·삭제 프로젝트 제외).

    설계값 보유 여부와 무관하게 보여준다 — 선택 시 설계값 조회에서 없으면 안내한다.
    """
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT p.id AS project_id, p.name AS name
            FROM projects p
            WHERE p.company_id = %s
              AND p.id <> %s
              AND p.deleted_at IS NULL
            ORDER BY p.updated_at DESC
            LIMIT 5
            """,
            (company_id, str(exclude_project_id)),
        )
        rows = await cursor.fetchall()
    return [{"project_id": row["project_id"], "name": row["name"]} for row in rows]


async def get_project_standard_cross_section(
    connection: aiomysql.Connection, company_id: int, project_id: UUID
) -> dict[str, Any] | None:
    """회사 스코프로 특정 프로젝트의 표준횡단 설정값을 반환한다(타 회사 접근 차단).

    company_id 조건으로 남의 회사 프로젝트 값은 조회되지 않는다(권한 강제).
    """
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT l.data
            FROM longitudinal_sections l
            JOIN projects p ON p.id = l.project_id
            WHERE l.project_id = %s
              AND p.company_id = %s
              AND p.deleted_at IS NULL
              AND JSON_EXTRACT(l.data, '$.options.standard_cross_section') IS NOT NULL
            ORDER BY l.id DESC
            LIMIT 1
            """,
            (str(project_id), company_id),
        )
        row = await cursor.fetchone()
    if not row or not row[0]:
        return None
    data = row[0]
    if isinstance(data, str):
        data = json.loads(data)
    options = data.get("options") if isinstance(data, dict) else None
    standard = options.get("standard_cross_section") if isinstance(options, dict) else None
    return standard if isinstance(standard, dict) else None


async def get_latest_grade_options(
    connection: aiomysql.Connection, project_id: UUID
) -> dict[str, Any] | None:
    """최신 종단면 data에 저장된 계획선 **사용자 입력값**을 반환한다 (없으면 None).

    해석이 끝난 기준값(`grade_options`)이 아니라 사용자가 명시 입력한 값
    (`grade_overrides`)만 돌려준다. 전자를 폴백으로 되쓰면 등급·지형 구분을 바꿔도
    옛 기본값이 법정값을 이겨 갱신되지 않는다.
    """
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT data
            FROM longitudinal_sections
            WHERE project_id = %s
            ORDER BY id DESC
            LIMIT 1
            """,
            (str(project_id),),
        )
        row = await cursor.fetchone()
    if not row or not row[0]:
        return None
    data = row[0]
    if isinstance(data, str):
        data = json.loads(data)
    overrides = data.get("grade_overrides") if isinstance(data, dict) else None
    return overrides if isinstance(overrides, dict) else None


async def get_route_generation_source(
    connection: aiomysql.Connection, project_id: UUID, route_id: int
) -> dict[str, Any] | None:
    """종횡단 재생성에 필요한 경로 GeoJSON 경로와 좌표계를 route_id로 조회한다."""
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            """
            SELECT r.route_data_path,
                   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.id = %s AND r.project_id = %s
            LIMIT 1
            """,
            (route_id, str(project_id)),
        )
        row = await cursor.fetchone()
    if not row or not row["route_data_path"]:
        return None
    return {
        "route_data_path": str(row["route_data_path"]),
        "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 update_cross_section_design(
    connection: aiomysql.Connection,
    *,
    route_id: int,
    chainage_m: float,
    design: dict[str, Any],
    project_id: UUID | None = None,
) -> bool:
    """측점 하나의 data.design(잠정 설계 지정·단면적)을 병합 저장한다.

    기존 data 요약을 보존하고 design 키만 갱신한다. 대상 측점을 chainage 근사로
    찾으며(부동소수 오차 허용), 갱신 여부를 반환한다.
    구조물(비정규) 측점은 B05 확정이 파일만 쓰고 DB 행을 만들지 않으므로, 행이 없고
    project_id가 오면 새 행을 삽입한다(upsert — 구조물 측점 설계 저장 보장).
    """
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT id, data FROM cross_sections
            WHERE route_id = %s AND ABS(chainage_m - %s) < 0.01
            ORDER BY id DESC
            LIMIT 1
            """,
            (route_id, chainage_m),
        )
        row = await cursor.fetchone()
        if not row:
            if project_id is None:
                return False
            await cursor.execute(
                """
                INSERT INTO cross_sections (project_id, route_id, chainage_m, data, status)
                VALUES (%s, %s, %s, %s, 'DRAFT')
                """,
                (
                    str(project_id),
                    route_id,
                    chainage_m,
                    json.dumps({"design": design}, ensure_ascii=False),
                ),
            )
            return True
        data = row[1]
        if isinstance(data, str):
            data = json.loads(data)
        if not isinstance(data, dict):
            data = {}
        data["design"] = design
        await cursor.execute(
            "UPDATE cross_sections SET data = %s WHERE id = %s",
            (json.dumps(data, ensure_ascii=False), int(row[0])),
        )
    return True


async def merge_cross_section_design_by_round(
    connection: aiomysql.Connection,
    *,
    route_id: int,
    chainage_int: int,
    patch: dict[str, Any],
) -> bool:
    """정수 chainage(m) 측점의 data.design에 patch를 병합 저장한다.

    B08 도면 확정(전체 설계 재계산 결과) 및 확정 해제(status 되돌림)에서 사용한다.
    도면 ID가 정수 m라 ROUND로 매칭한다. 갱신 여부를 반환한다.
    """
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT id, data FROM cross_sections
            WHERE route_id = %s AND ROUND(chainage_m) = %s
            ORDER BY id DESC
            LIMIT 1
            """,
            (route_id, chainage_int),
        )
        row = await cursor.fetchone()
        if not row:
            return False
        data = row[1]
        if isinstance(data, str):
            data = json.loads(data)
        if not isinstance(data, dict):
            data = {}
        design = data.get("design")
        if not isinstance(design, dict):
            design = {}
        design.update(patch)
        data["design"] = design
        await cursor.execute(
            "UPDATE cross_sections SET data = %s WHERE id = %s",
            (json.dumps(data, ensure_ascii=False), int(row[0])),
        )
    return True


async def get_cross_section_design(
    connection: aiomysql.Connection, route_id: int, chainage_int: int
) -> dict[str, Any] | None:
    """정수 chainage(m)에 해당하는 측점의 잠정 설계(data.design)를 반환한다.

    B08이 도면 ID(cross_{정수m}m)로 조회하므로 ROUND로 근사 매칭한다.
    """
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT data FROM cross_sections
            WHERE route_id = %s AND ROUND(chainage_m) = %s
            ORDER BY id DESC
            LIMIT 1
            """,
            (route_id, chainage_int),
        )
        row = await cursor.fetchone()
    if not row or not row[0]:
        return None
    data = row[0]
    if isinstance(data, str):
        data = json.loads(data)
    design = data.get("design") if isinstance(data, dict) else None
    return design if isinstance(design, dict) and design.get("ground_type") else None


async def get_cross_section_designs(
    connection: aiomysql.Connection, route_id: int
) -> list[dict[str, Any]]:
    """경로 측점별 저장된 설계 지정(data.design)을 chainage와 함께 반환한다.

    상세 조회가 파일 기반이라 DB에만 있는 잠정 설계 지정을 화면 복원용으로 얹기 위함.
    """
    async with connection.cursor() as cursor:
        await cursor.execute(
            "SELECT chainage_m, data FROM cross_sections WHERE route_id = %s", (route_id,)
        )
        rows = await cursor.fetchall()
    designs: list[dict[str, Any]] = []
    for row in rows:
        data = row[1]
        if isinstance(data, str):
            data = json.loads(data)
        design = data.get("design") if isinstance(data, dict) else None
        if isinstance(design, dict) and design.get("ground_type"):
            designs.append({"chainage_m": float(row[0]), "design": design})
    return designs


async def count_cross_sections_without_design(
    connection: aiomysql.Connection, route_id: int
) -> int:
    """지반유형(data.design.ground_type)이 아직 지정되지 않은 측점 수를 반환한다.

    확정 게이팅에 사용한다. JSON 함수 대신 애플리케이션에서 판정해
    MariaDB JSON 함수 가용성에 의존하지 않는다.
    """
    async with connection.cursor() as cursor:
        await cursor.execute("SELECT data FROM cross_sections WHERE route_id = %s", (route_id,))
        rows = await cursor.fetchall()
    missing = 0
    for row in rows:
        data = row[0]
        if isinstance(data, str):
            data = json.loads(data)
        design = data.get("design") if isinstance(data, dict) else None
        if not isinstance(design, dict) or not design.get("ground_type"):
            missing += 1
    return missing


async def get_cross_sections_missing_design_chainages(
    connection: aiomysql.Connection, route_id: int
) -> list[float]:
    """지반유형이 아직 지정되지 않은 측점의 chainage(m) 목록을 반환한다.

    확정 시 기본값(토사/좌절토)으로 일괄 채우기 위해 사용한다.
    """
    async with connection.cursor() as cursor:
        await cursor.execute(
            "SELECT chainage_m, data FROM cross_sections WHERE route_id = %s", (route_id,)
        )
        rows = await cursor.fetchall()
    chainages: list[float] = []
    for row in rows:
        data = row[1]
        if isinstance(data, str):
            data = json.loads(data)
        design = data.get("design") if isinstance(data, dict) else None
        if not isinstance(design, dict) or not design.get("ground_type"):
            chainages.append(float(row[0]))
    return chainages


async def merge_cross_section_design_patch(
    connection: aiomysql.Connection,
    *,
    route_id: int,
    chainage_m: float,
    patch: dict[str, Any],
) -> bool:
    """측점 하나(부동소수 chainage 근사 매칭)의 data.design에 patch를 병합한다.

    확정 시 프론트 세션 보관값(암 경계선 오프셋 등)을 기존 설계를 보존한 채 얹기
    위해 사용한다. design이 없던 측점이면 patch만으로 design을 만든다.
    """
    if not patch:
        return False
    async with connection.cursor() as cursor:
        await cursor.execute(
            """
            SELECT id, data FROM cross_sections
            WHERE route_id = %s AND ABS(chainage_m - %s) < 0.01
            ORDER BY id DESC
            LIMIT 1
            """,
            (route_id, chainage_m),
        )
        row = await cursor.fetchone()
        if not row:
            return False
        data = row[1]
        if isinstance(data, str):
            data = json.loads(data)
        if not isinstance(data, dict):
            data = {}
        design = data.get("design")
        if not isinstance(design, dict):
            design = {}
        design.update(patch)
        data["design"] = design
        await cursor.execute(
            "UPDATE cross_sections SET data = %s WHERE id = %s",
            (json.dumps(data, ensure_ascii=False), int(row[0])),
        )
    return True


async def _patch_latest_longitudinal_data(
    connection: aiomysql.Connection,
    route_id: int,
    apply_patch: Callable[[dict[str, Any]], None],
) -> bool:
    """경로 최신 종단면 행의 data(JSON)를 읽어 apply_patch로 고친 뒤 되쓴다.

    행이 없으면 아무것도 하지 않고 False를 반환한다.
    """
    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 False
        data = row[1]
        if isinstance(data, str):
            data = json.loads(data)
        if not isinstance(data, dict):
            data = {}
        apply_patch(data)
        await cursor.execute(
            "UPDATE longitudinal_sections SET data = %s WHERE id = %s",
            (json.dumps(data, ensure_ascii=False), int(row[0])),
        )
    return True


async def merge_longitudinal_section_options(
    connection: aiomysql.Connection,
    *,
    route_id: int,
    options_patch: dict[str, Any],
) -> bool:
    """경로 최신 종단면 data.options에 patch를 병합 저장한다.

    표준 횡단면 설정 등 확정 시점 옵션을 기존 생성 옵션 스냅샷을 보존한 채 갱신한다.
    갱신 여부를 반환한다.
    """
    if not options_patch:
        return False

    def apply_patch(data: dict[str, Any]) -> None:
        options = data.get("options")
        if not isinstance(options, dict):
            options = {}
        options.update(options_patch)
        data["options"] = options

    return await _patch_latest_longitudinal_data(connection, route_id, apply_patch)


async def merge_longitudinal_section_data(
    connection: aiomysql.Connection,
    *,
    route_id: int,
    data_patch: dict[str, Any],
) -> bool:
    """경로 최신 종단면 data의 **최상위 키**에 patch를 병합 저장한다.

    생성 옵션이 아닌 산출 결과(유토곡선 등)를 보관하는 데 쓴다. 갱신 여부를 반환한다.
    """
    if not data_patch:
        return False
    return await _patch_latest_longitudinal_data(
        connection, route_id, lambda data: data.update(data_patch)
    )


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,),
        )