Python 极速文件服务器

import io import json import mimetypes import os import shutil import socket import socketserver # 引入 socketserver 用于多线程 import urllib.parse import zipfile from http.server import BaseHTTPRequestHandler, HTTPServer # --- 配置 --- PORT = 8000 BASE_DIR = os.path.abspath(os.getcwd()) BUFFER_SIZE = 4 * 1024 * 1024 def parse_byte_range(range_header, file_size): if not range_header or not range_header.startswith("bytes=") or file_size <= 0: return None range_value = range_header.removeprefix("bytes=").split(",", 1)[0].strip() if "-" not in range_value: return None start_text, end_text = range_value.split("-", 1) try: if start_text == "": suffix_length = int(end_text) if suffix_length <= 0: return None start = max(file_size - suffix_length, 0) end = file_size - 1 else: start = int(start_text) end = int(end_text) if end_text else file_size - 1 except ValueError: return None if start < 0 or end < start or start >= file_size: return None return start, min(end, file_size - 1) def copy_file_range(source, target, length): remaining = length while remaining > 0: chunk = source.read(min(BUFFER_SIZE, remaining)) if not chunk: break target.write(chunk) remaining -= len(chunk) # --- 前端 HTML/CSS/JS --- HTML_CONTENT = """ Python 极速文件服务器

点击选择或拖拽多个文件到此处

支持大文件极速流式上传，无限大小

已选中 0 项

	文件名称	大小	操作

正在准备... 0%

""" # 核心修复 1: 引入 socketserver.ThreadingMixIn，实现多线程 HTTP 服务 # 彻底解决现代浏览器建立多个连接导致的"单线程阻塞假死"问题 class ThreadingHTTPServer(socketserver.ThreadingMixIn, HTTPServer): daemon_threads = True # 确保主进程退出时，子线程自动结束 class FastHTTPServer(ThreadingHTTPServer): def server_bind(self): super().server_bind() # 核心修复 3: 增加容错，避免部分 OS 拒绝底层网络配置导致服务器启动失败 try: self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8 * 1024 * 1024) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 8 * 1024 * 1024) self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) except Exception: pass # 如果系统不支持强行接管 TCP 栈，则静默忽略，不影响核心功能 class SimpleFileManager(BaseHTTPRequestHandler): def get_real_path(self, req_path): decoded_path = urllib.parse.unquote(req_path) decoded_path = os.path.normpath(decoded_path) if os.path.isabs(decoded_path): return decoded_path return os.path.abspath(os.path.join(BASE_DIR, decoded_path)) def do_GET(self): parsed = urllib.parse.urlparse(self.path) query = urllib.parse.parse_qs(parsed.query) if parsed.path == "/": self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() # 使用 json.dumps 安全转义路径，防止路径中的单双引号导致 JS 崩溃 safe_initial_path = json.dumps(BASE_DIR.replace("\\", "/")) init_html = HTML_CONTENT.replace("{{INITIAL_PATH}}", safe_initial_path) self.wfile.write(init_html.encode("utf-8")) elif parsed.path == "/api/list": target_dir = self.get_real_path(query.get("path", ["/"])[0]) if not os.path.exists(target_dir) or not os.path.isdir(target_dir): self.send_json({"error": "该目录不存在或无法访问"}) return files = [] try: for item in os.listdir(target_dir): if item.startswith("."): continue full = os.path.join(target_dir, item) is_dir = os.path.isdir(full) size = os.path.getsize(full) if not is_dir else 0 files.append( { "name": item, "is_dir": is_dir, "size": self.human_size(size) if not is_dir else "", } ) files.sort(key=lambda x: (not x["is_dir"], x["name"].lower())) self.send_json({"path": target_dir.replace("\\", "/"), "files": files}) except Exception as e: self.send_json({"error": f"没有权限读取该目录: {str(e)}"}) elif parsed.path == "/api/download": file_path = self.get_real_path(query.get("path", [""])[0]) self.serve_file(file_path) elif parsed.path == "/api/zip": base_path = self.get_real_path(query.get("path", ["/"])[0]) file_names = query.get("files", [""])[0].split(",") zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: for name in file_names: full_path = os.path.join(base_path, name) if os.path.isfile(full_path): zf.write(full_path, name) zip_buffer.seek(0) self.send_response(200) self.send_header("Content-Type", "application/zip") self.send_header( "Content-Disposition", 'attachment; filename="batch_download.zip"' ) self.end_headers() self.wfile.write(zip_buffer.read()) else: # 补齐缺漏的路由回退，防止游览器请求如 /favicon.ico 时连接挂起 self.send_error(404, "Not Found") def do_POST(self): if self.path == "/api/upload": try: name = urllib.parse.unquote(self.headers.get("X-File-Name", "unknown")) target_dir = self.get_real_path(self.headers.get("X-Target-Dir", "/")) os.makedirs(target_dir, exist_ok=True) save_path = os.path.join(target_dir, name) content_length = int(self.headers.get("Content-Length", 0)) with open(save_path, "wb") as f: remaining = content_length while remaining > 0: chunk = self.rfile.read(min(remaining, BUFFER_SIZE)) if not chunk: break f.write(chunk) remaining -= len(chunk) self.send_json({"status": "ok"}) except Exception as e: print(f"上传异常: {e}") self.send_error(500) def serve_file(self, path): if os.path.isfile(path): file_size = os.path.getsize(path) byte_range = parse_byte_range(self.headers.get("Range"), file_size) status_code = 206 if byte_range else 200 start, end = byte_range if byte_range else (0, file_size - 1) content_length = end - start + 1 if file_size else 0 self.send_response(status_code) mime, _ = mimetypes.guess_type(path) self.send_header("Content-Type", mime or "application/octet-stream") self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Length", str(content_length)) if byte_range: self.send_header("Content-Range", f"bytes {start}-{end}/{file_size}") safe_filename = urllib.parse.quote(os.path.basename(path)) self.send_header( "Content-Disposition", f"attachment; filename*=UTF-8''{safe_filename}" ) self.end_headers() try: with open(path, "rb") as f: if byte_range: f.seek(start) copy_file_range(f, self.wfile, content_length) else: shutil.copyfileobj(f, self.wfile, length=BUFFER_SIZE) except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): return else: self.send_error(404) def send_json(self, data): self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps(data).encode("utf-8")) def human_size(self, n): for unit in ["B", "KB", "MB", "GB"]: if n < 1024: return f"{n:.1f} {unit}" n /= 1024 return f"{n:.1f} TB" def log_message(self, format, *args): pass if __name__ == "__main__": hostname = socket.gethostname() try: local_ip = socket.gethostbyname(hostname) except: local_ip = "127.0.0.1" server_address = ("0.0.0.0", PORT) httpd = FastHTTPServer(server_address, SimpleFileManager) print("=" * 45) print(f"🚀 Python 极速文件服务器 (多线程版) 已启动！") print(f"🔗 本机访问: http://localhost:{PORT}") print(f"🌐 局域网访问: http://{local_ip}:{PORT}") print(f"🛑 终止服务请在终端按 Ctrl+C") print("=" * 45) try: httpd.serve_forever() except KeyboardInterrupt: print("\n👋 服务已停止。")