"""임시 보관함 Raw SQL 저장소. 프로젝트 업로드와 같은 청크 세션 테이블(`upload_sessions`)을 쓰되, 프로젝트 대신 `temp_batch_id`로 묶는다. 묶음·파일 메타는 `temp_upload_batches`/`temp_upload_files`. """ import json from typing import Any import aiomysql from config.config_system import TEMP_UPLOAD_RETENTION_DAYS # 묶음이 "완료"로 넘어가려면 있어야 하는 파일 종류. B03 필수 슬롯과 같은 기준이다. REQUIRED_TEMP_FILE_TYPES = frozenset({"csv", "prj", "tfw"}) POINT_CLOUD_FILE_TYPES = frozenset({"las", "laz"}) def is_batch_required_complete(file_types: set[str]) -> bool: """필수 파일(csv·prj·tfw + las/laz 1종)이 모두 찼는지.""" return REQUIRED_TEMP_FILE_TYPES.issubset(file_types) and bool( file_types & POINT_CLOUD_FILE_TYPES ) async def create_temp_batch( connection: aiomysql.Connection, *, batch_id: str, user_id: int, name: str, memo: str | None, ) -> None: """보관함 묶음을 만든다(업로드 시작 상태).""" async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO temp_upload_batches (id, user_id, name, memo, status) VALUES (%s, %s, %s, %s, 'uploading') """, (batch_id, user_id, name, memo), ) async def get_temp_batch( connection: aiomysql.Connection, *, batch_id: str, user_id: int, ) -> dict[str, Any]: """본인 소유 묶음만 조회한다(남의 보관함 접근 차단).""" async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( """ SELECT id, user_id, name, memo, status, completed_at, expires_at, linked_project_id, created_at FROM temp_upload_batches WHERE id = %s AND user_id = %s """, (batch_id, user_id), ) row = await cursor.fetchone() if not row: raise LookupError("임시 보관함 묶음을 찾을 수 없습니다.") return dict(row) async def list_temp_batches( connection: aiomysql.Connection, *, user_id: int, ) -> list[dict[str, Any]]: """내 보관함 묶음 목록(최신순) — **아직 쓰지 않은 자료만**. 프로젝트에 연결된 묶음은 파일이 프로젝트로 옮겨져 내용이 비어 있다. 목록에 남겨 두면 빈 껍데기가 영구히 쌓여 화면만 지저분해진다(2026-08-08 사용자 지시로 제외). 행 자체는 이력용으로 DB에 남긴다 — 어느 자료가 어느 프로젝트로 갔는지 추적할 근거다. """ async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( """ SELECT id, name, memo, status, completed_at, expires_at, linked_project_id, created_at FROM temp_upload_batches WHERE user_id = %s AND status <> 'linked' ORDER BY created_at DESC """, (user_id,), ) return [dict(row) for row in await cursor.fetchall()] async def list_temp_batch_files( connection: aiomysql.Connection, *, batch_ids: list[str], ) -> dict[str, list[dict[str, Any]]]: """묶음별 저장 완료 파일을 한 번에 읽는다(목록 화면 N+1 방지).""" if not batch_ids: return {} placeholders = ", ".join(["%s"] * len(batch_ids)) async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( f""" SELECT batch_id, file_type, original_filename, relative_path, file_size_bytes, crs_epsg, metadata FROM temp_upload_files WHERE batch_id IN ({placeholders}) ORDER BY id """, tuple(batch_ids), ) rows = [dict(row) for row in await cursor.fetchall()] grouped: dict[str, list[dict[str, Any]]] = {} for row in rows: grouped.setdefault(str(row["batch_id"]), []).append(row) return grouped async def list_temp_batch_sessions( connection: aiomysql.Connection, *, batch_ids: list[str], ) -> dict[str, list[dict[str, Any]]]: """묶음별 진행 중 청크 세션(리스트 행 진행률 표시용).""" if not batch_ids: return {} placeholders = ", ".join(["%s"] * len(batch_ids)) async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( f""" SELECT temp_batch_id, id, original_filename, file_size_bytes, total_chunks, completed_chunks FROM upload_sessions WHERE temp_batch_id IN ({placeholders}) AND status = 'in_progress' ORDER BY updated_at DESC """, tuple(batch_ids), ) rows = [dict(row) for row in await cursor.fetchall()] grouped: dict[str, list[dict[str, Any]]] = {} for row in rows: grouped.setdefault(str(row["temp_batch_id"]), []).append(row) return grouped async def upsert_temp_batch_file( connection: aiomysql.Connection, *, batch_id: str, file_type: str, original_filename: str, relative_path: str, file_size_bytes: int, crs_epsg: int | None, metadata: dict[str, Any], ) -> None: """같은 종류를 다시 올리면 교체한다(슬롯당 1개 규칙).""" async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO temp_upload_files ( batch_id, file_type, original_filename, relative_path, file_size_bytes, crs_epsg, metadata ) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE original_filename = VALUES(original_filename), relative_path = VALUES(relative_path), file_size_bytes = VALUES(file_size_bytes), crs_epsg = VALUES(crs_epsg), metadata = VALUES(metadata) """, ( batch_id, file_type, original_filename, relative_path, file_size_bytes, crs_epsg, json.dumps(metadata, ensure_ascii=False, default=str), ), ) async def get_temp_batch_file( connection: aiomysql.Connection, *, batch_id: str, file_type: str, ) -> dict[str, Any]: """묶음 안 파일 1건(종류로 지정). 삭제 전 실제 저장 경로를 알아내는 용도.""" async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( """ SELECT batch_id, file_type, original_filename, relative_path, file_size_bytes FROM temp_upload_files WHERE batch_id = %s AND file_type = %s """, (batch_id, file_type), ) row = await cursor.fetchone() if not row: raise LookupError("보관함 파일을 찾을 수 없습니다.") return dict(row) async def delete_temp_batch_file( connection: aiomysql.Connection, *, batch_id: str, file_type: str, ) -> None: """묶음 안 파일 1건 삭제(행만 — 실제 파일은 라우터가 지운다).""" async with connection.cursor() as cursor: await cursor.execute( "DELETE FROM temp_upload_files WHERE batch_id = %s AND file_type = %s", (batch_id, file_type), ) async def mark_temp_batch_incomplete( connection: aiomysql.Connection, *, batch_id: str, ) -> None: """필수 파일이 빠지면 다시 '업로드 중'으로 되돌린다. 만료일(`expires_at`)은 손대지 않는다 — 보관 기한은 **처음 완료 시점** 기준이고, 파일을 지웠다 다시 올리는 것으로 기한이 늘어나면 안 된다. """ async with connection.cursor() as cursor: await cursor.execute( """ UPDATE temp_upload_batches SET status = 'uploading' WHERE id = %s AND status = 'completed' """, (batch_id,), ) async def get_temp_batch_file_types( connection: aiomysql.Connection, *, batch_id: str, ) -> set[str]: """묶음에 들어 있는 파일 종류 집합.""" async with connection.cursor() as cursor: await cursor.execute( "SELECT file_type FROM temp_upload_files WHERE batch_id = %s", (batch_id,), ) return {str(row[0]).lower() for row in await cursor.fetchall()} async def mark_temp_batch_completed( connection: aiomysql.Connection, *, batch_id: str, ) -> None: """필수 파일이 다 찼을 때 완료로 올리고 만료일을 찍는다. 만료 기준은 **파일이 다 올라온 시점**이다(사용자 지시). 이미 완료된 묶음에 파일을 교체해도 처음 완료 시각을 유지해 보관 기간이 무한정 늘어나지 않게 한다. """ async with connection.cursor() as cursor: await cursor.execute( """ UPDATE temp_upload_batches SET status = 'completed', completed_at = COALESCE(completed_at, NOW()), expires_at = COALESCE( expires_at, DATE_ADD(NOW(), INTERVAL %s DAY) ) WHERE id = %s AND status IN ('uploading', 'failed', 'completed') """, (TEMP_UPLOAD_RETENTION_DAYS, batch_id), ) async def mark_temp_batch_linked( connection: aiomysql.Connection, *, batch_id: str, project_id: str, ) -> None: """프로젝트로 옮긴 묶음 — 파일은 지우고 이력만 남긴다.""" async with connection.cursor() as cursor: await cursor.execute( """ UPDATE temp_upload_batches SET status = 'linked', linked_project_id = %s, expires_at = NULL WHERE id = %s """, (project_id, batch_id), ) await cursor.execute("DELETE FROM temp_upload_files WHERE batch_id = %s", (batch_id,)) async def delete_temp_batch( connection: aiomysql.Connection, *, batch_id: str, user_id: int, ) -> None: """묶음 행 삭제(파일·세션은 FK CASCADE로 함께 정리).""" async with connection.cursor() as cursor: await cursor.execute( "DELETE FROM temp_upload_batches WHERE id = %s AND user_id = %s", (batch_id, user_id), ) async def create_temp_upload_session( connection: aiomysql.Connection, *, session_id: str, batch_id: str, original_filename: str, file_size_bytes: int, chunk_size_bytes: int, total_chunks: int, ) -> None: """보관함용 청크 세션 생성 — project_id 없이 temp_batch_id로 묶는다.""" async with connection.cursor() as cursor: await cursor.execute( """ INSERT INTO upload_sessions ( id, project_id, temp_batch_id, original_filename, file_size_bytes, chunk_size_bytes, total_chunks, completed_chunks, status, created_at, updated_at ) VALUES (%s, NULL, %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, batch_id, original_filename, file_size_bytes, chunk_size_bytes, total_chunks, ), ) async def get_temp_upload_session( connection: aiomysql.Connection, *, batch_id: str, session_id: str, ) -> dict[str, Any]: """보관함 청크 세션 조회.""" async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( """ SELECT id, temp_batch_id, original_filename, file_size_bytes, chunk_size_bytes, total_chunks, completed_chunks, status FROM upload_sessions WHERE id = %s AND temp_batch_id = %s """, (session_id, batch_id), ) row = await cursor.fetchone() if not row: raise LookupError("업로드 세션을 찾을 수 없습니다.") return dict(row) async def list_expired_temp_batches( connection: aiomysql.Connection, ) -> list[dict[str, Any]]: """보관 기한이 지난 묶음(정리 작업용).""" async with connection.cursor(aiomysql.DictCursor) as cursor: await cursor.execute( """ SELECT id, user_id, name, expires_at FROM temp_upload_batches WHERE status <> 'linked' AND expires_at IS NOT NULL AND expires_at <= NOW() """ ) return [dict(row) for row in await cursor.fetchall()]