改造python3中的http.server为简单的文件上传下载服务
改造
修改python3中的http.server.SimpleHTTPRequestHandler，实现简单的文件上传下载服务
simple_http_file_server.py：
# !/usr/bin/env python3import datetime
import email
import html
import http.server
import io
import mimetypes
import os
import posixpath
import re
import shutil
import sys
import urllib.error
import urllib.parse
import urllib.request
import socket
from http import HTTPStatus
import threading
import contextlib__version__ = "0.1"
__all__ = ["MySimpleHTTPRequestHandler"]class MySimpleHTTPRequestHandler(http.server.BaseHTTPRequestHandler):server_version = "SimpleHTTP/" + __version__extensions_map = _encodings_map_default = {'.gz': 'application/gzip','.Z': 'application/octet-stream','.bz2': 'application/x-bzip2','.xz': 'application/x-xz',}def __init__(self, *args, directory=None, **kwargs):if directory is None:directory = os.getcwd()self.directory = os.fspath(directory)super().__init__(*args, **kwargs)def do_GET(self):f = self.send_head()if f:try:self.copyfile(f, self.wfile)finally:f.close()def do_HEAD(self):f = self.send_head()if f:f.close()def send_head(self):path = self.translate_path(self.path)f = Noneif os.path.isdir(path):parts = urllib.parse.urlsplit(self.path)if not parts.path.endswith('/'):# redirect browser - doing basically what apache doesself.send_response(HTTPStatus.MOVED_PERMANENTLY)new_parts = (parts[0], parts[1], parts[2] + '/',parts[3], parts[4])new_url = urllib.parse.urlunsplit(new_parts)self.send_header("Location", new_url)self.end_headers()return Nonefor index in "index.html", "index.htm":index = os.path.join(path, index)if os.path.exists(index):path = indexbreakelse:return self.list_directory(path)ctype = self.guess_type(path)if path.endswith("/"):self.send_error(HTTPStatus.NOT_FOUND, "File not found")return Nonetry:f = open(path, 'rb')except OSError:self.send_error(HTTPStatus.NOT_FOUND, "File not found")return Nonetry:fs = os.fstat(f.fileno())# Use browser cache if possibleif ("If-Modified-Since" in self.headersand "If-None-Match" not in self.headers):# compare If-Modified-Since and time of last file modificationtry:ims = email.utils.parsedate_to_datetime(self.headers["If-Modified-Since"])except (TypeError, IndexError, OverflowError, ValueError):# ignore ill-formed valuespasselse:if ims.tzinfo is None:# obsolete format with no timezone, cf.# https://tools.ietf.org/html/rfc7231#section-7.1.1.1ims = ims.replace(tzinfo=datetime.timezone.utc)if ims.tzinfo is datetime.timezone.utc:# compare to UTC datetime of last modificationlast_modif = datetime.datetime.fromtimestamp(fs.st_mtime, datetime.timezone.utc)# remove microseconds, like in If-Modified-Sincelast_modif = last_modif.replace(microsecond=0)if last_modif <= ims:self.send_response(HTTPStatus.NOT_MODIFIED)self.end_headers()f.close()return Noneself.send_response(HTTPStatus.OK)self.send_header("Content-type", ctype)self.send_header("Content-Length", str(fs[6]))self.send_header("Last-Modified",self.date_time_string(fs.st_mtime))self.end_headers()return fexcept:f.close()raisedef list_directory(self, path):try:list_dir = os.listdir(path)except OSError:self.send_error(HTTPStatus.NOT_FOUND, "No permission to list_dir directory")return Nonelist_dir.sort(key=lambda a: a.lower())r = []try:display_path = urllib.parse.unquote(self.path, errors='surrogatepass')except UnicodeDecodeError:display_path = urllib.parse.unquote(path)display_path = html.escape(display_path, quote=False)enc = sys.getfilesystemencoding()form = """<h1>文件上传</h1>\n<form ENCTYPE="multipart/form-data" method="post">\n<input name="file" type="file"/>\n<input type="submit" value="upload"/>\n</form>\n"""title = 'Directory listing for %s' % display_pathr.append('<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01//EN" ''"http://www.w3.org/TR/html4/strict.dtd">')r.append('<html>\n<head>')r.append('<meta http-equiv="Content-Type" ''content="text/html; charset=%s">' % enc)r.append('<title>%s</title>\n</head>' % title)r.append('<body>%s\n<h1>%s</h1>' % (form, title))r.append('<hr>\n<ul>')for name in list_dir:fullname = os.path.join(path, name)displayname = linkname = name# Append / for directories or @ for symbolic linksif os.path.isdir(fullname):displayname = name + "/"linkname = name + "/"if os.path.islink(fullname):displayname = name + "@"# Note: a link to a directory displays with @ and links with /r.append('<li><a href="%s">%s</a></li>' % (urllib.parse.quote(linkname, errors='surrogatepass'),html.escape(displayname, quote=False)))r.append('</ul>\n<hr>\n</body>\n</html>\n')encoded = '\n'.join(r).encode(enc, 'surrogate escape')f = io.BytesIO()f.write(encoded)f.seek(0)self.send_response(HTTPStatus.OK)self.send_header("Content-type", "text/html; charset=%s" % enc)self.send_header("Content-Length", str(len(encoded)))self.end_headers()return fdef translate_path(self, path):# abandon query parameterspath = path.split('?', 1)[0]path = path.split('#', 1)[0]# Don't forget explicit trailing slash when normalizing. Issue17324trailing_slash = path.rstrip().endswith('/')try:path = urllib.parse.unquote(path, errors='surrogatepass')except UnicodeDecodeError:path = urllib.parse.unquote(path)path = posixpath.normpath(path)words = path.split('/')words = filter(None, words)path = self.directoryfor word in words:if os.path.dirname(word) or word in (os.curdir, os.pardir):# Ignore components that are not a simple file/directory namecontinuepath = os.path.join(path, word)if trailing_slash:path += '/'return pathdef copyfile(self, source, outputfile):shutil.copyfileobj(source, outputfile)def guess_type(self, path):base, ext = posixpath.splitext(path)if ext in self.extensions_map:return self.extensions_map[ext]ext = ext.lower()if ext in self.extensions_map:return self.extensions_map[ext]guess, _ = mimetypes.guess_type(path)if guess:return guessreturn 'application/octet-stream'def do_POST(self):r, info = self.deal_post_data()self.log_message('%s, %s => %s' % (r, info, self.client_address))enc = sys.getfilesystemencoding()res = ['<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01//EN" ''"http://www.w3.org/TR/html4/strict.dtd">','<html>\n<head>','<meta http-equiv="Content-Type" content="text/html; charset=%s">' % enc,'<title>%s</title>\n</head>' % "Upload Result Page",'<body><h1>%s</h1>\n' % "Upload Result"]if r:res.append('<p>SUCCESS: %s</p>\n' % info)else:res.append('<p>FAILURE: %s</p>' % info)res.append('<a href=\"%s\">back</a>' % self.headers['referer'])res.append('</body></html>')encoded = '\n'.join(res).encode(enc, 'surrogate escape')f = io.BytesIO()f.write(encoded)length = f.tell()f.seek(0)self.send_response(200)self.send_header("Content-type", "text/html")self.send_header("Content-Length", str(length))self.end_headers()if f:self.copyfile(f, self.wfile)f.close()def deal_post_data(self):content_type = self.headers['content-type']if not content_type:return False, "Content-Type header doesn't contain boundary"boundary = content_type.split("=")[1].encode()remain_bytes = int(self.headers['content-length'])line = self.rfile.readline()remain_bytes -= len(line)if boundary not in line:return False, "Content NOT begin with boundary"line = self.rfile.readline()remain_bytes -= len(line)fn = re.findall(r'Content-Disposition.*name="file"; filename="(.*)"', line.decode())if not fn:return False, "Can't find out file name..."path = self.translate_path(self.path)fn = os.path.join(path, fn[0])line = self.rfile.readline()remain_bytes -= len(line)line = self.rfile.readline()remain_bytes -= len(line)try:out = open(fn, 'wb')except IOError:return False, "Can't create file to write, do you have permission to write?"preline = self.rfile.readline()remain_bytes -= len(preline)while remain_bytes > 0:line = self.rfile.readline()remain_bytes -= len(line)if boundary in line:preline = preline[0:-1]if preline.endswith(b'\r'):preline = preline[0:-1]out.write(preline)out.close()return True, "File '%s' upload success!" % fnelse:out.write(preline)preline = linereturn False, "Unexpect Ends of data."def _get_best_family(*address):infos = socket.getaddrinfo(*address,type=socket.SOCK_STREAM,flags=socket.AI_PASSIVE,)family, type, proto, canonname, sockaddr = next(iter(infos))return family, sockaddrdef serve_forever(port=8000, bind=None, directory="."):"""This runs an HTTP server on port 8000 (or the port argument)."""# ensure dual-stack is not disabled; ref #38907class DualStackServer(http.server.ThreadingHTTPServer):def server_bind(self):# suppress exception when protocol is IPv4with contextlib.suppress(Exception):self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)return super().server_bind()def finish_request(self, request, client_address):self.RequestHandlerClass(request, client_address, self,directory=directory)HandlerClass=MySimpleHTTPRequestHandlerServerClass=DualStackServerprotocol="HTTP/1.0"ServerClass.address_family, addr = _get_best_family(bind, port)HandlerClass.protocol_version = protocolwith ServerClass(addr, HandlerClass) as httpd:host, port = httpd.socket.getsockname()[:2]url_host = f'[{host}]' if ':' in host else hostprint(f"Serving HTTP on {host} port {port} "f"(http://{url_host}:{port}/) ...")try:httpd.serve_forever()except KeyboardInterrupt:print("\nKeyboard interrupt received, exiting.")sys.exit(0)if __name__ == '__main__':import argparseparser = argparse.ArgumentParser()parser.add_argument('--bind', '-b', metavar='ADDRESS',help='specify alternate bind address ''(default: all interfaces)')parser.add_argument('--directory', '-d', default=os.getcwd(),help='specify alternate directory ''(default: current directory)')parser.add_argument('port', action='store', default=8000, type=int,nargs='?',help='specify alternate port (default: 8000)')args = parser.parse_args()# 在主线程中执行# serve_forever(# port=args.port,# bind=args.bind,# directory = args.directory# )# 在子线程中执行，这样主线程可以执行异步工作thread1 = threading.Thread(name='t1',target= serve_forever,kwargs={"port" : args.port,"bind" : args.bind,"directory" : args.directory})thread1.start()# thread1.join()使用帮助：
>python3 simple_http_file_server.py --help
usage: simple_http_file_server.py [-h] [--cgi] [--bind ADDRESS] [--directory DIRECTORY] [port]positional arguments:
port specify alternate port (default: 8000)optional arguments:
-h, --help show this help message and exit
--cgi run as CGI server
--bind ADDRESS, -b ADDRESS
specify alternate bind address (default: all interfaces)
--directory DIRECTORY, -d DIRECTORY
specify alternate directory (default: current directory)
使用示例：
>python3 simple_http_file_server.py -d ~ 12345
Serving HTTP on :: port 12345 (http://[::]:12345/) ...
::ffff:127.0.0.1 - - [30/Nov/2023 09:35:06] "GET / HTTP/1.1" 200 -。。。
访问：
浏览器中访问：http://127.0.0.1:12345/
参考：
Python3 实现简单HTTP服务器（附带文件上传）_python3 -m http.server-CSDN博客
相关文章：
改造python3中的http.server为简单的文件上传下载服务
改造 修改python3中的http.server.SimpleHTTPRequestHandler，实现简单的文件上传下载服务 simple_http_file_server.py： # !/usr/bin/env python3import datetime import email import html import http.server import io import mimetypes import os …...
Fiddler抓包工具之fiddler的composer可以简单发送http协议的请求
一，composer的详解 右侧Composer区域，是测试接口的界面： 相关说明： 1.请求方式：点开可以勾选请求协议是get、post等 2.url地址栏：输入请求的url地址 3.请求头：第三块区域可以输入请求头信息…...
14、pytest像用参数一样使用fixture
官方实例 # content of test_fruit.py import pytestclass Fruit:def __init__(self, name):self.name nameself.cubed Falsedef cube(self):self.cubed Trueclass FruitSalad:def __init__(self, *fruit_bowl):self.fruit fruit_bowlself._cube_fruit()def _cube_fruit(s…...
C++ Primer Plus第十三章笔记
目录 基类 构造函数：访问权限的考虑 1.2 派生类和基类之间的特殊关系 继承：is-a关系 多态公有继承 静态联编和动态联编 指针和引用类型的兼容性 虚成员函数和动态联编 虚函数的注意事项 构造函数 析构函数 友元 没有重新定义 重新定义将隐…...
【JavaEE】单例模式
作者主页：paper jie_博客 本文作者：大家好，我是paper jie，感谢你阅读本文，欢迎一建三连哦。 本文于《JavaEE》专栏，本专栏是针对于大学生，编程小白精心打造的。笔者用重金(时间和精力)打造&…...
第十五届蓝桥杯模拟赛（第二期 C++）
俺自己做的噢，还未核实答案，若有差错，望斧正。 第一题 小蓝要在屏幕上放置一行文字，每个字的宽度相同。小蓝发现，如果每个字的宽为 36 像素，一行正好放下 30 个字，字符之间和前后都没有任何空隙…...
关于Unity中字典在Inspector的显示
字典在Inspector的显示 方法一：实现ISerializationCallbackReceiver接口 《unity3D游戏开发第二版》记录 在编辑面板中可以利用序列化监听接口特性对字典进行序列化。 主要继承ISerializationCallbackReceiver接口 实现OnAfterDeserialize() OnBeforeSerialize() …...
使用Plex结合cpolar搭建本地私人媒体站并实现远程访问
文章目录 1.前言2. Plex网站搭建2.1 Plex下载和安装2.2 Plex网页测试2.3 cpolar的安装和注册 3. 本地网页发布3.1 Cpolar云端设置3.2 Cpolar本地设置 4. 公网访问测试5. 结语 1.前言 用手机或者平板电脑看视频，已经算是生活中稀松平常的场景了，特别是各…...
svn合并冲突时每个选项的含义
合并冲突时每个选项的含义 - 这个图片是 TortoiseSVN（一个Subversion（SVN）客户端）的合并冲突解决对话框。当你尝试合并两个版本的文件并且出现差异时，你需要解决这些差异。这个对话框提供了几个选项来处理合并冲突&…...
指针、数组与函数例题3
1、字符串复制 题目描述 设计函数实现字符串复制功能，每个字符串长度不超过100，不要使用系统提供的strcpy函数 输入要求 从键盘读入一个字符串到数组b中，以换行符结束 输出要求 将内容复制到另一个数组a中，并分别输出数组a和…...
ThreeJs样例 webgl_shadow_contact 分析
webgl_shadow_contact 官方样例中，对阴影的渲染比较特殊，很值得借鉴，学习渲染阴影的思路；这个例子中对阴影的渲染，并没有使用任何光源，没有用shadowmap的常规方式 渲染阴影；而是使用了深度材质T…...
Nginx（缓冲区）
先来思考一个问题，接入Nginx的项目一般请求流程为：“客户端→Nginx→服务端”，在这个过程中存在两个连接：“客户端→Nginx、Nginx→服务端”，那么两个不同的连接速度不一致，就会影响用户的体验（…...
MQTT协议理解并实践
MQTT是一个轻量的发布订阅模式消息传输协议，专门针对低带宽和不稳定网络环境的物联网应用设计 MQTT协议根据主题来分发消息进行通信，支持通配符匹配，可以低开销的使用数百万Topic进行一对一，一对多双向通信。 协议特点 1. 开放…...
实现一个简单的网络通信下（udp）
时间过去好久了，先回忆一下上一篇博客的代码！！ 目前来看，我们客户端发一条消息，我服务器收到这一条消息之后呢，服务器也知道了是谁给我发来的消息，紧接这就把这条消息放进buffer当中，…...
Linux中office环境LibreOffice_7.6.2下载
阿里云盘：LibreOffice_7.6.2 使用：下载的文件为exe文件，双击exe文件即可获取到文件 LibreOffice_7.6.2安装： 解压：tar -zxvf LibreOffice_7.6.2_Linux_x86-64_rpm.tar.gz 移动到RPMS目录：cd LibreOffice_7…...
Linux快捷控制
Linux快捷控制 工具安装 yum -y install lrzsz wget curl net-tools git防火墙 systemctl status firewalld.service systemctl stop firewalld.service systemctl disable firewalld.service宝塔 yum install -y wget && wget -O install.sh https://download.bt.…...
免费插件集-illustrator插件-Ai插件-重复复制-单一对象页面排版
文章目录 1.介绍2.安装3.通过窗口>扩展>知了插件>重复复制4.总结 1.介绍 本文介绍一款免费插件，加强illustrator使用人员工作效率，进行制卡专用分层分色。首先从下载网址下载这款插件 https://download.csdn.net/download/m0_67316550/8789050…...
GO基础之变量与常量
标识符与关键字 标识符 在编程语言中标识符就是程序员定义的具有特殊意义的词，比如变量名、常量名、函数名等等。 Go语言中标识符由字母数字和_(下划线）组成，并且只能以字母和_开头。 举几个例子：abc, _, _123, a123。 关键字 关键…...
Docker Compose简单入门
Docker Compose 简介 Docker Compose 是一个编排多容器发布式部署的工具，提供命令集管理容器化应用的完整开发周期，包括服务构建，启动和停止。 Docker Compose 真正的作用是在一个文件（docker-compose.yml）中定义并运…...
使用 PHPMailer 实现邮件的实时发送
💂 个人网站:【 海拥】【神级代码资源网站】【办公神器】🤟 基于Web端打造的：👉轻量化工具创作平台💅 想寻找共同学习交流的小伙伴，请点击【全栈技术交流群】 今天我们利用GitHub上20K星星的项目 PHPMailer…...
Python爬虫（二）：爬虫完整流程
爬虫完整流程详解（7大核心步骤实战技巧） 一、爬虫完整工作流程 以下是爬虫开发的完整流程，我将结合具体技术点和实战经验展开说明： 1. 目标分析与前期准备 网站技术分析： 使用浏览器开发者工具（F12&…...
Rust 异步编程
Rust 异步编程 引言 Rust 是一种系统编程语言，以其高性能、安全性以及零成本抽象而著称。在多核处理器成为主流的今天，异步编程成为了一种提高应用性能、优化资源利用的有效手段。本文将深入探讨 Rust 异步编程的核心概念、常用库以及最佳实践。 异步编程基础 什么是异步…...
Axios请求超时重发机制
Axios 超时重新请求实现方案 在 Axios 中实现超时重新请求可以通过以下几种方式： 1. 使用拦截器实现自动重试 import axios from axios;// 创建axios实例 const instance axios.create();// 设置超时时间 instance.defaults.timeout 5000;// 最大重试次数 cons…...
QT： `long long` 类型转换为 `QString` 2025.6.5
在 Qt 中，将 long long 类型转换为 QString 可以通过以下两种常用方法实现： 方法 1：使用 QString::number() 直接调用 QString 的静态方法 number()，将数值转换为字符串： long long value 1234567890123456789LL; …...
使用Matplotlib创建炫酷的3D散点图：数据可视化的新维度
文章目录 基础实现代码代码解析进阶技巧1. 自定义点的大小和颜色2. 添加图例和样式美化3. 真实数据应用示例实用技巧与注意事项完整示例（带样式）应用场景在数据科学和可视化领域，三维图形能为我们提供更丰富的数据洞察。本文将手把手教你如何使用Python的Matplotlib库创建引…...
C/C++ 中附加包含目录、附加库目录与附加依赖项详解
在 C/C 编程的编译和链接过程中，附加包含目录、附加库目录和附加依赖项是三个至关重要的设置，它们相互配合，确保程序能够正确引用外部资源并顺利构建。虽然在学习过程中，这些概念容易让人混淆，但深入理解它们的作用和联…...
4. TypeScript 类型推断与类型组合
一、类型推断 (一) 什么是类型推断 TypeScript 的类型推断会根据变量、函数返回值、对象和数组的赋值和使用方式，自动确定它们的类型。 这一特性减少了显式类型注解的需要，在保持类型安全的同时简化了代码。通过分析上下文和初始值，TypeSc…...
前端中slice和splic的区别
1. slice slice 用于从数组中提取一部分元素，返回一个新的数组。 特点： 不修改原数组：slice 不会改变原数组，而是返回一个新的数组。提取数组的部分：slice 会根据指定的开始索引和结束索引提取数组的一部分。不包含…...
抽象类和接口（全）
一、抽象类 1.概念：如果⼀个类中没有包含⾜够的信息来描绘⼀个具体的对象，这样的类就是抽象类。 像是没有实际⼯作的⽅法,我们可以把它设计成⼀个抽象⽅法，包含抽象⽅法的类我们称为抽象类。 2.语法 在Java中，⼀个类如果被 abs…...
Ubuntu系统复制（U盘-电脑硬盘）
所需环境 电脑自带硬盘：1块 (1T) U盘1：Ubuntu系统引导盘（用于“U盘2”复制到“电脑自带硬盘”） U盘2：Ubuntu系统盘（1T，用于被复制） ！！！建议“电脑…...