Aislo/common_util/common_util_storage.py

"""프로젝트 영구저장소 경로 유틸리티."""

import os
from pathlib import PurePosixPath

from config.config_system import (
    PROJECT_STORAGE_STAGE_DIRS,
    STORAGE_BASE_DIR,
    TEMP_UPLOAD_DIR_NAME,
)

PROJECT_STORAGE_LAYOUT_V2 = (
    ("B03_FileInput", "input"),
    ("B04_PreProcess", "processed"),
    ("B04_PreProcess", "models"),
    ("B05_Profile", "route"),
    ("B06_Section", "longitudinal"),
    ("B06_Section", "cross_sections"),
    ("B08_DesignDetail", "structures"),
    ("B07_Quantity", "quantities"),
    ("B09_Estimation", "v1"),
)


def get_project_stage_path(project_root: str, stage: str) -> str:
    """프로젝트 루트 아래의 허용된 워크플로우 단계 폴더를 생성한다."""
    if stage not in PROJECT_STORAGE_STAGE_DIRS:
        raise ValueError(f"허용되지 않은 프로젝트 저장 단계입니다: {stage}")

    root = os.path.abspath(project_root)
    path = os.path.abspath(os.path.join(root, stage))
    if os.path.commonpath((root, path)) != root:
        raise ValueError("단계 저장 경로가 프로젝트 루트를 벗어났습니다.")

    os.makedirs(path, exist_ok=True)
    return path


def ensure_project_storage_layout(project_root: str) -> None:
    """신규 워크플로우 저장소 하위 폴더를 누락분만 생성한다."""
    root = os.path.abspath(project_root)
    for stage, subdir in PROJECT_STORAGE_LAYOUT_V2:
        if stage not in PROJECT_STORAGE_STAGE_DIRS:
            raise ValueError(f"허용되지 않은 프로젝트 저장 단계입니다: {stage}")
        path = os.path.abspath(os.path.join(root, stage, subdir))
        if os.path.commonpath((root, path)) != root:
            raise ValueError("단계 저장 경로가 프로젝트 루트를 벗어났습니다.")
        os.makedirs(path, exist_ok=True)


def resolve_temp_batch_path(user_id: int, batch_id: str, *, create: bool = True) -> str:
    """임시 보관함 묶음 폴더(`storage/tmp/{user_id}/{batch_id}`)를 돌려준다.

    내부 구조는 프로젝트 저장소와 똑같이 `B03_FileInput/input/...`을 쓴다 — 그래야 청크
    저장·병합 엔진(`resolve_upload_destination`, `resolve_chunk_session_dir`)을 그대로
    재사용할 수 있고, 나중에 프로젝트로 옮길 때도 같은 상대 경로로 이어 붙기만 하면 된다.
    """
    if not str(user_id).isdigit():
        raise ValueError("임시 보관함 사용자 식별자가 올바르지 않습니다.")
    if not batch_id or any(sep in batch_id for sep in ("/", "\\", "..")):
        raise ValueError("임시 보관함 묶음 식별자가 올바르지 않습니다.")

    temp_root = os.path.abspath(os.path.join(STORAGE_BASE_DIR, TEMP_UPLOAD_DIR_NAME))
    path = os.path.abspath(os.path.join(temp_root, str(user_id), batch_id))
    if os.path.commonpath((temp_root, path)) != temp_root or path == temp_root:
        raise ValueError("임시 보관함 경로가 보관함 루트를 벗어났습니다.")
    if create:
        os.makedirs(path, exist_ok=True)
    return path


def temp_upload_root() -> str:
    """임시 보관함 루트(`storage/tmp`) 절대 경로."""
    return os.path.abspath(os.path.join(STORAGE_BASE_DIR, TEMP_UPLOAD_DIR_NAME))


def resolve_stored_project_path(relative_path: str) -> str:
    """DB의 storage 기준 상대 경로를 검증해 실제 프로젝트 경로로 변환한다."""
    normalized = PurePosixPath(relative_path.replace("\\", "/"))
    if normalized.is_absolute() or ".." in normalized.parts:
        raise ValueError("프로젝트 저장 경로는 안전한 상대 경로여야 합니다.")
    if not normalized.parts or normalized.parts[0] != "storage":
        raise ValueError("프로젝트 저장 경로는 storage/로 시작해야 합니다.")

    storage_root = os.path.abspath(STORAGE_BASE_DIR)
    path = os.path.abspath(os.path.join(storage_root, *normalized.parts[1:]))
    if os.path.commonpath((storage_root, path)) != storage_root or path == storage_root:
        raise ValueError("프로젝트 저장 경로가 저장소 루트를 벗어났습니다.")
    os.makedirs(path, exist_ok=True)
    ensure_project_storage_layout(path)
    return path