基于Python打造一个可视化FTP服务器

本文主要是介绍基于Python打造一个可视化FTP服务器，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

《基于Python打造一个可视化FTP服务器》在日常办公和团队协作中,文件共享是一个不可或缺的需求,所以本文将使用Python+Tkinter+pyftpdlib开发一款可视化FTP服务器,有需要的小...

1. 概述

在日常办公和团队协作中，文件共享是一个不可或缺的需求。然而，市场上的FTP服务器软件通常配置复杂、体积庞大，或者收费昂贵。有没有一种简单高效的方式，让你可以轻松搭建自己的FTP服务器，实现高效的文件共享呢？

答案就是——使用 python + Tkinter + pyftpdlib 开发一款可视化FTP服务器，让你无需命令行操作，也能轻松管理FTP服务器！本文将详细介绍这款FTP服务器的功能、实现原理，以及如何使用它来搭建属于自己的文件共享中心。

基于Python打造一个可视化FTP服务器

2. 功能介绍

该FTP服务器具备以下核心功能：

GUI 操作：基于 Tkinter 图形化界面，无需命令行操作，所有设置一目了然。

多IP支持：自动检测本机有效IP地址，支持局域网和公网访问。

多用户管理：支持添加多个用户，分配不同的访问权限。

权限控制：提供详细的权限配置（上传、下载、删除、重命名等）。

实时连接监控：显示当前在线用户数量，连接、断开的日志实时更新。

目录选择：支持自定义FTP根目录，灵活管理共享文件。

操作日志：日志窗口实时记录服务器状态，便于调试和管理。

3. 如何使用

1.运行程序

首先，你需要确保已经安装了 Python 环境，并安装了 pyftpdlib 依赖包。

pip install pyftpdlib

然后运行 FTPServerGUI.py 文件，启动FTP服务器图形化界面。

python FTPServerGUI.py

2.选择FTP目录

在软件界面中，点击 “选择FTP目录” 按钮，选择你要共享的文件夹。

该目录即为FTP服务器的根目录，所有FTP用户只能在该目录及其子目录下操作。

3.配置用户信息

输入用户名和密码，确保用户可以安全访问服务器。

4.设定访问权限

勾选用户权限：

上传 (w)：允许用户上传文件
下载 ®：允许用户下载文件
删除 (d)：允许用户删除文件
创建目录 (m)：允许用户在FTP目录下创建文件夹
修改权限 (M)：允许用户更改文件或文件夹权限

根据实际需求，自由组合用户的访问权限。

5.启动服务器

点击 “启动服务器” 按钮，FTP服务器即刻运行。

此时，日志窗口会显示服务器状态，并提供可用的FTP访问地址。

[状态] 服务已启动于 0.0.0.0:21
[提示] 可用地址: 192.168.1.100

6.连接FTP服务器

你可以使用 Windows 资源管理器、FileZilla、或者命令行连接FTP服务器。

方式1：Windows 资源管理器

在地址栏输入：

ftp://192.168.1.100

然后输入用户名和密码，即可访问FTP文件。

方式2：FileZilla

使用 FileChina编程Zilla 连接服务器，输入服务器IP、端口（21）、用户名、密码，即可管理文件。

方式3：命令行FTP

ftp 192.168.1.100

输入用户名和密码后，可以使用 FTP 命令操作文件。

ls   # 列出文件
get 文件名  # 下载文件
put 文件名  # 上传文件
exit  # 退出FTP

4. 代码解析

FTP服务器的核心功能基于 pyftpdlib 实现，以下是关键代码解析：

1.服务器启动逻辑

self.server = FTPServer(("0.0.0.0", port), handler)
server_thread = threading.Thread(target=self.server.serve_forever)
server_thread.daemon = True
server_thread.start()

FTPServer(("0.0.0.0", port), handler): 监听 0.0.0.0，接受所有IP访问。

server_thread 采用多线程模式，避免UI界面卡死。

2.用户权限管理

perm = ''.join([k for k, v in self.perm_vars.items() if v.get()])
authorizer.add_user(user, password, directory, perm=perm)

perm_vars 存储用户权限，如 w（写）、r（读）、d（删除）等。

通过 add_user 方法添加用户及其权限。

3.连接状态监控

def update_connection_count(self, delta: int):
    self.connection_count += delta
    self.root.after(0, self.conn_label.config, {"text": f"当前连接: {self.connection_count}"})

update_connection_count 线程安全地更新当前在线用户数量。

使用 self.root.after(0, func, args) 使界面线程安全更新。

5. 运行效果

基于Python打造一个可视化FTP服务器

6.相关源码

import tkinter as tk
from tkinter import ttk, filedialog, messagebox, scrolledtext
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import FTPServer
import threading
import socket
import webbrowser
from typing import List

