"""B03 input_files 테이블의 aiomysql Raw SQL 접근.""" import json from pathlib import PurePosixPath from typing import Any from uuid import UUID import aiomysql async def create_input_file( connection: aiomysql.Connection, *, project_id: UUID, file_type: str, original_filename: str, relative_path: str, file_size_bytes: int, upload_by: int | None, crs_epsg: int | None, metadata: dict[str, Any], ) -> int: """업로드 원본 파일 메타데이터를 저장하고 생성된 ID를 반환한다.""" normalized_path = PurePosixPath(relative_path) if normalized_path.is_absolute() or ".." in normalized_path.parts: raise ValueError("DB에는 프로젝트 루트 기준 상대 경로만 저장할 수 있습니다.") if normalized_path.parts[:2] != ("B03_FileInput", "input"): raise ValueError("입력 파일 경로는 B03_FileInput/input 아래여야 합니다.") async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO input_files ( project_id, file_type, original_filename, raw_file_path, file_size_mb, upload_by, crs_epsg, metadata, status ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, 'UPLOADED') """, ( str(project_id), file_type, original_filename, normalized_path.as_posix(), file_size_bytes / (1024 * 1024), upload_by, crs_epsg, json.dumps(metadata, ensure_ascii=False), ), ) input_file_id = cursor.lastrowid if not input_file_id: raise RuntimeError("input_files 레코드 생성 결과에 ID가 없습니다.") return int(input_file_id) async def get_project_storage_relative_path( connection: aiomysql.Connection, project_id: UUID, ) -> str: """프로젝트의 검증된 저장소 상대 경로를 조회한다.""" async with connection.cursor() as cursor: await cursor.execute( """ SELECT storage_path FROM projects WHERE id = %s AND deleted_at IS NULL """, (str(project_id),), ) row = await cursor.fetchone() if not row or not row[0]: raise LookupError("프로젝트 또는 프로젝트 저장 경로를 찾을 수 없습니다.") normalized_path = PurePosixPath(str(row[0]).replace("\\", "/")) if normalized_path.is_absolute() or ".." in normalized_path.parts: raise ValueError("프로젝트 저장 경로는 안전한 상대 경로여야 합니다.") return normalized_path.as_posix() async def create_upload_session( connection: aiomysql.Connection, *, session_id: str, project_id: UUID, original_filename: str, file_size_bytes: int, chunk_size_bytes: int, total_chunks: int, ) -> None: """청크 업로드 세션을 생성하거나 동일 세션 ID를 갱신한다.""" async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO upload_sessions ( id, project_id, original_filename, file_size_bytes, chunk_size_bytes, total_chunks, completed_chunks, status, created_at, updated_at ) VALUES (%s, %s, %s, %s, %s, %s, 0, 'in_progress', NOW(), NOW()) ON DUPLICATE KEY UPDATE original_filename = VALUES(original_filename), file_size_bytes = VALUES(file_size_bytes), chunk_size_bytes = VALUES(chunk_size_bytes), total_chunks = VALUES(total_chunks), status = 'in_progress', updated_at = NOW() """, ( session_id, str(project_id), original_filename, file_size_bytes, chunk_size_bytes, total_chunks, ), ) async def get_upload_session( connection: aiomysql.Connection, *, project_id: UUID, session_id: str, ) -> dict[str, Any]: """청크 업로드 세션을 조회한다.""" async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( """ SELECT id, project_id, original_filename, file_size_bytes, chunk_size_bytes, total_chunks, completed_chunks, status FROM upload_sessions WHERE id = %s AND project_id = %s """, (session_id, str(project_id)), ) row = await cursor.fetchone() if not row: raise LookupError("업로드 세션을 찾을 수 없습니다.") return dict(row) async def upsert_upload_chunk( connection: aiomysql.Connection, *, session_id: str, chunk_index: int, chunk_hash: str, size_bytes: int, stored_at: str, ) -> int: """청크 저장 정보를 기록하고 완료 청크 수를 반환한다.""" async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO upload_chunks ( session_id, chunk_index, chunk_hash, size_bytes, stored_at, completed_at ) VALUES (%s, %s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE chunk_hash = VALUES(chunk_hash), size_bytes = VALUES(size_bytes), stored_at = VALUES(stored_at), completed_at = NOW() """, (session_id, chunk_index, chunk_hash, size_bytes, stored_at), ) await cursor.execute( """ SELECT COUNT(*) FROM upload_chunks WHERE session_id = %s """, (session_id,), ) row = await cursor.fetchone() completed_chunks = int(row[0]) if row else 0 await cursor.execute( """ UPDATE upload_sessions SET completed_chunks = %s, updated_at = NOW() WHERE id = %s """, (completed_chunks, session_id), ) return completed_chunks async def list_completed_chunk_indexes( connection: aiomysql.Connection, *, session_id: str, ) -> list[int]: """완료된 청크 인덱스 목록을 반환한다.""" async with connection.cursor() as cursor: await cursor.execute( """ SELECT chunk_index FROM upload_chunks WHERE session_id = %s ORDER BY chunk_index ASC """, (session_id,), ) rows = await cursor.fetchall() return [int(row[0]) for row in rows] async def mark_upload_session_completed( connection: aiomysql.Connection, *, session_id: str, ) -> None: """업로드 세션을 완료 상태로 표시한다.""" async with connection.cursor() as cursor: await cursor.execute( """ UPDATE upload_sessions SET status = 'completed', updated_at = NOW() WHERE id = %s """, (session_id,), ) async def mark_upload_session_failed( connection: aiomysql.Connection, *, session_id: str, ) -> None: """업로드 세션을 실패 상태로 표시한다.""" async with connection.cursor() as cursor: await cursor.execute( """ UPDATE upload_sessions SET status = 'failed', updated_at = NOW() WHERE id = %s """, (session_id,), )