之前介绍了 droppy 文件共享服务的搭建。但在一些场景中,我们需要在命令行或在 Python 代码中,临时上传和下载文件。这时可以用一个更简单的策略:使用 flask 编写一个临时的 API。
服务端配置
以下是一个简单的 Flask 应用程序代码示例,用于处理文件的上传和下载。
# server.py
from flask import Flask, request, send_from_directory, jsonify
import os
# 创建 'uploads' 目录,如果不存在
os.makedirs('uploads', exist_ok=True)
app = Flask(__name__)
@app.route('/upload', methods=['POST'])
def upload_file():
if 'file' not in request.files:
return 'No file part', 400
file = request.files['file']
if file.filename == '':
return 'No selected file', 400
# 保存文件到服务器
file.save(os.path.join('uploads', file.filename))
return 'File uploaded successfully', 200
@app.route('/files', methods=['GET'])
def list_files():
# 获取 'uploads' 目录中的文件列表
files = os.listdir('uploads')
return jsonify(files), 200
@app.route('/download/<filename>', methods=['GET'])
def download_file(filename):
# 下载指定的文件
try:
return send_from_directory('uploads', filename, as_attachment=True)
except FileNotFoundError:
return 'File not found', 404
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
运行 python server.py
命令启动服务。
客户端交互
在代码中上传:将 file_path
填写为实际需要上传的文件,并相应修改 host
地址。
import requests
# 要上传的文件路径
file_path = 'path_to_your_file.txt'
with open(file_path, 'rb') as f:
files = {'file': f}
response = requests.post('http://server1_ip:5000/upload', files=files)
print(response.text)
也可以写成 Python 脚本形式,编写命令行工具 client.py
,这里包括三个函数:
upload
上传文件list_files
查看文件列表download
下载文件
完整代码:
import requests
import click
from loguru import logger
@click.group()
def cli():
"""文件管理客户端."""
pass
@cli.command()
@click.option('--file', 'file_path', required=True, type=click.Path(exists=True), help='要上传的文件路径')
@click.option('--server-host', 'server_host', required=True, type=str, help='服务器的主机地址(包含IP和端口)')
def upload(file_path, server_host):
"""上传文件到指定服务器."""
url = f"http://{server_host}/upload"
logger.info(f"正在上传 {file_path} 到 {url}")
with open(file_path, 'rb') as f:
files = {'file': f}
try:
response = requests.post(url, files=files)
response.raise_for_status()
logger.info(f"服务器返回: {response.text}")
except requests.RequestException as e:
logger.error(f"发生错误: {e}")
@cli.command()
@click.option('--server-host', 'server_host', required=True, type=str, help='服务器的主机地址(包含IP和端口)')
def list_files(server_host):
"""查看服务器上的文件列表."""
url = f"http://{server_host}/files"
logger.info(f"正在从 {url} 检索文件列表")
try:
response = requests.get(url)
response.raise_for_status()
files = response.json()
if files:
logger.info("服务器上的文件列表:")
for file in files:
logger.info(f"- {file}")
else:
logger.info("服务器上没有可用文件。")
except requests.RequestException as e:
logger.error(f"发生错误: {e}")
@cli.command()
@click.option('--filename', 'filename', required=True, type=str, help='要下载的文件名')
@click.option('--server-host', 'server_host', required=True, type=str, help='服务器的主机地址(包含IP和端口)')
def download(filename, server_host):
"""从服务器下载文件."""
url = f"http://{server_host}/download/{filename}"
logger.info(f"正在下载 {filename} 从 {url}")
try:
response = requests.get(url, stream=True)
response.raise_for_status()
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
logger.info(f"{filename} 下载成功")
except requests.RequestException as e:
logger.error(f"发生错误: {e}")
if __name__ == '__main__':
cli()
使用示例
在命令行中执行以下命令来上传、查看和下载文件:
上传文件:
python client.py upload --file path_to_your_file --server-host localhost:5000
查看文件列表:
python client.py list_files --server-host localhost:5000
下载文件:
python client.py download --filename file_name --server-host localhost:5000
以上,我们用 Python 的 flask 和 click 搭建一个简单的命令行文件共享服务。