基于Python搭建局域网大文件分享传输工具

 更新时间:2024年12月31日 16:11:48   作者:hvinsion  
这篇文章主要为大家详细介绍了如何基于Python搭建局域网大文件分享传输工具,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下

1.简介

由于单位不让用优盘、禁止用飞秋、也不准使共享,禁用FTP,也禁止搭建网站,且目前局域网内用的IM不支持1G以上文件传输,于是在找适合内网的大文件传输方法。开始搭建网盘,但上传一次、下载一次,效率还是有点低。后来用Everything的HTTP服务器,结果把整个硬盘都分享了,太不安全。于是就自己做了这个简单的文件分享工具。资源已打包可自行下载。

HTTP分享,利用Flask搭建了一个Web服务器,选择要分享的文件启动后,将生成的链接发给对方,对方就可以通过浏览器下载。
FTP分享用pyftpdlib搭建了一个FTP服务器,同样生成链接后,对方可以通过一些浏览器或FTP客户端下载,也可以通过程序提供的工具下载文件。

2.运行效果

3.相关源码

from PyQt5.QtWidgets import QApplication, QRadioButton, QMainWindow, QLabel, QComboBox, QLineEdit, QPushButton, QMessageBox, QToolButton, QFileDialog, QDialog
from PyQt5.QtCore import QDir, Qt, QThread, pyqtSignal
from PyQt5 import QtGui, QtCore, QtWidgets
from PyQt5.QtGui import QIcon,QFont
import sys
import os 
import random
import netifaces
from flask import Flask, send_file
from concurrent.futures import ThreadPoolExecutor
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import ThreadedFTPServer
from ftplib import FTP
from urllib.parse import urlparse

class HelpWindow(QDialog):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("帮助")
        self.setFixedSize(300, 280)
        self.center()
        self.setWindowFlags(Qt.WindowCloseButtonHint | Qt.WindowTitleHint)
        self.label = QLabel(self)
        self.label.setText("1、选择文件传输协议;2、输入端口号(建议10000~60000,一般默认即可); \
                            3、选择文件所在目录;4、选择要分享的文件;5、点击启动按钮,\
                           若防火墙弹出警示点击允许访问即可;6、将生成的链接复制发送给接收方, \
                           接收方通过浏览器访问即可下载。若通过FTP协议分享还可通过FTP客户端软件  \
                            或本程序的“接收FTP文件”按钮下载。注意:传输完成后请及时关闭程序, \
                            以免造成数据泄露和网络安全隐患。")
        self.label.setGeometry(10, 10, 280, 260)
        self.label.setWordWrap(True)
        font = QFont()
        font.setPointSize(10)   
        font.setBold(True)   
        self.label.setFont(font)
        
    def center(self):
        screen = QApplication.desktop().screenGeometry()
        window = self.geometry()
        x = (screen.width() - window.width()) // 2
        y = (screen.height() - window.height()) // 2
        self.move(x, y)

