wechat cos upload util

wxUploadUtil.py

import requests
import json
import time
import os


def get_access_token(appid: str, secret: str, config) -> str:
    """
    获取调用凭证
    https://developers.weixin.qq.com/miniprogram/dev/api-backend/open-api/access-token/auth.getAccessToken.html
    args:
        apppid: 小程序appid
        secret: 小程序appSecret

    return: access_token
    """
    # 检查本地token
    token = config["access_token"]
    try:
        now = int(time.time())
        if None == token or (now - config["create_time"]) > config["expires_in"]:
            resp = requests.get(
                "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={0}&secret={1}".format(
                    appid, secret
                )
            )
            resp = resp.json()

            if None == resp["access_token"]:
                raise Exception(resp["errmsg"])
            # 缓存处理
            config["access_token"] = resp["access_token"]
            config["create_time"] = int(time.time())
            config["expires_in"] = resp["expires_in"]

        return config["access_token"]
    except Exception as e:
        print("获取access token失败")
        raise e


def get_upload_file_info(access_token: str, env: str, path: str):
    """
    上传文件信息
    https://developers.weixin.qq.com/miniprogram/dev/wxcloudrun/src/development/storage/service/upload.html
    args:
        access_token: 
        env: 云托管的路径
        path: 存放位置
    return:
        json: {
            errcode,
            errmsg,
            url,
            token,
            authorization,
            file_id,
            cos_file_id
        }
    """
    try:
        url = "https://api.weixin.qq.com/tcb/uploadfile?access_token={0}".format(
            access_token
        )
        headers = {"Content-Type": "application/json;charset=UTF-8"}
        data = {"env": env, "path": path}
        resp = requests.post(url, json=data, headers=headers)
        # 转json
        resp = resp.json()
        if 0 != resp["errcode"]:
            print(resp["errmsg"])
            print(resp)
            raise Exception(resp["errmsg"])
        # 返回上传文件信息
        return resp
    except Exception as e:
        print("获取文件上传链接失败")
        raise e


def upload_file(url, request_path, authorization, token, file_id, file):
    """
    上传文件
    https://developers.weixin.qq.com/miniprogram/dev/wxcloudrun/src/development/storage/service/upload.html
    args:
        url: 上传图片路径 由get_upload获得的url
        request_path: 上传路径，和get_upload_file_info内部的一致
        authorization: info返回的token
        file_id: cos_file_id
        file: 文件的二进制数据
    return: None
    """
    try:
        form = {
            "key": request_path,
            "Signature": authorization,
            "x-cos-security-token": token,
            "x-cos-meta-fileid": file_id,
            "file": file,
        }
        resp = requests.post(url=url, files=form)
        return resp
    except Exception as e:
        print("上传文件失败")
        raise e


def get_download_link(access_token, env, file_id):
    """
    获取文件下载链接
    https://developers.weixin.qq.com/miniprogram/dev/wxcloudrun/src/development/storage/service/download.html
    args:
        access_token: 调用token
        env: 云环境env
        file_id: file_id
    return: 文件下载链接
    """
    try:
        url = "https://api.weixin.qq.com/tcb/batchdownloadfile?access_token={0}".format(
            access_token
        )
        #headers = {"Content-Type": "application/json"}
        data = {
            "env": env,
            "file_list": [
                {
                    "fileid": file_id
                    # max_age
                }
            ],
        }
        resp = requests.post(url, json=data)
        resp = resp.json()
        if 0 != resp["errcode"]:
            print(resp["errmsg"])
            raise Exception(resp["errmsg"])
        return resp["file_list"][0]["download_url"]
    except Exception as e:
        print("获取下载链接失败")
        raise e


def load_config(path="./config.json"):
    """
    获取配置，主要是读取缓存的access_token
    args: 读取路径
    return: config
    """
    with open(path, "r") as f:
        config = json.load(f)
    return config


def save_config(config, path="./config.json"):
    """
    保持配置
    args: 读取路径
    """
    with open(path, "w") as f:
        json.dump(config, f, indent=4)


def wx_upload_file(config, filepath, uploadpath):
    """
    上传文件
    args:
        config: 配置
        filepath: 文件的路径
        uploadpath: 上传的路径，不能以/开头，以/结尾
    return: url
    """
    realpath = os.path.realpath(filepath)
    filename = os.path.basename(realpath)
    with open(realpath, "rb") as f:
        data = f.read()
    # 获取access_token
    token = get_access_token(config["appid"], config["secret"], config)
    # 获取上传文件链接
    info = get_upload_file_info(token, config["env"], uploadpath + filename)
    upload_file(
        info["url"],
        uploadpath + filename,
        info["authorization"],
        info["token"],
        info["cos_file_id"],
        data,
    )
    download_url = get_download_link(token, config["env"], info['file_id'])
    return (download_url, info['file_id'])


if __name__ == "__main__":
    config = load_config("./config.json")
    #print(wx_upload_file(config, "121.jpg", "temp3/"))

    # 有url
    # print(wx_upload_file(config,'./temp.jpg','temp3/'))
    #
    # # 无url
    path = "./154190339_附件/"
    filenames = os.listdir(path)
    for filename in filenames:
        print(filename)
        image = os.path.join(path, filename)
        url = wx_upload_file(config, image, "temp3/")
        print(url)

    save_config(config, "./config.json")

如果上传文件路径或文件名有中文需要修改requests源码

requests/models.py的prepare_body里的complexjson.dumps参数，加一个ensure_ascii=False

config.json

{
    "appid": "",
    "secret": "",
    "env": "",
    "create_time": 1656343740,
    "expires_in": 7200,
    "access_token": ""
}