|
|
import base64
|
|
|
import time
|
|
|
import oss2
|
|
|
from oss2.models import PolicyConditions # 确保正确导入 PolicyConditions
|
|
|
from typing import Dict
|
|
|
|
|
|
class AliUtil:
|
|
|
def __init__(self):
|
|
|
# 从配置中读取阿里云 OSS 的配置
|
|
|
self.endpoint = "your_endpoint" # 替换为你的 OSS endpoint
|
|
|
self.access_key_id = "your_access_key_id" # 替换为你的 AccessKeyId
|
|
|
self.access_key_secret = "your_access_key_secret" # 替换为你的 AccessKeySecret
|
|
|
|
|
|
def get_signature(self, bucket_name: str) -> Dict[str, str]:
|
|
|
"""
|
|
|
获取临时访问 OSS 的签名
|
|
|
:param bucket_name: bucket 名称，由前端传入（上传不同文件到不同文件夹）
|
|
|
:return: 包含签名信息的字典
|
|
|
"""
|
|
|
resp_map = {}
|
|
|
# host 的格式为 bucketName.endpoint
|
|
|
host = f"https://{bucket_name}.{self.endpoint}"
|
|
|
# callbackUrl 为上传回调服务器的 URL，请将下面的 IP 和 Port 配置为您自己的真实信息。
|
|
|
callback_url = ""
|
|
|
dir = "" # 用户上传文件时指定的前缀。
|
|
|
|
|
|
# 初始化 OSS 客户端
|
|
|
auth = oss2.Auth(self.access_key_id, self.access_key_secret)
|
|
|
bucket = oss2.Bucket(auth, self.endpoint, bucket_name)
|
|
|
|
|
|
try:
|
|
|
# 设置过期时间为半小时 (1800 秒)
|
|
|
expire_time = 60 * 30
|
|
|
expire_end_time = int(time.time()) + expire_time
|
|
|
|
|
|
# 设置 Policy 条件
|
|
|
policy_conditions = PolicyConditions()
|
|
|
policy_conditions.add_condition(PolicyConditions.CONTENT_LENGTH_RANGE, 0, 1048576000) # 文件大小限制
|
|
|
policy_conditions.add_condition(PolicyConditions.STARTS_WITH, PolicyConditions.KEY, dir) # 文件前缀
|
|
|
|
|
|
# 生成 Post Policy
|
|
|
post_policy = bucket._make_post_policy(expire_end_time, policy_conditions)
|
|
|
encoded_policy = base64.b64encode(post_policy.encode('utf-8')).decode('utf-8')
|
|
|
|
|
|
# 计算签名
|
|
|
post_signature = bucket._make_post_signature(post_policy)
|
|
|
|
|
|
# 构建返回的签名信息
|
|
|
resp_map["accessid"] = self.access_key_id
|
|
|
resp_map["policy"] = encoded_policy
|
|
|
resp_map["signature"] = post_signature
|
|
|
resp_map["dir"] = dir
|
|
|
resp_map["host"] = host
|
|
|
resp_map["expire"] = str(expire_end_time)
|
|
|
|
|
|
# 构建回调信息
|
|
|
callback_body = {
|
|
|
"callbackUrl": callback_url,
|
|
|
"callbackBody": "filename=${object}&size=${size}&mimeType=${mimeType}&height=${imageInfo.height}&width=${imageInfo.width}",
|
|
|
"callbackBodyType": "application/x-www-form-urlencoded"
|
|
|
}
|
|
|
base64_callback_body = base64.b64encode(str(callback_body).encode('utf-8')).decode('utf-8')
|
|
|
resp_map["callback"] = base64_callback_body
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"Error: {e}")
|
|
|
|
|
|
return resp_map |
