python-使用http快速上传下载文件
·
python-使用http快速上传下载文件
日常工作中，日常给服务器上传下载文件主要通过sftp的方式，但是无奈总是被安全部门管制，只能放弃使用ssh，而使用http的方式还是比较通用的。
1. 下载文件
从服务器下载文件，可以在服务器上面起个httpserver下载即可，这个简单
a. 通过python模块
#cd /download
#python3 -m http.server 80
b. 起个nginx
#systemctl restart nginx
2. 上传文件
通过 开发python 服务
#python3 http_upload_server 80
http_upload_server 代码如下：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from http.server import HTTPServer, BaseHTTPRequestHandler
import cgi
import os
import shutil
import json
from datetime import datetime
from urllib.parse import urlparse, parse_qs
class FileUploadHandler(BaseHTTPRequestHandler):
# 存储上传文件的目录
UPLOAD_DIR = "uploads"
def do_GET(self):
"""处理GET请求，显示简单的上传页面和测试接口"""
parsed_path = urlparse(self.path)
if parsed_path.path == '/':
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
html = f"""
<html>
<head><title>File Upload Server (Python3)</title></head>
<body>
<h1>File Upload Server - Python3</h1>
<p>使用curl上传文件:</p>
<pre>curl -F "file=@yourfile.txt" http://{self.server.server_name}:{self.server.server_port}/upload</pre>
<h3>测试上传接口</h3>
<form action="/test_upload" method="post" enctype="multipart/form-data">
<input type="file" name="file">
<input type="submit" value="Test Upload">
</form>
<h3>查看已上传文件</h3>
<a href="/files">查看文件列表</a>
</body>
</html>
"""
self.wfile.write(html.encode('utf-8'))
elif parsed_path.path == '/files':
self.show_file_list()
else:
self.send_error(404, "File not found")
def do_POST(self):
"""处理POST请求"""
parsed_path = urlparse(self.path)
if parsed_path.path == '/upload':
self.handle_upload()
elif parsed_path.path == '/test_upload':
self.handle_test_upload()
else:
self.send_error(404, "Path not found")
def handle_upload(self):
"""处理文件上传"""
try:
# 创建上传目录
if not os.path.exists(self.UPLOAD_DIR):
os.makedirs(self.UPLOAD_DIR)
# 解析multipart/form-data
content_type = self.headers.get('content-type')
if not content_type or 'multipart/form-data' not in content_type:
self.send_error(400, "Only multipart/form-data supported")
return
form = cgi.FieldStorage(
fp=self.rfile,
headers=self.headers,
environ={'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': content_type}
)
# 检查文件字段
if 'file' not in form:
self.send_error(400, "No file field found")
return
file_item = form['file']
if not file_item.file:
self.send_error(400, "Invalid file")
return
# 获取文件名
filename = file_item.filename or "uploaded_file"
filename = os.path.basename(filename)
# 生成安全文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
safe_filename = f"{timestamp}_{filename}"
file_path = os.path.join(self.UPLOAD_DIR, safe_filename)
# 保存文件
with open(file_path, 'wb') as f:
if file_item.file:
shutil.copyfileobj(file_item.file, f)
file_size = os.path.getsize(file_path)
# 返回成功响应
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
response = {
'status': 'success',
'filename': filename,
'saved_as': safe_filename,
'size': file_size,
'path': file_path,
'timestamp': datetime.now().isoformat()
}
self.wfile.write(json.dumps(response).encode('utf-8'))
# 日志
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] File uploaded: {filename} ({file_size} bytes) from {self.client_address[0]}")
except Exception as e:
self.send_error(500, f"Upload failed: {str(e)}")
print(f"Upload error: {e}")
def handle_test_upload(self):
"""处理测试上传（使用requests模拟）"""
try:
# 这里可以添加requests上传测试逻辑
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
response = {
'status': 'test_success',
'message': 'Test endpoint ready for requests client testing',
'timestamp': datetime.now().isoformat()
}
self.wfile.write(json.dumps(response).encode('utf-8'))
except Exception as e:
self.send_error(500, f"Test upload failed: {str(e)}")
def show_file_list(self):
"""显示已上传文件列表"""
try:
if not os.path.exists(self.UPLOAD_DIR):
files = []
else:
files = os.listdir(self.UPLOAD_DIR)
files = [f for f in files if os.path.isfile(os.path.join(self.UPLOAD_DIR, f))]
files.sort(key=lambda x: os.path.getmtime(os.path.join(self.UPLOAD_DIR, x)), reverse=True)
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
html = f"""
<html>
<head><title>Uploaded Files</title></head>
<body>
<h1>已上传文件列表</h1>
<p>文件数量: {len(files)}</p>
<ul>
"""
for file in files:
file_path = os.path.join(self.UPLOAD_DIR, file)
file_size = os.path.getsize(file_path)
file_time = datetime.fromtimestamp(os.path.getmtime(file_path)).strftime('%Y-%m-%d %H:%M:%S')
html += f'<li>{file} ({file_size} bytes) - {file_time}</li>'
html += """
</ul>
<p><a href="/">返回首页</a></p>
</body>
</html>
"""
self.wfile.write(html.encode('utf-8'))
except Exception as e:
self.send_error(500, f"Error reading file list: {str(e)}")
def log_message(self, format, *args):
"""自定义日志格式"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"{self.client_address[0]} - - [{timestamp}] {format % args}")
def run_server(port=8000):
"""启动HTTP服务器"""
server_address = ('', port)
httpd = HTTPServer(server_address, FileUploadHandler)
print(f"File Upload Server (Python3) running on port {port}")
print(f"Upload directory: {FileUploadHandler.UPLOAD_DIR}")
print("\nUpload test commands:")
print(f" curl -F \"file=@filename.txt\" http://localhost:{port}/upload")
print(f" curl -F \"file=@filename.txt\" http://your-ip:{port}/upload")
print("\nPress Ctrl+C to stop the server")
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\nServer stopped")
if __name__ == '__main__':
import sys
port = 8000
if len(sys.argv) > 1:
try:
port = int(sys.argv[1])
except ValueError:
print("Invalid port number, using default port 8000")
run_server(port)
客户端通过curl即可上传
curl -F "file=@filename.txt" http://server-ip:{port}/upload
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
3
0- 0
已为社区贡献1条内容
所有评论(0)