小程序|微信小程序上传图片到阿里云存储 python|微信小程序|阿里云存储

文章目录

前言
一、微信小程序上传方法
二、阿里云OSS
- 1.配置跨域访问（参考文档）
- 2.获取上传签名（重点）
三、微信小程序封装上传方法
- 测试
总结

前言小程序业务，涉及到上传图片的功能，刚开始使用的是腾讯低代码平台，发现上传很方便，无需关注逻辑，但是有很多潜在的问题，无法定制开发。之前使用过阿里云的OSS，感觉还可以，就打算直接上手，在微信小程序环境内上传文件到阿里云存储。
【小程序|微信小程序上传图片到阿里云存储】本次是通过服务器生成签名，客户端拿着签名直接进行上传操作，减少客户端操作
一、微信小程序上传方法微信小程序文件上传官方文档。由于小程序内部环境，暂时无法使用其他方式进行上传。

是通过调用wx.uploadFile(Object object)方法进行上传，示例如下：

wx.chooseImage({ success (res) { const tempFilePaths = res.tempFilePaths wx.uploadFile({ url: 'https://example.weixin.qq.com/upload', //仅为示例，非真实的接口地址 filePath: tempFilePaths[0], name: 'file', formData: { 'user': 'test' }, success (res){ const data = https://www.it610.com/article/res.data //do something } }) } })

二、阿里云OSS 阿里云官方 微信小程序实践 官方文档
有了文档，那么，接下来就可以根据文档进行定制操作
1.配置跨域访问（参考文档） 2.获取上传签名（重点）获取签名有多种方式，因为我使用的是Python环境，本次就通过Python来实现如何在服务器获取上传签名

服务器签名准备所需要的数据

bucket_name_endpoint Bucket域名 callback_url上传完成之后，阿里云会回调该地址 access_key_id访问授权key access_key_secret访问授权secret

access_key_id和access_key_secret 为了安全，建议使用RAM授权特定存储权限

RAM权限策略如下：

{ "Statement": [ { "Action": "oss:*", "Effect": "Allow", "Resource": [ "acs:oss:*:*:devapp-storage", "acs:oss:*:*:devapp-storage/*" ] } ], "Version": "1" }

devapp-storage 是本人的存储名称，这个名称要更换为自己的存储名

分析下具体流程

小程序请求服务器，获取上传token
小程序根据token，进行上传文件
上传成功之后，如果配置回调，则阿里云回调服务器
进行数据库的存储以及资源展示

服务器代码如下：

#!/usr/bin/env python # -*- coding:utf-8 -*- # filename: alioss # date: 2022/5/26import datetime import hmac import json import logging import time import urllib.request from hashlib import sha1 as sha import base64 import urllib.request from Crypto.Signature import PKCS1_v1_5 from Crypto.Hash import MD5 from Crypto.PublicKey import RSAlogger = logging.getLogger(__name__)def verify_auth(auth_str, authorization_base64, pub_key): """ 校验签名是否正确（MD5 + RAS） :param auth_str: 文本信息 :param authorization_base64: 签名信息 :param pub_key: 公钥 :return: 若签名验证正确返回 True 否则返回 False """ pub_key_load = RSA.importKey(pub_key) auth_md5 = MD5.new(auth_str.encode()) result = False try: result = PKCS1_v1_5.new(pub_key_load).verify(auth_md5, base64.b64decode(authorization_base64.encode())) except Exception as e: logger.error(f"oss callback authorization verify failed! Exception:{e}") return resultdef get_iso_8601(expire): gmt = datetime.datetime.utcfromtimestamp(expire).isoformat() gmt += 'Z' return gmtdef get_pub_key(pub_key_url_base64): """ 抽取出 public key 公钥 """ pub_key_url = base64.b64decode(pub_key_url_base64.encode()).decode() if pub_key_url.find("://gosspublic.alicdn.com/") == -1: raise Exception('public key error') ## 可配置缓存，将公钥缓存下来 url_reader = urllib.request.urlopen(pub_key_url) pub_key = url_reader.read() return pub_keyclass AliOss(object):def __init__(self, access_key_id, access_key_secret, bucket_name_endpoint, callback_url): """ :param access_key_id: :param access_key_secret: :param bucket_name_endpoint:Bucket 域名 :param callback_url:回调地址 """ self.bucket_name_endpoint = bucket_name_endpoint self.callback_url = callback_url self.access_key_id = access_key_id self.access_key_secret = access_key_secretdef make_token(self, upload_dir, expire_time=300, content_max_length=10 * 1024 * 1024): """ :param upload_dir:上传目录 :param expire_time:token生效时间 :param content_max_length:最大上传文件大小 :return: """ expire_time = int(time.time()) + expire_time policy_dict = { 'expiration': get_iso_8601(expire_time), 'conditions': [ ['starts-with', '$key', upload_dir], ["content-length-range", 0, content_max_length] ] } policy = json.dumps(policy_dict).strip() policy_encode = base64.b64encode(policy.encode()) h = hmac.new(self.access_key_secret.encode(), policy_encode, sha) sign_result = base64.encodebytes(h.digest()).strip()callback_dict = { 'callbackUrl': self.callback_url, 'callbackBody': 'filename=${object}&size=${size}&mimeType=${mimeType}', 'callbackBodyType': 'application/x-www-form-urlencoded' } callback_param = json.dumps(callback_dict).strip() base64_callback_body = base64.b64encode(callback_param.encode())token_dict = { 'access_key_id': self.access_key_id, 'host': self.bucket_name_endpoint, 'policy': policy_encode.decode(), 'signature': sign_result.decode(), 'expire': expire_time, 'callback': base64_callback_body.decode() } return token_dict@staticmethod def callback_verify(request_headers, request_body):pub_key_url = ''try: pub_key_url_base64 = request_headers['HTTP_X_OSS_PUB_KEY_URL'] pub_key = get_pub_key(pub_key_url_base64) except Exception as e: logger.error(f"Get pub key failed! pub_key_url {pub_key_url} Exception:{e}") return {"status": 400, "msg": "Get pub key failed!"}# get authorization authorization_base64 = request_headers['HTTP_AUTHORIZATION']# get callback body content_length = int(request_headers['CONTENT_LENGTH']) callback_body = request_body[:content_length]query_string = request_headers['QUERY_STRING'] path_info = request_headers['PATH_INFO'] query_string = '?' + query_string if query_string else '' auth_str = path_info + query_string + '\n' + callback_body.decode() result = verify_auth(auth_str, authorization_base64, pub_key)if not result: return {"status": 400, "msg": "authorization verify failed!"}return {"status": 200, "Status": "OK"}

客户端请求token方法，客户端通过post请求，携带文件名参数

class Upload2View(APIView):def post(self, request): """ 获取上传的token :param request: :return: """ filename = request.data.get('filename', '') f_type = filename.split(".")[-1] if not f_type or (f_type and f_type not in settings.FILE_UPLOAD_ALLOW): return ApiResponse(code=1001, msg='上传类型错误')upload_dir = f"{request.user.pk}/" random_file = make_from_user_uuid(request.user.username) upload_key = f"{upload_dir}{random_file}.{f_type}{settings.FILE_UPLOAD_TMP_KEY}"access_key_id = "L******************"# 填写自己的 access_key_secret = "e2a5m1**************"# 填写自己的 bucket_name_endpoint = "devapp-storage.************ncs.com"# 填写自己的 callback_url = f"https://*********/api/v1/server/callback"# 填写自己的alioss = AliOss(access_key_id, access_key_secret, bucket_name_endpoint, callback_url) upload_token = alioss.make_token(upload_dir)data = https://www.it610.com/article/{"upload_token": upload_token, "upload_key": upload_key }return ApiResponse(data=https://www.it610.com/article/data)

阿里云回调方法,阿里云是通过post进行回调

class AliOSSCallBackView(APIView): authentication_classes = []def post(self, request): logging.error(request.META) access_token = request.query_params.get('access_token') result = AliOss.callback_verify(request.META, request.body) response = ApiResponse(**result) if response.status_code == 200: upload_key = request.data.get('filename') size = request.data.get('size') PictureInfo.objects.create(pic_uid_name=upload_key, pic_size=size) # 可以进行一些操作，比如数据入库 return response

三、微信小程序封装上传方法根据服务器生成的签名数据，通过下面方法就行封装，操作测试，记得把域名校验关闭

function uploadFile(filePath, upload_token, upload_key, success, fail) { if (!filePath) { fail && fail(); return; }const { access_key_id, policy, signature, host, callback } = upload_token//小程序直传oss wx.uploadFile({ url: `https://${host}`, filePath: filePath, name: 'file', //必须填file header: { "Content-Type": "multipart/form-data" }, formData: { 'key': upload_key, 'policy': policy, 'OSSAccessKeyId': access_key_id, 'signature': signature, 'callback': callback, // 'x-oss-security-token': security_token, 'success_action_status': '200' }, success: function (res) { if (res.statusCode != 200) { fail && fail(new Error('上传错误:' + JSON.stringify(res))) return; } success(res); }, fail: function (err) { fail && fail(err); }, })}module.exports = uploadFile;

测试小程序上传