|
|
from __future__ import annotations
|
|
|
|
|
|
import os
|
|
|
from typing import Dict, Optional
|
|
|
|
|
|
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[str] = None) -> StorageInterface:
|
|
|
"""
|
|
|
获取存储实例（单例模式）
|
|
|
|
|
|
Args:
|
|
|
storage_type: 存储类型，为None时使用默认类型
|
|
|
|
|
|
Returns:
|
|
|
存储实例
|
|
|
"""
|
|
|
config = current_app.config.get("FILE_STORAGE", {})
|
|
|
default_type = config.get("DEFAULT_STORAGE_TYPE", StorageTypeEnum.LOCAL.value)
|
|
|
storage_type = storage_type or default_type
|
|
|
|
|
|
if storage_type not in cls._instances:
|
|
|
cls._instances[storage_type] = cls._create_storage(storage_type, config)
|
|
|
return cls._instances[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:
|
|
|
# MinIO 可以后续添加
|
|
|
raise NotImplementedError("MinIO 适配器尚未实现")
|
|
|
|
|
|
raise ValueError(f"未支持的存储类型: {storage_type}")
|
|
|
|