class DownloadThread(QThread):
    finished = pyqtSignal(bool)
    ftp = FTP()

    def __init__(self, ftp_url):
        super().__init__()
        self.ftp_url = ftp_url

    def run(self):
        parsed_url = urlparse(self.ftp_url)
        username = parsed_url.username
        password = parsed_url.password
        host = parsed_url.hostname
        port = parsed_url.port
        filename = parsed_url.path.rsplit('/', 1)[-1]
        if len(filename)==0:
            downloadfile = False
        else:
            downloadfile = True
        try:
            self.ftp.connect(host, port)
            self.ftp.encoding = 'utf-8'
            self.ftp.login(user=username, passwd=password)

            current_dir = os.getcwd()
            current_dir = current_dir.replace("\\", "/")
            random_int = random.randint(100, 999)
            mulu = "接收文件_" + str(random_int)
            LocalDir = current_dir + "/" + mulu
            if not os.path.exists(LocalDir):
                os.makedirs(LocalDir)
            if downloadfile:#下载一个文件
                Local = os.path.join(LocalDir, filename)
                self.DownLoadFile(Local,filename)
            else:#下载整个目录
                self.DownLoadFileTree(LocalDir,"/")
            self.ftp.quit()
            self.finished.emit(True)
        except Exception as e:
            self.finished.emit(False)

    def DownLoadFile(self, LocalFile, RemoteFile):  # 下载单个文件
        with open(LocalFile, 'wb') as file_handler:
            self.ftp.retrbinary('RETR ' + RemoteFile, file_handler.write)
        file_handler.close()
        return True

    def DownLoadFileTree(self, LocalDir, RemoteDir):  # 下载整个目录下的文件
        print("远程文件夹remoteDir:", RemoteDir)
        if not os.path.exists(LocalDir):
            os.makedirs(LocalDir)
        self.ftp.cwd(RemoteDir)
        RemoteNames = self.ftp.nlst()
        print("远程文件目录:", RemoteNames)
        for file in RemoteNames:
            Local = os.path.join(LocalDir, file)
            #print("正在下载", self.ftp.nlst(file))
            try:
                self.ftp.cwd(file)
                self.ftp.cwd("..")
                if not os.path.exists(Local):
                    os.makedirs(Local)
                self.DownLoadFileTree(Local, file)
            except:
                self.DownLoadFile(Local, file)
        self.ftp.cwd("..")
        return
    
