实现在局域网(同一WIFI下) 文件上传与下载
该模块通过实现标准GET在BaseHTTPServer上构建
和HEAD请求。(将所有代码粘贴到同一个py文件中，即可使用)
所需包
基于python3版本实现，python2版本无涉猎
import os
import sys
import argparse
import posixpath
try:
from html import escape
except ImportError:
from cgi import escape
import shutil
import mimetypes
import re
import signal
from io import StringIO, BytesIO
if sys.version_info.major == 3:
# Python3
from urllib.parse import quote
from urllib.parse import unquote
from http.server import HTTPServer
from http.server import BaseHTTPRequestHandler
基本类 简单HTTP服务类
带有GET/HEAD/POST命令的简单HTTP请求处理程序。
这将提供当前目录中的文件及其子目录。
文件的MIME类型由调用.gues_type（）方法。
并且可以接收上传的文件由客户提供。
GET/HEAD/POST请求是相同的，除了HEAD请求忽略文件的实际内容。
class SimpleHTTPRequestHandler(BaseHTTPRequestHandler):
server_version = "simple_http_server/" + __version__
def do_GET(self):
"""Serve a GET request."""
fd = self.send_head()
if fd:
shutil.copyfileobj(fd, self.wfile)
fd.close()
def do_HEAD(self):
"""Serve a HEAD request."""
fd = self.send_head()
if fd:
fd.close()
def do_POST(self):
"""Serve a POST request."""
r, info = self.deal_post_data()
print(r, info, "by: ", self.client_address)
f = BytesIO()
f.write(b'<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">')
f.write(b"<html>\n<title>Upload Result Page</title>\n")
f.write(b"<body>\n<h2>Upload Result Page</h2>\n")
f.write(b"<hr>\n")
if r:
f.write(b"<strong>Success:</strong>")
else:
f.write(b"<strong>Failed:</strong>")
f.write(info.encode('utf-8'))
f.write(b"<br><a href=\".\">back</a>")
f.write(b"<hr><small>Powered By: freelamb, check new version at ")
# 原始代码地址 可以参考
f.write(b"<a href=\"https://github.com/freelamb/simple_http_server\">")
f.write(b"here</a>.</small></body>\n</html>\n")
length = f.tell()
f.seek(0)
self.send_response(200)
self.send_header("Content-type", "text/html;charset=utf-8")
self.send_header("Content-Length", str(length))
self.end_headers()
if f:
shutil.copyfileobj(f, self.wfile)
f.close()
def deal_post_data(self):
boundary = self.headers["Content-Type"].split("=")[1].encode('utf-8')
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('utf-8'))
if not fn:
return False, "Can't find out file name..."
path = translate_path(self.path)
fn = os.path.join(path, fn[0])
while os.path.exists(fn):
fn += "_"
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?"
pre_line = self.rfile.readline()
remain_bytes -= len(pre_line)
while remain_bytes > 0:
line = self.rfile.readline()
remain_bytes -= len(line)
if boundary in line:
pre_line = pre_line[0:-1]
if pre_line.endswith(b'\r'):
pre_line = pre_line[0:-1]
out.write(pre_line)
out.close()
return True, "File '%s' upload success!" % fn
else:
out.write(pre_line)
pre_line = line
return False, "Unexpect Ends of data."
def send_head(self):
"""
GET和HEAD命令的通用代码。
这将发送响应代码和MIME标头。
返回值要么是文件对象
（除非命令是HEAD，否则调用方必须将其复制到输出文件中，
并且在任何情况下都必须由调用方关闭），
要么是None，在这种情况下，调用方无需进一步操作。
"""
path = translate_path(self.path)
if os.path.isdir(path):
if not self.path.endswith('/'):
# redirect browser - doing basically what apache does
self.send_response(301)
self.send_header("Location", self.path + "/")
self.end_headers()
return None
for index in "index.html", "index.htm":
index = os.path.join(path, index)
if os.path.exists(index):
path = index
break
else:
return self.list_directory(path)
content_type = self.guess_type(path)
try:
#始终以二进制模式读取。以文本模式打开文件可能会导致
#换行翻译，使内容的实际大小
#传输*小于*内容长度！
f = open(path, 'rb')
except IOError:
self.send_error(404, "File not found")
return None
self.send_response(200)
self.send_header("Content-type", content_type)
fs = os.fstat(f.fileno())
self.send_header("Content-Length", str(fs[6]))
self.send_header("Last-Modified", self.date_time_string(fs.st_mtime))
self.end_headers()
return f
def list_directory(self, path):
"""
帮助程序生成目录列表（缺少index.html）。
返回值为file对象或None（表示错误）。
无论哪种情况，都会发送标头接口与send_head（）相同。
"""
try:
list_dir = os.listdir(path)
except os.error:
self.send_error(404, "No permission to list directory")
return None
list_dir.sort(key=lambda a: a.lower())
f = BytesIO()
display_path = escape(unquote(self.path))
f.write(b'<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">')
f.write(b"<html>\n<title>Directory listing for %s</title>\n" % display_path.encode('utf-8'))
f.write(b"<body>\n<h2>Directory listing for %s</h2>\n" % display_path.encode('utf-8'))
f.write(b"<hr>\n")
f.write(b"<form ENCTYPE=\"multipart/form-data\" method=\"post\">")
f.write(b"<input name=\"file\" type=\"file\"/>")
f.write(b"<input type=\"submit\" value=\"upload\"/></form>\n")
f.write(b"<hr>\n<ul>\n")
for name in list_dir:
fullname = os.path.join(path, name)
display_name = linkname = name
# Append / for directories or @ for symbolic links
if os.path.isdir(fullname):
display_name = name + "/"
linkname = name + "/"
if os.path.islink(fullname):
display_name = name + "@"
# Note: a link to a directory displays with @ and links with /
f.write(
b'<li><a href="%s">%s</a>\n' % (quote(linkname).encode('utf-8'), escape(display_name).encode('utf-8')))
f.write(b"</ul>\n<hr>\n</body>\n</html>\n")
length = f.tell()
f.seek(0)
self.send_response(200)
self.send_header("Content-type", "text/html;charset=utf-8")
self.send_header("Content-Length", str(length))
self.end_headers()
return f
def guess_type(self, path):
"""
参数是PATH（文件名）。
返回值是表单类型/子类型的字符串，
可用于MIME内容类型标头。
默认实现在self.extensions_map表中查找文件的扩展名，默认使用application/octet流；
"""
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]
else:
return self.extensions_map['']
if not mimetypes.inited:
mimetypes.init() # try to read system mime.types
extensions_map = mimetypes.types_map.copy()
extensions_map.update({
'': 'application/octet-stream', # Default
'.py': 'text/plain',
'.c': 'text/plain',
'.h': 'text/plain',
})
文件路径处理
将/分隔的PATH转换为本地文件名语法。
对本地文件系统有特殊意义的组件（例如驱动器或目录名），那么可能会被阻止或诊断。
def translate_path(path):
# abandon query parameters
path = path.split('?', 1)[0]
path = path.split('#', 1)[0]
path = posixpath.normpath(unquote(path))
words = path.split('/')
words = filter(None, words)
# 获取你的py文件存放的路径
path = os.getcwd()
# 可在此自定义路径（如果有其路径）
path = path+"/file_xxx/xxx"
for word in words:
drive, word = os.path.splitdrive(word)
head, word = os.path.split(word)
if word in (os.curdir, os.pardir):
continue
path = os.path.join(path, word)
return path
信息提醒
如果HTTP Server被关闭
def signal_handler(signal, frame):
print("You choose to stop me.")
exit()
HTTP Server 初始化
设置HTTP Server初始数值，基于自己电脑设置。
对于IP来说，双方ip设置应一样。
# 版本设置 自定义
__version__ = "0.3.1"
def _argparse():
parser = argparse.ArgumentParser()
# 一般用户电脑用做服务器，每次开机后，ip4地址可能会发生变化
ip = input("请输入IP地址:")
parser.add_argument('--bind', '-b', metavar='ADDRESS', default=ip,
help='Specify alternate bind address [default: all interfaces]')
parser.add_argument('--version', '-v', action='version', version=__version__)
parser.add_argument('port', action='store', default=8000, type=int, nargs='?',
help='Specify alternate port [default: 8000]')
return parser.parse_args()
启用HTTP Server
py程序启动后，会输出网址，点击后，会自动进入HTTP服务，可以进行文件传输操作。
def main():
args = _argparse()
# print(args)
server_address = (args.bind, args.port)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
httpd = HTTPServer(server_address, SimpleHTTPRequestHandler)
server = httpd.socket.getsockname()
print(
"server_version: " + SimpleHTTPRequestHandler.server_version + ", python_version: " + SimpleHTTPRequestHandler.sys_version)
print("sys encoding: " + sys.getdefaultencoding())
print("Serving http on: " + str(server[0]) + ", port: " + str(server[1]) + " ... (http://" + server[0] + ":" + str(
server[1]) + "/)")
httpd.serve_forever()
if __name__ == '__main__':
main()
