Andy Stewart
6 years ago
parent
b4be789467
commit
320840214b
7 changed files with 144 additions and 421 deletions
@ -0,0 +1,112 @@ |
||||
#!/usr/bin/env python3 |
||||
# -*- coding: utf-8 -*- |
||||
|
||||
# Copyright (C) 2018 Andy Stewart |
||||
# |
||||
# Author: Andy Stewart <lazycat.manatee@gmail.com> |
||||
# Maintainer: Andy Stewart <lazycat.manatee@gmail.com> |
||||
# |
||||
# This program is free software: you can redistribute it and/or modify |
||||
# it under the terms of the GNU General Public License as published by |
||||
# the Free Software Foundation, either version 3 of the License, or |
||||
# any later version. |
||||
# |
||||
# This program is distributed in the hope that it will be useful, |
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
||||
# GNU General Public License for more details. |
||||
# |
||||
# You should have received a copy of the GNU General Public License |
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
||||
|
||||
from PyQt5 import QtGui, QtCore |
||||
from PyQt5.QtCore import Qt |
||||
from PyQt5.QtGui import QColor, QFont |
||||
from PyQt5.QtWidgets import QWidget, QLabel, QVBoxLayout |
||||
from core.utils import get_local_ip, get_free_port |
||||
import subprocess |
||||
import os |
||||
import qrcode |
||||
import signal |
||||
|
||||
from core.buffer import Buffer |
||||
|
||||
class AppBuffer(Buffer): |
||||
def __init__(self, buffer_id, url, config_dir, argument, emacs_var_dict, module_path, call_emacs): |
||||
Buffer.__init__(self, buffer_id, url, argument, emacs_var_dict, module_path, call_emacs, False, QColor(0, 0, 0, 255)) |
||||
|
||||
self.add_widget(FileUploaderWidget(url, QColor(0, 0, 0, 255))) |
||||
|
||||
def destroy_buffer(self): |
||||
os.kill(self.buffer_widget.background_process.pid, signal.SIGKILL) |
||||
|
||||
class Image(qrcode.image.base.BaseImage): |
||||
def __init__(self, border, width, box_size): |
||||
self.border = border |
||||
self.width = width |
||||
self.box_size = box_size |
||||
size = (width + border * 2) * box_size |
||||
self._image = QtGui.QImage(size, size, QtGui.QImage.Format_RGB16) |
||||
self._image.fill(QtCore.Qt.white) |
||||
|
||||
def pixmap(self): |
||||
return QtGui.QPixmap.fromImage(self._image) |
||||
|
||||
def drawrect(self, row, col): |
||||
painter = QtGui.QPainter(self._image) |
||||
painter.fillRect( |
||||
(col + self.border) * self.box_size, |
||||
(row + self.border) * self.box_size, |
||||
self.box_size, self.box_size, |
||||
QtCore.Qt.black) |
||||
|
||||
def save(self, stream, kind=None): |
||||
pass |
||||
|
||||
class FileUploaderWidget(QWidget): |
||||
def __init__(self, url, color): |
||||
QWidget.__init__(self) |
||||
url = os.path.expanduser(url) |
||||
|
||||
self.setStyleSheet("background-color: black") |
||||
|
||||
self.file_name_font = QFont() |
||||
self.file_name_font.setPointSize(24) |
||||
|
||||
self.file_name_label = QLabel(self) |
||||
self.file_name_label.setText("Your file will be share at\n{0}".format(url)) |
||||
self.file_name_label.setFont(self.file_name_font) |
||||
self.file_name_label.setAlignment(Qt.AlignCenter) |
||||
self.file_name_label.setStyleSheet("color: #eee") |
||||
|
||||
self.qrcode_label = QLabel(self) |
||||
|
||||
self.notify_font = QFont() |
||||
self.notify_font.setPointSize(12) |
||||
self.notify_label = QLabel(self) |
||||
self.notify_label.setText("Scan QR code above to upload a file from your smartphone.\nMake sure the smartphone is connected to the same WiFi network as this computer.") |
||||
self.notify_label.setFont(self.notify_font) |
||||
self.notify_label.setAlignment(Qt.AlignCenter) |
||||
self.notify_label.setStyleSheet("color: #eee") |
||||
|
||||
layout = QVBoxLayout(self) |
||||
layout.setContentsMargins(0, 0, 0, 0) |
||||
layout.addStretch() |
||||
layout.addWidget(self.qrcode_label, 0, Qt.AlignCenter) |
||||
layout.addSpacing(20) |
||||
layout.addWidget(self.file_name_label, 0, Qt.AlignCenter) |
||||
layout.addSpacing(40) |
||||
layout.addWidget(self.notify_label, 0, Qt.AlignCenter) |
||||
layout.addStretch() |
||||
|
||||
self.port = get_free_port() |
||||
self.local_ip = get_local_ip() |
||||
self.address = "http://{0}:{1}".format(self.local_ip, self.port) |
||||
|
||||
self.qrcode_label.setPixmap(qrcode.make(self.address, image_factory=Image).pixmap()) |
||||
|
||||
self.background_process = subprocess.Popen( |
||||
"cd {0} && filebrowser --noauth -d /tmp/filebrowser.db --address {1} -p {2}".format(url, self.local_ip, self.port), |
||||
stdout=subprocess.PIPE, |
||||
stderr=subprocess.STDOUT, |
||||
shell=True) |
||||
@ -1,391 +0,0 @@ |
||||
#!/usr/bin/env python3 |
||||
# -*- coding: utf-8 -*- |
||||
|
||||
# Copyright (C) 2018 Andy Stewart |
||||
# |
||||
# Author: Andy Stewart <lazycat.manatee@gmail.com> |
||||
# Maintainer: Andy Stewart <lazycat.manatee@gmail.com> |
||||
# |
||||
# This program is free software: you can redistribute it and/or modify |
||||
# it under the terms of the GNU General Public License as published by |
||||
# the Free Software Foundation, either version 3 of the License, or |
||||
# any later version. |
||||
# |
||||
# This program is distributed in the hope that it will be useful, |
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
||||
# GNU General Public License for more details. |
||||
# |
||||
# You should have received a copy of the GNU General Public License |
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
||||
|
||||
from PyQt5 import QtGui, QtCore |
||||
from PyQt5.QtCore import Qt |
||||
from PyQt5.QtGui import QColor, QFont |
||||
from PyQt5.QtWidgets import QWidget, QLabel, QVBoxLayout |
||||
from core.utils import get_local_ip |
||||
from io import BytesIO |
||||
import html |
||||
import http.server |
||||
import mimetypes |
||||
import os |
||||
import posixpath |
||||
import qrcode |
||||
import re |
||||
import shutil |
||||
import socket |
||||
import sys |
||||
import threading |
||||
import urllib |
||||
|
||||
from core.buffer import Buffer |
||||
|
||||
class AppBuffer(Buffer): |
||||
def __init__(self, buffer_id, url, config_dir, argument, emacs_var_dict, module_path, call_emacs): |
||||
Buffer.__init__(self, buffer_id, url, argument, emacs_var_dict, module_path, call_emacs, False, QColor(0, 0, 0, 255)) |
||||
|
||||
self.add_widget(FileUploaderWidget(url, QColor(0, 0, 0, 255))) |
||||
|
||||
class SimpleHTTPRequestHandler(http.server.BaseHTTPRequestHandler): |
||||
|
||||
"""Simple HTTP request handler with GET/HEAD/POST commands. |
||||
|
||||
This serves files from the current directory and any of its |
||||
subdirectories. The MIME type for files is determined by |
||||
calling the .guess_type() method. And can reveive file uploaded |
||||
by client. |
||||
|
||||
The GET/HEAD/POST requests are identical except that the HEAD |
||||
request omits the actual contents of the file. |
||||
|
||||
""" |
||||
|
||||
def do_GET(self): |
||||
"""Serve a GET request.""" |
||||
f = self.send_head() |
||||
if f: |
||||
self.copyfile(f, self.wfile) |
||||
f.close() |
||||
|
||||
def do_HEAD(self): |
||||
"""Serve a HEAD request.""" |
||||
f = self.send_head() |
||||
if f: |
||||
f.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()) |
||||
f.write(("<br><a href=\"%s\">back</a>" % self.headers['referer']).encode()) |
||||
f.write(b"<hr><small>Powerd By: bones7456, check new version at ") |
||||
f.write(b"<a href=\"http://li2z.cn/?s=SimpleHTTPServerWithUpload\">") |
||||
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") |
||||
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() |
||||
remainbytes = int(self.headers['content-length']) |
||||
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) |
||||
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() |
||||
remainbytes -= len(line) |
||||
line = self.rfile.readline() |
||||
remainbytes -= 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() |
||||
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 |
||||
return (False, "Unexpect Ends of data.") |
||||
|
||||
def send_head(self): |
||||
"""Common code for GET and HEAD commands. |
||||
|
||||
This sends the response code and MIME headers. |
||||
|
||||
Return value is either a file object (which has to be copied |
||||
to the outputfile by the caller unless the command was HEAD, |
||||
and must be closed by the caller under all circumstances), or |
||||
None, in which case the caller has nothing further to do. |
||||
|
||||
""" |
||||
path = self.translate_path(self.path) |
||||
f = None |
||||
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) |
||||
ctype = 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", ctype) |
||||
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 = os.listdir(path) |
||||
except os.error: |
||||
self.send_error(404, "No permission to list directory") |
||||
return None |
||||
list.sort(key=lambda a: a.lower()) |
||||
f = BytesIO() |
||||
displaypath = html.escape(urllib.parse.unquote(self.path)) |
||||
f.write(b'<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">') |
||||
f.write(("<html>\n<head>{0} {1}<title>Directory listing for {2}</title>\n</head>\n".format( |
||||
'''<meta http-equiv="Content-Type" content="text/html; charset=utf-8" />''', |
||||
'''<meta name="viewport" content="width=device-width, initial-scale=1">\n''', |
||||
displaypath)).encode()) |
||||
f.write(("<body>\n<h2>Directory listing for %s</h2>\n" % displaypath).encode()) |
||||
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: |
||||
fullname = os.path.join(path, name) |
||||
displayname = linkname = name |
||||
# Append / for directories or @ for symbolic links |
||||
if 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 / |
||||
f.write(('<li><a href="%s">%s</a>\n' |
||||
% (urllib.parse.quote(linkname), html.escape(displayname))).encode()) |
||||
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") |
||||
self.send_header("Content-Length", str(length)) |
||||
self.end_headers() |
||||
return f |
||||
|
||||
def translate_path(self, path): |
||||
"""Translate a /-separated PATH to the local filename syntax. |
||||
|
||||
Components that mean special things to the local file system |
||||
(e.g. drive or directory names) are ignored. (XXX They should |
||||
probably be diagnosed.) |
||||
|
||||
""" |
||||
global upload_dir |
||||
|
||||
# abandon query parameters |
||||
path = path.split('?', 1)[0] |
||||
path = path.split('#', 1)[0] |
||||
path = posixpath.normpath(urllib.parse.unquote(path)) |
||||
words = path.split('/') |
||||
words = [_f for _f in words if _f] |
||||
# path = os.getcwd() |
||||
path = upload_dir |
||||
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 copyfile(self, source, outputfile): |
||||
"""Copy all data between two file objects. |
||||
|
||||
The SOURCE argument is a file object open for reading |
||||
(or anything with a read() method) and the DESTINATION |
||||
argument is a file object open for writing (or |
||||
anything with a write() method). |
||||
|
||||
The only reason for overriding this would be to change |
||||
the block size or perhaps to replace newlines by CRLF |
||||
-- note however that this the default server uses this |
||||
to copy binary data as well. |
||||
|
||||
""" |
||||
shutil.copyfileobj(source, outputfile) |
||||
|
||||
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', |
||||
}) |
||||
|
||||
class Image(qrcode.image.base.BaseImage): |
||||
def __init__(self, border, width, box_size): |
||||
self.border = border |
||||
self.width = width |
||||
self.box_size = box_size |
||||
size = (width + border * 2) * box_size |
||||
self._image = QtGui.QImage(size, size, QtGui.QImage.Format_RGB16) |
||||
self._image.fill(QtCore.Qt.white) |
||||
|
||||
def pixmap(self): |
||||
return QtGui.QPixmap.fromImage(self._image) |
||||
|
||||
def drawrect(self, row, col): |
||||
painter = QtGui.QPainter(self._image) |
||||
painter.fillRect( |
||||
(col + self.border) * self.box_size, |
||||
(row + self.border) * self.box_size, |
||||
self.box_size, self.box_size, |
||||
QtCore.Qt.black) |
||||
|
||||
def save(self, stream, kind=None): |
||||
pass |
||||
|
||||
class FileUploaderWidget(QWidget): |
||||
def __init__(self, url, color): |
||||
QWidget.__init__(self) |
||||
url = os.path.expanduser(url) |
||||
|
||||
self.setStyleSheet("background-color: black") |
||||
|
||||
self.file_name_font = QFont() |
||||
self.file_name_font.setPointSize(24) |
||||
|
||||
self.file_name_label = QLabel(self) |
||||
self.file_name_label.setText("Your file will be uploaded to\n{0}".format(url)) |
||||
self.file_name_label.setFont(self.file_name_font) |
||||
self.file_name_label.setAlignment(Qt.AlignCenter) |
||||
self.file_name_label.setStyleSheet("color: #eee") |
||||
|
||||
self.qrcode_label = QLabel(self) |
||||
|
||||
self.notify_font = QFont() |
||||
self.notify_font.setPointSize(12) |
||||
self.notify_label = QLabel(self) |
||||
self.notify_label.setText("Scan QR code above to upload a file from your smartphone.\nMake sure the smartphone is connected to the same WiFi network as this computer.") |
||||
self.notify_label.setFont(self.notify_font) |
||||
self.notify_label.setAlignment(Qt.AlignCenter) |
||||
self.notify_label.setStyleSheet("color: #eee") |
||||
|
||||
layout = QVBoxLayout(self) |
||||
layout.setContentsMargins(0, 0, 0, 0) |
||||
layout.addStretch() |
||||
layout.addWidget(self.qrcode_label, 0, Qt.AlignCenter) |
||||
layout.addSpacing(20) |
||||
layout.addWidget(self.file_name_label, 0, Qt.AlignCenter) |
||||
layout.addSpacing(40) |
||||
layout.addWidget(self.notify_label, 0, Qt.AlignCenter) |
||||
layout.addStretch() |
||||
|
||||
self.port = "8000" |
||||
self.local_ip = get_local_ip() |
||||
self.address = "http://{0}:{1}".format(self.local_ip, self.port) |
||||
|
||||
self.qrcode_label.setPixmap(qrcode.make(self.address, image_factory=Image).pixmap()) |
||||
|
||||
global upload_dir |
||||
upload_dir = url |
||||
|
||||
self.sender_thread = threading.Thread(target=self.run_http_server, name='LoopThread') |
||||
self.sender_thread.start() |
||||
|
||||
def run_http_server(self): |
||||
http.server.test(SimpleHTTPRequestHandler, http.server.HTTPServer) |
||||
|
||||
def before_destroy_buffer(self): |
||||
self.message_to_emacs.emit("Stop file receiver server: http://{0}:{1}".format(self.local_ip, self.port)) |
||||
self.sender_thread.stop() |
||||
|
After Width: | Height: | Size: 167 KiB |
|
Before Width: | Height: | Size: 180 KiB |
Loading…
Reference in new issue