Aislo/B03_FileInput/B03_FileInput_Repository_Temp.py

"""임시 보관함 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]]:
    """내 보관함 묶음 목록(최신순). 프로젝트로 옮긴 묶음도 이력으로 함께 준다."""
    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
            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()]