41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
import boto3
|
|
from botocore.client import Config
|
|
|
|
|
|
# minio config
|
|
# MinIO配置
|
|
MINIO_ENDPOINT = "s3.sansenhoshi.top" # 不含 http/https
|
|
MINIO_ACCESS_KEY = "JDMynACSjPaN8JRwriwS"
|
|
MINIO_SECRET_KEY = "JRl4bIGxeiqwqvfeBTlAWQbUKMpNNHSZPm6ne93j"
|
|
MINIO_BUCKET = "s-file-trans"
|
|
MINIO_REGION = "bot"
|
|
MINIO_SECURE = True # False 则用 HTTP
|
|
|
|
# 可访问的公网地址前缀
|
|
MINIO_PUBLIC_DOMAIN = f"https://{MINIO_ENDPOINT}/{MINIO_BUCKET}"
|
|
|
|
|
|
s3_client = boto3.client(
|
|
"s3",
|
|
endpoint_url=f"{'https' if MINIO_SECURE else 'http'}://{MINIO_ENDPOINT}",
|
|
aws_access_key_id=MINIO_ACCESS_KEY,
|
|
aws_secret_access_key=MINIO_SECRET_KEY,
|
|
region_name=MINIO_REGION,
|
|
config=Config(signature_version="s3v4"),
|
|
)
|
|
|
|
|
|
def upload_to_s3(file_path) -> str:
|
|
file_key = f"cache/{file_path.name}" # 你可以换成别的目录
|
|
try:
|
|
s3_client.upload_file(
|
|
str(file_path),
|
|
MINIO_BUCKET,
|
|
file_key,
|
|
ExtraArgs={"ACL": "public-read"} # 如果你设置桶为私有，可以移除这行
|
|
)
|
|
return f"{MINIO_PUBLIC_DOMAIN}/{file_key}"
|
|
except Exception as e:
|
|
logger.exception(f"上传到 MinIO 失败: {e}")
|
|
return ""
|