class FTPWindow(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("下载FTP文件")
        self.setFixedSize(400, 150)

        self.label_ftp_url = QLabel("FTP链接:", self)
        self.label_ftp_url.setGeometry(20, 20, 80, 30)

        self.textbox_ftp_url = QLineEdit(self)
        self.textbox_ftp_url.setGeometry(100, 20, 280, 30)

        self.button_download = QPushButton("下载", self)
        self.button_download.setGeometry(160, 65, 80, 30)
        self.button_download.clicked.connect(self.download_ftp_file)

        self.label_ftp_info = QLabel("输入包含文件名的链接下载指定文件;\n输入不包含文件名只到端口号的链接下载整个目录。", self)
        self.label_ftp_info.setGeometry(20, 110, 380, 40)

    def download_ftp_file(self):
        ftp_url = self.textbox_ftp_url.text()
        if ftp_url == "":
            return
        self.button_download.setEnabled(False)
        self.thread = DownloadThread(ftp_url)
        self.thread.finished.connect(self.show_message_box)
        self.thread.start()

    def show_message_box(self, success):
        if success:
            QMessageBox.information(self, "提示", "文件下载成功!", QMessageBox.Ok)
        else:
            QMessageBox.critical(self, "错误", "文件下载失败!", QMessageBox.Ok)
        self.button_download.setEnabled(True)


class MyWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("文件分享")
        self.setWindowIcon(QIcon('icon.ico'))
        self.setMaximumSize(self.size())
        self.setFixedSize(400, 300)

        self.tcp_label = QLabel('选择传输方式:', self)
        self.tcp_label.move(20, 10)
        self.http_button = QRadioButton('HTTP', self)
        self.http_button.setChecked(True)
        self.http_button.move(150, 10)
        self.http_button.toggled.connect(self.onProtocolChanged)
        self.ftp_button = QRadioButton('FTP', self)
        self.ftp_button.move(250, 10)
        self.ftp_button.toggled.connect(self.onProtocolChanged)

        self.ip_label = QLabel('当前IP地址:', self)
        self.ip_label.move(20, 40)
        self.ip_text = QLabel(self.get_wired_ip(), self)
        self.ip_text.setGeometry(120, 43, 250, 25)

        self.port_label = QLabel('端 口 :', self)
        self.port_label.move(20, 75)

        self.textbox = QLineEdit(self)
        self.textbox.setGeometry(80, 75, 300, 30)
        self.textbox.setValidator(QtGui.QIntValidator(10000, 65500))
        self.textbox.setText(str(random.randint(10000, 65500)))  

        self.folder_label = QLabel('文件夹:', self)
        self.folder_label.setGeometry(20, 115, 50, 30)

        self.textfol = QLineEdit(self)
        self.textfol.setGeometry(80, 115, 220, 30)
        self.textfol.setText(QDir.currentPath())
        self.textfol.setReadOnly(True)

        self.folder_button = QPushButton('选择文件夹', self)
        self.folder_button.setGeometry(300, 115, 80, 30)
        self.folder_button.clicked.connect(self.onFolderClicked)

        self.file_label = QLabel('文 件 :', self)
        self.file_label.setGeometry(20, 155, 50, 30)

        self.combobox = QComboBox(self)
        self.combobox.setGeometry(80, 155, 300, 30)

        self.updateFileList()
        self.updateFileList()

        self.button = QPushButton(self)
        self.button.setText("启  动")
        self.button.setGeometry(20, 195, 360, 30)
        self.button.clicked.connect(self.show_selection)

        self.textboxa = QLineEdit(self)
        self.textboxa.setGeometry(20, 235, 360, 30)
        self.textboxa.mousePressEvent = self.select_text

        self.hidden_button = QPushButton("接收\nFTP\n文件", self)
        self.hidden_button.setVisible(False)  
        self.hidden_button.setGeometry(320, 10, 60, 55)
        self.hidden_button.clicked.connect(self.show_ftp_window)

        self.labela = QLabel(self)
        self.labela.setText("@hvinsion")
        self.labela.setGeometry(320, 280, 360, 20)

        self.labelb = QPushButton(self)
        self.labelb.setText("说明")
        self.labelb.setToolTip("帮助")  
        self.labelb.setGeometry(10, self.height() - 30, 30, 30) 
        self.labelb.clicked.connect(self.open_help_window) 

        self.help_window = HelpWindow()

        self.thread_pool = ThreadPoolExecutor(max_workers=5)

    def open_help_window(self):
        self.help_window.show()

    def select_text(self, event):
        self.textboxa.selectAll()
        self.textboxa.setFocus()

    def show_ftp_window(self):
        ftp_window = FTPWindow(self)
        ftp_window.exec_()

    def onProtocolChanged(self):
        if self.ftp_button.isChecked():
            self.hidden_button.setVisible(True)
        else:
            self.hidden_button.setVisible(False)

    def onFolderClicked(self):
        m = QFileDialog.getExistingDirectory(self, "选取文件夹", QDir.currentPath()) 
        self.textfol.setText(m)
        self.updateFileList()

    def updateFileList(self):
        folder_path = self.textfol.text()
        file_list = QDir(folder_path).entryList(QDir.Files)
        self.combobox.clear()
        self.combobox.addItems(file_list)

    def get_wired_ip(self):
        try:
            default_gateway = netifaces.gateways()['default']
            if default_gateway and netifaces.AF_INET in default_gateway:
                routingNicName = default_gateway[netifaces.AF_INET][1]
                for interface in netifaces.interfaces():
                    if interface == routingNicName:
                        routingIPAddr = netifaces.ifaddresses(interface).get(netifaces.AF_INET)
                        if routingIPAddr:
                            return routingIPAddr[0]['addr']
        except (KeyError, IndexError):
            pass
        return "未找到物理网卡IP"

    def start_flask_server(self):
        app = Flask(__name__)

        @app.route('/download/<filename>', methods=['GET'])
        def download_file(filename):
            file_path = os.path.join(self.textfol.text(), filename) 
            return send_file(file_path, as_attachment=True)

        port = int(self.textbox.text())
        self.textbox.setText(str(random.randint(10000, 65500)))
        app.run(host='0.0.0.0', port=port, threaded=True)
        

    def start_ftp_server(self):
        authorizer = DummyAuthorizer()
        authorizer.add_user('a', 'a', self.textfol.text(), perm='elradfmwM')

        class CustomFTPHandler(FTPHandler):
            def on_file_received(self, file):
                pass
            def on_file_deleted(self, file):
                pass
            def on_rename(self, old_file, new_file):
                pass

        handler = CustomFTPHandler
        handler.authorizer = authorizer

        address = ('0.0.0.0', int(self.textbox.text()))

        server = ThreadedFTPServer(address, handler)
        self.thread_pool.submit(server.serve_forever)
        self.textbox.setText(str(random.randint(10000, 65500)))

    def show_selection(self):
        ip_address = self.get_wired_ip()
        port_text = self.textbox.text()
        if int(port_text)<10000:
            QMessageBox.critical(self, "警告", "请勿使用10000以下端口号!", QMessageBox.Ok)
            self.textbox.setText(str(random.randint(10000, 65500)))
            return
        selected_file = self.combobox.currentText()
        if self.http_button.isChecked():
            self.thread_pool.submit(self.start_flask_server)
            self.textboxa.setText("http://" + ip_address + ":" + port_text + "/download/" + selected_file)
        elif self.ftp_button.isChecked():
            self.thread_pool.submit(self.start_ftp_server)
            self.textboxa.setText("ftp://a:a@" + ip_address + ":" + port_text + "/" + selected_file)
        else:
            QMessageBox.warning(self, "警告", "请选择传输方式!", QMessageBox.Ok)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    win = MyWindow()
    win.show()
    sys.exit(app.exec_())

到此这篇关于基于Python搭建局域网大文件分享传输工具的文章就介绍到这了,更多相关Python大文件传输内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • python并发和异步编程实例

    python并发和异步编程实例

    这篇文章主要为大家详细介绍了python并发和异步编程实例,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2018-11-11
  • Python搭建自己IP代理池的方法实现

    Python搭建自己IP代理池的方法实现

    本文主要介绍了Python搭建自己IP代理池的方法实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2023-02-02
  • Python条件语句与循环语句

    Python条件语句与循环语句

    这篇文章主要介绍了Python条件语句与循环语句,条件语句就是通过指定的表达式的运行结果来判断当前是执行还是跳过某些指定的语句块,循环语句就是对某些语句的重复执行,这个重复执行是通过指定表达式来控制的,下面来看具体内容及续航管案例吧,需要的朋友可以参考一下
    2021-11-11
  • 解决Python复杂zip文件的解压问题

    解决Python复杂zip文件的解压问题

    这篇文章主要介绍了Python复杂zip文件的解压,通过配合 shutil 与 os 标准库中的相关功能,实现将指定任意 zip 压缩包,完好地解压到指定的目录中,需要的朋友可以参考下
    2021-12-12
  • 关于Python网络爬虫requests库的介绍

    关于Python网络爬虫requests库的介绍

    这篇文章主要介绍了关于Python网络爬虫requests库,而很多时候这些数据存储在网页中,手动下载需要花费的时间太长,这时候我们就需要网络爬虫帮助我们自动爬取这些数据,需要的朋友可以参考下
    2023-04-04
  • python通过对字典的排序,对json字段进行排序的实例

    python通过对字典的排序,对json字段进行排序的实例

    今天小编就为大家分享一篇python通过对字典的排序,对json字段进行排序的实例,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-02-02
  • 如何将yolov5中的PANet层改为BiFPN详析

    如何将yolov5中的PANet层改为BiFPN详析

    现在yolov5的neck用的是PANet,在efficient论文中提出了BiFPN结构,还有更加不错的性能,下面这篇文章主要给大家介绍了关于如何将yolov5中的PANet层改为BiFPN的相关资料,需要的朋友可以参考下
    2022-06-06
  • python数据分析数据标准化及离散化详解

    python数据分析数据标准化及离散化详解

    这篇文章主要为大家详细介绍了python数据分析数据标准化及离散化,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2018-02-02
  • Python 3实战爬虫之爬取京东图书的图片详解

    Python 3实战爬虫之爬取京东图书的图片详解

    最近在学习python3,下面这篇文章主要给大家介绍了关于Python3实战爬虫之爬取京东图书图片的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面来一起看看吧。
    2017-10-10
  • 浅谈python处理json和redis hash的坑

    浅谈python处理json和redis hash的坑

    这篇文章主要介绍了浅谈python处理json和redis hash的坑,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-07-07

最新评论