228 lines
7.3 KiB
Python
228 lines
7.3 KiB
Python
"""
|
||
阿里云 OSS 简单上传工具
|
||
参考文档: https://help.aliyun.com/zh/oss/user-guide/simple-upload
|
||
|
||
支持:
|
||
- 上传本地文件
|
||
- 上传字节数据 / 字符串
|
||
- 上传文件流（file-like object）
|
||
- 自动根据文件后缀生成 OSS 对象路径
|
||
- 返回可公开访问的 URL（需确保 Bucket 已开启公共读或已配置签名）
|
||
|
||
前置条件:
|
||
1. pip install alibabacloud-oss-v2
|
||
2. 在 .env 中配置以下变量:
|
||
OSS_ACCESS_KEY_ID=<你的 AccessKey ID>
|
||
OSS_ACCESS_KEY_SECRET=<你的 AccessKey Secret>
|
||
OSS_BUCKET_NAME=<存储空间名称>
|
||
OSS_ENDPOINT=<Endpoint，如 https://oss-cn-hangzhou.aliyuncs.com>
|
||
OSS_REGION=<地域，如 cn-hangzhou>
|
||
OSS_URL_PREFIX=<可选，自定义域名前缀，如 https://cdn.example.com>
|
||
"""
|
||
|
||
import os
|
||
import uuid
|
||
import mimetypes
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Optional, Union, BinaryIO
|
||
|
||
import alibabacloud_oss_v2 as oss
|
||
from dotenv import load_dotenv
|
||
|
||
# ── 加载环境变量 ──────────────────────────────────────────────
|
||
load_dotenv()
|
||
|
||
# AccessKey 从系统环境变量读取（~/.bashrc 中 export 设置）
|
||
OSS_ACCESS_KEY_ID = os.environ.get("OSS_ACCESS_KEY_ID", "")
|
||
OSS_ACCESS_KEY_SECRET = os.environ.get("OSS_ACCESS_KEY_SECRET", "")
|
||
# 以下配置从 .env 文件读取
|
||
OSS_BUCKET_NAME = os.getenv("OSS_BUCKET_NAME", "")
|
||
OSS_ENDPOINT = os.getenv("OSS_ENDPOINT", "")
|
||
OSS_REGION = os.getenv("OSS_REGION", "")
|
||
# 可选：自定义域名前缀，用于拼接返回的公开 URL
|
||
OSS_URL_PREFIX = os.getenv("OSS_URL_PREFIX", "")
|
||
|
||
|
||
def _get_client() -> oss.Client:
|
||
"""创建并返回 OSS 客户端实例"""
|
||
credentials_provider = oss.credentials.StaticCredentialsProvider(
|
||
access_key_id=OSS_ACCESS_KEY_ID,
|
||
access_key_secret=OSS_ACCESS_KEY_SECRET,
|
||
)
|
||
cfg = oss.config.load_default()
|
||
cfg.credentials_provider = credentials_provider
|
||
cfg.region = OSS_REGION
|
||
cfg.endpoint = OSS_ENDPOINT
|
||
return oss.Client(cfg)
|
||
|
||
|
||
def _generate_object_key(filename: str, prefix: str = "uploads") -> str:
|
||
"""
|
||
根据文件名生成唯一的 OSS 对象 Key
|
||
格式: {prefix}/{日期}/{uuid}_{原始文件名}
|
||
"""
|
||
date_str = datetime.now().strftime("%Y%m%d")
|
||
unique_id = uuid.uuid4().hex[:8]
|
||
safe_name = Path(filename).name # 只取文件名，去掉路径
|
||
return f"{prefix}/{date_str}/{unique_id}_{safe_name}"
|
||
|
||
|
||
def _build_url(object_key: str) -> str:
|
||
"""根据对象 Key 构建可访问的 URL"""
|
||
if OSS_URL_PREFIX:
|
||
return f"{OSS_URL_PREFIX.rstrip('/')}/{object_key}"
|
||
# 默认使用 Bucket 域名拼接
|
||
endpoint = OSS_ENDPOINT.replace("https://", "").replace("http://", "")
|
||
return f"https://{OSS_BUCKET_NAME}.{endpoint}/{object_key}"
|
||
|
||
|
||
def upload_file(
|
||
file_path: str,
|
||
object_key: Optional[str] = None,
|
||
prefix: str = "uploads",
|
||
) -> dict:
|
||
"""
|
||
上传本地文件到 OSS
|
||
|
||
参数:
|
||
file_path: 本地文件的绝对路径
|
||
object_key: 自定义 OSS 对象名称，为 None 则自动生成
|
||
prefix: 对象 Key 的前缀目录
|
||
|
||
返回:
|
||
包含上传结果的字典:
|
||
{
|
||
"url": "文件访问地址",
|
||
"object_key": "OSS 对象路径",
|
||
"etag": "ETag",
|
||
"status_code": 200,
|
||
}
|
||
"""
|
||
if not os.path.isfile(file_path):
|
||
raise FileNotFoundError(f"文件不存在: {file_path}")
|
||
|
||
filename = os.path.basename(file_path)
|
||
if object_key is None:
|
||
object_key = _generate_object_key(filename, prefix)
|
||
|
||
# 检测文件类型，设置 Content-Type
|
||
content_type, _ = mimetypes.guess_type(file_path)
|
||
|
||
client = _get_client()
|
||
result = client.put_object_from_file(
|
||
oss.PutObjectRequest(
|
||
bucket=OSS_BUCKET_NAME,
|
||
key=object_key,
|
||
content_type=content_type,
|
||
),
|
||
file_path,
|
||
)
|
||
|
||
return {
|
||
"url": _build_url(object_key),
|
||
"object_key": object_key,
|
||
"etag": result.etag,
|
||
"status_code": result.status_code,
|
||
}
|
||
|
||
|
||
def upload_bytes(
|
||
data: Union[bytes, str],
|
||
filename: str,
|
||
object_key: Optional[str] = None,
|
||
prefix: str = "uploads",
|
||
content_type: Optional[str] = None,
|
||
) -> dict:
|
||
"""
|
||
上传字节数据或字符串到 OSS
|
||
|
||
参数:
|
||
data: 要上传的数据（bytes 或 str）
|
||
filename: 用于生成 Key 的文件名（如 "report.txt"）
|
||
object_key: 自定义 OSS 对象名称，为 None 则自动生成
|
||
prefix: 对象 Key 的前缀目录
|
||
content_type: 自定义 Content-Type
|
||
|
||
返回:
|
||
包含上传结果的字典
|
||
"""
|
||
if isinstance(data, str):
|
||
data = data.encode("utf-8")
|
||
|
||
if object_key is None:
|
||
object_key = _generate_object_key(filename, prefix)
|
||
|
||
if content_type is None:
|
||
content_type, _ = mimetypes.guess_type(filename)
|
||
|
||
client = _get_client()
|
||
result = client.put_object(
|
||
oss.PutObjectRequest(
|
||
bucket=OSS_BUCKET_NAME,
|
||
key=object_key,
|
||
body=data,
|
||
content_type=content_type,
|
||
)
|
||
)
|
||
|
||
return {
|
||
"url": _build_url(object_key),
|
||
"object_key": object_key,
|
||
"etag": result.etag,
|
||
"status_code": result.status_code,
|
||
}
|
||
|
||
|
||
def upload_fileobj(
|
||
fileobj: BinaryIO,
|
||
filename: str,
|
||
object_key: Optional[str] = None,
|
||
prefix: str = "uploads",
|
||
content_type: Optional[str] = None,
|
||
) -> dict:
|
||
"""
|
||
上传文件流（file-like object）到 OSS
|
||
|
||
参数:
|
||
fileobj: 文件流对象（如 open(path, 'rb') 或 FastAPI 的 UploadFile.file）
|
||
filename: 用于生成 Key 的文件名
|
||
object_key: 自定义 OSS 对象名称，为 None 则自动生成
|
||
prefix: 对象 Key 的前缀目录
|
||
content_type: 自定义 Content-Type
|
||
|
||
返回:
|
||
包含上传结果的字典
|
||
"""
|
||
data = fileobj.read()
|
||
return upload_bytes(
|
||
data=data,
|
||
filename=filename,
|
||
object_key=object_key,
|
||
prefix=prefix,
|
||
content_type=content_type,
|
||
)
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────
|
||
# 命令行入口：python -m utils.oss_uploader --file <路径>
|
||
# ────────────────────────────────────────────────────────────────
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description="阿里云 OSS 简单上传工具")
|
||
parser.add_argument("--file", required=True, help="要上传的本地文件路径")
|
||
parser.add_argument("--key", default=None, help="自定义 OSS 对象路径（可选）")
|
||
parser.add_argument(
|
||
"--prefix", default="uploads", help="对象 Key 前缀（默认: uploads）"
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
result = upload_file(args.file, object_key=args.key, prefix=args.prefix)
|
||
|
||
print("✅ 上传成功!")
|
||
print(f" 访问地址: {result['url']}")
|
||
print(f" 对象路径: {result['object_key']}")
|
||
print(f" ETag: {result['etag']}")
|
||
print(f" 状态码: {result['status_code']}")
|