需求：对方提供处理文件的接口，本地将待处理文件压缩后，通过http post multipart方式上传，等待处理完成后从相应连接下载结果
代码
import os import time import zipfile import requests from requests_toolbelt.multipart.encoder import MultipartEncoder class Func4Fuxi(object): def __init__(self): self.remote_result = 0 # 压缩文件 def zip_dir(self, dirname, zipfilename): filelist = [] if os.path.isfile(dirname): filelist.append(dirname) else: for root, dirs, files in os.walk(dirname): for name in files: filelist.append(os.path.join(root, name)) zf = zipfile.ZipFile(zipfilename, mode="w", compression=zipfile.zlib.DEFLATED, allowZip64=True) for tar in filelist: arcname = tar[len(dirname):] zf.write(tar, arcname) zf.close() # 解压文件 def unzip_file(self, zipfilename, unziptodir): if not os.path.exists(unziptodir): os.mkdir(unziptodir) zfobj = zipfile.ZipFile(zipfilename) for name in zfobj.namelist(): name = name.replace('\\', '/') if name.endswith('/'): os.mkdir(os.path.join(unziptodir, name)) else: ext_filename = os.path.join(unziptodir, name) ext_dir = os.path.dirname(ext_filename) if not os.path.exists(ext_dir): os.mkdir(ext_dir) outfile = open(ext_filename, 'wb') outfile.write(zfobj.read(name)) outfile.close() # 下载 def download_result(self, filename): filename.replace('\\', '/') file = filename.split('/')[-1] URL = '--------------' re = requests.get(URL+'?name='+file, stream=True) self.remote_result = re.status_code if self.remote_result == 200: print("find result,try to download") f = open("download_"+file, "wb") for chunk in re.iter_content(chunk_size=512): if chunk: f.write(chunk) print("download result success") return self.remote_result # 上传 def upload_zip(self, filename): self.remote_result = 0 filename.replace('\\', '/') file = filename.split('/')[-1] file_tup = (file, open(filename, 'rb'), 'application/zip') URL = '-----------------' #fields属性根据对方接口说明设置 m = MultipartEncoder( fields={'name': file, 'zipfile': file_tup} ) re = requests.post(URL, data=m, headers={'Content-Type': m.content_type}) self.remote_result = re.status_code if self.remote_result == 200: print("upload success") else: print("upload failed") return self.remote_result
