|
|
from __future__ import annotations
|
|
|
|
|
|
import os
|
|
|
from typing import Dict, Optional, Union
|
|
|
|
|
|
from flask import current_app
|
|
|
|
|
|
from .interface import StorageInterface
|
|
|
from .local import LocalStorage
|
|
|
from ..enums import StorageTypeEnum
|
|
|
|
|
|
|
|
|
class StorageManager:
|
|
|
"""存储管理器，负责根据类型创建存储实例"""
|
|
|
|
|
|
_instances: Dict[str, StorageInterface] = {}
|
|
|
|
|
|
@classmethod
|
|
|
def get_storage(cls, storage_type: Optional[Union[str, StorageTypeEnum]] = None) -> StorageInterface:
|
|
|
"""
|
|
|
获取存储实例（单例模式）
|
|
|
|
|
|
Args:
|
|
|
storage_type: 存储类型（支持字符串或 StorageTypeEnum），为None时使用默认类型
|
|
|
|
|
|
Returns:
|
|
|
存储实例
|
|
|
"""
|
|
|
# 标准化存储类型为字符串
|
|
|
storage_type_str = cls._normalize_storage_type(storage_type)
|
|
|
|
|
|
if storage_type_str not in cls._instances:
|
|
|
config = current_app.config.get("FILE_STORAGE", {})
|
|
|
cls._instances[storage_type_str] = cls._create_storage(storage_type_str, config)
|
|
|
return cls._instances[storage_type_str]
|
|
|
|
|
|
@staticmethod
|
|
|
def _normalize_storage_type(storage_type: Optional[Union[str, StorageTypeEnum]]) -> str:
|
|
|
"""
|
|
|
标准化存储类型为字符串
|
|
|
|
|
|
Args:
|
|
|
storage_type: 存储类型（字符串或 StorageTypeEnum）
|
|
|
|
|
|
Returns:
|
|
|
存储类型字符串
|
|
|
"""
|
|
|
# 如果未指定，使用默认类型
|
|
|
if storage_type is None:
|
|
|
config = current_app.config.get("FILE_STORAGE", {})
|
|
|
return config.get("DEFAULT_STORAGE_TYPE", StorageTypeEnum.LOCAL.value)
|
|
|
|
|
|
# 如果是 enum，转换为字符串
|
|
|
if isinstance(storage_type, StorageTypeEnum):
|
|
|
return storage_type.value
|
|
|
|
|
|
# 已经是字符串，直接返回
|
|
|
return storage_type
|
|
|
|
|
|
@staticmethod
|
|
|
def _create_storage(storage_type: str, config: dict) -> StorageInterface:
|
|
|
"""
|
|
|
创建存储实例
|
|
|
|
|
|
Args:
|
|
|
storage_type: 存储类型
|
|
|
config: 配置字典
|
|
|
|
|
|
Returns:
|
|
|
存储实例
|
|
|
"""
|
|
|
if storage_type == StorageTypeEnum.LOCAL.value:
|
|
|
local_config = config.get("LOCAL", {})
|
|
|
if not local_config.get("base_path"):
|
|
|
local_config["base_path"] = os.path.join(
|
|
|
current_app.config.get("BASE_DIR", current_app.root_path), "runtime", "uploads"
|
|
|
)
|
|
|
return LocalStorage(local_config)
|
|
|
|
|
|
elif storage_type == StorageTypeEnum.ALIYUN_OSS.value:
|
|
|
from .aliyun_oss import AliyunOSSStorage
|
|
|
oss_config = config.get("ALIYUN_OSS", {})
|
|
|
return AliyunOSSStorage(oss_config)
|
|
|
|
|
|
elif storage_type == StorageTypeEnum.TENCENT_COS.value:
|
|
|
from .tencent_cos import TencentCOSStorage
|
|
|
cos_config = config.get("TENCENT_COS", {})
|
|
|
return TencentCOSStorage(cos_config)
|
|
|
|
|
|
elif storage_type == StorageTypeEnum.QINIU_KODO.value:
|
|
|
from .qiniu_kodo import QiniuKodoStorage
|
|
|
kodo_config = config.get("QINIU_KODO", {})
|
|
|
return QiniuKodoStorage(kodo_config)
|
|
|
|
|
|
elif storage_type == StorageTypeEnum.AWS_S3.value:
|
|
|
# AWS S3 可以后续添加
|
|
|
raise NotImplementedError("AWS S3 适配器尚未实现")
|
|
|
|
|
|
elif storage_type == StorageTypeEnum.HUAWEI_OBS.value:
|
|
|
from .huawei_obs import HuaweiOBSStorage
|
|
|
obs_config = config.get("HUAWEI_OBS", {})
|
|
|
return HuaweiOBSStorage(obs_config)
|
|
|
|
|
|
elif storage_type == StorageTypeEnum.MINIO.value:
|
|
|
from .minio_storage import MinIOStorage
|
|
|
minio_config = config.get("MINIO", {})
|
|
|
return MinIOStorage(minio_config)
|
|
|
|
|
|
raise ValueError(f"未支持的存储类型: {storage_type}")
|
|
|
|