#!/usr/bin/env python3
"""简易文件上传服务 - 浏览器打开即可拖拽上传"""
import os
import cgi
import html
from http.server import HTTPServer, BaseHTTPRequestHandler
UPLOAD_DIR = os.path.expanduser("~/.openclaw/workspace-xiaoban/tmp/uploads")
PORT = 18888
HTML = r"""
上传文件到大麦
"""
class UploadHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == '/':
self._send_html(HTML)
elif self.path == '/files':
self._list_files()
else:
self.send_error(404)
def do_POST(self):
if self.path == '/upload':
self._handle_upload()
else:
self.send_error(404)
def _send_html(self, content, code=200):
body = content.encode('utf-8')
self.send_response(code)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Content-Length', len(body))
self.end_headers()
self.wfile.write(body)
def _list_files(self):
import json
import time
os.makedirs(UPLOAD_DIR, exist_ok=True)
files = []
for name in sorted(os.listdir(UPLOAD_DIR), reverse=True):
path = os.path.join(UPLOAD_DIR, name)
if os.path.isfile(path):
st = os.stat(path)
files.append({
'name': name,
'size': st.st_size,
'time': time.strftime('%m-%d %H:%M', time.localtime(st.st_mtime))
})
body = json.dumps(files[:20]).encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Content-Length', len(body))
self.end_headers()
self.wfile.write(body)
def _handle_upload(self):
content_type = self.headers.get('Content-Type', '')
if 'multipart/form-data' not in content_type:
self.send_error(400, '需要 multipart/form-data')
return
form = cgi.FieldStorage(
fp=self.rfile,
headers=self.headers,
environ={'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': content_type}
)
file_item = form['file']
if not file_item.filename:
self.send_error(400, '未选择文件')
return
safe_name = os.path.basename(file_item.filename)
os.makedirs(UPLOAD_DIR, exist_ok=True)
path = os.path.join(UPLOAD_DIR, safe_name)
# 避免覆盖，自动加序号
base, ext = os.path.splitext(safe_name)
counter = 1
while os.path.exists(path):
path = os.path.join(UPLOAD_DIR, f"{base}_{counter}{ext}")
counter += 1
with open(path, 'wb') as f:
f.write(file_item.file.read())
self.send_response(200)
self.send_header('Content-Type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(f'OK: {os.path.basename(path)}'.encode('utf-8'))
def log_message(self, format, *args):
pass # 静默日志
if __name__ == '__main__':
os.makedirs(UPLOAD_DIR, exist_ok=True)
server = HTTPServer(('0.0.0.0', PORT), UploadHandler)
print(f'上传服务已启动: http://115.190.225.235:{PORT}')
print(f'上传目录: {UPLOAD_DIR}')
try:
server.serve_forever()
except KeyboardInterrupt:
print('\n已停止')
server.server_close()