260710_0

B03_FileInput/B03_FileInput_Engine.py

@@ -1,6 +1,7 @@
 """B03 파일 입력 저장 엔진."""
 
 import os
+import shutil
 import tempfile
 from pathlib import Path
 
@@ -8,7 +9,7 @@ from fastapi import UploadFile
 
 from B03_FileInput.B03_FileInput_Schema import FileUploadDescriptor
 from common_util.common_util_storage import get_project_stage_path
-from config.config_system import UPLOAD_CHUNK_SIZE_BYTES, UPLOAD_MAX_MB
+from config.config_system import CHUNK_TEMP_DIR, UPLOAD_CHUNK_SIZE_BYTES, UPLOAD_MAX_MB
 
 
 def resolve_upload_destination(
@@ -58,3 +59,118 @@ async def save_upload_stream(upload: UploadFile, destination: Path) -> int:
     finally:
         if temporary_path is not None:
             temporary_path.unlink(missing_ok=True)
+
+
+def resolve_chunk_session_dir(project_root: str | Path, session_id: str) -> Path:
+    """청크 업로드 세션의 임시 저장 폴더를 B03 단계 내부에 만든다."""
+    stage_root = Path(get_project_stage_path(str(project_root), "B03_FileInput")).resolve()
+    chunk_root = (Path(project_root) / CHUNK_TEMP_DIR).resolve()
+    session_dir = (chunk_root / session_id).resolve()
+
+    if os.path.commonpath((stage_root, session_dir)) != str(stage_root):
+        raise ValueError("청크 임시 저장 경로가 B03 단계 폴더를 벗어났습니다.")
+
+    session_dir.mkdir(parents=True, exist_ok=True)
+    return session_dir
+
+
+async def save_upload_chunk(
+    upload: UploadFile,
+    session_dir: Path,
+    chunk_index: int,
+    expected_max_bytes: int = UPLOAD_CHUNK_SIZE_BYTES,
+) -> tuple[Path, int, str]:
+    """단일 청크를 임시 파일로 저장하고 SHA256 해시를 반환한다."""
+    import hashlib
+
+    if chunk_index < 0:
+        raise ValueError("청크 인덱스는 0 이상이어야 합니다.")
+
+    destination = (session_dir / f"{chunk_index:08d}.chunk").resolve()
+    if os.path.commonpath((session_dir, destination)) != str(session_dir):
+        raise ValueError("청크 파일 경로가 세션 폴더를 벗어났습니다.")
+
+    digest = hashlib.sha256()
+    written_bytes = 0
+    temporary_path: Path | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="wb",
+            dir=session_dir,
+            prefix=f".{chunk_index:08d}.",
+            suffix=".chunk.upload",
+            delete=False,
+        ) as temporary:
+            temporary_path = Path(temporary.name)
+            while chunk := await upload.read(min(1024 * 1024, expected_max_bytes)):
+                written_bytes += len(chunk)
+                if written_bytes > expected_max_bytes:
+                    raise ValueError("청크 크기가 허용 범위를 초과했습니다.")
+                digest.update(chunk)
+                temporary.write(chunk)
+            temporary.flush()
+            os.fsync(temporary.fileno())
+
+        if written_bytes == 0:
+            raise ValueError("빈 청크는 업로드할 수 없습니다.")
+        os.replace(temporary_path, destination)
+        temporary_path = None
+        return destination, written_bytes, digest.hexdigest()
+    finally:
+        if temporary_path is not None:
+            temporary_path.unlink(missing_ok=True)
+
+
+def merge_upload_chunks(
+    project_root: str | Path,
+    descriptor: FileUploadDescriptor,
+    session_id: str,
+    total_chunks: int,
+) -> Path:
+    """세션 청크를 순서대로 병합해 최종 입력 파일로 저장한다."""
+    if total_chunks <= 0:
+        raise ValueError("병합할 청크 개수가 올바르지 않습니다.")
+
+    project_root_path = Path(project_root)
+    session_dir = resolve_chunk_session_dir(project_root_path, session_id)
+    destination = resolve_upload_destination(project_root_path, descriptor)
+    temporary_path: Path | None = None
+    merged_bytes = 0
+    maximum_bytes = UPLOAD_MAX_MB * 1024 * 1024
+
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="wb",
+            dir=destination.parent,
+            prefix=f".{destination.name}.",
+            suffix=".merge",
+            delete=False,
+        ) as temporary:
+            temporary_path = Path(temporary.name)
+            for index in range(total_chunks):
+                chunk_path = session_dir / f"{index:08d}.chunk"
+                if not chunk_path.exists():
+                    raise ValueError(f"누락된 청크가 있습니다. index={index}")
+                with chunk_path.open("rb") as chunk_file:
+                    while data := chunk_file.read(1024 * 1024):
+                        merged_bytes += len(data)
+                        if merged_bytes > maximum_bytes:
+                            raise ValueError(f"파일 크기는 {UPLOAD_MAX_MB}MB를 초과할 수 없습니다.")
+                        temporary.write(data)
+            temporary.flush()
+            os.fsync(temporary.fileno())
+
+        if merged_bytes != descriptor.size_bytes:
+            raise ValueError("병합 파일 크기가 업로드 세션 정보와 일치하지 않습니다.")
+        os.replace(temporary_path, destination)
+        temporary_path = None
+        return destination
+    finally:
+        if temporary_path is not None:
+            temporary_path.unlink(missing_ok=True)
+
+
+def remove_chunk_session(project_root: str | Path, session_id: str) -> None:
+    """B03 임시 청크 세션 폴더를 제거한다."""
+    session_dir = resolve_chunk_session_dir(project_root, session_id)
+    shutil.rmtree(session_dir, ignore_errors=True)