146 lines
5.0 KiB
Python
146 lines
5.0 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 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, 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
|
|
return {
|
|
"id": int(row[0]),
|
|
"longitudinal_file_path": row[1],
|
|
"status": row[2],
|
|
"computed_at": row[3].isoformat() if row[3] else None,
|
|
}
|
|
|
|
|
|
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,),
|
|
)
|