python3中http協(xié)議提供文件服務(wù)器功能詳解
更新時(shí)間：2023年06月27日 16:43:55 作者：hsy12342611
http協(xié)議是互聯(lián)網(wǎng)的通用基礎(chǔ)協(xié)議，也可以利用其來(lái)開(kāi)發(fā)文件服務(wù)器，給客戶提供文件瀏覽，查看，下載，上傳等功能，這篇文章主要介紹了python3中http協(xié)議提供文件服務(wù)器功能,需要的朋友可以參考下
http協(xié)議是互聯(lián)網(wǎng)的通用基礎(chǔ)協(xié)議，也可以利用其來(lái)開(kāi)發(fā)文件服務(wù)器，給客戶提供文件瀏覽，查看，下載，上傳等功能。
1.python3自帶http文件服務(wù)
python3中http.server提供http文件服務(wù)，默認(rèn)端口是8000，可以進(jìn)行修改
運(yùn)行命令：
python3 -m http.server 12567
另外python2中可以使用SimpleHTTPServer來(lái)提供http文件服務(wù)，默認(rèn)端口是8000，ye可以進(jìn)行修改
運(yùn)行命令：
python -m SimpleHTTPServer 12567運(yùn)行效果如下：
2.python3從頭開(kāi)發(fā)http文件服務(wù)
http_server.py
# -*- coding: UTF-8 -*-
import os, time
import sys, socket
import posixpath
#escape：逃跑，用來(lái)讓特殊符號(hào)表示它本來(lái)的意思
try:
from html import escape
except ImportError:
from cgi import escape
import shutil
import mimetypes
import re
import signal
from io import StringIO, BytesIO
import codecs
from urllib.parse import quote
from urllib.parse import unquote
import urllib.parse
from http.server import HTTPServer
from http.server import BaseHTTPRequestHandler
"""urlencode quote unquote
urllib.parse.urlencode
urlencode函數(shù)，可以把key-value這樣的鍵值對(duì)轉(zhuǎn)換成我們想要的格式，返回的是a=1&b=2這樣的字符串
urllib.parse.quote
quote對(duì)一個(gè)字符串進(jìn)行urlencode轉(zhuǎn)換
urllib.parse.unquote
unquote與quote對(duì)應(yīng)，用來(lái)解碼quote處理后的結(jié)果
"""
"""內(nèi)存IO之StringIO和BytesIO
參考博客：
https://zhuanlan.zhihu.com/p/332651899
"""
"""HTTPServer
參考博客
https://www.cnblogs.com/jason-huawen/p/16241405.html
"""
"""
瀏覽器運(yùn)行：
http://127.0.0.1:18081
"""
class MyHTTPRequestHandler(BaseHTTPRequestHandler):
address = socket.gethostbyname(socket.gethostname())
treefile = "sesssion_cache.txt"
mylist = []
myspace = ""
def do_GET(self):
"""處理GET請(qǐng)求
"""
print(MyHTTPRequestHandler.address)
paths = unquote(self.path)
path = str(paths)
print("path is ", paths, path)
'''
self.send_error(404, "File not found")
'''
if self.remove_dir_or_file(path):
return
fd = self.send_head()
if fd:
#關(guān)鍵代碼: self.wfile用來(lái)向客戶端寫入數(shù)據(jù)
shutil.copyfileobj(fd, self.wfile)
if path == "/":
self.traversal_file(translate_path(self.path))
self.write_list(MyHTTPRequestHandler.treefile)
fd.close()
def do_HEAD(self):
"""處理HEAD請(qǐng)求
"""
fd = self.send_head()
if fd:
fd.close()
def do_POST(self):
"""處理POST請(qǐng)求
"""
r, info = self.deal_post_data()
#拼裝HTML文本
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><br>")
else:
f.write(b"<strong>Failed:</strong><br>")
for i in info:
print(r, i, "by: ", self.client_address)
f.write(i.encode('utf-8')+b"<br>")
f.write(b"<br><a href=\"%s\">back</a>" % self.headers['referer'].encode('ascii'))
f.write(b"</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()
#每次提交post請(qǐng)求之后更新目錄樹(shù)文件
self.p(translate_path(self.path))
self.write_list(MyHTTPRequestHandler.treefile)
def str_to_chinese(self,var):
not_end = True
while not_end:
start1 = var.find("\\x")
# print start1
if start1 > -1:
str1 = var[start1 + 2:start1 + 4]
print(str1)
start2 = var[start1 + 4:].find("\\x") + start1 + 4
if start2 > -1:
str2 = var[start2 + 2:start2 + 4]
start3 = var[start2 + 4:].find("\\x") + start2 + 4
if start3 > -1:
str3 = var[start3 + 2:start3 + 4]
else:
not_end = False
if start1 > -1 and start2 > -1 and start3 > -1:
str_all = str1 + str2 + str3
# print str_all
str_all = codecs.decode(str_all, "hex").decode('utf-8')
str_re = var[start1:start3 + 4]
# print str_all, " " ,str_re
var = var.replace(str_re, str_all)
# print var.decode('utf-8')
return var
def deal_post_data(self):
boundary = self.headers["Content-Type"].split("=")[1].encode('ascii')
print("boundary===", boundary)
remain_bytes = int(self.headers['content-length'])
print("remain_bytes===", remain_bytes)
res = []
line = self.rfile.readline()
while boundary in line and str(line, encoding = "utf-8")[-4:] != "--\r\n":
#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)
print("line!!!",line)
fn = re.findall(r'Content-Disposition.*name="file"; filename="(.*)"', str(line))
if not fn:
return False, "Can't find out file name..."
path = translate_path(self.path)
fname = fn[0]
#fname = fname.replace("\\", "\\\\")
fname = self.str_to_chinese(fname)
print("------",fname)
fn = os.path.join(path, fname)
while os.path.exists(fn):
fn += "_"
print("!!!!",fn)
dirname = os.path.dirname(fn)
if not os.path.exists(dirname):
os.makedirs(dirname)
line = self.rfile.readline()
remain_bytes -= len(line)
line = self.rfile.readline()
# b'\r\n'
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()
print("pre_line", pre_line)
remain_bytes -= len(pre_line)
print("remain_bytes", remain_bytes)
Flag = True
while remain_bytes > 0:
line = self.rfile.readline()
print("while line", line)
if boundary in line:
remain_bytes -= len(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()
res.append("File '%s' upload success!" % fn)
Flag = False
break
else:
out.write(pre_line)
pre_line = line
if pre_line is not None and Flag == True:
out.write(pre_line)
out.close()
res.append("File '%s' upload success!" % fn)
return True, res
def remove_dir_or_file(self, path):
plist = path.split("/", 2)
print("plist plist plist plist ", plist)
if len(plist) > 2 and plist[1] == "delete":
#路徑轉(zhuǎn)化
file = translate_path(plist[2])
print("======>>>>>>>>>> file", file)
if os.path.exists(file):
fdir = os.path.dirname(file)
print("======>>>>>>>>>> ", file, fdir)
#刪除文件
os.remove(file)
if not os.listdir(fdir):
#刪除目錄
os.removedirs(fdir)
time.sleep(0.5)
# 0.5s后重定向
self.send_response(302)
self.send_header('Location', "/")
self.end_headers()
return True
print("======>>>>>>>>>> file not exists ", file)
return False
def send_head(self):
"""發(fā)送HTTP頭
"""
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)
print("=================================")
content_type = self.guess_type(path)
try:
# Always read in binary mode. Opening files in text mode may cause
# newline translations, making the actual size of the content
# transmitted *less* than the content-length!
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):
"""Helper to produce a directory listing (absent index.html).
Return value is either a file object, or None (indicating an
error). In either case, the headers are sent, making the
interface the same as for 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('ascii'))
f.write(b"<body>\n<h2>Directory listing for %s</h2>\n" % display_path.encode('ascii'))
f.write(b"<hr>\n")
#上傳目錄
f.write(b"<h3>Directory Updating</h3>\n")
f.write(b"<form ENCTYPE=\"multipart/form-data\" method=\"post\">")
#@change=\"handleChange\" @click=\"handelClick\"
f.write(b"<input ref=\"input\" webkitdirectory multiple name=\"file\" type=\"file\"/>")
f.write(b"<input type=\"submit\" value=\"uploadDir\"/></form>\n")
f.write(b"<hr>\n")
#上傳文件
f.write(b"<h3>Files Updating</h3>\n")
f.write(b"<form ENCTYPE=\"multipart/form-data\" method=\"post\">")
f.write(b"<input ref=\"input\" multiple name=\"file\" type=\"file\"/>")
f.write(b"<input type=\"submit\" value=\"uploadFiles\"/></form>\n")
f.write(b"<hr>\n")
#表格
f.write(b"<table with=\"100%\">")
f.write(b"<tr><th>path</th>")
f.write(b"<th>size(Byte)</th>")
f.write(b"<th>modify time</th>")
f.write(b"</tr>")
# 根目錄下所有的內(nèi)容
for name in list_dir:
# 根目錄下的路徑
fullname = os.path.join(path, name)
# 目錄名/文件名
display_name = linkname = name
print("display_name ==> ", display_name)
if display_name.upper() == "HTTP_SERVER.PY":
continue
# 如果是文件夾的話
if os.path.isdir(fullname):
# 遍歷文件夾
for root, dirs, files in os.walk(fullname):
# root 表示當(dāng)前正在訪問(wèn)的文件夾路徑
# dirs 表示該文件夾下的子目錄名list
# files 表示該文件夾下的文件list
# 遍歷文件
for fi in files:
print("########", os.path.join(root, fi))
display_name = os.path.join(root, fi)
#刪除前面的xx個(gè)字符，取出相對(duì)路徑
relativePath = display_name[len(os.getcwd()):].replace('\\','/')
st = os.stat(display_name)
fsize = st.st_size
fmtime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(st.st_mtime))
f.write(b"<tr>")
f.write(b'<td><a href="%s" rel="external nofollow" rel="external nofollow" rel="external nofollow" >%s</a></td>' % (quote(relativePath).encode('utf-8'), escape(relativePath).encode('utf-8')))
f.write(b"<td>%d</td>" % fsize)
f.write(b"<td>%s</td>" % escape(fmtime).encode('ascii'))
f.write(b"<td><a href=\"/delete/%s\">delete</a>" % escape(relativePath).encode('utf-8'))
f.write(b"</tr>")
# 遍歷所有的文件夾名字，其實(shí)在上面walk已經(jīng)遍歷過(guò)了
# for d in dirs:
# print(d)
# 如果是鏈接文件
elif os.path.islink(fullname):
linkname = linkname + "/"
print("real===", linkname)
display_name = name + "@"
# Note: a link to a directory displays with @ and links with /
f.write(b'<li><a href="%s" rel="external nofollow" rel="external nofollow" rel="external nofollow" >%s</a>\n' % (quote(linkname).encode('ascii'), escape(display_name).encode('ascii')))
else:
#其他直接在根目錄下的文件直接顯示出來(lái)
st = os.stat(display_name)
fsize = st.st_size
fmtime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(st.st_mtime))
f.write(b"<tr>")
f.write(b'<td><a href="%s" rel="external nofollow" rel="external nofollow" rel="external nofollow" >%s</a></td>' % (quote(linkname).encode('utf-8'), escape(display_name).encode('utf-8')))
f.write(b"<td>%d</td>" % fsize)
f.write(b"<td>%s</td>" % escape(fmtime).encode('ascii'))
f.write(b"<td><a href=\"/delete/%s\">delete</a>" % escape(display_name).encode('utf-8'))
f.write(b"</tr>")
f.write(b"</table>")
f.write(b"\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):
"""Guess the type of a file.
Argument is a PATH (a filename).
Return value is a string of the form type/subtype,
usable for a MIME Content-type header.
The default implementation looks the file's extension
up in the table self.extensions_map, using application/octet-stream
as a default; however it would be permissible (if
slow) to look inside the data to make a better guess.
"""
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',
})
def traversal_file(self, url):
print("url:", url)
files = os.listdir(r''+url)
for file in files:
myfile = url + "http://" + file
size = os.path.getsize(myfile)
if os.path.isfile(myfile):
MyHTTPRequestHandler.mylist.append(str(MyHTTPRequestHandler.myspace)+"|____"+file +" "+ str(size)+"\n")
if os.path.isdir(myfile) :
MyHTTPRequestHandler.mylist.append(str(MyHTTPRequestHandler.myspace)+"|____"+file + "\n")
#get into the sub-directory,add "| "
MyHTTPRequestHandler.myspace = MyHTTPRequestHandler.myspace+"| "
self.traversal_file(myfile)
#when sub-directory of iteration is finished,reduce "| "
MyHTTPRequestHandler.myspace = MyHTTPRequestHandler.myspace[:-5]
def get_all_file_list(self):
listofme = []
for root, dirs, files in os.walk(translate_path(self.path)):
files.sort()
for fi in files:
display_name = os.path.join(root, fi)
#刪除前面的XXX個(gè)字符，取出相對(duì)當(dāng)前目錄的路徑
relativePath = display_name[len(os.getcwd()):].replace('\\','/')
st = os.stat(display_name)
fsize = st.st_size
fmtime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(st.st_mtime))
listofme.append(relativePath+"\t")
listofme.append(str(fsize)+"\t")
listofme.append(str(fmtime)+"\t\n")
return listofme
def write_list(self,url):
f = open(url,'w')
f.write("http://"+str(MyHTTPRequestHandler.address)+":8001/ directory tree\n")
MyHTTPRequestHandler.mylist.sort()
f.writelines(MyHTTPRequestHandler.mylist)
f.write("\nFile Path\tFile Size\tFile Modify Time\n")
f.writelines(self.get_all_file_list())
MyHTTPRequestHandler.mylist = []
MyHTTPRequestHandler.myspace = ""
f.close()
print("write_list end")
def translate_path(path):
path = path.split('?', 1)[0]
path = path.split('#', 1)[0]
path = posixpath.normpath(unquote(path))
words = path.split('/')
words = filter(None, words)
path = os.getcwd()
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
def signal_handler(signal, frame):
print(signal, frame)
exit()
def main():
print('python version:', sys.version_info.major, sys.version_info.minor)
if sys.argv[1:]:
port = int(sys.argv[1])
else:
port = 18081
#server_address = ('', port)
address = ('127.0.0.1', port)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGQUIT, signal_handler)
#忽略ctrl+z：SIGTSTP（掛起信號(hào)）
signal.signal(signal.SIGTSTP, signal.SIG_IGN)
server = HTTPServer(address, MyHTTPRequestHandler)
server_info = server.socket.getsockname()
print("server info: " + str(server_info[0]) + ", port: " + str(server_info[1]) + " ...")
server.serve_forever()
def test_url_code():
values = {'username': 'hello 你好', 'password': 'pass'}
data = urllib.parse.urlencode(values)
print("data = ", data)
name = '狄仁杰'
name = urllib.parse.quote(name)
print("name = ", name)
dname = urllib.parse.unquote(name)
print("dname = " + dname)
if __name__ == '__main__':
test_url_code()
main()運(yùn)行效果如下：
到此這篇關(guān)于python3中http協(xié)議提供文件服務(wù)器功能詳解的文章就介紹到這了,更多相關(guān)python3 http文件服務(wù)器內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
關(guān)于AnacondaNavigator?Jupyter?Notebook更換Python內(nèi)核的問(wèn)題
因?yàn)樾掳惭b的Anaconda?Navigator默認(rèn)安裝了一個(gè)Python，Jupyter?Notebook默認(rèn)使用的內(nèi)核就是這個(gè)Python，跟我系統(tǒng)安裝好的Python沖突了，下面小編給大家介紹AnacondaNavigator?Jupyter?Notebook更換Python內(nèi)核的問(wèn)題,需要的朋友可以參考下2022-02-02
今天小編就為大家分享一篇使用Python進(jìn)行目錄的對(duì)比方法，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-11-11
python實(shí)現(xiàn)簡(jiǎn)易SSL的項(xiàng)目實(shí)踐
本文主要介紹了python實(shí)現(xiàn)簡(jiǎn)易SSL的項(xiàng)目實(shí)踐,包括CA.py、server.py和client.py三個(gè)模塊,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2025-02-02
Python快速轉(zhuǎn)換numpy數(shù)組中Nan和Inf的方法實(shí)例說(shuō)明
今天小編就為大家分享一篇關(guān)于Python快速轉(zhuǎn)換numpy數(shù)組中Nan和Inf的方法實(shí)例說(shuō)明，小編覺(jué)得內(nèi)容挺不錯(cuò)的，現(xiàn)在分享給大家，具有很好的參考價(jià)值，需要的朋友一起跟隨小編來(lái)看看吧2019-02-02
這篇文章主要介紹了Python3操作CSV文件使用包過(guò)程解析,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-04-04
大家都知道ipython是一個(gè)python的交互式shell，比默認(rèn)的python shell好用得多，IPython有許多種安裝方式，這主要和使用什么操作系統(tǒng)有關(guān)。本文給大家介紹的是在windows下ipython的安裝與使用，有需要的朋友們可以參考學(xué)習(xí)。2016-10-10
python在協(xié)程中增加任務(wù)實(shí)例操作
在本篇文章里小編給大家整理的是一篇關(guān)于python在協(xié)程中增加任務(wù)實(shí)例操作內(nèi)容，有興趣的朋友們可以學(xué)習(xí)下。2021-02-02
for循環(huán)用于迭代序列（即列表、元組、字典、集合或字符串）。for 語(yǔ)句的寫法如從對(duì)象開(kāi)始按順序給變量賦值，元素個(gè)數(shù)重復(fù)這個(gè)過(guò)程。對(duì)象可以是列表（數(shù)組）、元組、字符串等。本文將詳細(xì)講解Python中for定義迭代方法詳解，需要的可以了解一下2022-04-04
這篇文章主要介紹了python實(shí)現(xiàn)磁盤日志清理的示例，幫助大家更好的理解和使用python，感興趣的朋友可以了解下2020-11-11
Python實(shí)現(xiàn)調(diào)整Excel的內(nèi)容高度
本文介紹了一種自動(dòng)調(diào)整Excel行高的Python實(shí)現(xiàn)方法,代碼實(shí)現(xiàn)了自動(dòng)計(jì)算行高、批量處理多個(gè)Excel文件的功能并提供了進(jìn)度提示,希望對(duì)大家有所幫助2026-05-05