93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
"""
|
||
MinIO SDK
|
||
对象存储封装，支持配置注入
|
||
"""
|
||
import os
|
||
import time
|
||
import random
|
||
import string
|
||
from typing import Optional
|
||
|
||
try:
|
||
from minio import Minio
|
||
MINIO_AVAILABLE = True
|
||
except ImportError:
|
||
MINIO_AVAILABLE = False
|
||
Minio = None
|
||
|
||
|
||
class MinioSDK:
|
||
"""MinIO 对象存储 SDK"""
|
||
|
||
def __init__(
|
||
self,
|
||
end_point: str,
|
||
access_key: str,
|
||
secret_key: str,
|
||
bucket: str = "hackrobot",
|
||
use_ssl: bool = True,
|
||
port: Optional[int] = None,
|
||
region: str = "china",
|
||
public_url: Optional[str] = None,
|
||
upload_prefix: str = "joins",
|
||
):
|
||
if not MINIO_AVAILABLE:
|
||
raise ImportError("minio 未安装，请执行: pip install minio")
|
||
self.end_point = end_point
|
||
self.port = port or (443 if use_ssl else 80)
|
||
self.access_key = access_key
|
||
self.secret_key = secret_key
|
||
self.bucket = bucket
|
||
self.use_ssl = use_ssl
|
||
self.region = region
|
||
self.public_url = public_url or f"https://{end_point}/{bucket}"
|
||
self.upload_prefix = upload_prefix.rstrip("/")
|
||
self._client: Optional[Minio] = None
|
||
|
||
def get_client(self) -> "Minio":
|
||
"""获取 MinIO 客户端"""
|
||
if self._client is None:
|
||
endpoint = f"{self.end_point}:{self.port}" if self.port not in (80, 443) else self.end_point
|
||
self._client = Minio(
|
||
endpoint,
|
||
access_key=self.access_key,
|
||
secret_key=self.secret_key,
|
||
secure=self.use_ssl,
|
||
region=self.region,
|
||
)
|
||
return self._client
|
||
|
||
def generate_object_key(self, ext: str) -> str:
|
||
"""生成上传用的对象键"""
|
||
ext = ext.lstrip(".")
|
||
suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
|
||
return f"{self.upload_prefix}/{int(time.time() * 1000)}_{suffix}.{ext}"
|
||
|
||
def upload_buffer(
|
||
self,
|
||
data: bytes,
|
||
object_key: str,
|
||
content_type: str = "application/octet-stream",
|
||
) -> dict:
|
||
"""
|
||
上传文件到 MinIO
|
||
:return: {"url": str, "object_key": str}
|
||
"""
|
||
client = self.get_client()
|
||
from io import BytesIO
|
||
stream = BytesIO(data)
|
||
client.put_object(
|
||
self.bucket,
|
||
object_key,
|
||
stream,
|
||
len(data),
|
||
content_type=content_type,
|
||
)
|
||
url = f"{self.public_url.rstrip('/')}/{object_key}"
|
||
return {"url": url, "object_key": object_key}
|
||
|
||
@property
|
||
def enabled(self) -> bool:
|
||
"""是否启用（需配置 access_key/secret_key）"""
|
||
return bool(self.end_point and self.access_key and self.secret_key)
|