django上传文件到微信小程序云平台

一、获取access_token

1.安装redis

下载安装包

由于这里是在win10环境下进行测试的，所以要安装redis win10版 ，在生产环境中请安装对应版本。

在https://github.com/MSOpenTech/redis/releases下载redis的win10安装包，安装包里通常包括以下文件：

win+r打开cmd窗口，切换到redis安装包目录，这里假设为D:\redis，运行以下命令：

redis-server.exe redis.windows.conf

将会出现以下界面：

这时候另启一个cmd窗口，并切换到redis安装包目录下，也就是D:\redis下，原来的不要关闭，不然就无法访问服务端了。

安装服务

在另外打开的cmd窗口中输入

redis-server --service-install redis.windows.conf

就能把redis添加到win10的服务中。

启动服务

redis-server --service-start

停止服务

redis-server --service-stop

测试服务

在D:\redis目录下运行

redis-cli.exe -h 127.0.0.1 -p 6379

设置键值对 set myKey abc

取出键值对 get myKey

这时就成功了。

2.access_token获取

首先在settings.py中创建一个redis全局连接池

# redis连接池
redis_pool = redis.ConnectionPool(max_connections=10, host='127.0.0.1', port=6379, decode_responses=True)

然后编写一个全局函数文件（最好放在settings.py所在目录的子目录下），添加一个get_access_token.py文件

import logging
import time

import redis
import requests

from mywebsite.settings import redis_pool, GET_WX_ACCESS_TOKEN_URL

# 获取微信文件上传token
log_location = "get_wx_file_upload_access_token.py: "

#获取接口调用凭证的URL，appid即小程序appid，APPSECRET是小程序的APPSECRET。
GET_WX_ACCESS_TOKEN_URL = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=APPID&secret=APPSECRET"
请求参数"

#获取access_token
def get_mini_access_token():
# access_token过期时间
wx_access_token_invalid_time = 0
wx_access_token = ""
r = requests.get(GET_WX_ACCESS_TOKEN_URL)
# redis连接池
red = redis.Redis(connection_pool=redis_pool)
response = eval(r.text)
try:
# print("push_data_to_mini", response["access_token"])
wx_access_token = response["access_token"]
# 设置过期微信access_token过期时间
wx_access_token_invalid_time = time.time() + response["expires_in"]
# 存储到redis
red.set("wx_access_token", wx_access_token)
red.set("wx_access_token_invalid_time", wx_access_token_invalid_time)
logging.info(log_location + "获取access_token成功")

except requests.exceptions.ConnectionError:
logging.error(log_location + "获取access_token错误，连接失败")
else:
try:
errmsg = response["errmsg"]
except KeyError:
pass
else:
logging.error(log_location + "获取access_token失败,", response["errmsg"])

return wx_access_token_invalid_time, wx_access_token

#access_token有效性校验
def get_valid_wx_access_token():
# redis连接
red = redis.Redis(connection_pool=redis_pool)
wx_access_token = red.get("wx_access_token")
wx_access_token_invalid_time = red.get("wx_access_token_invalid_time")
# 检查wx_access_token是否存在
if wx_access_token:
time_now = time.time()
# access_token过期
if time_now > float(wx_access_token_invalid_time):
# print(" wx_access_token过期")
logging.info(log_location + "wx_access_token过期 ")
get_mini_access_token()
wx_access_token = red.get("wx_access_token")
return wx_access_token
else:
# print(" wx_access_token没有过期")
wx_access_token = red.get("wx_access_token")
return wx_access_token

else:
get_mini_access_token()
wx_access_token = red.get("wx_access_token")
return wx_access_token

注意：微信小程序的appid和appsecret是敏感信息，请加密存储，这里仅作为示例。

二、获取文件上传URL和认证参数

根据微信官方的要求

上传文件需要access_token，就是前面获取的access_token，env和path最好放在settings.py中，因为这是全局参数，放在settings.py中方便更改。新建get_file_upload_param.py文件：

import ast
import logging

import redis
import requests

from mywebsite.global_functions.get_wx_file_upload_access_token import get_valid_wx_access_token
from mywebsite.settings import GET_MINI_UPLOAD_FILE_URL, redis_pool, MINI_TEST_ENV, \
UPLOAD_FILE_PATH

log_location = "get_wx_file_upload_param.py: "

# 获取小程序文件上传参数
def get_wx_file_upload_param(env, path):
# 获取接口调用token
wx_access_token, wx_access_token_invalid_time = get_valid_wx_access_token()
# 获取文件上传链接的URL
url = GET_MINI_UPLOAD_FILE_URL + wx_access_token

try:
para = {"env": env, "path": path}
wx_response = requests.post(url, json=para).text
mini_upload_file_params = ast.literal_eval(wx_response)
mini_upload_file_params.pop("errcode")
mini_upload_file_params.pop("errmsg")
return mini_upload_file_params

except requests.exceptions.RequestException as e:
logging.error(log_location + "获取文件上传参数失败")
return ""

三、数据编码上传

根据微信官方文件上传要求：

需要对文件进行编码。首先安装文件编码工具requests_toolbelt：

pip install requests_toolbelt

接下来添加一个push_data.py（推荐和前面的文件放在一个文件夹中，因为这些都是全局使用的函数文件）

# 推送文件到微信小程序云平台
def push_file_mini(path, env, file):
successful = False
# 获取文件上传post请求参数
content = get_wx_file_upload_param(env, path)
# 修改字典的键
content["key"] = path
content["Signature"] = content.pop("authorization")
content["x-cos-security-token"] = content.pop("token")
content["x-cos-meta-fileid"] = content.pop("cos_file_id").replace("\\", "")
# 文件上传URL
file_upload_url = content["url"].replace("\\", "")
m = MultipartEncoder(fields={
"key": path + file.name,#加上path可以传送到云平台指定目录下
"Signature": content["Signature"],
"x-cos-security-token": content["x-cos-security-token"],
"x-cos-meta-fileid": content["x-cos-meta-fileid"],
"file": (file.name, file, 'image/octet-stream')
})
try:
r = requests.post(file_upload_url, data=m, headers={'Content-Type': m.content_type})
# print(r.status_code)
if r.status_code == 204:
successful = True
except requests.exceptions.RequestException as e:
logging.error(log_location + "推送文件到微信小程序云平台失败。", e)

return successful

path = UPLOAD_FILE_PATH
env = MINI_TEST_ENV

with open("4.jpg", "rb") as file:
print(push_file_mini(path, env, file))

结束语

文件推送到小程序云平台是一个耗时操作，最好作为一个异步任务执行。

• 点赞
• 收藏
• 分享
• 文章举报