75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
import io
|
||
from datetime import timedelta
|
||
|
||
from minio import Minio
|
||
from config import settings
|
||
|
||
# MinIO 客户端
|
||
minio_client = Minio(
|
||
settings.MINIO_ENDPOINT,
|
||
access_key=settings.MINIO_ACCESS_KEY,
|
||
secret_key=settings.MINIO_SECRET_KEY,
|
||
secure=settings.MINIO_USE_SSL,
|
||
)
|
||
|
||
|
||
async def init_buckets():
|
||
"""初始化 MinIO 存储桶（应用启动时调用）"""
|
||
buckets = [
|
||
settings.MINIO_BUCKET_TEMPLATES, # 原始模板文件
|
||
settings.MINIO_BUCKET_UPLOADS, # 用户上传的参考文件
|
||
settings.MINIO_BUCKET_OUTPUTS, # 生成的文档
|
||
]
|
||
for bucket in buckets:
|
||
if not minio_client.bucket_exists(bucket):
|
||
minio_client.make_bucket(bucket)
|
||
print(f"[MinIO] 创建存储桶: {bucket}")
|
||
|
||
|
||
def get_file_url(bucket: str, object_name: str) -> str:
|
||
"""获取文件的公开访问 URL"""
|
||
if settings.MINIO_USE_SSL:
|
||
protocol = "https"
|
||
else:
|
||
protocol = "http"
|
||
return f"{protocol}://{settings.MINIO_ENDPOINT}/{bucket}/{object_name}"
|
||
|
||
|
||
def get_presigned_url(bucket: str, object_name: str, expires: int = 3600) -> str:
|
||
"""获取预签名下载 URL（带过期时间）"""
|
||
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,
|
||
)
|
||
|
||
|
||
def delete_object(bucket: str, object_name: str):
|
||
minio_client.remove_object(bucket, object_name)
|