临时使用网络通过http传输文件非常的方便。默认共享当前文件夹也可在启动时指定共享的文件夹。也可上传文件。python win32/64 3.6/3.7测试通过。运行后会提示本机ip在同一局域网下在浏览器内输入网址即可。如果本机有外网ip一样可用。使用curl可上传文件。curl -T 本地文件 http://192.168.1.99/目标文件名import copy, datetime, email.utils, html, http.client import io, mimetypes, os, posixpath, select, shutil import socket, socketserver, sys, time import urllib.parse, urllib.request, urllib.error from functools import partial import http.server from http import HTTPStatus import re from io import BytesIO class mSimpleHTTPRequestHandler(http.server.BaseHTTPRequestHandler): def __init__(self, *args, directoryNone, **kwargs): if directory is None: directory os.getcwd() self.directory directory super().__init__(*args, **kwargs) # ---------------- GET / HEAD ---------------- 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() # ---------------- PUT ---------------- # 用法 curl -T 本地文件 http://ip:port/目标路径/文件名 # 不需要 -F、不需要字段名、不需要 Referer def do_PUT(self): path self.translate_path(self.path) if os.path.isdir(path): self.send_error(HTTPStatus.BAD_REQUEST, PUT to a directory is not allowed) return try: length int(self.headers.get(Content-Length, 0)) except (TypeError, ValueError): self.send_error(HTTPStatus.BAD_REQUEST, Invalid Content-Length) return parent os.path.dirname(path) if parent and not os.path.isdir(parent): try: os.makedirs(parent, exist_okTrue) except OSError as e: self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(e)) return try: with open(path, wb) as out: remaining length while remaining 0: chunk self.rfile.read(min(65536, remaining)) if not chunk: break out.write(chunk) remaining - len(chunk) except OSError as e: self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(e)) return if remaining ! 0: # 客户端提前断开或 Content-Length 不匹配 self.send_error(HTTPStatus.BAD_REQUEST, Incomplete upload (client disconnected?)) return body bOK\n self.send_response(HTTPStatus.CREATED) self.send_header(Content-Type, text/plain; charsetutf-8) self.send_header(Content-Length, str(len(body))) self.end_headers() self.wfile.write(body) # ---------------- POST (浏览器表单上传) ---------------- def do_POST(self): r, info self.deal_post_data() print((r, info, by: , self.client_address)) referer self.headers.get(referer, /) f BytesIO() f.write(b!DOCTYPE html PUBLIC -//W3C//DTD HTML 3.2 Final//EN) f.write(bhtml\ntitleUpload Result Page/title\n) f.write(bbody\nh2Upload Result Page/h2\n) f.write(bhr\n) if r: f.write(bstrongSuccess:/strong) else: f.write(bstrongFailed:/strong) f.write(info.encode()) f.write((bra href%sback/a % html.escape(referer, quoteTrue)).encode()) f.write(bhrsmallPowerd By: bones7456, check new version at ) f.write(ba href\http://li2z.cn/?sSimpleHTTPServerWithUpload\) f.write(bhere/a./small/body\n/html\n) 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.get(content-type) if not content_type or boundary not in content_type: return (False, Content-Type header doesnt contain boundary) boundary content_type.split(boundary, 1)[1].strip().strip().encode() remainbytes int(self.headers.get(content-length, 0)) line self.rfile.readline() remainbytes - len(line) if boundary not in line: return (False, Content NOT begin with boundary) line self.rfile.readline() remainbytes - len(line) # 放宽不再限定 namefile只要带 filename 即可 fn re.findall(rfilename(.*?), line.decode(utf-8, replace)) if not fn: return (False, Cant find out file name...) path self.translate_path(self.path) if not os.path.isdir(path): return (False, Target is not a directory: %s % path) filename os.path.basename(fn[0]) # 防止 ../ 之类的路径穿越 fn os.path.join(path, filename) line self.rfile.readline() # 空行 remainbytes - len(line) line self.rfile.readline() remainbytes - len(line) try: out open(fn, wb) except IOError: return (False, Cant create file to write, do you have permission to write?) preline self.rfile.readline() remainbytes - len(preline) while remainbytes 0: line self.rfile.readline() remainbytes - 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! % fn) else: out.write(preline) preline line out.close() return (False, Unexpect Ends of data.) # ---------------- 目录列举 ---------------- def send_head(self): path self.translate_path(self.path) f None if os.path.isdir(path): parts urllib.parse.urlsplit(self.path) if not parts.path.endswith(/): self.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 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) ctype self.guess_type(path) try: f open(path, rb) except OSError: self.send_error(HTTPStatus.NOT_FOUND, File not found) return None try: fs os.fstat(f.fileno()) if (If-Modified-Since in self.headers and If-None-Match not in self.headers): try: ims email.utils.parsedate_to_datetime( self.headers[If-Modified-Since]) except (TypeError, IndexError, OverflowError, ValueError): pass else: if ims.tzinfo is None: ims ims.replace(tzinfodatetime.timezone.utc) if ims.tzinfo is datetime.timezone.utc: last_modif datetime.datetime.fromtimestamp( fs.st_mtime, datetime.timezone.utc) last_modif last_modif.replace(microsecond0) if last_modif ims: self.send_response(HTTPStatus.NOT_MODIFIED) self.end_headers() f.close() return None self.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 f except Exception: f.close() raise def list_directory(self, path): try: list os.listdir(path) except OSError: self.send_error(HTTPStatus.NOT_FOUND, No permission to list directory) return None list.sort(keylambda a: a.lower()) r [] try: displaypath urllib.parse.unquote(self.path, errorssurrogatepass) except UnicodeDecodeError: displaypath urllib.parse.unquote(path) displaypath html.escape(displaypath, quoteFalse) enc sys.getfilesystemencoding() title Directory listing for %s -- %s % (displaypath, get_host_ip()) r.append(!DOCTYPE HTML PUBLIC -//W3C//DTD HTML 4.01//EN http://www.w3.org/TR/html4/strict.dtd) r.append(html\nhead) r.append(meta http-equivContent-Type contenttext/html; charset%s % enc) r.append(title%s/title\n/head % title) r.append(body\nh1%s/h1 % title) r.append(hr\nul) r.append(form ENCTYPEmultipart/form-data methodpost) r.append(input namefile typefile/) r.append(input typesubmit valueupload//form\n) r.append(hr\nul\n) for name in list: fullname os.path.join(path, name) displayname linkname name if os.path.isdir(fullname): displayname name / linkname name / if os.path.islink(fullname): displayname name r.append(lia href%s%s/a/li % (urllib.parse.quote(linkname, errorssurrogatepass), html.escape(displayname, quoteFalse))) r.append(/ul\nhr\n pcurl 上传: codecurl -T 本地文件 http:// get_host_ip() : str(self.server.server_address[1]) /目标文件名/code/p\n /body\n/html\n) encoded \n.join(r).encode(enc, surrogateescape) 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 f def translate_path(self, path): path path.split(?, 1)[0] path path.split(#, 1)[0] trailing_slash path.rstrip().endswith(/) try: path urllib.parse.unquote(path, errorssurrogatepass) except UnicodeDecodeError: path urllib.parse.unquote(path) path posixpath.normpath(path) words path.split(/) words filter(None, words) path self.directory for word in words: if os.path.dirname(word) or word in (os.curdir, os.pardir): continue path os.path.join(path, word) if trailing_slash: path / return path def 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] else: return self.extensions_map[] if not mimetypes.inited: mimetypes.init() extensions_map mimetypes.types_map.copy() extensions_map.update({ : application/octet-stream, .py: text/plain, .c: text/plain, .h: text/plain, }) def test(HandlerClasshttp.server.BaseHTTPRequestHandler, ServerClasshttp.server.ThreadingHTTPServer, protocolHTTP/1.0, port80, bind): server_address (bind, port) HandlerClass.protocol_version protocol with ServerClass(server_address, HandlerClass) as httpd: sa httpd.socket.getsockname() serve_message Serving HTTP on {host} port {port} (http://{host}:{port}/) ... print(serve_message.format(hostsa[0], portsa[1])) try: httpd.serve_forever() except KeyboardInterrupt: print(\nKeyboard interrupt received, exiting.) sys.exit(0) def get_host_ip(): try: s socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect((8.8.8.8, 80)) ip s.getsockname()[0] except Exception: ip 127.0.0.1 finally: try: s.close() except Exception: pass return ip if __name__ __main__: import argparse print(get_host_ip()) parser argparse.ArgumentParser() parser.add_argument(--bind, -b, default, metavarADDRESS, helpSpecify alternate bind address [default: all interfaces]) parser.add_argument(--directory, -d, defaultos.getcwd(), helpSpecify alternative directory [default:current directory]) parser.add_argument(port, actionstore, default80, typeint, nargs?, helpSpecify alternate port [default: 80]) args parser.parse_args() handler_class partial(mSimpleHTTPRequestHandler, directoryargs.directory) test(HandlerClasshandler_class, portargs.port, bindargs.bind)
