实现文件上传api

Author: gusibi

example/wxapp_cloud_file.py

@@ -0,0 +1,21 @@
+#! -*- coding: utf-8 -*-
+from os import environ, path
+from pprint import pprint
+
+from weixin import WxAppCloudAPI
+
+appid = environ.get("WXAPP_APPID", "appid")
+secret = environ.get("WXAPP_SECRET", "secret")
+env = "test-id"
+
+app_cloud = WxAppCloudAPI(
+    appid=appid, app_secret=secret, grant_type="client_credential"
+)
+token = app_cloud.client_credential_for_access_token().get("access_token")
+
+cloud_api = WxAppCloudAPI(access_token=token)
+
+path = "test/file.py"
+filepath = "./file.py"
+resp = cloud_api.upload_file(json_body={"env": env, "path": path, "filepath": filepath})
+print(resp)

weixin/client.py

@@ -10,9 +10,12 @@
 Description: Weixin OAuth2
 """
 
+import requests
+import xmltodict
+
 from . import oauth2
-from .bind import bind_method
-from .helper import genarate_signature
+from .bind import bind_method, WeixinAPIError
+from .helper import genarate_signature, smart_bytes, smart_str
 
 
 SUPPORTED_FORMATS = ["", "json"]
@@ -317,13 +320,53 @@ def __init__(self, *args, **kwargs):
         response_type="entry",
     )
 
-    # 上传文件 获取文件上传链接
-    upload_file = bind_method(
+    pre_upload_file = bind_method(
         path="/tcb/uploadfile",
         method="POST",
         accepts_parameters=["json_body"],
         response_type="entry",
     )
+    # 上传文件 获取文件上传链接
+    def upload_file(self, json_body=None):
+        """
+        获取文件下载链接
+        参数：json_body  字典
+            json_body 结构：
+                env string 云环境ID
+                path string 文件云存储的路径(包括文件名),比如：  test/file.txt
+                filepath string 文件本地路径
+        upload_file(json_body={"env": "envid", "path": "test/file.txt", "filepath": "/home/user/test.txt"})
+        """
+        try:
+            with open(json_body["filepath"], "rb") as f:
+                file_data = f.read()
+        except Exception as e:
+            raise WeixinAPIError("400", "000000", e)
+
+        filepath = json_body.pop("filepath", None)
+        pre_resp = self.pre_upload_file(json_body=json_body)
+        if pre_resp.get("errcode") == 0 and pre_resp.get("errmsg") == "ok":
+            files = {
+                "key": json_body.get("path"),
+                "Signature": pre_resp.get("authorization"),
+                "x-cos-security-token": pre_resp.get("token"),
+                "x-cos-meta-fileid": pre_resp.get("cos_file_id"),
+                "file": file_data,  # file 一定要放到最后，血泪的教训
+            }
+            # encode
+            params = {smart_str(k): v for k, v in files.items()}
+            resp = requests.post(pre_resp.get("url"), files=params)
+            status_code = resp.status_code
+            if status_code == 204:
+                return {"errcode": 0, "errmsg": "ok"}
+            else:
+                content_obj = xmltodict.parse(resp.content)
+                results = content_obj.get("Error", {})
+                raise WeixinAPIError(status_code, results["Code"], results["Message"])
+        else:
+            raise WeixinAPIError(
+                pre_resp.status_code, pre_resp["errcode"], pre_resp["errmsg"]
+            )
 
     # 获取文件下载链接
     batch_download_file = bind_method(

weixin/config.py

@@ -1,9 +1,9 @@
 # -*- coding: utf-8 -*-
 from __future__ import unicode_literals
 
-AUTO_REPLY_CONTENT = '''
+AUTO_REPLY_CONTENT = """
 欢迎关注在行！
 「在行」是一个知识服务的平台。我们为想要解决问题的人找到行家，一对一面对面答疑解惑、出谋划策；我们为愿意分享知识和经验的人找到学员，让头脑中的资源发挥更大的价值。
 期待你的加入！
 www.zaih.com
