补充运行说明并实现基础导出能力

This commit is contained in:
zwt13703
2026-07-02 15:13:57 +08:00
parent a2bad58591
commit b313083766
7 changed files with 414 additions and 57 deletions
+34 -1
View File
@@ -1,3 +1,6 @@
import io
from datetime import timedelta
from minio import Minio
from config import settings
@@ -34,4 +37,34 @@ def get_file_url(bucket: str, object_name: str) -> str:
def get_presigned_url(bucket: str, object_name: str, expires: int = 3600) -> str:
"""获取预签名下载 URL(带过期时间)"""
return minio_client.presigned_get_object(bucket, object_name, expires=expires)
return minio_client.presigned_get_object(bucket, object_name, expires=timedelta(seconds=expires))
def split_bucket_path(file_path: str) -> tuple[str, str]:
if "/" not in file_path:
raise ValueError("非法的 MinIO 文件路径")
return file_path.split("/", 1)
def download_object_bytes(bucket: str, object_name: str) -> bytes:
response = minio_client.get_object(bucket, object_name)
try:
return response.read()
finally:
response.close()
response.release_conn()
def upload_bytes(
bucket: str,
object_name: str,
content: bytes,
content_type: str = "application/octet-stream",
):
minio_client.put_object(
bucket,
object_name,
io.BytesIO(content),
len(content),
content_type=content_type,
)