class FTPServerGUI:
    def __init__(self, root):
        self.root = root
        self.server = None
        self.connection_count = 0
        self.setup_ui()
        self.show_filtered_ips()
        self.display_welcome_msg()

    def setup_ui(self):
        # 主窗口设置
        self.root.title("FTP文件共享服务器")
        self.root.geometry("560x500")
        self.root.minsize(560, 480)

        # 配置网格布局权重
        self.root.columnconfigure(0, weight=1)
        self.root.columnconfigure(1, weight=1)
        self.root.rowconfigure(4, weight=1)  # 日志区域行

        # 顶部信息栏
        top_info = ttk.Frame(self.root)
        top_info.grid(row=0, column=0, columnspan=2, padx=10, pady=5, sticky="ew")

        # 当前连接数显示
        self.conn_label = ttk.Label(top_info, text="当前连接: 0", foreground="green")
        self.conn_label.pack(side="right", padx=10)

        # IP地址显示
        ttk.Label(top_info, text="服务器IP:").pack(side="left")
        self.ip_label = ttk.Label(top_info, text="正在获取...", foreground="blue"js)
        self.ip_label.pack(side="left", padx=5)

        # 配置框架
        config_frame = ttk.LabelFrame(self.root, text="服务器配置")
        config_frame.grid(row=1, column=0, columnspan=2, padx=10, pady=5, sticky="ew")

        # 输入字段布局
        input_grid = ttk.Frame(config_frame)
        input_grid.pack(fill="x", padx=5, pady=5)

        # 端口配置
        ttk.Label(input_grid, text="端口:").grid(row=0, column=0, padx=5, sticky="w")
        self.port_entry = ttk.Entry(input_grid, width=8)
        self.port_entry.grid(row=0, column=1, sticky="w")
        self.port_entry.insert(0, "21")

        # 用户配置
        ttk.Label(input_grid, text="用户:").grid(row=0, column=2, padx=(15,5), sticky="w")
        self.user_entry = ttk.Entry(input_grid, width=12)
        self.user_entry.grid(row=0, column=3, sticky="w")
        #self.user_entry.insert(0, "user")

        # 密码配置
        ttk.Label(input_grid, text="密码:").grid(row=0, column=4, padx=(15,5), sticky="w")
        self.pass_entry = ttk.Entry(input_grid, width=12, show="*")
        self.pass_entry.grid(row=0, column=5, sticky="w")
        #self.pass_entry.insert(0, "123")

        # 目录选择
        dir_frame = ttk.Frame(config_frame)
        dir_frame.pack(fill="x", padx=5, pady=5)
        ttk.Button(dir_frame, text="选择FTP目录", command=self.select_directory).pack(side="left")
        self.dir_label = ttk.Label(dir_frame, text="未选择目录", foreground="gray")
        self.dir_label.pack(side="left", padx=10)

        # 权限设置框架
        perm_frame = ttk.LabelFrame(self.root, text="用户权限设置")
        perm_frame.grid(row=2, column=0, columnspan=2, padx=10, pady=5, sticky="nsew")

        # 三列布局
        self.perm_vars = {
            'e': tk.BooleanVar(value=True),
            'l': tk.BooleanVar(value=True),
            'r': tk.BooleanVar(value=True),
            'a': tk.BooleanVar(),
            'd': tk.BooleanVar(),
            'f': tk.BooleanVar(),
            'm': tk.BooleanVar(value=True),
            'M': tk.BooleanVar(),
            'w': tk.BooleanVar(value=True),
        }

        cols = [ttk.Frame(perm_frame) for _ in range(3)]
        for i, col in enumerate(cols):
            col.grid(row=0, column=i, sticky="nsew", padx=5)
            perm_frame.columnconfigure(i, weight=1)

        # 权限项分布
        permissions = [
        ("上传文件 (w)", 'w', 0),
        ("下载文件 (r)", 'r', 0),
        ("删除文件 (d)", 'd', 0),
        ("查看列表 (l)", 'l', 1),
        ("切换目录 (e)", 'e', 1),
        ("创建目录 (m)", 'm', 1China编程),
        ("修改权限 (M)", 'M', 2),
        ("重命名 (f)", 'f', 2),
            ("追加文件 (a)", 'a', 2)
javascript        ]

        for text, key, col_idx in permissions:
            ttk.Checkbutton(cols[col_idx], text=text, variable=self.perm_vars[key]).pack(anchor="w", pady=2)

        # 控制按钮
        btn_frame = ttk.Frame(self.root)
        btn_frame.grid(row=3, column=0, columnspan=2, pady=10)
        self.start_btn = ttk.Button(btn_frame, text="启动服务器", command=self.toggle_server)
        self.start_btn.pack(side="left", padx=5)

        # 日志区域
        log_frame = ttk.LabelFrame(self.root, text="服务器日志")
        log_frame.grid(row=4, column=0, columnspan=2, padx=10, pady=5, sticky="nsew")
        self.log_text = scrolledtext.ScrolledText(log_frame, wrap=tk.WORD, height=8)
        self.log_text.pack(expand=True, fill="both")

        # 博客链接
        blog_link = ttk.Label(self.root, text="探客白泽", foreground="blue", cursor="hand2")
        blog_link.grid(row=5, column=1, padx=10, pady=5, sticky="se")
        blog_link.bind("<Button-1>", lambda e: webbrowser.open("https://blog.csdn.net/Clay_K?spm=1011.2415.3001.10640"))

        # 布局权重配置
        perm_frame.rowconfigure(0, weight=1)
        log_frame.rowconfigure(0, weight=1)
        log_frame.columnconfigure(0, weight=1)

    def show_filtered_ips(self) -> List[str]:
        try:
            ips = socket.gethostbyname_ex(socket.gethostname())[2]
            filtered_ips = [
                ip for ip in ips 
                if ip != "127.0.0.1" 
                and not ip.startswith("169.254.")
            ]

            self.ip_label.config(text=", ".join(filtered_ips) if filtered_ips else "未找到有效IP")
            return filtered_ips
        except Exception as e:
            self.ip_label.config(text=f"获取IP失败: {str(e)}")
            return []

    def is_valid_ip(self, ip: str) -> bool:
        if ip.startswith("127.") or ip == "::1":
            return False
        if ip.startswith("169.254."):  # APIPA地址
            return False
        if ip.startswith("172.17."):  # docker默认地址
            return False
        if ip.startswith("192.168.") or ip.startswith("10."):  # 内网地址
            return True  # 保留显示内网地址
        return True

    def select_directory(self):
        directory = filedialog.askdirectory()
        if directory:
            self.dir_label.config(text=directory, foreground="black")

    def update_connection_count(self, delta: int):
        """线程安全更新连接数"""
        self.connection_count += delta
        self.root.after(0, self.conn_label.config, 
                      {"text": f"当前连接: {self.connection_count}"})

    def log_message(self, msg: str):
        self.log_text.insert(tk.END, msg + "\n")
        self.log_text.see(tk.END)

    def thread_safe_log(self, msg: str):
        self.root.after(0, self.log_message, msg)

    def toggle_server(self):
        if self.server is None:
            self.start_server()
        else:
            self.stop_server()

    def start_server(self):
        port = int(self.port_entry.get())
        user = self.user_entry.get()
        password = self.pass_entry.get()
        directory = self.dir_label.cget("text")

        if not all([user, password, directory]) or directory == "未选择目录":
            messagebox.showerror("错误", "请填写所有必填字段！")
            return

        perm = ''.join([k for k, v in self.perm_vars.items() if v.get()])
        if not perm:
            messagebox.showerror("错误", "请至少选择一个权限！")
            return

        class CustomHandler(FTPHandler):
            gui = self  # 引用GUI实例

            def on_connect(self):
                self.gui.update_connection_count(1)
                self.gui.thread_safe_log(f"[连接] {self.remote_ip}:{self.remote_port} 已连接")

            def on_disconnect(self):
                self.gui.update_connection_count(-1)
                self.gui.thread_safe_log(f"[连接] {self.remote_ip}:{self.remote_port} 断开连接")

            def on_login(self, username):
                self.gui.thread_safe_log(f"[认证] 用户 {username} 登录成功")

            def on_login_failed(self, username, password):
                self.gui.thread_safe_log(f"[认证] 登录失败 用户名: {username}")

        try:
            authorizer = DummyAuthorizer()
            authorizer.add_user(user, password, directory, perm=perm)

            handler = CustomHandler
            handler.authorizer = authorizer

            self.server = FTPServer(("0.0.0.0", port), handler)
            self.thread_safe_log(f"[状态] 服务已启动于 0.0.0.0:{port}")
            self.thread_safe_log(f"[提示] 可用地址: {self.ip_label.cget('text')}")
            self.start_btn.config(text="停止服务器")

            server_thread = threading.Thread(target=self.server.serve_forever)
            server_thread.daemon = True
            server_thread.start()
        except Exception as e:
            self.thread_safe_log(f"[错误] 启动失败: {str(e)}")

    def stop_server(self):
        if self.server:
            self.server.close_all()
            self.server = None
            seyEuSmlf.connection_count = 0
            self.conn_label.config(text="当前连接: 0")
            self.thread_safe_log("[状态] 服务器已停止")
            self.start_btn.config(text="启动服务器")

    def display_welcome_msg(self):
        """显示欢迎信息"""
        welcome_text = """=== FTP文件共享服务器 ===
【使用指南】
1. 选择FTP目录 -> 设置用户权限 -> 启动服务
2. 客户端连接地址显示在顶部IP栏
3. 日志区域实时显示连接状态
【常用命令帮助】
USER [用户名]   - 登录认证
PASS [密码]     - 输入密码
LIST            - 列出文件列表
RETR [文件名]   - 下载文件
STOR [文件名]   - 上传文件
QUIT/EXIT       - 断开连接
当前版本功能：
• 多IP地址自动检测
• 实时连接数统计
• 细粒度权限控制
• 操作日志记录
"""
        self.log_text.configure(state='normal')
        self.log_text.delete(1.0, tk.END)
        self.log_text.insert(tk.END, welcome_text)

        # 设置文本样式
        self.log_text.tag_configure("title", foreground="#2E75B6", font=('微软雅黑', 10, 'bold'))
        self.log_text.tag_add("title", "1.0", "1.19")

if __name__ == "__main__":
    root = tk.Tk()
    app = FTPServerGUI(root)
    root.mainloop()