"""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_PreProcess" def _validate_stage_path(relative_path: str) -> str: """B04_PreProcess 아래의 안전한 상대 경로인지 검증하고 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, generation_params, created_at FROM surface_models WHERE project_id = %s ORDER BY created_at DESC """, (str(project_id),), ) rows = await cursor.fetchall() models: list[dict[str, Any]] = [] for row in rows: generation_params = row[5] if isinstance(generation_params, str): generation_params = json.loads(generation_params) models.append( { "id": int(row[0]), "model_type": row[1], "status": row[2], "resolution_m": row[3], "model_file_path": row[4], "generation_params": generation_params, "created_at": row[6].isoformat() if row[6] else None, } ) return models async def clear_confirmed_surface_models(connection: aiomysql.Connection, project_id: UUID) -> None: """재분석 시 기존 확정을 해제하여 새 분석 결과의 재확정을 요구한다.""" async with connection.cursor() as cursor: await cursor.execute( """ UPDATE surface_models SET status = 'COMPLETE' WHERE project_id = %s AND status = 'CONFIRMED' """, (str(project_id),), ) async def confirm_surface_model( connection: aiomysql.Connection, project_id: UUID, model_id: int ) -> None: """프로젝트 내 단일 모델만 CONFIRMED가 되도록 갱신한다.""" async with connection.cursor() as cursor: await cursor.execute( """ UPDATE surface_models SET status = 'COMPLETE' WHERE project_id = %s AND status = 'CONFIRMED' """, (str(project_id),), ) await cursor.execute( """ UPDATE surface_models SET status = 'CONFIRMED' WHERE id = %s AND project_id = %s AND status = 'COMPLETE' """, (model_id, str(project_id)), ) if cursor.rowcount != 1: raise LookupError("확정할 지표면 모델을 찾을 수 없습니다.") async def update_project_status( connection: aiomysql.Connection, project_id: UUID, status: str ) -> None: async with connection.cursor() as cursor: await cursor.execute( """ UPDATE projects SET status = %s, updated_at = NOW() WHERE id = %s AND deleted_at IS NULL """, (status, str(project_id)), ) async def delete_project_surface_models(connection: aiomysql.Connection, project_id: UUID) -> int: """프로젝트의 기존 지표면 모델 행을 모두 제거한다. 모든 분석 세대가 동일 파일 경로를 재사용하므로 재분석 시 구세대 행을 남기면 존재하지 않는 파일을 가리키게 된다. terrain_layers는 FK CASCADE, routes.surface_model_id는 FK SET NULL로 함께 정리된다. """ async with connection.cursor() as cursor: await cursor.execute( "DELETE FROM surface_models WHERE project_id = %s", (str(project_id),), ) return int(cursor.rowcount) async def save_surface_analysis_to_db( connection: aiomysql.Connection, *, project_id: UUID, input_file_id: int, analysis_result: dict[str, Any], source_filters: list[str], ) -> list[int]: """WF1 분석 결과를 DB에 저장한다 (기존 모델 행은 교체). 이 함수는 트랜잭션을 시작하거나 종료하지 않는다. 호출자는 같은 커넥션에서 begin/commit/rollback을 한 번만 수행해야 한다. """ input_file = await get_input_file(connection, project_id, input_file_id) await delete_project_surface_models(connection, project_id) processed = analysis_result["processed"] processed_cloud_id = await create_processed_point_cloud( connection, input_file_id=input_file_id, project_id=project_id, process_type="structured", processed_file_path=processed["processed_file_path"], converted_format=None, converted_file_path=processed["converted_file_path"], point_count=processed["point_count"], bounds=processed["bounds"], statistics=processed["statistics"], classification_summary=None, processing_params={"filters": source_filters}, ) surface_model_ids: list[int] = [] for model in analysis_result["models"]: model_id = await create_surface_model( connection, project_id=project_id, model_type=model["model_type"], source_file_id=input_file_id, processed_cloud_id=processed_cloud_id, crs_epsg=input_file["crs_epsg"], resolution_m=model["resolution_m"], model_file_path=model["model_file_path"], generation_params=model["generation_params"], ) surface_model_ids.append(model_id) for layer in model["layers"]: await create_terrain_layer( connection, surface_model_id=model_id, layer_name=layer["layer_name"], geometry_type=layer["geometry_type"], layer_file_path=layer["file_path"], file_format=layer["file_format"], file_size_mb=None, statistics=None, ) return surface_model_ids