252 lines
8.9 KiB
Python
252 lines
8.9 KiB
Python
"""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 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 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 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,),
|
|
)
|