-'''
+"""

weixin/helper.py

@@ -5,7 +5,7 @@
 File:   client.py
 Author: goodspeed
 Email:  cacique1103@gmail.com
-Github: https://github.com/zongxiao
+Github: https://github.com/gusibi
 Date:   2015-02-11
 Description: Weixin helpers
 """
@@ -20,32 +20,28 @@
 
 PY2 = sys.version_info[0] == 2
 
-_always_safe = ('abcdefghijklmnopqrstuvwxyz'
-                'ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-+')
+_always_safe = "abcdefghijklmnopqrstuvwxyz" "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-+"
 
 safe_char = _always_safe
 
-error_dict = {
-    'AppID 参数错误': {
-        'errcode': 40013,
-        'errmsg': 'invalid appid'
-    }
-}
+error_dict = {"AppID 参数错误": {"errcode": 40013, "errmsg": "invalid appid"}}
 
 
 if PY2:
     text_type = unicode
     iteritems = lambda d, *args, **kwargs: d.iteritems(*args, **kwargs)
 
-    def to_native(x, charset=sys.getdefaultencoding(), errors='strict'):
+    def to_native(x, charset=sys.getdefaultencoding(), errors="strict"):
         if x is None or isinstance(x, str):
             return x
         return x.encode(charset, errors)
+
+
 else:
     text_type = str
     iteritems = lambda d, *args, **kwargs: iter(d.items(*args, **kwargs))
 
-    def to_native(x, charset=sys.getdefaultencoding(), errors='strict'):
+    def to_native(x, charset=sys.getdefaultencoding(), errors="strict"):
         if x is None or isinstance(x, str):
             return x
         return x.decode(charset, errors)
@@ -60,15 +56,18 @@ def to_native(x, charset=sys.getdefaultencoding(), errors='strict'):
 
 try:
     import hashlib
+
     md5_constructor = hashlib.md5
     md5_hmac = md5_constructor
     sha_constructor = hashlib.sha1
     sha_hmac = sha_constructor
 except ImportError:
     import md5
+
     md5_constructor = md5.new
     md5_hmac = md5
     import sha
+
     sha_constructor = sha.new
     sha_hmac = sha
 
@@ -79,6 +78,7 @@ class Promise(object):
     the closure of the lazy function. It can be used to recognize
     promises in code.
     """
+
     pass
 
 
@@ -89,11 +89,10 @@ def __init__(self, obj, *args):
 
     def __str__(self):
         original = UnicodeDecodeError.__str__(self)
-        return '%s. You passed in %r (%s)' % (original, self.obj,
-                                              type(self.obj))
+        return "%s. You passed in %r (%s)" % (original, self.obj, type(self.obj))
 
 
-def smart_text(s, encoding='utf-8', strings_only=False, errors='strict'):
+def smart_text(s, encoding="utf-8", strings_only=False, errors="strict"):
     """
     Returns a text object representing 's' -- unicode on Python 2 and str on
     Python 3. Treats bytestrings using the 'encoding' codec.
@@ -105,9 +104,14 @@ def smart_text(s, encoding='utf-8', strings_only=False, errors='strict'):
     return force_text(s, encoding, strings_only, errors)
 
 
-_PROTECTED_TYPES = six.integer_types + (type(None), float, Decimal,
-                                        datetime.datetime, datetime.date,
-                                        datetime.time)
+_PROTECTED_TYPES = six.integer_types + (
+    type(None),
+    float,
+    Decimal,
+    datetime.datetime,
+    datetime.date,
+    datetime.time,
+)
 
 
 def is_protected_type(obj):
@@ -118,7 +122,7 @@ def is_protected_type(obj):
     return isinstance(obj, _PROTECTED_TYPES)
 
 
-def force_text(s, encoding='utf-8', strings_only=False, errors='strict'):
+def force_text(s, encoding="utf-8", strings_only=False, errors="strict"):
     """
     Similar to smart_text, except that lazy instances are resolved to
     strings, rather than kept as lazy objects.
@@ -136,7 +140,7 @@ def force_text(s, encoding='utf-8', strings_only=False, errors='strict'):
                     s = six.text_type(s, encoding, errors)
                 else:
                     s = six.text_type(s)
-            elif hasattr(s, '__unicode__'):
+            elif hasattr(s, "__unicode__"):
                 s = six.text_type(s)
             else:
                 s = six.text_type(bytes(s), encoding, errors)
@@ -154,12 +158,11 @@ def force_text(s, encoding='utf-8', strings_only=False, errors='strict'):
             # working unicode method. Try to handle this without raising a
             # further exception by individually forcing the exception args
             # to unicode.
-            s = ' '.join(force_text(arg, encoding, strings_only, errors)
-                         for arg in s)
+            s = " ".join(force_text(arg, encoding, strings_only, errors) for arg in s)
     return s
 
 
-def smart_bytes(s, encoding='utf-8', strings_only=False, errors='strict'):
+def smart_bytes(s, encoding="utf-8", strings_only=False, errors="strict"):
     """
     Returns a bytestring version of 's', encoded as specified in 'encoding'.
     If strings_only is True, don't convert (some) non-string-like objects.
@@ -170,18 +173,18 @@ def smart_bytes(s, encoding='utf-8', strings_only=False, errors='strict'):
     return force_bytes(s, encoding, strings_only, errors)
 
 
-def force_bytes(s, encoding='utf-8', strings_only=False, errors='strict'):
+def force_bytes(s, encoding="utf-8", strings_only=False, errors="strict"):
     """
     Similar to smart_bytes, except that lazy instances are resolved to
     strings, rather than kept as lazy objects.
     If strings_only is True, don't convert (some) non-string-like objects.
     """
     # Handle the common case first for performance reasons.
     if isinstance(s, bytes):
-        if encoding == 'utf-8':
+        if encoding == "utf-8":
             return s
         else:
-            return s.decode('utf-8', errors).encode(encoding, errors)
+            return s.decode("utf-8", errors).encode(encoding, errors)
     if strings_only and is_protected_type(s):
         return s
     if isinstance(s, Promise):
@@ -197,13 +200,14 @@ def force_bytes(s, encoding='utf-8', strings_only=False, errors='strict'):
                 # An Exception subclass containing non-ASCII data that doesn't
                 # know how to print itself properly. We shouldn't raise a
                 # further exception.
-                return b' '.join(force_bytes(arg, encoding,
-                                             strings_only, errors)
-                                 for arg in s)
+                return b" ".join(
+                    force_bytes(arg, encoding, strings_only, errors) for arg in s
+                )
             return six.text_type(s).encode(encoding, errors)
     else:
         return s.encode(encoding, errors)
 
+
 if six.PY3:
     smart_str = smart_text
     force_str = force_text
@@ -228,31 +232,32 @@ def force_bytes(s, encoding='utf-8', strings_only=False, errors='strict'):
 def genarate_js_signature(params):
     keys = params.keys()
     keys.sort()
-    params_str = b''
+    params_str = b""
     for key in keys:
-        params_str += b'%s=%s&' % (smart_str(key), smart_str(params[key]))
+        params_str += b"%s=%s&" % (smart_str(key), smart_str(params[key]))
     params_str = params_str[:-1]
     return sha1(params_str).hexdigest()
 
 
 def genarate_signature(params):
     sorted_params = sorted([v for k, v in params.items()])
-    params_str = smart_str(''.join(sorted_params))
-    return sha1(str(params_str).encode('utf-8')).hexdigest()
+    params_str = smart_str("".join(sorted_params))
+    return sha1(str(params_str).encode("utf-8")).hexdigest()
 
 
 def get_encoding(html=None, headers=None):
     try:
         import chardet
+
         if html:
-            encoding = chardet.detect(html).get('encoding')
+            encoding = chardet.detect(html).get("encoding")
             return encoding
     except ImportError:
         pass
     if headers:
-        content_type = headers.get('content-type')
+        content_type = headers.get("content-type")
         try:
-            encoding = content_type.split(' ')[1].split('=')[1]
+            encoding = content_type.split(" ")[1].split("=")[1]
             return encoding
         except IndexError:
             pass
@@ -275,7 +280,7 @@ def iter_multi_items(mapping):
             yield item
 
 
-def url_quote(string, charset='utf-8', errors='strict', safe='/:', unsafe=''):
+def url_quote(string, charset="utf-8", errors="strict", safe="/:", unsafe=""):
     """
     URL encode a single string with a given encoding.
 
@@ -302,12 +307,12 @@ def url_quote(string, charset='utf-8', errors='strict', safe='/:', unsafe=''):
         if char in safe:
             rv.append(char)
         else:
-            rv.extend(('%%%02X' % char).encode('ascii'))
+            rv.extend(("%%%02X" % char).encode("ascii"))
     return to_native(bytes(rv))
 
 
-def url_quote_plus(string, charset='utf-8', errors='strict', safe=''):
-    return url_quote(string, charset, errors, safe + ' ', '+').replace(' ', '+')
+def url_quote_plus(string, charset="utf-8", errors="strict", safe=""):
+    return url_quote(string, charset, errors, safe + " ", "+").replace(" ", "+")
 
 
 def _url_encode_impl(obj, charset, encode_keys, sort, key):
@@ -321,40 +326,40 @@ def _url_encode_impl(obj, charset, encode_keys, sort, key):
             key = text_type(key).encode(charset)
         if not isinstance(value, bytes):
             value = text_type(value).encode(charset)
-        yield url_quote_plus(key) + '=' + url_quote_plus(value)
+        yield url_quote_plus(key) + "=" + url_quote_plus(value)
 
 
-def url_encode(obj, charset='utf-8', encode_keys=False, sort=False, key=None,
-               separator=b'&'):
-    separator = to_native(separator, 'ascii')
+def url_encode(
+    obj, charset="utf-8", encode_keys=False, sort=False, key=None, separator=b"&"
+):
+    separator = to_native(separator, "ascii")
     return separator.join(_url_encode_impl(obj, charset, encode_keys, sort, key))
 
 
 class WeixiErrorParser(html_parser.HTMLParser):
-
     def __init__(self):
         html_parser.HTMLParser.__init__(self)
         self.recording = 0
         self.data = []
 
     def handle_starttag(self, tag, attrs):
-        if tag != 'h4':
+        if tag != "h4":
             return
         if self.recording:
             self.recording += 1
         self.recording = 1
 
     def handle_endtag(self, tag):
-        if tag == 'h4' and self.recording:
+        if tag == "h4" and self.recording:
             self.recording -= 1
 
     def handle_data(self, data):
         if self.recording:
             self.data.append(data)
 
 
-def error_parser(error_html, encoding='gbk'):
-    html = text_type(error_html, encoding or 'gbk')
+def error_parser(error_html, encoding="gbk"):
+    html = text_type(error_html, encoding or "gbk")
     error_parser = WeixiErrorParser()
     error_parser.feed(html)
     if error_parser.data: