Falsk 大文件上传/下载(send_from_directory)
2024-10-21 15:39:41
下载接口:
服务端flask下载接口
@app.route("/api/download/", methods=["POST"])
def download():
try:
logger.debug("download start")
param = request.get_json(force=True).get('param')
logger.debug("download({})".format(param))
file_path = param.get('file_path')
file_name = param.get('file_name')
base_dir = u"C:\\Program Files\\download"
file_dir = os.path.join(base_dir, file_path)
file_abs_path = os.path.join(file_dir, file_name)
logger.debug("send_from_directory file_dir ({}) ,file_name({})".format(file_dir, file_name))
response_file = send_from_directory(file_dir, filename=file_name, as_attachment=False)
response = make_response(response_file)
response.headers["Content-Disposition"] = "attachment; filename={}".format(file_name.encode().decode("latin-1"))
response.headers["Content-Length"] = os.stat(file_abs_path).st_size
response.headers["Content-Type"] = "application/octet-stream"
return response
except:
logger.debug("download failed ({})".format(traceback.format_exc()))
response = {"status": 9999}
return jsonify(response)
客户端下载接口
url = "https://{}:{}/api/download/".format(server_ip, get_https_port())
data = {"param": {
"file_path": file_path,
"file_name": file_name
}}
save_to = "/home" try:
logger.debug("download, url[%s] data:%s" % (url, data))
download_headers = {"Content-Type": "application/json"}
res = requests.post(url=url, headers=download_headers, data=json.dumps(data), verify=False)
logger.debug('download response status[%s] headers:%s' % (res.status_code, res.headers)) if not os.path.exists(save_to):
os.makedirs(save_to) save_path = os.path.join(save_to, file_name) with open(save_path, "wb") as fp:
for chunk in res.iter_content(1024):
if not chunk:
break
fp.write(chunk)
上传接口:
服务接口上传文件接口
@app.route("/api/upload/", methods=["POST"])
def upload():
if request.method == "POST":
try:
logger.debug("upload vars ({})".format(request.files))
file = request.files['file']
content = file.read()
# bdecode(content) path = "/home"
file_path = os.path.join(path, file_name)
if os.path.exists(file_path):
os.remove(file_path) with open(file_path, "wb") as f:
f.write(content)
f.flush()
f.close()
return jsonify({"status": 0, "data": []})
except:
logger.debug("save upload file fail ({})".format(traceback.format_exc()))
return jsonify({"status": 1, "data": []})
客户端上传接口:
url = "https://{}:{}/api/download/".format("192.168.11.200", "8000")
file_path = "/home/dddd/aaa.txt"
file_name = os.path.split(file_path)[-1]
send_data = {}
headers = {
'Connection': 'keep-alive',
'User-Agent': 'P2pclient with Linuxos',
'Accept-Language': 'en-US,en;q=0.5',
'Accept-Encoding': 'gzip, deflate',
}
if not send_data:
with open(file_path, 'rb')as fp:
file_list = {'file': (file_name, fp, ''), "uuid": timestamp_uuid, "sign": sign,
"Authorization": "token"}
response = requests.post(url, headers=headers, timeout=15,
files=file_list, verify=False)
else:
with open(file_path, 'rb')as fp:
file_list = {'file': (file_name, fp, ''), "uuid": timestamp_uuid, "sign": sign,
"Authorization": "token"}
response = requests.post(url, headers=headers, timeout=15,
data=send_data, files=file_list, verify=False)
logger.debug('upload {} to {} response {}'.format(file_path, url, response.__dict__))
data = response.json()
if data.get('status') == 0:
logger.debug('upload {} to {} successed!'.format(file_path, url))
return True
else:
logger.debug('upload {} to {} failed, try again ... {}'.format(file_path, url, i))
return False
最新文章
- 新手如何在gdb中存活
- .htaccess 基础教程(二)
- JavaScript中String的math方法与RegExp的exec方法的区别
- try,catch,throw-----C++
- iOS开发——源代码管理——svn 命令行下常用的几个命令
- C#访问postgresql数据库
- Chart For Asp.Net ----Overview
- UML视图(九)部署图
- hdu 4800 Josephina and RPG
- 数据库导出到excel
- 二、第一个ExtJS程序:helloExtJS
- Webuploader 大文件分片上传
- .net Mongo Driver 1.0与2.0的对比与2.0的优化
- linux 上 mysql 的使用
- InnoDB中锁的算法(3)
- python for dl
- 【python】try...except...后中断程序继续运行
- Cisco 4507R+E四引擎VSS故障解决
- 百度云盘下载插进-油猴Tampermonkey
- [LeetCode] Read N Characters Given Read4 I &; II