概述
uniapp使用临时签名上传腾讯云oss对象储存方案，支持小程序、app、h5；
前端不依赖腾讯云SDK工具类；
后端使用python实现，需要安装qcloud-python-sts；
其中计算文件md5值使用了条件编译，因为每个环境获取ArrayBuffer方案不一样都不兼容；
pip install qcloud-python-sts==3.1.6
那些踩过的坑🕳
- 官方方案小程序SDK，但是小程序SDK在APP环境由于无法获取file://类型地址的文件；
- 官方方案JS-SDK，此方案由于要使用file对象然而APP端无法使用Blob工具类；
uniapp实现
import SparkMD5 from "spark-md5";
import {
api_getBucketAndRegionSelf
} from "@/api/common";
import {
OSS_BASE_URL
} from '../config';
async function putObjectAutoPath(keyPath, file) {
console.log("getMD5FileName")
try {
console.log("getMD5FileName")
const md5FileName = await getMD5FileName(file);
const datePath = getDatePath();
const uploadPath = `${datePath}${keyPath.trim() ? `${keyPath.trim()}/` : ''}${md5FileName}`;
console.log("上传路径为：" + uploadPath);
console.log("图片路径=>" + file);
const res = await api_getBucketAndRegionSelf(uploadPath);
console.log(res)
const formData = {
key: res.data.cosKey,
policy: res.data.policy,
success_action_status: 200,
'q-sign-algorithm': res.data.qSignAlgorithm,
'q-ak': res.data.qAk,
'q-key-time': res.data.qKeyTime,
'q-signature': res.data.qSignature,
'x-cos-security-token': res.data.securityToken
};
const uploadResult = await uploadFile('https://' + res.data.cosHost, file, formData);
console.log('上传成功:', uploadResult);
return OSS_BASE_URL + res.data.cosKey;
} catch (error) {
console.error('上传失败:', error);
throw error;
}
}
function getDatePath() {
const date = new Date();
const year = date.getFullYear();
const month = String(date.getMonth() + 1).padStart(2, "0");
const day = String(date.getDate()).padStart(2, "0");
return `/${year}/${month}/${day}/`;
}
function calculateMD5(file) {
return new Promise((resolve, reject) => {
console.log("执行md5值计算H5", file);
const xhr = new XMLHttpRequest();
xhr.open('GET', file, true);
xhr.responseType = 'blob';
xhr.onload = function() {
if (xhr.status === 200) {
const blob = xhr.response;
const reader = new FileReader();
reader.onload = (e) => {
const binary = e.target.result;
const spark = new SparkMD5.ArrayBuffer();
spark.append(binary);
resolve(spark.end());
};
reader.onerror = reject;
reader.readAsArrayBuffer(blob);
} else {
reject(new Error('Failed to fetch blob'));
}
};
xhr.onerror = reject;
xhr.send();
console.log("执行md5值计算MP");
const fs = uni.getFileSystemManager();
fs.readFile({
filePath: file,
encoding: 'base64',
success: (res) => {
const binary = uni.base64ToArrayBuffer(res.data);
const spark = new SparkMD5.ArrayBuffer();
spark.append(binary);
resolve(spark.end());
},
fail: reject,
});
console.log("执行md5值计算APP");
plus.io.resolveLocalFileSystemURL(file, (entry) => {
entry.file((fileObj) => {
const reader = new plus.io.FileReader();
reader.readAsDataURL(file);
reader.onloadend = (evt) => {
const binary = uni.base64ToArrayBuffer(evt.target
.result);
const spark = new SparkMD5.ArrayBuffer();
spark.append(binary);
resolve(spark.end());
};
reader.onerror = reject;
});
}, reject);
});
}
async function getMD5FileName(file) {
const md5 = await calculateMD5(file);
console.log(md5)
return;
const fileType = file.substring(file.lastIndexOf("."));
return `${md5}${fileType}`;
}
function uploadFile(url, filePath, formData) {
return new Promise((resolve, reject) => {
uni.uploadFile({
url: url,
filePath: filePath,
name: 'file',
formData: formData,
success: (res) => {
if (res.statusCode === 200) {
resolve(res);
} else {
reject(new Error(`上传失败，状态码：${res.statusCode}, 响应信息：${res.data}`));
}
},
error: (err) => {
console.log("图片上传失败=》" + res)
reject(err);
},
});
});
}
export {
putObjectAutoPath
};
python实现的
import json
from sts.sts import Sts
import hashlib
import hmac
import base64
import time
from datetime import datetime, timedelta
secret_id = ''
secret_key = ''
bucket = ''
region = ''
def get_temporary_credential():
"""
获取临时密钥
:return:
"""
config = {
'duration_seconds': 1800,
'secret_id': secret_id,
'secret_key': secret_key,
'bucket': bucket,
'region': region,
'allow_prefix': ['*'],
'allow_actions': [
'name/cos:PutObject',
'name/cos:PostObject',
'name/cos:InitiateMultipartUpload',
'name/cos:ListMultipartUploads',
'name/cos:ListParts',
'name/cos:UploadPart',
'name/cos:CompleteMultipartUpload'
],
}
try:
sts = Sts(config)
response = sts.get_credential()
print(response)
response['bucket'] = bucket
response['region'] = region
return response
except Exception as e:
raise Exception("腾讯OSS临时密钥获取异常！")
def get_bucketAndRegion():
"""
获取bucket 桶id 和region地域
:return:
"""
data = {
"bucket": bucket,
"region": region
}
return data
def get_temporary_credential_self_upload(keyPath):
"""
获取腾讯云oss凭证 适用于POST上传请求【不依赖腾讯SDK】
"""
credentials_data = get_temporary_credential().get("credentials")
tmp_secret_id = credentials_data.get("tmpSecretId")
tmp_secret_key = credentials_data.get("tmpSecretKey")
session_token = credentials_data.get("sessionToken")
cos_host = f"{bucket}.cos.{region}.myqcloud.com"
cos_key = keyPath
now = int(time.time())
exp = now + 900
q_key_time = f"{now};{exp}"
q_sign_algorithm = 'sha1'
policy = {
'expiration': (datetime.utcfromtimestamp(exp)).isoformat() + 'Z',
'conditions': [
{'q-sign-algorithm': q_sign_algorithm},
{'q-ak': tmp_secret_id},
{'q-sign-time': q_key_time},
{'bucket': bucket},
{'key': cos_key},
]
}
policy_encoded = base64.b64encode(json.dumps(policy).encode()).decode()
sign_key = hmac.new(tmp_secret_key.encode(), q_key_time.encode(), hashlib.sha1).hexdigest()
string_to_sign = hashlib.sha1(json.dumps(policy).encode()).hexdigest()
q_signature = hmac.new(sign_key.encode(), string_to_sign.encode(), hashlib.sha1).hexdigest()
return {
'cosHost': cos_host,
'cosKey': cos_key,
'policy': policy_encoded,
'qSignAlgorithm': q_sign_algorithm,
'qAk': tmp_secret_id,
'qKeyTime': q_key_time,
'qSignature': q_signature,
'securityToken': session_token
}
