BBIT-Kai
54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
from datetime import timedelta
|
|
|
|
from minio import Minio
|
|
|
|
# MinIO 客户端初始化
|
|
minio_client = Minio(
|
|
"ai.ronsunny.cn:9000",
|
|
access_key="minioadmin",
|
|
secret_key="minioadmin",
|
|
region="Chengdu",
|
|
secure=True,
|
|
)
|
|
|
|
|
|
def push_file(bucket_name, object_name, file_bytes, contents, content_type):
|
|
minio_client.put_object(
|
|
bucket_name,
|
|
object_name,
|
|
file_bytes,
|
|
length=len(contents),
|
|
content_type=content_type,
|
|
)
|
|
|
|
|
|
def get_upload_token(user_id, bucket_name, object_name, xpires=timedelta(minutes=15)):
|
|
return minio_client.presigned_put_object(
|
|
bucket_name=bucket_name, object_name=object_name, expires=xpires
|
|
)
|
|
|
|
|
|
def get_temp_url(bucket_name, object_name):
|
|
# 如果 object_name 为 None 或空字符串，则返回默认图片
|
|
if not object_name or not bucket_name:
|
|
bucket_name = "system"
|
|
object_name = "favicon.ico"
|
|
|
|
# 使用 presigned_get_object 获取临时 URL
|
|
return minio_client.presigned_get_object(
|
|
bucket_name, object_name, expires=timedelta(seconds=3600)
|
|
)
|
|
|
|
|
|
def get_temp_url_dict(bucket_name, object_dict, object_name):
|
|
# 如果 object_name 为 None 或空字符串，则返回默认图片
|
|
if not object_name:
|
|
bucket_name = "system"
|
|
object_dict = "default"
|
|
object_name = "favicon.ico"
|
|
|
|
# 使用 presigned_get_object 获取临时 URL
|
|
return minio_client.presigned_get_object(
|
|
bucket_name, object_dict + "/" + object_name, expires=timedelta(seconds=3600)
|
|
)
|