"""B04 지표면 분석 결과의 aiomysql Raw SQL 접근. processed_point_cloud(변환 포인트클라우드), surface_models(지표면 모델), terrain_layers(지형 레이어) 테이블에 메타데이터와 상대 경로를 기록한다. 공간 데이터는 MariaDB JSON 컬럼에 GeoJSON 문자열로 저장한다. """ import json from pathlib import PurePosixPath from typing import Any from uuid import UUID import aiomysql _STAGE_ROOT = "B04_wf1_Surface" def _validate_stage_path(relative_path: str) -> str: """B04_wf1_Surface 아래의 안전한 상대 경로인지 검증하고 posix 문자열로 반환한다.""" 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"B04 산출물 경로는 {_STAGE_ROOT} 아래여야 합니다.") return normalized.as_posix() async def create_processed_point_cloud( connection: aiomysql.Connection, *, input_file_id: int, project_id: UUID, process_type: str, processed_file_path: str | None, converted_format: str | None, converted_file_path: str | None, point_count: int | None, bounds: dict[str, Any] | None, statistics: dict[str, Any] | None, classification_summary: dict[str, Any] | None, processing_params: dict[str, Any] | None, status: str = "COMPLETE", ) -> int: """변환 포인트클라우드 메타데이터를 저장하고 생성된 ID를 반환한다.""" processed_rel = _validate_stage_path(processed_file_path) if processed_file_path else None converted_rel = _validate_stage_path(converted_file_path) if converted_file_path else None stats = statistics or {} async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO processed_point_cloud ( input_file_id, project_id, process_type, processed_file_path, converted_format, converted_file_path, point_count, min_z, max_z, mean_z, x_min, x_max, y_min, y_max, density_per_sqm, classification_summary, processing_params, status ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( input_file_id, str(project_id), process_type, processed_rel, converted_format, converted_rel, point_count, stats.get("min_z"), stats.get("max_z"), stats.get("mean_z"), (bounds or {}).get("x_min"), (bounds or {}).get("x_max"), (bounds or {}).get("y_min"), (bounds or {}).get("y_max"), stats.get("density_per_sqm"), json.dumps(classification_summary, ensure_ascii=False) if classification_summary is not None else None, json.dumps(processing_params, ensure_ascii=False) if processing_params is not None else None, status, ), ) new_id = cursor.lastrowid if not new_id: raise RuntimeError("processed_point_cloud 레코드 생성 결과에 ID가 없습니다.") return int(new_id) async def create_surface_model( connection: aiomysql.Connection, *, project_id: UUID, model_type: str, source_file_id: int | None, processed_cloud_id: int | None, crs_epsg: int | None, resolution_m: float | None, model_file_path: str | None, generation_params: dict[str, Any] | None, status: str = "COMPLETE", ) -> int: """지표면 모델 메타데이터를 저장하고 생성된 ID를 반환한다.""" model_rel = _validate_stage_path(model_file_path) if model_file_path else None async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO surface_models ( project_id, model_type, source_file_id, processed_cloud_id, status, crs_epsg, resolution_m, model_file_path, generation_params, completed_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP) """, ( str(project_id), model_type, source_file_id, processed_cloud_id, status, crs_epsg, resolution_m, model_rel, json.dumps(generation_params, ensure_ascii=False) if generation_params is not None else None, ), ) new_id = cursor.lastrowid if not new_id: raise RuntimeError("surface_models 레코드 생성 결과에 ID가 없습니다.") return int(new_id) async def create_terrain_layer( connection: aiomysql.Connection, *, surface_model_id: int, layer_name: str, geometry_type: str, layer_file_path: str | None, file_format: str | None, file_size_mb: float | None, statistics: dict[str, Any] | None, ) -> int: """지형 레이어 메타데이터를 저장하고 생성된 ID를 반환한다.""" layer_rel = _validate_stage_path(layer_file_path) if layer_file_path else None async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO terrain_layers ( surface_model_id, layer_name, geometry_type, layer_file_path, file_format, file_size_mb, statistics ) VALUES (%s, %s, %s, %s, %s, %s, %s) """, ( surface_model_id, layer_name, geometry_type, layer_rel, file_format, file_size_mb, json.dumps(statistics, ensure_ascii=False) if statistics is not None else None, ), ) new_id = cursor.lastrowid if not new_id: raise RuntimeError("terrain_layers 레코드 생성 결과에 ID가 없습니다.") return int(new_id) async def get_input_file( connection: aiomysql.Connection, project_id: UUID, input_file_id: int ) -> dict[str, Any]: """프로젝트의 특정 입력 파일 경로·좌표계를 조회한다.""" async with connection.cursor() as cursor: await cursor.execute( """ SELECT id, file_type, raw_file_path, crs_epsg FROM input_files WHERE id = %s AND project_id = %s """, (input_file_id, str(project_id)), ) row = await cursor.fetchone() if not row: raise LookupError("입력 파일을 찾을 수 없습니다.") return { "id": int(row[0]), "file_type": row[1], "raw_file_path": row[2], "crs_epsg": row[3], } async def list_project_point_cloud_inputs( connection: aiomysql.Connection, project_id: UUID ) -> list[dict[str, Any]]: """프로젝트의 LAS/LAZ 원본 입력 파일 목록을 최신순으로 조회한다.""" async with connection.cursor() as cursor: await cursor.execute( """ SELECT id, file_type, original_filename, raw_file_path, file_size_mb, crs_epsg, status, upload_at FROM input_files WHERE project_id = %s AND file_type IN ('las', 'laz') ORDER BY upload_at DESC, id DESC """, (str(project_id),), ) rows = await cursor.fetchall() return [ { "id": int(row[0]), "file_type": row[1], "original_filename": row[2], "raw_file_path": row[3], "file_size_mb": float(row[4]) if row[4] is not None else None, "crs_epsg": row[5], "status": row[6], "created_at": row[7].isoformat() if row[7] else None, } for row in rows ] async def list_surface_models( connection: aiomysql.Connection, project_id: UUID ) -> list[dict[str, Any]]: """프로젝트의 지표면 모델 목록을 최신순으로 조회한다.""" async with connection.cursor() as cursor: await cursor.execute( """ SELECT id, model_type, status, resolution_m, model_file_path, created_at FROM surface_models WHERE project_id = %s ORDER BY created_at DESC """, (str(project_id),), ) rows = await cursor.fetchall() return [ { "id": int(row[0]), "model_type": row[1], "status": row[2], "resolution_m": row[3], "model_file_path": row[4], "created_at": row[5].isoformat() if row[5] else None, } for row in rows ]