refactor(B03): 임시 보관함 라우터 700줄 초과 분리 — 청크·보조

749줄 한 파일을 셋으로 나눔 (동작 불변, 순수 이동). - `B03_FileInput_Router_Temp.py` 478줄 — 묶음 생성·목록·삭제·파일 업로드·프로젝트 연결 - `B03_FileInput_Router_Temp_Chunks.py` 294줄 — 세션·조각·진행·마무리 (prefix 없는 APIRouter 를 본체가 `include_router` 로 붙여 `/api/temp-uploads/...` 경로 불변) - `B03_FileInput_Router_Temp_Support.py` 46줄 — 조각 수·묶음 폴더·상태 갱신 (순환 방지) 검증: 라우트 10개(temp 9 + attach 1) 경로·메서드 동일, 공용 브라우저에서 실제 청크 흐름 create 200 → upload-sessions 200 → chunks 200 → finalize 200 → delete 200, ruff check 통과, tmp/tests 378 passed Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-04 10:34:01 +09:00
parent c4aac6f299
commit 3ec6527cc2
3 changed files with 350 additions and 281 deletions
@@ -0,0 +1,294 @@
+"""임시 보관함 청크 업로드 엔드포인트 — 세션 생성·조각 전송·진행 조회·마무리.
+
+라우터가 700줄을 넘어 떼어냈다(2026-09-04). 본체가 `include_router` 로 붙이므로
+여기 라우터에는 prefix 를 두지 않는다 — 두면 `/api/temp-uploads` 가 두 번 붙는다.
+"""
+
+import asyncio
+import logging
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from fastapi import APIRouter, Depends, File, Form, UploadFile
+from fastapi.responses import JSONResponse
+
+from B03_FileInput.B03_FileInput_Engine import (
+    merge_upload_chunks,
+    remove_chunk_session,
+    resolve_chunk_session_dir,
+    save_upload_chunk,
+)
+from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata
+from B03_FileInput.B03_FileInput_Repository import (
+    list_completed_chunk_indexes,
+    mark_upload_session_completed,
+    mark_upload_session_failed,
+    upsert_upload_chunk,
+)
+from B03_FileInput.B03_FileInput_Repository_Temp import (
+    create_temp_upload_session,
+    get_temp_batch,
+    get_temp_upload_session,
+    upsert_temp_batch_file,
+)
+from B03_FileInput.B03_FileInput_Router_Errors import lookup_error_response
+from B03_FileInput.B03_FileInput_Router_Temp_Support import (
+    _batch_root,
+    _refresh_batch_status,
+    _total_chunks,
+)
+from B03_FileInput.B03_FileInput_Schema import (
+    ChunkSessionCreateRequest,
+    ChunkSessionCreateResponse,
+    ChunkUploadResponse,
+    FileUploadDescriptor,
+    UploadFinalizeRequest,
+    UploadStatusResponse,
+)
+from B03_FileInput.B03_FileInput_Schema_Temp import (
+    TempFileUploadResponse,
+    TempFileUploadResult,
+)
+from common_util.common_util_auth import verify_session
+from config.config_db import get_db_pool
+from config.config_system import (
+    UPLOAD_CHUNK_SIZE_BYTES,
+)
+
+logger = logging.getLogger(__name__)
+router = APIRouter(tags=["B03 Temp Upload"])
+
+
+@router.post("/{batch_id}/upload-sessions", response_model=ChunkSessionCreateResponse)
+async def create_batch_upload_session(
+    batch_id: str,
+    payload: ChunkSessionCreateRequest,
+    session: dict[str, Any] = Depends(verify_session),
+) -> ChunkSessionCreateResponse | JSONResponse:
+    """대용량 파일(LAS/LAZ) 청크 세션을 만든다 — 프로젝트 업로드와 같은 규칙."""
+    user_id = int(session["user_id"])
+    chunk_size_bytes = min(payload.chunk_size_bytes, UPLOAD_CHUNK_SIZE_BYTES)
+    total_chunks = _total_chunks(payload.size_bytes, chunk_size_bytes)
+    session_id = str(uuid4())
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            await _batch_root(connection, batch_id=batch_id, user_id=user_id)
+            await create_temp_upload_session(
+                connection,
+                session_id=session_id,
+                batch_id=batch_id,
+                original_filename=payload.original_filename,
+                file_size_bytes=payload.size_bytes,
+                chunk_size_bytes=chunk_size_bytes,
+                total_chunks=total_chunks,
+            )
+            await connection.commit()
+        return ChunkSessionCreateResponse(
+            project_id=batch_id,
+            upload_session_id=session_id,
+            original_filename=payload.original_filename,
+            file_size_bytes=payload.size_bytes,
+            chunk_size_bytes=chunk_size_bytes,
+            total_chunks=total_chunks,
+        )
+    except LookupError as exc:
+        return lookup_error_response(exc, logger, context="B03 보관함", batch_id=batch_id)
+    except (OSError, ValueError) as exc:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
+    except Exception:
+        logger.exception("임시 보관함 청크 세션 생성 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "업로드 세션 생성 중 오류가 발생했습니다."},
+        )
+
+
+@router.post("/{batch_id}/chunks", response_model=ChunkUploadResponse)
+async def upload_batch_chunk(
+    batch_id: str,
+    session_id: str = Form(...),
+    chunk_index: int = Form(...),
+    chunk_data: UploadFile = File(...),
+    session: dict[str, Any] = Depends(verify_session),
+) -> ChunkUploadResponse | JSONResponse:
+    """청크 한 조각을 보관함 묶음 폴더에 저장한다."""
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            batch_root = await _batch_root(connection, batch_id=batch_id, user_id=user_id)
+            upload_session = await get_temp_upload_session(
+                connection, batch_id=batch_id, session_id=session_id
+            )
+            if chunk_index < 0 or chunk_index >= int(upload_session["total_chunks"]):
+                return JSONResponse(
+                    status_code=400,
+                    content={"status": "error", "message": "청크 인덱스가 범위를 벗어났습니다."},
+                )
+            session_dir = resolve_chunk_session_dir(batch_root, session_id)
+            chunk_path, size_bytes, chunk_hash = await save_upload_chunk(
+                chunk_data,
+                session_dir,
+                chunk_index,
+                expected_max_bytes=int(upload_session["chunk_size_bytes"]),
+            )
+            completed_chunks = await upsert_upload_chunk(
+                connection,
+                session_id=session_id,
+                chunk_index=chunk_index,
+                chunk_hash=chunk_hash,
+                size_bytes=size_bytes,
+                stored_at=chunk_path.relative_to(batch_root).as_posix(),
+            )
+        return ChunkUploadResponse(
+            upload_session_id=session_id,
+            chunk_index=chunk_index,
+            completed_chunks=completed_chunks,
+            total_chunks=int(upload_session["total_chunks"]),
+            chunk_hash=chunk_hash,
+        )
+    except LookupError as exc:
+        return lookup_error_response(exc, logger, context="B03 보관함", batch_id=batch_id)
+    except (OSError, ValueError) as exc:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
+    except Exception:
+        logger.exception("임시 보관함 청크 업로드 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "청크 업로드 중 오류가 발생했습니다."},
+        )
+    finally:
+        await chunk_data.close()
+
+
+@router.get("/{batch_id}/upload-status/{session_id}", response_model=UploadStatusResponse)
+async def get_batch_upload_status(
+    batch_id: str,
+    session_id: str,
+    session: dict[str, Any] = Depends(verify_session),
+) -> UploadStatusResponse | JSONResponse:
+    """이어올리기용 — 이미 올라간 청크 번호를 돌려준다."""
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            await get_temp_batch(connection, batch_id=batch_id, user_id=user_id)
+            upload_session = await get_temp_upload_session(
+                connection, batch_id=batch_id, session_id=session_id
+            )
+            completed_indexes = await list_completed_chunk_indexes(
+                connection, session_id=session_id
+            )
+        return UploadStatusResponse(
+            upload_session_id=session_id,
+            upload_status=str(upload_session["status"]),
+            original_filename=str(upload_session["original_filename"]),
+            file_size_bytes=int(upload_session["file_size_bytes"]),
+            chunk_size_bytes=int(upload_session["chunk_size_bytes"]),
+            total_chunks=int(upload_session["total_chunks"]),
+            completed_chunks=len(completed_indexes),
+            completed_chunk_indexes=completed_indexes,
+        )
+    except LookupError as exc:
+        return lookup_error_response(exc, logger, context="B03 보관함", batch_id=batch_id)
+    except Exception:
+        logger.exception("임시 보관함 업로드 상태 조회 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "업로드 상태 조회 중 오류가 발생했습니다."},
+        )
+
+
+@router.post("/{batch_id}/finalize", response_model=TempFileUploadResponse)
+async def finalize_batch_upload(
+    batch_id: str,
+    payload: UploadFinalizeRequest,
+    session: dict[str, Any] = Depends(verify_session),
+) -> TempFileUploadResponse | JSONResponse:
+    """청크를 병합해 보관함에 저장하고, 필수 파일이 다 차면 완료로 올린다."""
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    final_path: Path | None = None
+    try:
+        async with pool.acquire() as connection:
+            batch_root = await _batch_root(connection, batch_id=batch_id, user_id=user_id)
+            upload_session = await get_temp_upload_session(
+                connection, batch_id=batch_id, session_id=payload.session_id
+            )
+            if int(upload_session["total_chunks"]) != payload.total_chunks:
+                return JSONResponse(
+                    status_code=400,
+                    content={"status": "error", "message": "세션 청크 개수가 일치하지 않습니다."},
+                )
+            completed_indexes = await list_completed_chunk_indexes(
+                connection, session_id=payload.session_id
+            )
+            if completed_indexes != list(range(payload.total_chunks)):
+                return JSONResponse(
+                    status_code=400,
+                    content={"status": "error", "message": "아직 업로드되지 않은 청크가 있습니다."},
+                )
+
+            descriptor = FileUploadDescriptor(
+                original_filename=str(upload_session["original_filename"]),
+                size_bytes=int(upload_session["file_size_bytes"]),
+            )
+            final_path = merge_upload_chunks(
+                batch_root, descriptor, payload.session_id, payload.total_chunks
+            )
+            metadata = await asyncio.to_thread(analyze_input_metadata, final_path)
+            relative_path = final_path.relative_to(batch_root).as_posix()
+            file_type = final_path.suffix.lower().lstrip(".")
+            crs_epsg = metadata.get("epsg")
+
+            await connection.begin()
+            try:
+                await upsert_temp_batch_file(
+                    connection,
+                    batch_id=batch_id,
+                    file_type=file_type,
+                    original_filename=descriptor.original_filename,
+                    relative_path=relative_path,
+                    file_size_bytes=int(upload_session["file_size_bytes"]),
+                    crs_epsg=int(crs_epsg) if crs_epsg is not None else None,
+                    metadata=metadata,
+                )
+                await mark_upload_session_completed(connection, session_id=payload.session_id)
+                required_complete = await _refresh_batch_status(connection, batch_id=batch_id)
+                await connection.commit()
+            except Exception:
+                await connection.rollback()
+                await mark_upload_session_failed(connection, session_id=payload.session_id)
+                raise
+
+        remove_chunk_session(batch_root, payload.session_id)
+        return TempFileUploadResponse(
+            batch_id=batch_id,
+            files=[
+                TempFileUploadResult(
+                    batch_id=batch_id,
+                    file_type=file_type,
+                    original_filename=descriptor.original_filename,
+                    relative_path=relative_path,
+                    size_bytes=int(upload_session["file_size_bytes"]),
+                    metadata=metadata,
+                )
+            ],
+            required_complete=required_complete,
+        )
+    except LookupError as exc:
+        return lookup_error_response(exc, logger, context="B03 보관함", batch_id=batch_id)
+    except (OSError, ValueError) as exc:
+        if final_path is not None:
+            final_path.unlink(missing_ok=True)
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
+    except Exception:
+        if final_path is not None:
+            final_path.unlink(missing_ok=True)
+        logger.exception("임시 보관함 병합 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "업로드 최종 처리 중 오류가 발생했습니다."},
+        )