Aislo/B03_FileInput/B03_FileInput_Engine.py

"""B03 파일 입력 저장 엔진."""

import os
import shutil
import tempfile
from pathlib import Path

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 CHUNK_TEMP_DIR, UPLOAD_CHUNK_SIZE_BYTES, UPLOAD_MAX_MB

# shapefile은 형제 파일이 **같은 폴더·같은 basename**이라야 열린다. 확장자별 폴더에
# 흩어 두면 노선 자체를 못 읽는다. 그래서 세트는 `input/shp/`에 통째로 모은다.
# 부수 효과가 하나 더 있다 — 노선 PRJ가 이 폴더 안에 남으므로 지형 PRJ(`input/prj/`)와
# 파일명 정렬 운에 기대지 않고 갈린다 (2026-08-31).
_SHAPEFILE_FOLDER = "shp"
_SHAPEFILE_OWNED_EXT = frozenset({"shp", "shx", "dbf", "cpg"})


def reconcile_shapefile_members(stage_root: Path) -> list[Path]:
    """먼저 도착해 다른 폴더에 앉은 shapefile 형제 파일을 `input/shp/`로 모은다.

    파일이 오는 순서를 강제할 수 없다 — PRJ가 SHP보다 먼저 오면 확장자 규칙만으로는
    `input/prj/`에 앉는다. SHP가 들어온 시점에 같은 basename의 형제를 끌어온다.
    """
    input_root = stage_root / "input"
    shapefile_dir = input_root / _SHAPEFILE_FOLDER
    if not shapefile_dir.exists():
        return []
    moved: list[Path] = []
    for shape_path in shapefile_dir.glob("*.shp"):
        for extension in ("shx", "dbf", "cpg", "prj"):
            stray = input_root / extension / f"{shape_path.stem}.{extension}"
            target = shapefile_dir / stray.name
            if stray.exists() and not target.exists():
                shutil.move(str(stray), str(target))
                moved.append(target)
    return moved


def resolve_upload_destination(
    project_root: str | Path,
    descriptor: FileUploadDescriptor,
) -> Path:
    """검증된 파일의 B03 입력 저장 경로를 생성해 반환한다."""
    stage_root = Path(get_project_stage_path(str(project_root), "B03_FileInput")).resolve()
    reconcile_shapefile_members(stage_root)

    source_name = Path(descriptor.original_filename)
    file_type = source_name.suffix.lower().lstrip(".")
    folder = file_type
    if file_type in _SHAPEFILE_OWNED_EXT:
        folder = _SHAPEFILE_FOLDER
    elif file_type == "prj":
        # 노선 shapefile의 짝 PRJ만 세트 폴더로. 지형 PRJ는 `input/prj/`에 남는다.
        if (stage_root / "input" / _SHAPEFILE_FOLDER / f"{source_name.stem}.shp").exists():
            folder = _SHAPEFILE_FOLDER
    destination = (stage_root / "input" / folder / descriptor.original_filename).resolve()

    if os.path.commonpath((stage_root, destination)) != str(stage_root):
        raise ValueError("업로드 저장 경로가 B03 단계 폴더를 벗어났습니다.")

    destination.parent.mkdir(parents=True, exist_ok=True)
    return destination


async def save_upload_stream(upload: UploadFile, destination: Path) -> int:
    """업로드 스트림을 크기 제한 내에서 임시 파일에 기록한 뒤 교체한다."""
    maximum_bytes = UPLOAD_MAX_MB * 1024 * 1024
    written_bytes = 0
    temporary_path: Path | None = None

    try:
        with tempfile.NamedTemporaryFile(
            mode="wb",
            dir=destination.parent,
            prefix=f".{destination.name}.",
            suffix=".upload",
            delete=False,
        ) as temporary:
            temporary_path = Path(temporary.name)
            while chunk := await upload.read(UPLOAD_CHUNK_SIZE_BYTES):
                written_bytes += len(chunk)
                if written_bytes > maximum_bytes:
                    raise ValueError(f"파일 크기는 {UPLOAD_MAX_MB}MB를 초과할 수 없습니다.")
                temporary.write(chunk)
            temporary.flush()
            os.fsync(temporary.fileno())

        if written_bytes == 0:
            raise ValueError("빈 파일은 업로드할 수 없습니다.")
        os.replace(temporary_path, destination)
        temporary_path = None
        return written_bytes
    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)